Allow listing tables to be created via TableFactories #4112

avantgardnerio · 2022-11-04T20:39:02Z

Which issue does this PR close?

Closes #4111.

Rationale for this change

Described in issue.

What changes are included in this PR?

Described in issue.

Are there any user-facing changes?

They can register CSVs, parquets, jsons and avros from the datafusion-cli.

avantgardnerio · 2022-11-04T20:40:12Z

@andygrove and @alamb sorry for the immediate follow-on PR. I did not realize we could merge Ballista PRs pointed at git refs of Datafusion or I would have done these two as a single PR.

avantgardnerio · 2022-11-04T20:40:39Z

datafusion-cli/src/main.rs

+                .ok_or(DataFusionError::Internal("Schema not found!".to_string()))?;
+            let lister = schema.as_any().downcast_ref::<ListingSchemaProvider>();
+            if let Some(lister) = lister {
+                lister.refresh(&ctx.state()).await?;


Register tables on startup.

avantgardnerio · 2022-11-04T20:41:23Z

datafusion-cli/src/main.rs

+    table_factories.insert("csv".to_string(), Arc::new(ListingTableFactory::new(FileType::CSV)));
+    table_factories.insert("parquet".to_string(), Arc::new(ListingTableFactory::new(FileType::PARQUET)));
+    table_factories.insert("avro".to_string(), Arc::new(ListingTableFactory::new(FileType::AVRO)));
+    table_factories.insert("json".to_string(), Arc::new(ListingTableFactory::new(FileType::JSON)));


@alamb this is getting closer to something I think you asked for earlier: treating ListingTables and custom tables the same way and just registering factories for table types.

avantgardnerio · 2022-11-04T20:41:54Z

datafusion/core/src/catalog/listing_schema.rs

@@ -72,7 +74,7 @@ impl ListingSchemaProvider {
    }

    /// Reload table information from ObjectStore
-    pub async fn refresh(&self) -> datafusion_common::Result<()> {
+    pub async fn refresh(&self, state: &SessionState) -> datafusion_common::Result<()> {


Needed to add SessionState to allow ListingTables to load their schema.

avantgardnerio · 2022-11-04T20:42:45Z

datafusion/core/src/catalog/listing_schema.rs

@@ -100,13 +102,20 @@ impl ListingSchemaProvider {
                .ok_or_else(|| {
                    DataFusionError::Internal("Cannot parse file name!".to_string())
                })?;
-            let table_name = table.to_str().ok_or_else(|| {
+            let table_name = file_name.split('.').collect_vec()[0];


Deltatables are always folders, but .csvs should have their extension removed from their table name (and unfortunately the Path method to do this is marked unstable).

Which method? Maybe we can contribute something back upstream to object_store 🤔

https://doc.rust-lang.org/std/path/struct.PathBuf.html#method.file_prefix

For some reason the github UI didn't let me respond in thread until now :/

avantgardnerio · 2022-11-04T20:43:37Z

avantgardnerio · 2022-11-04T20:43:51Z

@andygrove I think you were requesting this feature ☝️

alamb

The code looks good to me -- is there any way we can write a test for this functionality?

alamb · 2022-11-05T11:22:05Z

datafusion/core/src/catalog/listing_schema.rs

@@ -100,13 +102,20 @@ impl ListingSchemaProvider {
                .ok_or_else(|| {
                    DataFusionError::Internal("Cannot parse file name!".to_string())
                })?;
-            let table_name = table.to_str().ok_or_else(|| {
+            let table_name = file_name.split('.').collect_vec()[0];


Which method? Maybe we can contribute something back upstream to object_store 🤔

alamb · 2022-11-05T11:23:34Z

datafusion/core/src/datasource/listing_table_factory.rs

+    ) -> datafusion_common::Result<Arc<dyn TableProvider>> {
+        let file_extension = self.file_type.get_ext();
+
+        let file_format: Arc<dyn FileFormat> = match self.file_type {


🤔 it would be really neat to somehow combine the logic in ListingTable and ListingTableFactory (or maybe datafusion-cli could just use the factory -- not sure)

combine the logic

I think this is the duplicated fragment. I would love to combine those two and only have register_table instead of create_listing_table() and create_custom_table().

alamb · 2022-11-05T11:25:46Z

I know tests are somewhat painful to write, but if we don't have them I worry about breaking the functionality at some future time

Maybe making a test as an example in datafusion-examples https://github.com/apache/arrow-datafusion/tree/master/datafusion-examples/examples would be a good way to

document a bit more how to use these features,
Ensure whatever APIs you need are properly exposed in the API
Ensure we don't break the feature when we reorganize things in the future

avantgardnerio · 2022-11-05T15:53:10Z

write a test

Done. I also filed this: #4114 . It's my own personal fault, so I understand if we don't want to inflict that on everyone.

avantgardnerio · 2022-11-05T15:56:08Z

Which method?

https://doc.rust-lang.org/std/path/struct.PathBuf.html#method.file_prefix

avantgardnerio · 2022-11-05T16:50:25Z

datafusion-cli/src/main.rs

@@ -142,11 +147,31 @@ pub async fn main() -> Result<()> {
 }

 fn create_runtime_env() -> Result<RuntimeEnv> {
+    let mut table_factories: HashMap<String, Arc<dyn TableProviderFactory>> =


Add TableFactorys for default formats.

avantgardnerio · 2022-11-05T16:51:24Z

datafusion-cli/src/object_storage.rs

+            Err(e) => format!("{}", e),
+            Ok(_) => "".to_string()
+        };
+        assert_eq!("".to_string(), msg); // Fail with error message


I refactored this test to have a better failure message because it was failing for me locally. The fact that CI passed makes me think it's not running there because datafusion-cli is excluded from the workspace - I don't know why this is, but would propose we include it.

Could you please file a ticket to do so? Thank you!

avantgardnerio · 2022-11-05T16:52:30Z

datafusion/core/src/execution/context.rs

@@ -175,6 +175,26 @@ impl SessionContext {
        Self::with_config(SessionConfig::new())
    }

+    /// Finds any ListSchemaProviders and instructs them to reload tables from "disk"
+    pub async fn refresh_catalogs(&self) -> Result<()> {


Moving this code to a common function on the context which we can use from datafusion-cli, tests, Ballista, etc.

makes sense to me

avantgardnerio · 2022-11-05T16:53:00Z

datafusion/core/src/execution/context.rs

@@ -1764,7 +1784,7 @@ impl ContextProvider for SessionState {
            Ok(schema) => {
                let provider = schema.table(resolved_ref.table).ok_or_else(|| {
                    DataFusionError::Plan(format!(
-                        "'{}.{}.{}' not found",
+                        "table '{}.{}.{}' not found",


not found is a hard error message to ctrl-f for. Adding the word table will hopefully make this statistically more likely to be found.

avantgardnerio · 2022-11-05T16:53:43Z

datafusion/core/src/execution/context.rs

+| Customer#000000002 |
+| Customer#000000003 |
+| Customer#000000004 |
+--------------------+"#;


The proof is in the pudding. Can't select from a table without registering it first, so this must be auto-registered.

alamb

Looks great to me -- thank you @avantgardnerio

I also merged this PR to master locally and all the tests still pass for me

I left a few comments, but I don't think any of them are required to merge this PR

alamb · 2022-11-07T20:07:28Z

datafusion-cli/src/object_storage.rs

+            Err(e) => format!("{}", e),
+            Ok(_) => "".to_string()
+        };
+        assert_eq!("".to_string(), msg); // Fail with error message


Could you please file a ticket to do so? Thank you!

alamb · 2022-11-07T20:10:40Z

datafusion/core/src/execution/context.rs

@@ -175,6 +175,26 @@ impl SessionContext {
        Self::with_config(SessionConfig::new())
    }

+    /// Finds any ListSchemaProviders and instructs them to reload tables from "disk"


Suggested change

/// Finds any ListSchemaProviders and instructs them to reload tables from "disk"

/// Invokes `ListingSchemaProvider::reload()` for all registered providers

alamb · 2022-11-07T20:10:47Z

datafusion/core/src/execution/context.rs

@@ -175,6 +175,26 @@ impl SessionContext {
        Self::with_config(SessionConfig::new())
    }

+    /// Finds any ListSchemaProviders and instructs them to reload tables from "disk"
+    pub async fn refresh_catalogs(&self) -> Result<()> {


makes sense to me

alamb · 2022-11-07T20:13:29Z

datafusion/core/src/execution/context.rs

+            plan_and_collect(&ctx, "select c_name from default.customer limit 3;")
+                .await?;
+
+        let actual = arrow::util::pretty::pretty_format_batches(&result)


You might also consider using assert_batches_eq here in this test

avantgardnerio · 2022-11-07T20:45:03Z

I don't think any of them are required to merge this PR

Then if it's alright with you, I'd love to get it merged rather than wait for another round of CI checks. Thanks for thoroughly reviewing it!

ursabot · 2022-11-07T21:02:01Z

Benchmark runs are scheduled for baseline = c1fc732 and contender = 7b5842b. 7b5842b is a master commit associated with this PR. Results will be available as each benchmark for each run completes.
Conbench compare runs links:
[Skipped ⚠️ Benchmarking of arrow-datafusion-commits is not supported on ec2-t3-xlarge-us-east-2] ec2-t3-xlarge-us-east-2
[Skipped ⚠️ Benchmarking of arrow-datafusion-commits is not supported on test-mac-arm] test-mac-arm
[Skipped ⚠️ Benchmarking of arrow-datafusion-commits is not supported on ursa-i9-9960x] ursa-i9-9960x
[Skipped ⚠️ Benchmarking of arrow-datafusion-commits is not supported on ursa-thinkcentre-m75q] ursa-thinkcentre-m75q
Buildkite builds:
Supported benchmarks:
ec2-t3-xlarge-us-east-2: Supported benchmark langs: Python, R. Runs only benchmarks with cloud = True
test-mac-arm: Supported benchmark langs: C++, Python, R
ursa-i9-9960x: Supported benchmark langs: Python, R, JavaScript
ursa-thinkcentre-m75q: Supported benchmark langs: C++, Java

Allow listing tables to be created via TableFactories

423d80b

github-actions bot added the core Core DataFusion crate label Nov 4, 2022

avantgardnerio commented Nov 4, 2022

View reviewed changes

alamb reviewed Nov 5, 2022

View reviewed changes

Add a test

c2ab352

avantgardnerio commented Nov 5, 2022

View reviewed changes

avantgardnerio requested a review from alamb November 5, 2022 16:54

alamb approved these changes Nov 7, 2022

View reviewed changes

alamb merged commit 7b5842b into apache:master Nov 7, 2022

avantgardnerio deleted the bg_listing_factory branch November 7, 2022 21:47

avantgardnerio mentioned this pull request Nov 8, 2022

Use ListingSchema in Ballista apache/datafusion-ballista#500

Open

alamb mentioned this pull request Nov 10, 2022

An asynchronous version of CatalogList/CatalogProvider/SchemaProvider #3777

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Allow listing tables to be created via TableFactories #4112

Allow listing tables to be created via TableFactories #4112

avantgardnerio commented Nov 4, 2022

avantgardnerio commented Nov 4, 2022

avantgardnerio Nov 4, 2022

avantgardnerio Nov 4, 2022

avantgardnerio Nov 4, 2022

avantgardnerio Nov 4, 2022

alamb Nov 5, 2022

avantgardnerio Nov 5, 2022

avantgardnerio commented Nov 4, 2022

avantgardnerio commented Nov 4, 2022 •

edited

Loading

alamb left a comment

alamb Nov 5, 2022

alamb Nov 5, 2022

avantgardnerio Nov 5, 2022

alamb commented Nov 5, 2022 •

edited

Loading

avantgardnerio commented Nov 5, 2022

avantgardnerio commented Nov 5, 2022

avantgardnerio Nov 5, 2022

avantgardnerio Nov 5, 2022

alamb Nov 7, 2022

avantgardnerio Nov 5, 2022

alamb Nov 7, 2022

avantgardnerio Nov 5, 2022

avantgardnerio Nov 5, 2022

alamb left a comment

alamb Nov 7, 2022

alamb Nov 7, 2022

alamb Nov 7, 2022

alamb Nov 7, 2022

avantgardnerio commented Nov 7, 2022

ursabot commented Nov 7, 2022

	/// Finds any ListSchemaProviders and instructs them to reload tables from "disk"
	/// Invokes `ListingSchemaProvider::reload()` for all registered providers

Allow listing tables to be created via TableFactories #4112

Allow listing tables to be created via TableFactories #4112

Conversation

avantgardnerio commented Nov 4, 2022

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Are there any user-facing changes?

avantgardnerio commented Nov 4, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

avantgardnerio commented Nov 4, 2022

avantgardnerio commented Nov 4, 2022 • edited Loading

alamb left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

alamb commented Nov 5, 2022 • edited Loading

avantgardnerio commented Nov 5, 2022

avantgardnerio commented Nov 5, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

alamb left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

avantgardnerio commented Nov 7, 2022

ursabot commented Nov 7, 2022

avantgardnerio commented Nov 4, 2022 •

edited

Loading

alamb commented Nov 5, 2022 •

edited

Loading