Break datafusion-catalog code into its own crate

@goldmedal

Is your feature request related to a problem or challenge?

As we start thinking of adding new catalog support in DataFusion for example, @goldmedal tried to do this for DynamicFileCatalog in #11035 , the current code structure requires that any such implementation be in datafusion/core which is already quite large and monolithic

Also, I think long term it would be a more sustainable structure if we can move ListingTable out of the core as well

However, this move is not likely possible until we move table provider and related catalog traits out of the core.

Also, if we could split up datafusion core more, it would be easier for people to use DataFusion as a smaller embedded library (aka not bring in file format support if they didn't need it)

Describe the solution you'd like

Ideally I would like the following traits / code moved into a datafusion-catalog crate:

TableProvider
CatalogProvider
SchemaProvider
MemoryCatalogProvider
InformationSchemaProvider

Describe alternatives you've considered

No response

Additional context

No response

Part of #10782

I looked into this --

One of the major challenges is that SessionState's constructor basically installs the "pre-provided" functionality (like data sources, etc

datafusion/datafusion/core/src/execution/session_state.rs

Lines 190 to 295 in 78055fe

    
               pub fn new_with_config_rt_and_catalog_list( 
        
                   config: SessionConfig, 
        
                   runtime: Arc<RuntimeEnv>, 
        
                   catalog_list: Arc<dyn CatalogProviderList>, 
        
               ) -> Self { 
        
                   let session_id = Uuid::new_v4().to_string(); 
        
                   // Create table_factories for all default formats 
        
                   let mut table_factories: HashMap<String, Arc<dyn TableProviderFactory>> = 
        
                       HashMap::new(); 
        
                   #[cfg(feature = "parquet")] 
        
                   table_factories.insert("PARQUET".into(), Arc::new(DefaultTableFactory::new())); 
        
                   table_factories.insert("CSV".into(), Arc::new(DefaultTableFactory::new())); 
        
                   table_factories.insert("JSON".into(), Arc::new(DefaultTableFactory::new())); 
        
                   table_factories.insert("NDJSON".into(), Arc::new(DefaultTableFactory::new())); 
        
                   table_factories.insert("AVRO".into(), Arc::new(DefaultTableFactory::new())); 
        
                   table_factories.insert("ARROW".into(), Arc::new(DefaultTableFactory::new())); 
        
                   if config.create_default_catalog_and_schema() { 
        
                       let default_catalog = MemoryCatalogProvider::new(); 
        
                       default_catalog 
        
                           .register_schema( 
        
                               &config.options().catalog.default_schema, 
        
                               Arc::new(MemorySchemaProvider::new()), 
        
                           ) 
        
                           .expect("memory catalog provider can register schema"); 
        
                       Self::register_default_schema( 
        
                           &config, 
        
                           &table_factories, 
        
                           &runtime, 
        
                           &default_catalog, 
        
                       ); 
        
                       catalog_list.register_catalog( 
        
                           config.options().catalog.default_catalog.clone(), 
        
                           Arc::new(default_catalog), 
        
                       ); 
        
                   } 
        
                   let mut new_self = SessionState { 
        
                       session_id, 
        
                       analyzer: Analyzer::new(), 
        
                       optimizer: Optimizer::new(), 
        
                       physical_optimizers: PhysicalOptimizer::new(), 
        
                       query_planner: Arc::new(DefaultQueryPlanner {}), 
        
                       catalog_list, 
        
                       table_functions: HashMap::new(), 
        
                       scalar_functions: HashMap::new(), 
        
                       aggregate_functions: HashMap::new(), 
        
                       window_functions: HashMap::new(), 
        
                       serializer_registry: Arc::new(EmptySerializerRegistry), 
        
                       file_formats: HashMap::new(), 
        
                       table_options: TableOptions::default_from_session_config(config.options()), 
        
                       config, 
        
                       execution_props: ExecutionProps::new(), 
        
                       runtime_env: runtime, 
        
                       table_factories, 
        
                       function_factory: None, 
        
                   }; 
        
                   #[cfg(feature = "parquet")] 
        
                   if let Err(e) = 
        
                       new_self.register_file_format(Arc::new(ParquetFormatFactory::new()), false) 
        
                   { 
        
                       log::info!("Unable to register default ParquetFormat: {e}") 
        
                   }; 
        
                   if let Err(e) = 
        
                       new_self.register_file_format(Arc::new(JsonFormatFactory::new()), false) 
        
                   { 
        
                       log::info!("Unable to register default JsonFormat: {e}") 
        
                   }; 
        
                   if let Err(e) = 
        
                       new_self.register_file_format(Arc::new(CsvFormatFactory::new()), false) 
        
                   { 
        
                       log::info!("Unable to register default CsvFormat: {e}") 
        
                   }; 
        
                   if let Err(e) = 
        
                       new_self.register_file_format(Arc::new(ArrowFormatFactory::new()), false) 
        
                   { 
        
                       log::info!("Unable to register default ArrowFormat: {e}") 
        
                   }; 
        
                   if let Err(e) = 
        
                       new_self.register_file_format(Arc::new(AvroFormatFactory::new()), false) 
        
                   { 
        
                       log::info!("Unable to register default AvroFormat: {e}") 
        
                   }; 
        
                   // register built in functions 
        
                   functions::register_all(&mut new_self) 
        
                       .expect("can not register built in functions"); 
        
                   // register crate of array expressions (if enabled) 
        
                   #[cfg(feature = "array_expressions")] 
        
                   functions_array::register_all(&mut new_self) 
        
                       .expect("can not register array expressions"); 
        
                   functions_aggregate::register_all(&mut new_self) 
        
                       .expect("can not register aggregate functions"); 
        
                   new_self

One way to handle this would be to allow constructing SessionState with the minimal built in features, and have a function in SessionContext like SessionContext::register_built_ins that would register things like listing table, information schema, etc.

That way it would still be easy to use DataFusion with a minimal SessionState but also easily register all the built in extensions 🤔

I made #11183 to start breaking apart the API and implementation -- there is still a ways to go

@alamb

@alamb I took a stab at moving parquet functionality into a datafusion-parquet crate (#11188) , and I ran into similar challenges you highlight here. I think to accomplish these goals core will need to be refactored into a number of different crates to avoid circular dependencies and allow core to still offer a batteries included experience.

I think to accomplish these goals core will need to be refactored into a number of different crates to avoid circular dependencies and allow core to still offer a batteries included experience.

I agree with this entirely

The center of the knot is SessionState I think -- figuring out how to get that out of the core is likely key to breaking things up reasonably

The more I think about this the more I think trying to make SessionState a container that doesn't have all the optional features (like parquet support) by default makes sense

So like

let session_state = SessionState::new();
// no table providers, etc
// install standard built in table providers
SessionContex::install_built_in(&mut session_state);
// now session_state has them here

The more I think about this the more I think trying to make SessionState a container that doesn't have all the optional features (like parquet support) by default makes sense

So like
let session_state = SessionState::new();
// no table providers, etc
// install standard built in table providers
SessionContex::install_built_in(&mut session_state);
// now session_state has them here

I filed #11320 to track this idea

I think this is basically complete, so closing it down

Navigation Menu

Repository navigation

Break datafusion-catalog code into its own crate #11182

Is your feature request related to a problem or challenge?

Describe the solution you'd like

Describe alternatives you've considered

Additional context

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

	pub fn new_with_config_rt_and_catalog_list(
	config: SessionConfig,
	runtime: Arc<RuntimeEnv>,
	catalog_list: Arc<dyn CatalogProviderList>,
	) -> Self {
	let session_id = Uuid::new_v4().to_string();

	// Create table_factories for all default formats
	let mut table_factories: HashMap<String, Arc<dyn TableProviderFactory>> =
	HashMap::new();
	#[cfg(feature = "parquet")]
	table_factories.insert("PARQUET".into(), Arc::new(DefaultTableFactory::new()));
	table_factories.insert("CSV".into(), Arc::new(DefaultTableFactory::new()));
	table_factories.insert("JSON".into(), Arc::new(DefaultTableFactory::new()));
	table_factories.insert("NDJSON".into(), Arc::new(DefaultTableFactory::new()));
	table_factories.insert("AVRO".into(), Arc::new(DefaultTableFactory::new()));
	table_factories.insert("ARROW".into(), Arc::new(DefaultTableFactory::new()));

	if config.create_default_catalog_and_schema() {
	let default_catalog = MemoryCatalogProvider::new();

	default_catalog
	.register_schema(
	&config.options().catalog.default_schema,
	Arc::new(MemorySchemaProvider::new()),
	)
	.expect("memory catalog provider can register schema");

	Self::register_default_schema(
	&config,
	&table_factories,
	&runtime,
	&default_catalog,
	);

	catalog_list.register_catalog(
	config.options().catalog.default_catalog.clone(),
	Arc::new(default_catalog),
	);
	}

	let mut new_self = SessionState {
	session_id,
	analyzer: Analyzer::new(),
	optimizer: Optimizer::new(),
	physical_optimizers: PhysicalOptimizer::new(),
	query_planner: Arc::new(DefaultQueryPlanner {}),
	catalog_list,
	table_functions: HashMap::new(),
	scalar_functions: HashMap::new(),
	aggregate_functions: HashMap::new(),
	window_functions: HashMap::new(),
	serializer_registry: Arc::new(EmptySerializerRegistry),
	file_formats: HashMap::new(),
	table_options: TableOptions::default_from_session_config(config.options()),
	config,
	execution_props: ExecutionProps::new(),
	runtime_env: runtime,
	table_factories,
	function_factory: None,
	};

	#[cfg(feature = "parquet")]
	if let Err(e) =
	new_self.register_file_format(Arc::new(ParquetFormatFactory::new()), false)
	{
	log::info!("Unable to register default ParquetFormat: {e}")
	};

	if let Err(e) =
	new_self.register_file_format(Arc::new(JsonFormatFactory::new()), false)
	{
	log::info!("Unable to register default JsonFormat: {e}")
	};

	if let Err(e) =
	new_self.register_file_format(Arc::new(CsvFormatFactory::new()), false)
	{
	log::info!("Unable to register default CsvFormat: {e}")
	};

	if let Err(e) =
	new_self.register_file_format(Arc::new(ArrowFormatFactory::new()), false)
	{
	log::info!("Unable to register default ArrowFormat: {e}")
	};

	if let Err(e) =
	new_self.register_file_format(Arc::new(AvroFormatFactory::new()), false)
	{
	log::info!("Unable to register default AvroFormat: {e}")
	};

	// register built in functions
	functions::register_all(&mut new_self)
	.expect("can not register built in functions");

	// register crate of array expressions (if enabled)
	#[cfg(feature = "array_expressions")]
	functions_array::register_all(&mut new_self)
	.expect("can not register array expressions");

	functions_aggregate::register_all(&mut new_self)
	.expect("can not register aggregate functions");

	new_self

Break datafusion-catalog code into its own crate #11182

Description

Is your feature request related to a problem or challenge?

Describe the solution you'd like

Describe alternatives you've considered

Additional context

Activity

alamb commented on Jun 30, 2024

alamb commented on Jun 30, 2024

devinjdangelo commented on Jul 1, 2024

alamb commented on Jul 1, 2024

alamb commented on Jul 1, 2024

alamb commented on Jul 7, 2024

alamb commented on Oct 15, 2025

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions