diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index 86fb33a82ce04..dab7dd959d914 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -222,6 +222,19 @@ jobs: # - name: Check datafusion (no-default-features) run: cargo xtask ci step check datafusion no-default + - name: Check datafusion (object_store) + run: cargo xtask ci step check datafusion object_store + - name: Check datafusion does not require object_store + run: | + tree_file=$(mktemp) + trap 'rm -f "$tree_file"' EXIT + cargo tree -p datafusion --no-default-features --features sql \ + --target all --edges all --prefix none --format '{p}' > "$tree_file" + if grep -q '^object_store v' "$tree_file"; then + cat "$tree_file" + echo "DataFusion without storage must not depend on object_store" >&2 + exit 1 + fi - name: Check datafusion (nested_expressions) run: cargo xtask ci step check datafusion nested_expressions - name: Check datafusion (array_expressions) diff --git a/Cargo.toml b/Cargo.toml index efeb580074138..037bb2523be3c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -123,7 +123,7 @@ criterion = "0.8" ctor = "1.0.7" dashmap = "6.2.1" datafusion = { path = "datafusion/core", version = "55.0.0", default-features = false } -datafusion-catalog = { path = "datafusion/catalog", version = "55.0.0" } +datafusion-catalog = { path = "datafusion/catalog", version = "55.0.0", default-features = false } datafusion-catalog-listing = { path = "datafusion/catalog-listing", version = "55.0.0" } datafusion-common = { path = "datafusion/common", version = "55.0.0", default-features = false } datafusion-common-runtime = { path = "datafusion/common-runtime", version = "55.0.0" } @@ -138,11 +138,19 @@ datafusion-execution = { path = "datafusion/execution", version = "55.0.0", defa datafusion-expr = { path = "datafusion/expr", version = "55.0.0", default-features = false } datafusion-expr-common = { path = "datafusion/expr-common", version = "55.0.0" } datafusion-ffi = { path = "datafusion/ffi", version = "55.0.0" } -datafusion-functions = { path = "datafusion/functions", version = "55.0.0" } -datafusion-functions-aggregate = { path = "datafusion/functions-aggregate", version = "55.0.0" } +# Select the existing expression defaults without implicitly enabling storage. +datafusion-functions = { path = "datafusion/functions", version = "55.0.0", default-features = false, features = [ + "datetime_expressions", + "encoding_expressions", + "math_expressions", + "regex_expressions", + "string_expressions", + "unicode_expressions", +] } +datafusion-functions-aggregate = { path = "datafusion/functions-aggregate", version = "55.0.0", default-features = false } datafusion-functions-aggregate-common = { path = "datafusion/functions-aggregate-common", version = "55.0.0" } datafusion-functions-nested = { path = "datafusion/functions-nested", version = "55.0.0", default-features = false } -datafusion-functions-table = { path = "datafusion/functions-table", version = "55.0.0" } +datafusion-functions-table = { path = "datafusion/functions-table", version = "55.0.0", default-features = false } datafusion-functions-window = { path = "datafusion/functions-window", version = "55.0.0" } datafusion-functions-window-common = { path = "datafusion/functions-window-common", version = "55.0.0" } datafusion-macros = { path = "datafusion/macros", version = "55.0.0" } @@ -150,15 +158,18 @@ datafusion-optimizer = { path = "datafusion/optimizer", version = "55.0.0", defa datafusion-physical-expr = { path = "datafusion/physical-expr", version = "55.0.0", default-features = false } datafusion-physical-expr-adapter = { path = "datafusion/physical-expr-adapter", version = "55.0.0", default-features = false } datafusion-physical-expr-common = { path = "datafusion/physical-expr-common", version = "55.0.0", default-features = false } -datafusion-physical-optimizer = { path = "datafusion/physical-optimizer", version = "55.0.0" } -datafusion-physical-plan = { path = "datafusion/physical-plan", version = "55.0.0" } +datafusion-physical-optimizer = { path = "datafusion/physical-optimizer", version = "55.0.0", default-features = false } +datafusion-physical-plan = { path = "datafusion/physical-plan", version = "55.0.0", default-features = false } datafusion-proto = { path = "datafusion/proto", version = "55.0.0", default-features = false } datafusion-proto-common = { path = "datafusion/proto-common", version = "55.0.0" } datafusion-proto-models = { path = "datafusion/proto-models", version = "55.0.0" } -datafusion-pruning = { path = "datafusion/pruning", version = "55.0.0" } -datafusion-session = { path = "datafusion/session", version = "55.0.0" } +datafusion-pruning = { path = "datafusion/pruning", version = "55.0.0", default-features = false } +datafusion-session = { path = "datafusion/session", version = "55.0.0", default-features = false } datafusion-spark = { path = "datafusion/spark", version = "55.0.0" } -datafusion-sql = { path = "datafusion/sql", version = "55.0.0" } +datafusion-sql = { path = "datafusion/sql", version = "55.0.0", default-features = false, features = [ + "unicode_expressions", + "unparser", +] } datafusion-substrait = { path = "datafusion/substrait", version = "55.0.0" } doc-comment = "0.3" diff --git a/datafusion/catalog-listing/Cargo.toml b/datafusion/catalog-listing/Cargo.toml index abe58f45994be..d4e7fdd97fbf3 100644 --- a/datafusion/catalog-listing/Cargo.toml +++ b/datafusion/catalog-listing/Cargo.toml @@ -33,10 +33,10 @@ all-features = true [dependencies] arrow = { workspace = true } async-trait = { workspace = true } -datafusion-catalog = { workspace = true } +datafusion-catalog = { workspace = true, features = ["object_store"] } datafusion-common = { workspace = true, features = ["object_store"] } -datafusion-datasource = { workspace = true } -datafusion-execution = { workspace = true } +datafusion-datasource = { workspace = true, features = ["object_store"] } +datafusion-execution = { workspace = true, features = ["object_store"] } datafusion-expr = { workspace = true } datafusion-physical-expr = { workspace = true } datafusion-physical-expr-adapter = { workspace = true } diff --git a/datafusion/catalog/Cargo.toml b/datafusion/catalog/Cargo.toml index 1009e9aee477b..f4195c2f4a161 100644 --- a/datafusion/catalog/Cargo.toml +++ b/datafusion/catalog/Cargo.toml @@ -30,6 +30,16 @@ version.workspace = true [package.metadata.docs.rs] all-features = true +[features] +default = ["object_store"] +object_store = [ + "dep:object_store", + "datafusion-datasource/object_store", + "datafusion-execution/object_store", + "datafusion-physical-plan/object_store", + "datafusion-session/object_store", +] + [dependencies] arrow = { workspace = true } async-trait = { workspace = true } @@ -45,7 +55,7 @@ datafusion-session = { workspace = true } futures = { workspace = true } itertools = { workspace = true } log = { workspace = true } -object_store = { workspace = true } +object_store = { workspace = true, optional = true } parking_lot = { workspace = true } tokio = { workspace = true } diff --git a/datafusion/catalog/src/lib.rs b/datafusion/catalog/src/lib.rs index 815bfe32fac72..db47a09e3f5d1 100644 --- a/datafusion/catalog/src/lib.rs +++ b/datafusion/catalog/src/lib.rs @@ -39,6 +39,7 @@ pub mod cte_worktable; pub mod default_table_source; pub mod empty; pub mod information_schema; +#[cfg(feature = "object_store")] pub mod listing_schema; pub mod memory; pub mod stream; diff --git a/datafusion/core/Cargo.toml b/datafusion/core/Cargo.toml index 222c0ec688b78..3da32308a3ab6 100644 --- a/datafusion/core/Cargo.toml +++ b/datafusion/core/Cargo.toml @@ -50,19 +50,42 @@ nested_expressions = ["datafusion-functions-nested"] # This feature is deprecated. Use the `nested_expressions` feature instead. array_expressions = ["nested_expressions"] # Used to enable the avro format -avro = ["datafusion-datasource-avro"] +avro = ["object_store", "datafusion-datasource-avro"] backtrace = ["datafusion-common/backtrace"] compression = [ "liblzma", "bzip2", "flate2", "zstd", - "datafusion-datasource-arrow/compression", + "datafusion-datasource-arrow?/compression", "datafusion-datasource/compression", ] crypto_expressions = ["datafusion-functions/crypto_expressions"] datetime_expressions = ["datafusion-functions/datetime_expressions"] +# Built-in file sources, object store registration, and file caches. +# Disable default features and omit this feature for storage-independent embedding. +object_store = [ + "dep:object_store", + "datafusion-common/object_store", + "datafusion-execution/object_store", + "datafusion-datasource/object_store", + "datafusion-catalog/object_store", + "dep:datafusion-catalog-listing", + "dep:datafusion-datasource-arrow", + "dep:datafusion-datasource-csv", + "dep:datafusion-datasource-json", + "datafusion-functions/object_store", + "datafusion-functions-aggregate/object_store", + "datafusion-functions-nested?/object_store", + "datafusion-functions-table/object_store", + "datafusion-physical-expr-adapter/object_store", + "datafusion-physical-optimizer/object_store", + "datafusion-physical-plan/object_store", + "datafusion-session/object_store", + "datafusion-sql?/object_store", +] default = [ + "object_store", "nested_expressions", "crypto_expressions", "datetime_expressions", @@ -79,7 +102,7 @@ encoding_expressions = ["datafusion-functions/encoding_expressions"] # Used for testing ONLY: causes all values to hash to the same value (test for collisions) force_hash_collisions = ["datafusion-physical-plan/force_hash_collisions", "datafusion-common/force_hash_collisions"] math_expressions = ["datafusion-functions/math_expressions"] -parquet = ["datafusion-common/parquet", "dep:parquet", "datafusion-datasource-parquet"] +parquet = ["object_store", "datafusion-common/parquet", "dep:parquet", "datafusion-datasource-parquet"] parquet_encryption = [ "parquet", "parquet/encryption", @@ -124,14 +147,14 @@ async-trait = { workspace = true } bzip2 = { workspace = true, optional = true } chrono = { workspace = true } datafusion-catalog = { workspace = true } -datafusion-catalog-listing = { workspace = true } -datafusion-common = { workspace = true, features = ["object_store"] } +datafusion-catalog-listing = { workspace = true, optional = true } +datafusion-common = { workspace = true } datafusion-common-runtime = { workspace = true } datafusion-datasource = { workspace = true } -datafusion-datasource-arrow = { workspace = true } +datafusion-datasource-arrow = { workspace = true, optional = true } datafusion-datasource-avro = { workspace = true, optional = true } -datafusion-datasource-csv = { workspace = true } -datafusion-datasource-json = { workspace = true } +datafusion-datasource-csv = { workspace = true, optional = true } +datafusion-datasource-json = { workspace = true, optional = true } datafusion-datasource-parquet = { workspace = true, optional = true } datafusion-execution = { workspace = true } datafusion-expr = { workspace = true, default-features = false } @@ -155,7 +178,7 @@ indexmap = { workspace = true } itertools = { workspace = true } liblzma = { workspace = true, optional = true } log = { workspace = true } -object_store = { workspace = true } +object_store = { workspace = true, optional = true } parking_lot = { workspace = true } parquet = { workspace = true, optional = true, default-features = true } serde = { version = "1.0", default-features = false, features = ["derive"], optional = true } diff --git a/datafusion/core/src/dataframe/mod.rs b/datafusion/core/src/dataframe/mod.rs index ed3dc5ea838b9..7ee47f9195b5d 100644 --- a/datafusion/core/src/dataframe/mod.rs +++ b/datafusion/core/src/dataframe/mod.rs @@ -22,8 +22,11 @@ mod parquet; use crate::arrow::record_batch::RecordBatch; use crate::arrow::util::pretty; +#[cfg(feature = "object_store")] use crate::datasource::file_format::csv::CsvFormatFactory; +#[cfg(feature = "object_store")] use crate::datasource::file_format::format_as_file_type; +#[cfg(feature = "object_store")] use crate::datasource::file_format::json::JsonFormatFactory; use crate::datasource::{ DefaultTableSource, MemTable, TableProvider, provider_as_source, @@ -42,7 +45,9 @@ use crate::physical_plan::{ }; use crate::prelude::SessionContext; use std::borrow::Cow; -use std::collections::{HashMap, HashSet}; +#[cfg(feature = "object_store")] +use std::collections::HashMap; +use std::collections::HashSet; use std::sync::Arc; use arrow::array::{Array, ArrayRef, Int64Array, StringArray}; @@ -50,10 +55,13 @@ use arrow::compute::{cast, concat}; use arrow::datatypes::{DataType, Field, Schema, SchemaRef}; use arrow::util::display::{ArrayFormatter, FormatOptions}; use arrow_schema::FieldRef; +#[cfg(feature = "object_store")] use datafusion_common::config::{CsvOptions, JsonOptions}; +#[cfg(feature = "object_store")] +use datafusion_common::not_impl_err; use datafusion_common::{ Column, DFSchema, DataFusionError, ParamValues, ScalarValue, SchemaError, - TableReference, UnnestOptions, exec_err, internal_datafusion_err, not_impl_err, + TableReference, UnnestOptions, exec_err, internal_datafusion_err, plan_datafusion_err, plan_err, unqualified_field_not_found, }; use datafusion_expr::select_expr::SelectExpr; @@ -133,6 +141,7 @@ impl DataFrameWriteOptions { } /// Build the options HashMap to pass to CopyTo for sink configuration. + #[cfg(feature = "object_store")] fn build_sink_options(&self) -> HashMap { let mut options = HashMap::new(); if let Some(single_file) = self.single_file_output { @@ -2060,6 +2069,7 @@ impl DataFrame { /// # Ok(()) /// # } /// ``` + #[cfg(feature = "object_store")] pub async fn write_csv( self, path: &str, @@ -2130,6 +2140,7 @@ impl DataFrame { /// # Ok(()) /// # } /// ``` + #[cfg(feature = "object_store")] pub async fn write_json( self, path: &str, diff --git a/datafusion/core/src/datasource/mod.rs b/datafusion/core/src/datasource/mod.rs index de54078aafef4..8a2762f146420 100644 --- a/datafusion/core/src/datasource/mod.rs +++ b/datafusion/core/src/datasource/mod.rs @@ -19,11 +19,16 @@ //! //! [`ListingTable`]: crate::datasource::listing::ListingTable +#[cfg(feature = "object_store")] pub mod dynamic_file; +#[cfg(feature = "object_store")] pub mod file_format; +#[cfg(feature = "object_store")] pub mod listing; +#[cfg(feature = "object_store")] pub mod listing_table_factory; mod memory_test; +#[cfg(feature = "object_store")] pub mod physical_plan; pub mod provider; mod view_test; @@ -42,11 +47,13 @@ pub use datafusion_catalog::empty; pub use datafusion_catalog::memory; pub use datafusion_catalog::stream; pub use datafusion_catalog::view; +#[cfg(feature = "object_store")] pub use datafusion_datasource::projection; pub use datafusion_datasource::schema_adapter; pub use datafusion_datasource::sink; pub use datafusion_datasource::source; pub use datafusion_datasource::table_schema; +#[cfg(feature = "object_store")] pub use datafusion_execution::object_store; pub use datafusion_physical_expr::create_ordering; diff --git a/datafusion/core/src/datasource/provider.rs b/datafusion/core/src/datasource/provider.rs index e574042813a7b..18a24eb0882c5 100644 --- a/datafusion/core/src/datasource/provider.rs +++ b/datafusion/core/src/datasource/provider.rs @@ -17,29 +17,40 @@ //! Data source traits +#[cfg(feature = "object_store")] use std::sync::Arc; +#[cfg(feature = "object_store")] use async_trait::async_trait; +#[cfg(feature = "object_store")] use datafusion_catalog::Session; +#[cfg(feature = "object_store")] use datafusion_expr::CreateExternalTable; pub use datafusion_expr::{TableProviderFilterPushDown, TableType}; +#[cfg(feature = "object_store")] use futures::future::BoxFuture; +#[cfg(feature = "object_store")] use crate::catalog::{TableProvider, TableProviderFactory}; +#[cfg(feature = "object_store")] use crate::datasource::listing_table_factory::ListingTableFactory; +#[cfg(feature = "object_store")] use crate::datasource::stream::StreamTableFactory; +#[cfg(feature = "object_store")] use crate::error::Result; /// The default [`TableProviderFactory`] /// /// If [`CreateExternalTable`] is unbounded calls [`StreamTableFactory::create`], /// otherwise calls [`ListingTableFactory::create`] +#[cfg(feature = "object_store")] #[derive(Debug, Default)] pub struct DefaultTableFactory { stream: StreamTableFactory, listing: ListingTableFactory, } +#[cfg(feature = "object_store")] impl DefaultTableFactory { /// Creates a new [`DefaultTableFactory`] pub fn new() -> Self { @@ -47,6 +58,7 @@ impl DefaultTableFactory { } } +#[cfg(feature = "object_store")] #[async_trait] impl TableProviderFactory for DefaultTableFactory { // Hand-written `#[async_trait]` expansion to reduce compile time. See @@ -66,6 +78,7 @@ impl TableProviderFactory for DefaultTableFactory { } } +#[cfg(feature = "object_store")] impl DefaultTableFactory { fn create_boxed<'a>( &'a self, diff --git a/datafusion/core/src/execution/context/mod.rs b/datafusion/core/src/execution/context/mod.rs index ff1ad25811440..b13a1d49083aa 100644 --- a/datafusion/core/src/execution/context/mod.rs +++ b/datafusion/core/src/execution/context/mod.rs @@ -20,25 +20,31 @@ use std::collections::HashSet; use std::fmt::Debug; use std::sync::{Arc, Weak}; +#[cfg(feature = "object_store")] use std::time::Duration; +#[cfg(feature = "object_store")] +use super::options::ArrowReadOptions; +#[cfg(feature = "object_store")] use super::options::ReadOptions; +#[cfg(feature = "object_store")] +use crate::catalog::listing_schema::ListingSchemaProvider; +#[cfg(feature = "object_store")] use crate::datasource::dynamic_file::DynamicListTableFactory; +#[cfg(feature = "object_store")] +use crate::datasource::listing::{ + ListingOptions, ListingTable, ListingTableConfig, ListingTableUrl, +}; use crate::execution::session_state::SessionStateBuilder; use crate::{ - catalog::listing_schema::ListingSchemaProvider, catalog::{ CatalogProvider, CatalogProviderList, TableProvider, TableProviderFactory, }, dataframe::DataFrame, - datasource::listing::{ - ListingOptions, ListingTable, ListingTableConfig, ListingTableUrl, - }, datasource::{MemTable, ViewTable, provider_as_source}, error::Result, execution::{ FunctionRegistry, - options::ArrowReadOptions, runtime_env::{RuntimeEnv, RuntimeEnvBuilder}, }, logical_expr::AggregateUDF, @@ -50,31 +56,36 @@ use crate::{ SetVariable, TableType, UNNAMED_TABLE, }, physical_expr::PhysicalExpr, - physical_plan::ExecutionPlan, variable::{VarProvider, VarType}, }; // backwards compatibility pub use crate::execution::session_state::SessionState; -use arrow::datatypes::{Schema, SchemaRef}; +use arrow::datatypes::Schema; +#[cfg(feature = "object_store")] +use arrow::datatypes::SchemaRef; use arrow::record_batch::RecordBatch; use datafusion_catalog::MemoryCatalogProvider; use datafusion_catalog::memory::MemorySchemaProvider; -use datafusion_catalog::{ - DynamicFileCatalog, TableFunction, TableFunctionImpl, UrlTableFactory, -}; +#[cfg(feature = "object_store")] +use datafusion_catalog::{DynamicFileCatalog, UrlTableFactory}; +use datafusion_catalog::{TableFunction, TableFunctionImpl}; +#[cfg(feature = "object_store")] use datafusion_catalog_listing::SchemaSource; use datafusion_common::config::{ConfigField, ConfigOptions}; +#[cfg(feature = "object_store")] +use datafusion_common::internal_datafusion_err; use datafusion_common::metadata::ScalarAndMetadata; use datafusion_common::{ DFSchema, DataFusionError, ParamValues, SchemaError, SchemaReference, TableReference, config::{ConfigExtension, TableOptions}, - exec_datafusion_err, exec_err, internal_datafusion_err, not_impl_err, - plan_datafusion_err, plan_err, schema_err, + exec_datafusion_err, exec_err, not_impl_err, plan_datafusion_err, plan_err, + schema_err, tree_node::{TreeNodeRecursion, TreeNodeVisitor}, }; pub use datafusion_execution::TaskContext; +#[cfg(feature = "object_store")] use datafusion_execution::cache::cache_manager::{ DEFAULT_FILE_STATISTICS_MEMORY_LIMIT, DEFAULT_LIST_FILES_CACHE_MEMORY_LIMIT, DEFAULT_LIST_FILES_CACHE_TTL, DEFAULT_METADATA_CACHE_LIMIT, @@ -99,15 +110,22 @@ use datafusion_optimizer::analyzer::type_coercion::TypeCoercion; use datafusion_optimizer::simplify_expressions::ExprSimplifier; use datafusion_optimizer::{Analyzer, OptimizerContext}; use datafusion_optimizer::{AnalyzerRule, OptimizerRule}; +#[cfg(feature = "object_store")] use datafusion_session::SessionStore; +#[cfg(any(test, feature = "object_store"))] +use crate::physical_plan::ExecutionPlan; use async_trait::async_trait; use chrono::{DateTime, Utc}; +#[cfg(feature = "object_store")] use object_store::ObjectStore; use parking_lot::RwLock; +#[cfg(feature = "object_store")] use url::Url; +#[cfg(feature = "object_store")] mod csv; +#[cfg(feature = "object_store")] mod json; #[cfg(feature = "parquet")] mod parquet; @@ -118,29 +136,34 @@ mod avro; /// DataFilePaths adds a method to convert strings and vector of strings to vector of [`ListingTableUrl`] URLs. /// This allows methods such [`SessionContext::read_csv`] and [`SessionContext::read_avro`] /// to take either a single file or multiple files. +#[cfg(feature = "object_store")] pub trait DataFilePaths { /// Parse to a vector of [`ListingTableUrl`] URLs. fn to_urls(self) -> Result>; } +#[cfg(feature = "object_store")] impl DataFilePaths for &str { fn to_urls(self) -> Result> { Ok(vec![ListingTableUrl::parse(self)?]) } } +#[cfg(feature = "object_store")] impl DataFilePaths for String { fn to_urls(self) -> Result> { Ok(vec![ListingTableUrl::parse(self)?]) } } +#[cfg(feature = "object_store")] impl DataFilePaths for &String { fn to_urls(self) -> Result> { Ok(vec![ListingTableUrl::parse(self)?]) } } +#[cfg(feature = "object_store")] impl

DataFilePaths for Vec

where P: AsRef, @@ -312,6 +335,7 @@ impl SessionContext { } /// Finds any [`ListingSchemaProvider`]s and instructs them to reload tables from "disk" + #[cfg(feature = "object_store")] pub async fn refresh_catalogs(&self) -> Result<()> { let cat_names = self.catalog_names().clone(); for cat_name in cat_names.iter() { @@ -411,6 +435,7 @@ impl SessionContext { /// # Ok(()) /// # } /// ``` + #[cfg(feature = "object_store")] pub fn enable_url_table(self) -> Self { let current_catalog_list = Arc::clone(self.state.read().catalog_list()); let factory = Arc::new(DynamicListTableFactory::new(SessionStore::new())); @@ -517,6 +542,7 @@ impl SessionContext { /// // All files with the file:// url prefix will be read from the local file system /// ctx.register_object_store(object_store_url.as_ref(), Arc::new(object_store)); /// ``` + #[cfg(feature = "object_store")] pub fn register_object_store( &self, url: &Url, @@ -528,6 +554,7 @@ impl SessionContext { /// Deregisters an [`ObjectStore`] associated with the specific URL prefix. /// /// See [`RuntimeEnv::deregister_object_store`] for more details. + #[cfg(feature = "object_store")] pub fn deregister_object_store(&self, url: &Url) -> Result> { self.runtime_env().deregister_object_store(url) } @@ -1183,18 +1210,22 @@ impl SessionContext { builder.with_max_temp_directory_size(directory_size as u64) } "temp_directory" => builder.with_temp_file_path(value), + #[cfg(feature = "object_store")] "metadata_cache_limit" => { let limit = Self::parse_capacity_limit(variable, value)?; builder.with_metadata_cache_limit(limit) } + #[cfg(feature = "object_store")] "list_files_cache_limit" => { let limit = Self::parse_capacity_limit(variable, value)?; builder.with_object_list_cache_limit(limit) } + #[cfg(feature = "object_store")] "list_files_cache_ttl" => { let duration = Self::parse_duration(variable, value)?; builder.with_object_list_cache_ttl(Some(duration)) } + #[cfg(feature = "object_store")] "file_statistics_cache_limit" => { let limit = Self::parse_capacity_limit(variable, value)?; builder.with_file_statistics_cache_limit(limit) @@ -1235,17 +1266,21 @@ impl SessionContext { "temp_directory" => { builder.disk_manager_builder = Some(DiskManagerBuilder::default()); } + #[cfg(feature = "object_store")] "metadata_cache_limit" => { builder = builder.with_metadata_cache_limit(DEFAULT_METADATA_CACHE_LIMIT); } + #[cfg(feature = "object_store")] "list_files_cache_limit" => { builder = builder .with_object_list_cache_limit(DEFAULT_LIST_FILES_CACHE_MEMORY_LIMIT); } + #[cfg(feature = "object_store")] "list_files_cache_ttl" => { builder = builder.with_object_list_cache_ttl(DEFAULT_LIST_FILES_CACHE_TTL); } + #[cfg(feature = "object_store")] "file_statistics_cache_limit" => { builder = builder.with_file_statistics_cache_limit( DEFAULT_FILE_STATISTICS_MEMORY_LIMIT, @@ -1361,6 +1396,7 @@ impl SessionContext { } } + #[cfg(feature = "object_store")] fn parse_duration(config_name: &str, duration: &str) -> Result { if duration.trim().is_empty() { return Err(plan_datafusion_err!( @@ -1404,6 +1440,7 @@ impl SessionContext { Ok(duration) } + #[cfg(feature = "object_store")] fn check_overflow( config_name: &str, mins: Option, @@ -1471,15 +1508,18 @@ impl SessionContext { Ok(false) } + #[cfg_attr(not(feature = "object_store"), expect(unused_variables))] fn invalidate_caches( &self, table_ref: &TableReference, table_type: TableType, ) -> Result<()> { if table_type == TableType::Base { + #[cfg(feature = "object_store")] if let Some(lfc) = self.runtime_env().cache_manager.get_list_files_cache() { lfc.drop_table_entries(table_ref)?; } + #[cfg(feature = "object_store")] if let Some(fsc) = self.runtime_env().cache_manager.get_file_statistic_cache() { fsc.drop_table_entries(table_ref)?; @@ -1712,6 +1752,7 @@ impl SessionContext { /// /// For more control such as reading multiple files, you can use /// [`read_table`](Self::read_table) with a [`ListingTable`]. + #[cfg(feature = "object_store")] async fn _read_type<'a, P: DataFilePaths>( &self, table_paths: P, @@ -1765,6 +1806,7 @@ impl SessionContext { /// [`read_table`](Self::read_table) with a [`ListingTable`]. /// /// For an example, see [`read_csv`](Self::read_csv) + #[cfg(feature = "object_store")] pub async fn read_arrow( &self, table_paths: P, @@ -1834,6 +1876,7 @@ impl SessionContext { /// This method is `async` because it might need to resolve the schema. /// /// [`ObjectStore`]: object_store::ObjectStore + #[cfg(feature = "object_store")] pub async fn register_listing_table( &self, table_ref: impl Into, @@ -1859,6 +1902,7 @@ impl SessionContext { Ok(()) } + #[cfg(feature = "object_store")] fn register_type_check( &self, table_paths: P, @@ -1884,6 +1928,7 @@ impl SessionContext { /// Registers an Arrow file as a table that can be referenced from /// SQL statements executed against this context. + #[cfg(feature = "object_store")] pub async fn register_arrow( &self, table_ref: impl Into, @@ -2375,6 +2420,7 @@ mod tests { use crate::catalog::SchemaProvider; use crate::execution::session_state::SessionStateBuilder; + use crate::physical_plan::ExecutionPlan; use crate::physical_planner::PhysicalPlanner; use async_trait::async_trait; use datafusion_expr::planner::TypePlanner; diff --git a/datafusion/core/src/execution/mod.rs b/datafusion/core/src/execution/mod.rs index 2e3e09685bcc7..6ec139d2778b3 100644 --- a/datafusion/core/src/execution/mod.rs +++ b/datafusion/core/src/execution/mod.rs @@ -26,5 +26,6 @@ mod session_state_defaults; pub use session_state_defaults::SessionStateDefaults; // backwards compatibility +#[cfg(feature = "object_store")] pub use crate::datasource::file_format::options; pub use datafusion_execution::*; diff --git a/datafusion/core/src/execution/session_state.rs b/datafusion/core/src/execution/session_state.rs index aa8ba4c3b733b..406239ae75013 100644 --- a/datafusion/core/src/execution/session_state.rs +++ b/datafusion/core/src/execution/session_state.rs @@ -24,6 +24,7 @@ use std::fmt::Debug; use std::sync::Arc; use crate::catalog::{CatalogProviderList, SchemaProvider, TableProviderFactory}; +#[cfg(feature = "object_store")] use crate::datasource::file_format::FileFormatFactory; #[cfg(feature = "sql")] use crate::datasource::provider_as_source; @@ -45,8 +46,8 @@ use datafusion_common::config::{ConfigExtension, ConfigOptions, TableOptions}; use datafusion_common::display::{PlanType, StringifiedPlan, ToStringifiedPlan}; use datafusion_common::tree_node::TreeNode; use datafusion_common::{ - DFSchema, DataFusionError, ResolvedTableReference, TableReference, config_err, - exec_err, plan_datafusion_err, + DFSchema, DataFusionError, ResolvedTableReference, TableReference, exec_err, + plan_datafusion_err, }; use datafusion_execution::TaskContext; use datafusion_execution::config::SessionConfig; @@ -85,15 +86,21 @@ use datafusion_sql::{ use async_trait::async_trait; use chrono::{DateTime, Utc}; +#[cfg(feature = "object_store")] +use datafusion_common::config_err; use futures::future::BoxFuture; use itertools::Itertools; -use log::{debug, info}; +use log::debug; +#[cfg(feature = "object_store")] +use log::info; +#[cfg(feature = "object_store")] use object_store::ObjectStore; #[cfg(feature = "sql")] use sqlparser::{ ast::{Expr as SQLExpr, ExprWithAlias as SQLExprWithAlias}, dialect::dialect_from_str, }; +#[cfg(feature = "object_store")] use url::Url; use uuid::Uuid; @@ -187,6 +194,7 @@ struct SessionStateInner { /// Deserializer registry for extensions. serializer_registry: Arc, /// Holds registered external FileFormat implementations + #[cfg(feature = "object_store")] file_formats: HashMap>, /// Session configuration config: SessionConfig, @@ -241,8 +249,10 @@ impl Debug for SessionState { .field("config", &self.inner.config) .field("runtime_env", &self.inner.runtime_env) .field("catalog_list", &self.inner.catalog_list) - .field("serializer_registry", &self.inner.serializer_registry) - .field("file_formats", &self.inner.file_formats) + .field("serializer_registry", &self.inner.serializer_registry); + #[cfg(feature = "object_store")] + let ret = ret.field("file_formats", &self.inner.file_formats); + let ret = ret .field("execution_props", &self.execution_props) .field("table_options", &self.inner.table_options) .field("table_factories", &self.inner.table_factories) @@ -967,6 +977,7 @@ impl SessionState { /// Adds or updates a [FileFormatFactory] which can be used with COPY TO or /// CREATE EXTERNAL TABLE statements for reading and writing files of custom /// formats. + #[cfg(feature = "object_store")] pub fn register_file_format( &mut self, file_format: Arc, @@ -996,6 +1007,7 @@ impl SessionState { /// Retrieves a [FileFormatFactory] based on file extension which has been registered /// via SessionContext::register_file_format. Extensions are not case sensitive. + #[cfg(feature = "object_store")] pub fn get_file_format_factory( &self, ext: &str, @@ -1129,6 +1141,7 @@ pub struct SessionStateBuilder { window_functions: Option>>, extension_types: Option, serializer_registry: Option>, + #[cfg(feature = "object_store")] file_formats: Option>>, config: Option, table_options: Option, @@ -1172,6 +1185,7 @@ impl SessionStateBuilder { window_functions: None, extension_types: None, serializer_registry: None, + #[cfg(feature = "object_store")] file_formats: None, table_options: None, config: None, @@ -1237,6 +1251,7 @@ impl SessionStateBuilder { window_functions: Some(existing.window_functions.into_values().collect_vec()), extension_types: Some(existing.extension_types), serializer_registry: Some(existing.serializer_registry), + #[cfg(feature = "object_store")] file_formats: Some(existing.file_formats.into_values().collect_vec()), config: Some(new_config), table_options: Some(existing.table_options), @@ -1262,6 +1277,7 @@ impl SessionStateBuilder { .get_or_insert_with(HashMap::new) .extend(SessionStateDefaults::default_table_factories()); + #[cfg(feature = "object_store")] self.file_formats .get_or_insert_with(Vec::new) .extend(SessionStateDefaults::default_file_formats()); @@ -1509,6 +1525,7 @@ impl SessionStateBuilder { } /// Set the map of [`FileFormatFactory`]s + #[cfg(feature = "object_store")] pub fn with_file_formats( mut self, file_formats: Vec>, @@ -1610,6 +1627,7 @@ impl SessionStateBuilder { /// .with_default_features() /// .build(); /// ``` + #[cfg(feature = "object_store")] pub fn with_object_store( mut self, url: &Url, @@ -1650,6 +1668,7 @@ impl SessionStateBuilder { window_functions, extension_types, serializer_registry, + #[cfg(feature = "object_store")] file_formats, table_options, config, @@ -1690,6 +1709,7 @@ impl SessionStateBuilder { extension_types: Arc::new(MemoryExtensionTypeRegistry::default()), serializer_registry: serializer_registry .unwrap_or_else(|| Arc::new(EmptySerializerRegistry)), + #[cfg(feature = "object_store")] file_formats: HashMap::new(), table_options: table_options.unwrap_or_else(|| { TableOptions::default_from_session_config(config.options()) @@ -1708,6 +1728,7 @@ impl SessionStateBuilder { execution_props: execution_props.unwrap_or_default(), }; + #[cfg(feature = "object_store")] if let Some(file_formats) = file_formats { for file_format in file_formats { if let Err(e) = state.register_file_format(file_format, false) { @@ -1911,6 +1932,7 @@ impl SessionStateBuilder { } /// Returns the current file_formats value + #[cfg(feature = "object_store")] pub fn file_formats(&mut self) -> &mut Option>> { &mut self.file_formats } @@ -1984,8 +2006,10 @@ impl Debug for SessionStateBuilder { .field("config", &self.config) .field("runtime_env", &self.runtime_env) .field("catalog_list", &self.catalog_list) - .field("serializer_registry", &self.serializer_registry) - .field("file_formats", &self.file_formats) + .field("serializer_registry", &self.serializer_registry); + #[cfg(feature = "object_store")] + let ret = ret.field("file_formats", &self.file_formats); + let ret = ret .field("execution_props", &self.execution_props) .field("table_options", &self.table_options) .field("table_factories", &self.table_factories) @@ -2177,6 +2201,12 @@ impl ContextProvider for SessionContextProvider<'_> { ) -> datafusion_common::Result< Arc, > { + #[cfg(not(feature = "object_store"))] + return datafusion_common::not_impl_err!( + "File formats require the object_store feature: {ext}" + ); + + #[cfg(feature = "object_store")] self.state .inner .file_formats diff --git a/datafusion/core/src/execution/session_state_defaults.rs b/datafusion/core/src/execution/session_state_defaults.rs index 59d3c440c7865..e1d06b35c3dac 100644 --- a/datafusion/core/src/execution/session_state_defaults.rs +++ b/datafusion/core/src/execution/session_state_defaults.rs @@ -15,16 +15,22 @@ // specific language governing permissions and limitations // under the License. +#[cfg(feature = "object_store")] use crate::catalog::listing_schema::ListingSchemaProvider; use crate::catalog::{CatalogProvider, TableProviderFactory}; +#[cfg(feature = "object_store")] use crate::datasource::file_format::FileFormatFactory; +#[cfg(feature = "object_store")] use crate::datasource::file_format::arrow::ArrowFormatFactory; #[cfg(feature = "avro")] use crate::datasource::file_format::avro::AvroFormatFactory; +#[cfg(feature = "object_store")] use crate::datasource::file_format::csv::CsvFormatFactory; +#[cfg(feature = "object_store")] use crate::datasource::file_format::json::JsonFormatFactory; #[cfg(feature = "parquet")] use crate::datasource::file_format::parquet::ParquetFormatFactory; +#[cfg(feature = "object_store")] use crate::datasource::provider::DefaultTableFactory; use crate::execution::context::SessionState; #[cfg(feature = "nested_expressions")] @@ -33,6 +39,7 @@ use crate::{functions, functions_aggregate, functions_table, functions_window}; use datafusion_catalog::TableFunction; use datafusion_catalog::{MemoryCatalogProvider, MemorySchemaProvider}; use datafusion_execution::config::SessionConfig; +#[cfg(feature = "object_store")] use datafusion_execution::object_store::ObjectStoreUrl; use datafusion_execution::runtime_env::RuntimeEnv; use datafusion_expr::planner::ExprPlanner; @@ -40,6 +47,7 @@ use datafusion_expr::registry::ExtensionTypeRegistrationRef; use datafusion_expr::{AggregateUDF, HigherOrderUDF, ScalarUDF, WindowUDF}; use std::collections::HashMap; use std::sync::Arc; +#[cfg(feature = "object_store")] use url::Url; /// Defaults that are used as part of creating a SessionState such as table providers, @@ -49,20 +57,27 @@ pub struct SessionStateDefaults {} impl SessionStateDefaults { /// returns a map of the default [`TableProviderFactory`]s pub fn default_table_factories() -> HashMap> { + #[cfg_attr(not(feature = "object_store"), expect(unused_mut))] let mut table_factories: HashMap> = HashMap::new(); #[cfg(feature = "parquet")] table_factories.insert("PARQUET".into(), Arc::new(DefaultTableFactory::new())); + #[cfg(feature = "object_store")] table_factories.insert("CSV".into(), Arc::new(DefaultTableFactory::new())); + #[cfg(feature = "object_store")] table_factories.insert("JSON".into(), Arc::new(DefaultTableFactory::new())); + #[cfg(feature = "object_store")] table_factories.insert("NDJSON".into(), Arc::new(DefaultTableFactory::new())); + #[cfg(feature = "object_store")] table_factories.insert("AVRO".into(), Arc::new(DefaultTableFactory::new())); + #[cfg(feature = "object_store")] table_factories.insert("ARROW".into(), Arc::new(DefaultTableFactory::new())); table_factories } /// returns the default MemoryCatalogProvider + #[cfg_attr(not(feature = "object_store"), expect(unused_variables))] pub fn default_catalog( config: &SessionConfig, table_factories: &HashMap>, @@ -77,6 +92,7 @@ impl SessionStateDefaults { ) .expect("memory catalog provider can register schema"); + #[cfg(feature = "object_store")] Self::register_default_schema(config, table_factories, runtime, &default_catalog); default_catalog @@ -145,6 +161,7 @@ impl SessionStateDefaults { } /// returns the list of default [`FileFormatFactory`]s + #[cfg(feature = "object_store")] pub fn default_file_formats() -> Vec> { let file_formats: Vec> = vec![ #[cfg(feature = "parquet")] @@ -187,6 +204,7 @@ impl SessionStateDefaults { } /// registers the default schema + #[cfg(feature = "object_store")] pub fn register_default_schema( config: &SessionConfig, table_factories: &HashMap>, @@ -229,6 +247,7 @@ impl SessionStateDefaults { } /// registers the default [`FileFormatFactory`]s + #[cfg(feature = "object_store")] pub fn register_default_file_formats(state: &mut SessionState) { let formats = SessionStateDefaults::default_file_formats(); for format in formats { diff --git a/datafusion/core/src/lib.rs b/datafusion/core/src/lib.rs index 28ea4b4490c3e..3315f9f9e077c 100644 --- a/datafusion/core/src/lib.rs +++ b/datafusion/core/src/lib.rs @@ -779,6 +779,7 @@ pub mod scalar; // Re-export dependencies that are part of DataFusion public API (e.g. via DataFusionError) pub use arrow; +#[cfg(feature = "object_store")] pub use object_store; #[cfg(feature = "parquet")] diff --git a/datafusion/core/src/physical_planner.rs b/datafusion/core/src/physical_planner.rs index 882956a63c114..d1381e3f040c1 100644 --- a/datafusion/core/src/physical_planner.rs +++ b/datafusion/core/src/physical_planner.rs @@ -21,8 +21,11 @@ use std::borrow::Cow; use std::collections::{HashMap, HashSet}; use std::sync::Arc; +#[cfg(feature = "object_store")] use crate::datasource::file_format::file_type_to_format; +#[cfg(feature = "object_store")] use crate::datasource::listing::ListingTableUrl; +#[cfg(feature = "object_store")] use crate::datasource::physical_plan::{FileOutputMode, FileSinkConfig}; use crate::datasource::{DefaultTableSource, source_as_provider}; use crate::error::{DataFusionError, Result}; @@ -76,8 +79,10 @@ use datafusion_common::{ use datafusion_common::{ TableReference, assert_eq_or_internal_err, assert_or_internal_err, }; +#[cfg(feature = "object_store")] use datafusion_datasource::file_groups::FileGroup; use datafusion_datasource::memory::MemorySourceConfig; +#[cfg(feature = "object_store")] use datafusion_expr::dml::{CopyTo, InsertOp}; use datafusion_expr::expr::{ Alias, GroupingSet, NullTreatment, WindowFunction, WindowFunctionParams, @@ -671,6 +676,7 @@ impl DefaultPhysicalPlanner { } // 1 Child + #[cfg(feature = "object_store")] LogicalPlan::Copy(CopyTo { input, output_url, @@ -772,6 +778,10 @@ impl DefaultPhysicalPlanner { ) .await? } + #[cfg(not(feature = "object_store"))] + LogicalPlan::Copy(_) => { + return not_impl_err!("COPY TO requires the object_store feature"); + } LogicalPlan::Dml(DmlStatement { target, op: WriteOp::Insert(insert_op), @@ -3367,6 +3377,7 @@ mod tests { use super::*; use crate::datasource::MemTable; + #[cfg(feature = "object_store")] use crate::datasource::file_format::options::CsvReadOptions; use crate::physical_plan::{ DisplayAs, DisplayFormatType, Partitioning, PlanProperties, diff --git a/datafusion/core/src/prelude.rs b/datafusion/core/src/prelude.rs index 31d9d7eb471f0..659e97efc1f61 100644 --- a/datafusion/core/src/prelude.rs +++ b/datafusion/core/src/prelude.rs @@ -28,6 +28,7 @@ pub use crate::dataframe; pub use crate::dataframe::DataFrame; pub use crate::execution::context::{SQLOptions, SessionConfig, SessionContext}; +#[cfg(feature = "object_store")] pub use crate::execution::options::{ AvroReadOptions, CsvReadOptions, JsonReadOptions, ParquetReadOptions, }; diff --git a/datafusion/core/src/test/mod.rs b/datafusion/core/src/test/mod.rs index f46a5a0749065..d3e13dfbe0607 100644 --- a/datafusion/core/src/test/mod.rs +++ b/datafusion/core/src/test/mod.rs @@ -19,47 +19,63 @@ #![allow(missing_docs)] +#[cfg(feature = "object_store")] use std::fs::File; +#[cfg(feature = "object_store")] use std::io::prelude::*; +#[cfg(feature = "object_store")] use std::io::{BufReader, BufWriter}; +#[cfg(feature = "object_store")] use std::path::Path; use std::sync::Arc; +#[cfg(feature = "object_store")] use crate::datasource::file_format::FileFormat; +#[cfg(feature = "object_store")] use crate::datasource::file_format::csv::CsvFormat; +#[cfg(feature = "object_store")] use crate::datasource::file_format::file_compression_type::FileCompressionType; +#[cfg(feature = "object_store")] use crate::datasource::physical_plan::CsvSource; use crate::datasource::{MemTable, TableProvider}; use crate::error::Result; use crate::logical_expr::LogicalPlan; +#[cfg(feature = "object_store")] use crate::test_util::{aggr_test_schema, arrow_test_data}; +#[cfg(feature = "object_store")] use datafusion_common::config::CsvOptions; use arrow::array::{self, Array, ArrayRef, Decimal128Builder, Int32Array}; use arrow::datatypes::{DataType, Field, Schema}; use arrow::record_batch::RecordBatch; -#[cfg(feature = "compression")] +#[cfg(all(feature = "compression", feature = "object_store"))] use datafusion_common::DataFusionError; +#[cfg(feature = "object_store")] use datafusion_datasource::TableSchema; +#[cfg(feature = "object_store")] use datafusion_datasource::source::DataSourceExec; -#[cfg(feature = "compression")] +#[cfg(all(feature = "compression", feature = "object_store"))] use bzip2::Compression as BzCompression; -#[cfg(feature = "compression")] +#[cfg(all(feature = "compression", feature = "object_store"))] use bzip2::write::BzEncoder; +#[cfg(feature = "object_store")] use datafusion_datasource::file_groups::FileGroup; +#[cfg(feature = "object_store")] use datafusion_datasource::file_scan_config::FileScanConfigBuilder; +#[cfg(feature = "object_store")] use datafusion_datasource_csv::partitioned_csv_config; -#[cfg(feature = "compression")] +#[cfg(all(feature = "compression", feature = "object_store"))] use flate2::Compression as GzCompression; -#[cfg(feature = "compression")] +#[cfg(all(feature = "compression", feature = "object_store"))] use flate2::write::GzEncoder; -#[cfg(feature = "compression")] +#[cfg(all(feature = "compression", feature = "object_store"))] use liblzma::write::XzEncoder; +#[cfg(feature = "object_store")] use object_store::local_unpartitioned_file; -#[cfg(feature = "compression")] +#[cfg(all(feature = "compression", feature = "object_store"))] use zstd::Encoder as ZstdEncoder; pub fn create_table_dual() -> Arc { @@ -80,6 +96,7 @@ pub fn create_table_dual() -> Arc { } /// Returns a [`DataSourceExec`] that scans "aggregate_test_100.csv" with `partitions` partitions +#[cfg(feature = "object_store")] pub fn scan_partitioned_csv( partitions: usize, work_dir: &Path, @@ -113,6 +130,7 @@ pub fn scan_partitioned_csv( } /// Returns file groups [`Vec`] for scanning `partitions` of `filename` +#[cfg(feature = "object_store")] pub fn partitioned_file_groups( path: &str, filename: &str, @@ -267,5 +285,6 @@ fn make_decimal() -> RecordBatch { RecordBatch::try_new(Arc::new(schema), vec![Arc::new(array)]).unwrap() } +#[cfg(feature = "object_store")] pub mod object_store; pub mod variable; diff --git a/datafusion/core/src/test_util/mod.rs b/datafusion/core/src/test_util/mod.rs index ab7edef885b40..66ad934c279b7 100644 --- a/datafusion/core/src/test_util/mod.rs +++ b/datafusion/core/src/test_util/mod.rs @@ -24,6 +24,7 @@ pub mod csv; use futures::Stream; use std::collections::HashMap; +#[cfg(feature = "object_store")] use std::fmt::Formatter; use std::fs::File; use std::future::ready; @@ -33,24 +34,31 @@ use std::sync::Arc; use std::task::{Context, Poll}; use crate::catalog::{TableProvider, TableProviderFactory}; +#[cfg(feature = "object_store")] use crate::dataframe::DataFrame; use crate::datasource::stream::{FileStreamProvider, StreamConfig, StreamTable}; use crate::datasource::{empty::EmptyTable, provider_as_source}; use crate::error::Result; +#[cfg(feature = "object_store")] use crate::execution::session_state::CacheFactory; use crate::logical_expr::{LogicalPlanBuilder, UNNAMED_TABLE}; use crate::physical_plan::ExecutionPlan; -use crate::prelude::{CsvReadOptions, SessionContext}; +#[cfg(feature = "object_store")] +use crate::prelude::CsvReadOptions; +use crate::prelude::SessionContext; -use crate::execution::{SendableRecordBatchStream, SessionState, SessionStateBuilder}; +use crate::execution::SendableRecordBatchStream; +#[cfg(feature = "object_store")] +use crate::execution::{SessionState, SessionStateBuilder}; use arrow::datatypes::{DataType, Field, Schema, SchemaRef}; use arrow::record_batch::RecordBatch; use datafusion_catalog::Session; -use datafusion_common::{DFSchemaRef, TableReference, plan_err}; -use datafusion_expr::{ - CreateExternalTable, Expr, LogicalPlan, SortExpr, TableType, - UserDefinedLogicalNodeCore, -}; +#[cfg(feature = "object_store")] +use datafusion_common::DFSchemaRef; +use datafusion_common::{TableReference, plan_err}; +use datafusion_expr::{CreateExternalTable, Expr, SortExpr, TableType}; +#[cfg(feature = "object_store")] +use datafusion_expr::{LogicalPlan, UserDefinedLogicalNodeCore}; use std::pin::Pin; use async_trait::async_trait; @@ -113,6 +121,7 @@ pub fn aggr_test_schema() -> SchemaRef { } /// Register session context for the aggregate_test_100.csv file +#[cfg(feature = "object_store")] pub async fn register_aggregate_csv( ctx: &SessionContext, table_name: &str, @@ -129,6 +138,7 @@ pub async fn register_aggregate_csv( } /// Create a table from the aggregate_test_100.csv file with the specified name +#[cfg(feature = "object_store")] pub async fn test_table_with_name(name: &str) -> Result { let ctx = SessionContext::new(); register_aggregate_csv(&ctx, name).await?; @@ -136,6 +146,7 @@ pub async fn test_table_with_name(name: &str) -> Result { } /// Create a table from the aggregate_test_100.csv file with the name "aggregate_test_100" +#[cfg(feature = "object_store")] pub async fn test_table() -> Result { test_table_with_name("aggregate_test_100").await } @@ -315,11 +326,13 @@ impl RecordBatchStream for BoundedStream { } } +#[cfg(feature = "object_store")] #[derive(Hash, Eq, PartialEq, PartialOrd, Debug)] struct CacheNode { input: LogicalPlan, } +#[cfg(feature = "object_store")] impl UserDefinedLogicalNodeCore for CacheNode { fn name(&self) -> &str { "CacheNode" @@ -354,8 +367,10 @@ impl UserDefinedLogicalNodeCore for CacheNode { } #[derive(Debug)] +#[cfg(feature = "object_store")] struct TestCacheFactory {} +#[cfg(feature = "object_store")] impl CacheFactory for TestCacheFactory { fn create( &self, @@ -369,6 +384,7 @@ impl CacheFactory for TestCacheFactory { } /// Create a test table registered to a session context with an associated cache factory +#[cfg(feature = "object_store")] pub async fn test_table_with_cache_factory() -> Result { let session_state = SessionStateBuilder::new() .with_cache_factory(Some(Arc::new(TestCacheFactory {}))) diff --git a/datafusion/datasource-arrow/Cargo.toml b/datafusion/datasource-arrow/Cargo.toml index 6f50135403d69..2e03a0414de8b 100644 --- a/datafusion/datasource-arrow/Cargo.toml +++ b/datafusion/datasource-arrow/Cargo.toml @@ -37,8 +37,8 @@ async-trait = { workspace = true } bytes = { workspace = true } datafusion-common = { workspace = true, features = ["object_store"] } datafusion-common-runtime = { workspace = true } -datafusion-datasource = { workspace = true } -datafusion-execution = { workspace = true } +datafusion-datasource = { workspace = true, features = ["object_store"] } +datafusion-execution = { workspace = true, features = ["object_store"] } datafusion-expr = { workspace = true } datafusion-physical-expr-common = { workspace = true } datafusion-physical-plan = { workspace = true } diff --git a/datafusion/datasource-avro/Cargo.toml b/datafusion/datasource-avro/Cargo.toml index 70b675d63f427..3b48aaef12159 100644 --- a/datafusion/datasource-avro/Cargo.toml +++ b/datafusion/datasource-avro/Cargo.toml @@ -45,7 +45,7 @@ arrow-avro = { workspace = true } async-trait = { workspace = true } bytes = { workspace = true } datafusion-common = { workspace = true, features = ["object_store"] } -datafusion-datasource = { workspace = true } +datafusion-datasource = { workspace = true, features = ["object_store"] } datafusion-physical-expr-adapter = { workspace = true } datafusion-physical-plan = { workspace = true } datafusion-proto-models = { workspace = true, optional = true } diff --git a/datafusion/datasource-csv/Cargo.toml b/datafusion/datasource-csv/Cargo.toml index 7e7195dfda9d5..bb58c4a77de75 100644 --- a/datafusion/datasource-csv/Cargo.toml +++ b/datafusion/datasource-csv/Cargo.toml @@ -44,8 +44,8 @@ async-trait = { workspace = true } bytes = { workspace = true } datafusion-common = { workspace = true, features = ["object_store"] } datafusion-common-runtime = { workspace = true } -datafusion-datasource = { workspace = true } -datafusion-execution = { workspace = true } +datafusion-datasource = { workspace = true, features = ["object_store"] } +datafusion-execution = { workspace = true, features = ["object_store"] } datafusion-expr = { workspace = true } datafusion-physical-expr-common = { workspace = true } datafusion-physical-plan = { workspace = true } diff --git a/datafusion/datasource-json/Cargo.toml b/datafusion/datasource-json/Cargo.toml index 04192083f583a..f972476237ad5 100644 --- a/datafusion/datasource-json/Cargo.toml +++ b/datafusion/datasource-json/Cargo.toml @@ -44,8 +44,8 @@ async-trait = { workspace = true } bytes = { workspace = true } datafusion-common = { workspace = true, features = ["object_store"] } datafusion-common-runtime = { workspace = true } -datafusion-datasource = { workspace = true } -datafusion-execution = { workspace = true } +datafusion-datasource = { workspace = true, features = ["object_store"] } +datafusion-execution = { workspace = true, features = ["object_store"] } datafusion-expr = { workspace = true } datafusion-physical-expr-common = { workspace = true } datafusion-physical-plan = { workspace = true } diff --git a/datafusion/datasource-parquet/Cargo.toml b/datafusion/datasource-parquet/Cargo.toml index a2589af19a6ee..3e4a4fc0f3dcf 100644 --- a/datafusion/datasource-parquet/Cargo.toml +++ b/datafusion/datasource-parquet/Cargo.toml @@ -37,8 +37,8 @@ async-trait = { workspace = true } bytes = { workspace = true } datafusion-common = { workspace = true, features = ["object_store", "parquet"] } datafusion-common-runtime = { workspace = true } -datafusion-datasource = { workspace = true } -datafusion-execution = { workspace = true } +datafusion-datasource = { workspace = true, features = ["object_store"] } +datafusion-execution = { workspace = true, features = ["object_store"] } datafusion-expr = { workspace = true } datafusion-functions = { workspace = true } datafusion-functions-aggregate-common = { workspace = true } @@ -47,7 +47,7 @@ datafusion-physical-expr-adapter = { workspace = true } datafusion-physical-expr-common = { workspace = true } datafusion-physical-plan = { workspace = true } datafusion-proto-models = { workspace = true, optional = true } -datafusion-pruning = { workspace = true } +datafusion-pruning = { workspace = true, features = ["object_store"] } datafusion-session = { workspace = true } futures = { workspace = true } itertools = { workspace = true } diff --git a/datafusion/datasource/Cargo.toml b/datafusion/datasource/Cargo.toml index f09447b694f52..636026c012cb7 100644 --- a/datafusion/datasource/Cargo.toml +++ b/datafusion/datasource/Cargo.toml @@ -33,12 +33,21 @@ all-features = true [features] backtrace = ["datafusion-common/backtrace"] compression = ["async-compression", "liblzma", "bzip2", "flate2", "zstd", "tokio-util"] -default = ["compression"] +default = ["compression", "object_store"] +object_store = [ + "dep:object_store", + "datafusion-common/object_store", + "datafusion-execution/object_store", + "datafusion-physical-expr-adapter/object_store", + "datafusion-physical-plan/object_store", + "datafusion-session/object_store", +] # Enables protobuf conversions for datasource types, source serialization hooks, # and the shared `FileScanConfig` <-> proto conversion. Off by default so # consumers that never serialize plans pay nothing. Mirrors the `proto` feature # on `datafusion-physical-plan`. proto = [ + "object_store", "dep:datafusion-proto-models", "datafusion-physical-plan/proto", ] @@ -56,7 +65,7 @@ async-trait = { workspace = true } bytes = { workspace = true } bzip2 = { workspace = true, optional = true } chrono = { workspace = true } -datafusion-common = { workspace = true, features = ["object_store"] } +datafusion-common = { workspace = true } datafusion-common-runtime = { workspace = true } datafusion-execution = { workspace = true } datafusion-expr = { workspace = true } @@ -72,7 +81,7 @@ glob = { workspace = true } itertools = { workspace = true } liblzma = { workspace = true, optional = true } log = { workspace = true } -object_store = { workspace = true } +object_store = { workspace = true, optional = true } parking_lot = { workspace = true } rand = { workspace = true } tempfile = { workspace = true, optional = true } diff --git a/datafusion/datasource/src/mod.rs b/datafusion/datasource/src/mod.rs index d0bf64d81c71f..df23658f2df6b 100644 --- a/datafusion/datasource/src/mod.rs +++ b/datafusion/datasource/src/mod.rs @@ -28,18 +28,29 @@ //! A table that uses the `ObjectStore` listing capability //! to get the list of files to process. +#[cfg(feature = "object_store")] pub mod boundary_stream; pub mod decoder; +#[cfg(feature = "object_store")] pub mod display; +#[cfg(feature = "object_store")] pub mod file; +#[cfg(feature = "object_store")] pub mod file_compression_type; +#[cfg(feature = "object_store")] pub mod file_format; +#[cfg(feature = "object_store")] pub mod file_groups; +#[cfg(feature = "object_store")] pub mod file_scan_config; +#[cfg(feature = "object_store")] pub mod file_sink_config; +#[cfg(feature = "object_store")] pub mod file_stream; pub mod memory; +#[cfg(feature = "object_store")] pub mod morsel; +#[cfg(feature = "object_store")] pub mod projection; /// Protobuf conversions for [`FileRange`], [`PartitionedFile`] and /// [`FileGroup`](crate::file_groups::FileGroup), gated on the `proto` feature. @@ -48,28 +59,46 @@ mod proto; pub mod schema_adapter; pub mod sink; pub mod source; +#[cfg(feature = "object_store")] mod statistics; pub mod table_schema; #[cfg(test)] pub mod test_util; +#[cfg(feature = "object_store")] pub mod url; +#[cfg(feature = "object_store")] pub mod write; +#[cfg(feature = "object_store")] pub use self::file::as_file_source; +#[cfg(feature = "object_store")] pub use self::url::ListingTableUrl; +#[cfg(feature = "object_store")] use crate::file_groups::FileGroup; +#[cfg(feature = "object_store")] use arrow::datatypes::SchemaRef; +#[cfg(feature = "object_store")] use chrono::TimeZone; +#[cfg(feature = "object_store")] use datafusion_common::stats::{Precision, is_known_empty}; +#[cfg(feature = "object_store")] use datafusion_common::{ColumnStatistics, Result, TableReference}; +#[cfg(feature = "object_store")] use datafusion_common::{ScalarValue, Statistics}; +#[cfg(feature = "object_store")] use datafusion_physical_expr::LexOrdering; +#[cfg(feature = "object_store")] use futures::Stream; +#[cfg(feature = "object_store")] use object_store::{ObjectMeta, path::Path}; +#[cfg(feature = "object_store")] pub use statistics::compute_all_files_statistics; +#[cfg(feature = "object_store")] use std::any::Any; +#[cfg(feature = "object_store")] use std::pin::Pin; +#[cfg(feature = "object_store")] use std::sync::Arc; pub use table_schema::{TableSchema, TableSchemaBuilder}; @@ -85,12 +114,14 @@ pub type FileExtensions = datafusion_common::extensions::Extensions; since = "54.0.0", note = "This type is unused and will be removed in a future release" )] +#[cfg(feature = "object_store")] pub type PartitionedFileStream = Pin> + Send + Sync + 'static>>; /// Only scan a subset of Row Groups from the Parquet file whose data "midpoint" /// lies within the [start, end) byte offsets. This option can be used to scan non-overlapping /// sections of a Parquet file in parallel. +#[cfg(feature = "object_store")] #[derive(Debug, Clone, PartialEq, Hash, Eq, PartialOrd, Ord)] pub struct FileRange { /// Range start @@ -99,6 +130,7 @@ pub struct FileRange { pub end: i64, } +#[cfg(feature = "object_store")] impl FileRange { /// returns true if this file range contains the specified offset pub fn contains(&self, offset: i64) -> bool { @@ -122,6 +154,7 @@ impl FileRange { /// - `distinct_count = 1` (single distinct value per file for each partition column) /// /// This enables query optimizers to use partition column bounds for pruning and planning. +#[cfg(feature = "object_store")] pub struct PartitionedFile { /// Path for the file (e.g. URL, filesystem path, etc) pub object_meta: ObjectMeta, @@ -176,6 +209,7 @@ pub struct PartitionedFile { pub arrow_schema: Option, } +#[cfg(feature = "object_store")] impl PartitionedFile { /// Create a simple file without metadata or partition pub fn new(path: impl Into, size: u64) -> Self { @@ -395,6 +429,7 @@ impl PartitionedFile { } } +#[cfg(feature = "object_store")] impl From for PartitionedFile { fn from(object_meta: ObjectMeta) -> Self { PartitionedFile { @@ -450,6 +485,7 @@ impl From for PartitionedFile { /// File 2: [40, 140] /// File 3: [60, 160] /// File 4: [80, 180] +#[cfg(feature = "object_store")] pub fn generate_test_files(num_files: usize, overlap_factor: f64) -> Vec { let mut files = Vec::with_capacity(num_files); if num_files == 0 { @@ -502,6 +538,7 @@ pub fn generate_test_files(num_files: usize, overlap_factor: f64) -> Vec bool { for group in file_groups { // Known-empty files contribute no rows and may not have min/max @@ -539,6 +576,7 @@ mod tests { use datafusion_execution::object_store::{ DefaultObjectStoreRegistry, ObjectStoreRegistry, }; + #[cfg(feature = "object_store")] use object_store::{local::LocalFileSystem, path::Path}; use std::{collections::HashMap, ops::Not, sync::Arc}; use url::Url; diff --git a/datafusion/datasource/src/source.rs b/datafusion/datasource/src/source.rs index 741010c595197..991a100bf0d8e 100644 --- a/datafusion/datasource/src/source.rs +++ b/datafusion/datasource/src/source.rs @@ -26,10 +26,10 @@ use datafusion_physical_expr::projection::ProjectionExprs; use datafusion_physical_plan::execution_plan::{ Boundedness, EmissionType, SchedulingType, }; +#[cfg(feature = "object_store")] +use datafusion_physical_plan::metrics::BaselineMetrics; use datafusion_physical_plan::metrics::SplitMetrics; -use datafusion_physical_plan::metrics::{ - BaselineMetrics, ExecutionPlanMetricsSet, MetricsSet, -}; +use datafusion_physical_plan::metrics::{ExecutionPlanMetricsSet, MetricsSet}; use datafusion_physical_plan::projection::ProjectionExec; use datafusion_physical_plan::stream::BatchSplitStream; use datafusion_physical_plan::{ @@ -38,7 +38,9 @@ use datafusion_physical_plan::{ }; use itertools::Itertools; +#[cfg(feature = "object_store")] use crate::file::FileSource; +#[cfg(feature = "object_store")] use crate::file_scan_config::FileScanConfig; use datafusion_common::config::ConfigOptions; use datafusion_common::tree_node::TreeNodeRecursion; @@ -483,10 +485,12 @@ impl ExecutionPlan for DataSourceExec { } fn metrics(&self) -> Option { + #[cfg_attr(not(feature = "object_store"), expect(unused_mut))] let mut metrics = self.data_source.metrics().clone_inner(); // Add `output_rows_skew` metric to the metrics set. // Done here because it's a derived metric from output_rows metric. + #[cfg(feature = "object_store")] if let Some(file_scan_config) = self.data_source.downcast_ref::() && file_scan_config.file_source().file_type() == "parquet" && let Some(output_rows_skew) = @@ -682,6 +686,7 @@ impl DataSourceExec { /// Returns `None` if /// 1. the datasource is not scanning files (`FileScanConfig`) /// 2. The [`FileScanConfig::file_source`] is not of type `T` + #[cfg(feature = "object_store")] pub fn downcast_to_file_source( &self, ) -> Option<(&FileScanConfig, &T)> { diff --git a/datafusion/execution/Cargo.toml b/datafusion/execution/Cargo.toml index c9d4acd3644ba..41ff5a2b14a8f 100644 --- a/datafusion/execution/Cargo.toml +++ b/datafusion/execution/Cargo.toml @@ -41,9 +41,11 @@ workspace = true name = "datafusion_execution" [features] -default = ["sql"] +default = ["sql", "object_store"] +object_store = ["dep:object_store", "datafusion-common/object_store"] parquet_encryption = [ + "object_store", "parquet/encryption", ] arrow_buffer_pool = [ @@ -62,7 +64,7 @@ datafusion-expr = { workspace = true, default-features = false } datafusion-physical-expr-common = { workspace = true, default-features = false } futures = { workspace = true } log = { workspace = true } -object_store = { workspace = true, features = ["fs"] } +object_store = { workspace = true, optional = true, features = ["fs"] } parking_lot = { workspace = true } parquet = { workspace = true, optional = true } pin-project-lite = { workspace = true } diff --git a/datafusion/execution/src/cache/mod.rs b/datafusion/execution/src/cache/mod.rs index f47a3f3ca49f3..84d2a2b5dfc5e 100644 --- a/datafusion/execution/src/cache/mod.rs +++ b/datafusion/execution/src/cache/mod.rs @@ -15,6 +15,7 @@ // specific language governing permissions and limitations // under the License. +#[cfg(feature = "object_store")] pub mod cache_manager; pub mod lru_queue; @@ -24,9 +25,12 @@ use datafusion_common::arrow::datatypes::{DataType, Schema}; use datafusion_common::heap_size::{DFHeapSize, DFHeapSizeCtx}; use datafusion_common::instant::Instant; use datafusion_common::{HashMap, TableReference}; +#[cfg(feature = "object_store")] use object_store::path::Path; use std::collections::hash_map::DefaultHasher; -use std::fmt::{Debug, Display, Formatter}; +#[cfg(feature = "object_store")] +use std::fmt::Display; +use std::fmt::{Debug, Formatter}; use std::hash::{Hash, Hasher}; use std::time::Duration; @@ -123,6 +127,7 @@ impl Debug for dyn Cache { } } +#[cfg(feature = "object_store")] impl CacheKey for Path { fn size(&self) -> usize { self.as_ref().heap_size(&mut DFHeapSizeCtx::default()) @@ -133,6 +138,7 @@ impl CacheKey for Path { } } +#[cfg(feature = "object_store")] impl CacheKey for TableScopedPath { fn size(&self) -> usize { DFHeapSize::heap_size(self, &mut DFHeapSizeCtx::default()) @@ -146,18 +152,21 @@ impl CacheKey for TableScopedPath { /// Each entry is scoped to its use within a specific table so that the cache /// can differentiate between identical paths in different tables, and /// table-level cache invalidation. +#[cfg(feature = "object_store")] #[derive(PartialEq, Eq, Hash, Clone, Debug)] pub struct TableScopedPath { pub table: Option, pub path: Path, } +#[cfg(feature = "object_store")] impl DFHeapSize for TableScopedPath { fn heap_size(&self, ctx: &mut DFHeapSizeCtx) -> usize { self.path.as_ref().heap_size(ctx) + self.table.heap_size(ctx) } } +#[cfg(feature = "object_store")] impl Display for TableScopedPath { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { if let Some(table) = &self.table { diff --git a/datafusion/execution/src/lib.rs b/datafusion/execution/src/lib.rs index 5af7064f1cb8b..cf5c41eb67007 100644 --- a/datafusion/execution/src/lib.rs +++ b/datafusion/execution/src/lib.rs @@ -32,6 +32,7 @@ pub mod cache; pub mod config; pub mod disk_manager; pub mod memory_pool; +#[cfg(feature = "object_store")] pub mod object_store; #[cfg(feature = "parquet_encryption")] pub mod parquet_encryption; diff --git a/datafusion/execution/src/runtime_env.rs b/datafusion/execution/src/runtime_env.rs index fcfe51267e65f..1e2503d7dd81c 100644 --- a/datafusion/execution/src/runtime_env.rs +++ b/datafusion/execution/src/runtime_env.rs @@ -24,20 +24,26 @@ use crate::{ memory_pool::{ GreedyMemoryPool, MemoryPool, TrackConsumersPool, UnboundedMemoryPool, }, - object_store::{DefaultObjectStoreRegistry, ObjectStoreRegistry}, }; +#[cfg(feature = "object_store")] use crate::cache::cache_manager::{CacheManager, CacheManagerConfig}; +#[cfg(feature = "object_store")] +use crate::object_store::{DefaultObjectStoreRegistry, ObjectStoreRegistry}; #[cfg(feature = "parquet_encryption")] use crate::parquet_encryption::{EncryptionFactory, EncryptionFactoryRegistry}; use datafusion_common::{Result, config::ConfigEntry}; +#[cfg(feature = "object_store")] use object_store::ObjectStore; +use std::path::PathBuf; use std::sync::Arc; +#[cfg(feature = "object_store")] +use std::time::Duration; use std::{ fmt::{Debug, Formatter}, num::NonZeroUsize, }; -use std::{path::PathBuf, time::Duration}; +#[cfg(feature = "object_store")] use url::Url; #[derive(Clone)] @@ -76,8 +82,10 @@ pub struct RuntimeEnv { /// Manage temporary files during query execution pub disk_manager: Arc, /// Manage temporary cache during query execution + #[cfg(feature = "object_store")] pub cache_manager: Arc, /// Object Store Registry + #[cfg(feature = "object_store")] pub object_store_registry: Arc, /// Parquet encryption factory registry #[cfg(feature = "parquet_encryption")] @@ -95,9 +103,13 @@ struct RuntimeConfigValues { max_temp_directory_size: Option, max_spill_merge_fan_in: Option, temp_directory: Option, + #[cfg(feature = "object_store")] metadata_cache_limit: Option, + #[cfg(feature = "object_store")] list_files_cache_limit: Option, + #[cfg(feature = "object_store")] list_files_cache_ttl: Option, + #[cfg(feature = "object_store")] file_statistics_cache_limit: Option, } @@ -113,9 +125,13 @@ impl RuntimeConfigValues { max_temp_directory_size, max_spill_merge_fan_in, temp_directory, + #[cfg(feature = "object_store")] metadata_cache_limit, + #[cfg(feature = "object_store")] list_files_cache_limit, + #[cfg(feature = "object_store")] list_files_cache_ttl, + #[cfg(feature = "object_store")] file_statistics_cache_limit, } = self; vec![ @@ -139,21 +155,25 @@ impl RuntimeConfigValues { value: temp_directory, description: "The path to the temporary file directory.", }, + #[cfg(feature = "object_store")] ConfigEntry { key: "datafusion.runtime.metadata_cache_limit".to_string(), value: metadata_cache_limit, description: "Maximum memory to use for file metadata cache such as Parquet metadata. Supports suffixes K (kilobytes), M (megabytes), and G (gigabytes) or '0' for 0. Example: '2G' for 2 gigabytes.", }, + #[cfg(feature = "object_store")] ConfigEntry { key: "datafusion.runtime.list_files_cache_limit".to_string(), value: list_files_cache_limit, description: "Maximum memory to use for list files cache. Supports suffixes K (kilobytes), M (megabytes), and G (gigabytes) or '0' for 0. Example: '2G' for 2 gigabytes.", }, + #[cfg(feature = "object_store")] ConfigEntry { key: "datafusion.runtime.list_files_cache_ttl".to_string(), value: list_files_cache_ttl, description: "TTL (time-to-live) of the entries in the list file cache. Supports units m (minutes), and s (seconds). Example: '2m' for 2 minutes.", }, + #[cfg(feature = "object_store")] ConfigEntry { key: "datafusion.runtime.file_statistics_cache_limit".to_string(), value: file_statistics_cache_limit, @@ -203,6 +223,7 @@ impl RuntimeEnv { /// // register the object store with the runtime environment /// runtime_env.register_object_store(&base_url, Arc::new(http_store)); /// ``` + #[cfg(feature = "object_store")] pub fn register_object_store( &self, url: &Url, @@ -213,6 +234,7 @@ impl RuntimeEnv { /// Deregisters a custom `ObjectStore` previously registered for a specific url. /// See [`ObjectStoreRegistry::deregister_store`] for more details. + #[cfg(feature = "object_store")] pub fn deregister_object_store(&self, url: &Url) -> Result> { self.object_store_registry.deregister_store(url) } @@ -220,6 +242,7 @@ impl RuntimeEnv { /// Retrieves a `ObjectStore` instance for a url by consulting the /// registry. See [`ObjectStoreRegistry::get_store`] for more /// details. + #[cfg(feature = "object_store")] pub fn object_store(&self, url: impl AsRef) -> Result> { self.object_store_registry.get_store(url.as_ref()) } @@ -269,6 +292,7 @@ impl RuntimeEnv { } } + #[cfg(feature = "object_store")] fn format_duration(duration: Duration) -> String { let total = duration.as_secs(); let mins = total / 60; @@ -304,42 +328,61 @@ impl RuntimeEnv { ) }; - let metadata_cache_limit = self.cache_manager.get_metadata_cache_limit(); - let metadata_cache_value = format_byte_size( - metadata_cache_limit - .try_into() - .expect("Metadata cache size conversion failed"), - ); - - let list_files_cache_limit = self.cache_manager.get_list_files_cache_limit(); - let list_files_cache_value = format_byte_size( - list_files_cache_limit - .try_into() - .expect("List files cache size conversion failed"), - ); - - let list_files_cache_ttl = self - .cache_manager - .get_list_files_cache_ttl() - .map(format_duration); - - let file_statistics_cache_limit = - self.cache_manager.get_file_statistic_cache_limit(); - let file_statistics_cache_value = format_byte_size( - file_statistics_cache_limit - .try_into() - .expect("File statistics cache size conversion failed"), - ); + #[cfg(feature = "object_store")] + let ( + metadata_cache_value, + list_files_cache_value, + list_files_cache_ttl, + file_statistics_cache_value, + ) = { + let metadata_cache_limit = self.cache_manager.get_metadata_cache_limit(); + let metadata_cache_value = format_byte_size( + metadata_cache_limit + .try_into() + .expect("Metadata cache size conversion failed"), + ); + + let list_files_cache_limit = self.cache_manager.get_list_files_cache_limit(); + let list_files_cache_value = format_byte_size( + list_files_cache_limit + .try_into() + .expect("List files cache size conversion failed"), + ); + + let list_files_cache_ttl = self + .cache_manager + .get_list_files_cache_ttl() + .map(format_duration); + + let file_statistics_cache_limit = + self.cache_manager.get_file_statistic_cache_limit(); + let file_statistics_cache_value = format_byte_size( + file_statistics_cache_limit + .try_into() + .expect("File statistics cache size conversion failed"), + ); + + ( + Some(metadata_cache_value), + Some(list_files_cache_value), + list_files_cache_ttl, + Some(file_statistics_cache_value), + ) + }; RuntimeConfigValues { memory_limit: memory_limit_value, max_temp_directory_size: Some(max_temp_dir_value), max_spill_merge_fan_in: Some(max_spill_merge_fan_in), temp_directory: temp_dir_value, - metadata_cache_limit: Some(metadata_cache_value), - list_files_cache_limit: Some(list_files_cache_value), + #[cfg(feature = "object_store")] + metadata_cache_limit: metadata_cache_value, + #[cfg(feature = "object_store")] + list_files_cache_limit: list_files_cache_value, + #[cfg(feature = "object_store")] list_files_cache_ttl, - file_statistics_cache_limit: Some(file_statistics_cache_value), + #[cfg(feature = "object_store")] + file_statistics_cache_limit: file_statistics_cache_value, } .into_config_entries() } @@ -365,8 +408,10 @@ pub struct RuntimeEnvBuilder { /// Defaults to using an [`UnboundedMemoryPool`] if `None` pub memory_pool: Option>, /// CacheManager to manage cache data + #[cfg(feature = "object_store")] pub cache_manager: CacheManagerConfig, /// ObjectStoreRegistry to get object store based on url + #[cfg(feature = "object_store")] pub object_store_registry: Arc, /// Parquet encryption factory registry #[cfg(feature = "parquet_encryption")] @@ -386,7 +431,9 @@ impl RuntimeEnvBuilder { disk_manager: Default::default(), disk_manager_builder: Default::default(), memory_pool: Default::default(), + #[cfg(feature = "object_store")] cache_manager: Default::default(), + #[cfg(feature = "object_store")] object_store_registry: Arc::new(DefaultObjectStoreRegistry::default()), #[cfg(feature = "parquet_encryption")] parquet_encryption_factory_registry: Default::default(), @@ -406,12 +453,14 @@ impl RuntimeEnvBuilder { } /// Customize cache policy + #[cfg(feature = "object_store")] pub fn with_cache_manager(mut self, cache_manager: CacheManagerConfig) -> Self { self.cache_manager = cache_manager; self } /// Customize object store registry + #[cfg(feature = "object_store")] pub fn with_object_store_registry( mut self, object_store_registry: Arc, @@ -458,23 +507,27 @@ impl RuntimeEnvBuilder { } /// Specify the limit of the file-embedded metadata cache, in bytes. + #[cfg(feature = "object_store")] pub fn with_metadata_cache_limit(mut self, limit: usize) -> Self { self.cache_manager = self.cache_manager.with_metadata_cache_limit(limit); self } /// Specifies the memory limit for the object list cache, in bytes. + #[cfg(feature = "object_store")] pub fn with_object_list_cache_limit(mut self, limit: usize) -> Self { self.cache_manager = self.cache_manager.with_list_files_cache_limit(limit); self } /// Specifies the duration entries in the object list cache will be considered valid. + #[cfg(feature = "object_store")] pub fn with_object_list_cache_ttl(mut self, ttl: Option) -> Self { self.cache_manager = self.cache_manager.with_list_files_cache_ttl(ttl); self } + #[cfg(feature = "object_store")] pub fn with_file_statistics_cache_limit(mut self, limit: usize) -> Self { self.cache_manager = self.cache_manager.with_file_statistics_cache_limit(limit); self @@ -486,7 +539,9 @@ impl RuntimeEnvBuilder { disk_manager, disk_manager_builder, memory_pool, + #[cfg(feature = "object_store")] cache_manager, + #[cfg(feature = "object_store")] object_store_registry, #[cfg(feature = "parquet_encryption")] parquet_encryption_factory_registry, @@ -503,7 +558,9 @@ impl RuntimeEnvBuilder { Ok(RuntimeEnv { memory_pool, disk_manager, + #[cfg(feature = "object_store")] cache_manager: CacheManager::try_new(&cache_manager)?, + #[cfg(feature = "object_store")] object_store_registry, #[cfg(feature = "parquet_encryption")] parquet_encryption_factory_registry, @@ -517,6 +574,7 @@ impl RuntimeEnvBuilder { /// Create a new RuntimeEnvBuilder from an existing RuntimeEnv pub fn from_runtime_env(runtime_env: &RuntimeEnv) -> Self { + #[cfg(feature = "object_store")] let cache_config = CacheManagerConfig { file_statistics_cache: runtime_env.cache_manager.get_file_statistic_cache(), file_statistics_cache_limit: runtime_env @@ -537,7 +595,9 @@ impl RuntimeEnvBuilder { disk_manager: Some(Arc::clone(&runtime_env.disk_manager)), disk_manager_builder: None, memory_pool: Some(Arc::clone(&runtime_env.memory_pool)), + #[cfg(feature = "object_store")] cache_manager: cache_config, + #[cfg(feature = "object_store")] object_store_registry: Arc::clone(&runtime_env.object_store_registry), #[cfg(feature = "parquet_encryption")] parquet_encryption_factory_registry: Arc::clone( @@ -553,9 +613,13 @@ impl RuntimeEnvBuilder { max_temp_directory_size: Some("100G".to_string()), max_spill_merge_fan_in: Some("0".to_string()), temp_directory: None, + #[cfg(feature = "object_store")] metadata_cache_limit: Some("50M".to_owned()), + #[cfg(feature = "object_store")] list_files_cache_limit: Some("1M".to_owned()), + #[cfg(feature = "object_store")] list_files_cache_ttl: None, + #[cfg(feature = "object_store")] file_statistics_cache_limit: Some("20M".to_owned()), } .into_config_entries() diff --git a/datafusion/functions-aggregate/Cargo.toml b/datafusion/functions-aggregate/Cargo.toml index ae6bf70088f58..44197e910a22b 100644 --- a/datafusion/functions-aggregate/Cargo.toml +++ b/datafusion/functions-aggregate/Cargo.toml @@ -109,4 +109,6 @@ name = "variance" harness = false [features] +object_store = ["datafusion-execution/object_store"] +default = ["object_store"] force_hash_collisions = ["datafusion-common/force_hash_collisions"] diff --git a/datafusion/functions-nested/Cargo.toml b/datafusion/functions-nested/Cargo.toml index cfa642f6205da..b43398c57ca9e 100644 --- a/datafusion/functions-nested/Cargo.toml +++ b/datafusion/functions-nested/Cargo.toml @@ -41,7 +41,12 @@ workspace = true name = "datafusion_functions_nested" [features] -default = ["sql"] +object_store = [ + "datafusion-execution/object_store", + "datafusion-functions/object_store", + "datafusion-functions-aggregate/object_store", +] +default = ["object_store", "sql"] sql = ["datafusion-expr/sql"] [dependencies] diff --git a/datafusion/functions-table/Cargo.toml b/datafusion/functions-table/Cargo.toml index fb02c2c5e2cb3..e2f3164fa3fce 100644 --- a/datafusion/functions-table/Cargo.toml +++ b/datafusion/functions-table/Cargo.toml @@ -40,6 +40,10 @@ workspace = true [lib] name = "datafusion_functions_table" +[features] +object_store = ["datafusion-catalog/object_store", "datafusion-physical-plan/object_store"] +default = ["object_store"] + [dependencies] arrow = { workspace = true } async-trait = { workspace = true } diff --git a/datafusion/functions/Cargo.toml b/datafusion/functions/Cargo.toml index 28765bdda9e2e..485e573825a65 100644 --- a/datafusion/functions/Cargo.toml +++ b/datafusion/functions/Cargo.toml @@ -38,11 +38,13 @@ all-features = true workspace = true [features] +object_store = ["datafusion-execution/object_store"] crypto_expressions = ["md-5", "sha2", "blake2", "blake3"] # enable datetime functions datetime_expressions = ["chrono-tz"] # Enable encoding by default so the doctests work. In general don't automatically enable all packages. default = [ + "object_store", "datetime_expressions", "encoding_expressions", "math_expressions", diff --git a/datafusion/physical-expr-adapter/Cargo.toml b/datafusion/physical-expr-adapter/Cargo.toml index 453c8bdaacb4a..e50650ac1f66c 100644 --- a/datafusion/physical-expr-adapter/Cargo.toml +++ b/datafusion/physical-expr-adapter/Cargo.toml @@ -15,6 +15,10 @@ rust-version = { workspace = true } name = "datafusion_physical_expr_adapter" path = "src/lib.rs" +[features] +object_store = ["datafusion-functions/object_store"] +default = ["object_store"] + [dependencies] arrow = { workspace = true } datafusion-common = { workspace = true } diff --git a/datafusion/physical-optimizer/Cargo.toml b/datafusion/physical-optimizer/Cargo.toml index cb03303ac3c3f..f1b981a31f27e 100644 --- a/datafusion/physical-optimizer/Cargo.toml +++ b/datafusion/physical-optimizer/Cargo.toml @@ -38,6 +38,13 @@ all-features = true workspace = true [features] +object_store = [ + "datafusion-execution/object_store", + "datafusion-physical-plan/object_store", + "datafusion-pruning/object_store", + "datafusion-session/object_store", +] +default = ["object_store"] recursive_protection = ["dep:recursive"] [dependencies] diff --git a/datafusion/physical-plan/Cargo.toml b/datafusion/physical-plan/Cargo.toml index 534cea8ea9cbb..f17a8b2507b95 100644 --- a/datafusion/physical-plan/Cargo.toml +++ b/datafusion/physical-plan/Cargo.toml @@ -38,6 +38,8 @@ all-features = true workspace = true [features] +object_store = ["datafusion-execution/object_store", "datafusion-functions/object_store"] +default = ["object_store"] force_hash_collisions = [] test_utils = ["arrow/test_utils"] tokio_coop = [] @@ -88,7 +90,7 @@ num-traits = { workspace = true } parking_lot = { workspace = true } pin-project-lite = { workspace = true } serde_json = { workspace = true, features = ["preserve_order"] } -tokio = { workspace = true } +tokio = { workspace = true, features = ["time"] } [dev-dependencies] arrow-data = { workspace = true } diff --git a/datafusion/proto/Cargo.toml b/datafusion/proto/Cargo.toml index 4b11bea01103f..1f8c529e419f4 100644 --- a/datafusion/proto/Cargo.toml +++ b/datafusion/proto/Cargo.toml @@ -57,16 +57,16 @@ avro = ["datafusion-datasource-avro"] [dependencies] arrow = { workspace = true } -datafusion-catalog = { workspace = true } +datafusion-catalog = { workspace = true, features = ["object_store"] } datafusion-catalog-listing = { workspace = true } datafusion-common = { workspace = true } -datafusion-datasource = { workspace = true, features = ["proto"] } +datafusion-datasource = { workspace = true, features = ["object_store", "proto"] } datafusion-datasource-arrow = { workspace = true, features = ["proto"] } datafusion-datasource-avro = { workspace = true, optional = true, features = ["proto"] } datafusion-datasource-csv = { workspace = true, features = ["proto"] } datafusion-datasource-json = { workspace = true, features = ["proto"] } datafusion-datasource-parquet = { workspace = true, optional = true, features = ["proto"] } -datafusion-execution = { workspace = true } +datafusion-execution = { workspace = true, features = ["object_store"] } datafusion-expr = { workspace = true, features = ["proto"] } datafusion-functions-table = { workspace = true } datafusion-physical-expr = { workspace = true, features = ["proto"] } diff --git a/datafusion/pruning/Cargo.toml b/datafusion/pruning/Cargo.toml index a914a0a079bcd..b40b0aeb67230 100644 --- a/datafusion/pruning/Cargo.toml +++ b/datafusion/pruning/Cargo.toml @@ -15,6 +15,13 @@ authors = { workspace = true } [lints] workspace = true +[features] +default = ["object_store"] +object_store = [ + "datafusion-datasource/object_store", + "datafusion-physical-plan/object_store", +] + [dependencies] arrow = { workspace = true } datafusion-common = { workspace = true, default-features = true } diff --git a/datafusion/pruning/src/lib.rs b/datafusion/pruning/src/lib.rs index 88a7fdda5e733..cd0367688e4d8 100644 --- a/datafusion/pruning/src/lib.rs +++ b/datafusion/pruning/src/lib.rs @@ -17,12 +17,14 @@ #![cfg_attr(test, allow(clippy::needless_pass_by_value))] +#[cfg(feature = "object_store")] mod file_pruner; mod in_list; mod primitive_in_list; mod pruning_predicate; mod string_in_list; +#[cfg(feature = "object_store")] pub use file_pruner::FilePruner; pub use pruning_predicate::{ MAX_IN_LIST_SIZE, PredicateRewriter, PruningPredicate, PruningPredicateBuilder, diff --git a/datafusion/session/Cargo.toml b/datafusion/session/Cargo.toml index e6c277a8b8493..e26215cf70d70 100644 --- a/datafusion/session/Cargo.toml +++ b/datafusion/session/Cargo.toml @@ -30,6 +30,10 @@ version.workspace = true [package.metadata.docs.rs] all-features = true +[features] +object_store = ["datafusion-execution/object_store", "datafusion-physical-plan/object_store"] +default = ["object_store"] + [dependencies] arrow-schema = { workspace = true } async-trait = { workspace = true } diff --git a/datafusion/spark/Cargo.toml b/datafusion/spark/Cargo.toml index f40708863d108..ee68f537b6465 100644 --- a/datafusion/spark/Cargo.toml +++ b/datafusion/spark/Cargo.toml @@ -30,7 +30,14 @@ edition = { workspace = true } all-features = true [features] -default = [] +default = ["object_store"] +object_store = [ + "datafusion?/object_store", + "datafusion-functions/object_store", + "datafusion-functions-aggregate/object_store", + "datafusion-functions-nested/object_store", + "datafusion-session/object_store", +] core = ["datafusion"] # Note: add additional linter rules in lib.rs. diff --git a/datafusion/sql/Cargo.toml b/datafusion/sql/Cargo.toml index 318f8934639aa..8ec26c1d48922 100644 --- a/datafusion/sql/Cargo.toml +++ b/datafusion/sql/Cargo.toml @@ -41,7 +41,8 @@ workspace = true name = "datafusion_sql" [features] -default = ["unicode_expressions", "unparser"] +object_store = ["datafusion-functions-nested/object_store"] +default = ["object_store", "unicode_expressions", "unparser"] unicode_expressions = [] unparser = [] recursive_protection = ["dep:recursive", "dep:stacker"] diff --git a/datafusion/substrait/Cargo.toml b/datafusion/substrait/Cargo.toml index a0f203cec8db6..dfc8c7e6d1d32 100644 --- a/datafusion/substrait/Cargo.toml +++ b/datafusion/substrait/Cargo.toml @@ -40,13 +40,13 @@ chrono = { workspace = true } datafusion = { workspace = true, features = ["sql"] } half = { workspace = true } itertools = { workspace = true } -object_store = { workspace = true } +object_store = { workspace = true, optional = true } # We need to match the version in substrait, so we don't use the workspace version here pbjson-types = { version = "0.8.0" } prost = { workspace = true } substrait = { version = "0.63.0", features = ["serde"] } url = { workspace = true } -tokio = { workspace = true, features = ["fs"] } +tokio = { workspace = true, features = ["fs", "io-util"] } [dev-dependencies] datafusion = { workspace = true, features = ["nested_expressions", "unicode_expressions"] } @@ -57,7 +57,7 @@ insta = { workspace = true } [features] default = ["physical"] -physical = ["datafusion/parquet"] +physical = ["datafusion/parquet", "dep:object_store"] protoc = ["substrait/protoc"] [package.metadata.docs.rs] diff --git a/datafusion/wasmtest/Cargo.toml b/datafusion/wasmtest/Cargo.toml index d6cea2e68d384..c8a6b090780b6 100644 --- a/datafusion/wasmtest/Cargo.toml +++ b/datafusion/wasmtest/Cargo.toml @@ -50,7 +50,7 @@ chrono = { version = "0.4", features = ["wasmbind"] } console_error_panic_hook = { version = "0.1.1", optional = true } datafusion = { workspace = true, features = ["compression", "parquet", "sql"] } datafusion-common = { workspace = true } -datafusion-execution = { workspace = true } +datafusion-execution = { workspace = true, features = ["object_store"] } datafusion-expr = { workspace = true } datafusion-optimizer = { workspace = true, default-features = true } datafusion-physical-plan = { workspace = true } diff --git a/docs/source/user-guide/crate-configuration.md b/docs/source/user-guide/crate-configuration.md index 09c65107e58c8..0df2f9f4bdf8f 100644 --- a/docs/source/user-guide/crate-configuration.md +++ b/docs/source/user-guide/crate-configuration.md @@ -52,6 +52,52 @@ datafusion = { git = "https://github.com/apache/datafusion", branch = "main", de More on [Cargo dependencies](https://doc.rust-lang.org/cargo/reference/specifying-dependencies.html#specifying-dependencies) +## Building without object_store + +The default-enabled `object_store` feature provides built-in file sources and +sinks, listing tables, object store registration, and file caches. To embed +DataFusion with your own `TableProvider` and `ExecutionPlan` implementations +without depending on the `object_store` crate, disable default features and +select the features you need: + +```toml +datafusion = { version = "55.0.0", default-features = false, features = ["sql", "nested_expressions", "string_expressions"] } +``` + +Custom table providers, in-memory tables, SQL planning, relational execution, +memory management, and local disk spilling remain available. Built-in file +read/write APIs and the object store registry are unavailable. File `COPY TO` +returns an error, and file cache settings are unavailable. + +The `parquet`, `avro`, and `parquet_encryption` features enable `object_store` +automatically. File format crates, `datafusion-proto`, and the `physical` feature +of `datafusion-substrait` also require it. Logical-only Substrait conversion can +be used with `datafusion-substrait`'s default features disabled. + +Direct dependencies on child crates, such as `datafusion-execution`, +`datafusion-datasource`, `datafusion-catalog`, `datafusion-physical-plan`, and +`datafusion-pruning`, also enable storage support by default. Disable default +features on those dependencies as well to opt out. Cargo unifies features across +all dependencies: any dependency that enables storage support can bring +`object_store` back into the graph. Check your application's enabled dependencies +with `cargo tree --target all --edges all`; an entry in `Cargo.lock` alone does +not mean an optional dependency is enabled. + +### Migrating existing builds with default features disabled + +Previously, built-in storage support was enabled even with +`default-features = false`. Existing applications that use those APIs must now +explicitly enable `object_store`, unless another selected feature, such as +`parquet`, already enables it: + +```toml +datafusion = { version = "55.0.0", default-features = false, features = ["sql", "object_store"] } +datafusion-execution = { version = "55.0.0", default-features = false, features = ["object_store"] } +``` + +This also applies to direct users of child crates. With storage enabled, existing +storage APIs and their concrete `object_store` types remain unchanged. + ## Optimizing Builds Here are several suggestions to get the Rust compiler to produce faster code when