make datafusion-catalog-listing and move some implementation of listing out of datafusion/core/datasource/listing (#14464)

* make datafusion_catalog_listing

* fix: this is a bit hacky

* fixes: prettier, taplo etc

* fixes: clippy

* minor: permalink commit hash -> main

* Tweak README

* fix:prettier + wasm

* prettier

* Put unit tests with code

---------

Co-authored-by: Andrew Lamb <[email protected]>

Author: logan-keede

Cargo.toml

@@ -22,6 +22,7 @@ members = [
     "datafusion/common",
     "datafusion/common-runtime",
     "datafusion/catalog",
+    "datafusion/catalog-listing",
     "datafusion/core",
     "datafusion/expr",
     "datafusion/expr-common",
@@ -100,6 +101,7 @@ ctor = "0.2.9"
 dashmap = "6.0.1"
 datafusion = { path = "datafusion/core", version = "45.0.0", default-features = false }
 datafusion-catalog = { path = "datafusion/catalog", version = "45.0.0" }
+datafusion-catalog-listing = { path = "datafusion/catalog-listing", version = "45.0.0" }
 datafusion-common = { path = "datafusion/common", version = "45.0.0", default-features = false }
 datafusion-common-runtime = { path = "datafusion/common-runtime", version = "45.0.0" }
 datafusion-doc = { path = "datafusion/doc", version = "45.0.0" }

datafusion-cli/Cargo.lock

@@ -0,0 +1,65 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+[package]
+name = "datafusion-catalog-listing"
+description = "datafusion-catalog-listing"
+authors.workspace = true
+edition.workspace = true
+homepage.workspace = true
+license.workspace = true
+readme.workspace = true
+repository.workspace = true
+rust-version.workspace = true
+version.workspace = true
+
+[dependencies]
+arrow = { workspace = true }
+arrow-schema = { workspace = true }
+async-compression = { version = "0.4.0", features = [
+    "bzip2",
+    "gzip",
+    "xz",
+    "zstd",
+    "tokio",
+], optional = true }
+chrono = { workspace = true }
+datafusion-catalog = { workspace = true }
+datafusion-common = { workspace = true, features = ["object_store"] }
+datafusion-execution = { workspace = true }
+datafusion-expr = { workspace = true }
+datafusion-physical-expr = { workspace = true }
+datafusion-physical-expr-common = { workspace = true }
+datafusion-physical-plan = { workspace = true }
+futures = { workspace = true }
+glob = "0.3.0"
+itertools = { workspace = true }
+log = { workspace = true }
+object_store = { workspace = true }
+url = { workspace = true }
+
+[dev-dependencies]
+async-trait = { workspace = true }
+tempfile = { workspace = true }
+tokio = { workspace = true }
+
+[lints]
+workspace = true
+
+[lib]
+name = "datafusion_catalog_listing"
+path = "src/mod.rs"

datafusion/catalog-listing/Cargo.toml

@@ -0,0 +1 @@
+../../LICENSE.txt

datafusion/catalog-listing/LICENSE.txt

@@ -0,0 +1 @@
+../../NOTICE.txt

datafusion/catalog-listing/NOTICE.txt

@@ -0,0 +1,30 @@
+<!---
+  Licensed to the Apache Software Foundation (ASF) under one
+  or more contributor license agreements.  See the NOTICE file
+  distributed with this work for additional information
+  regarding copyright ownership.  The ASF licenses this file
+  to you under the Apache License, Version 2.0 (the
+  "License"); you may not use this file except in compliance
+  with the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing,
+  software distributed under the License is distributed on an
+  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+  KIND, either express or implied.  See the License for the
+  specific language governing permissions and limitations
+  under the License.
+-->
+
+# DataFusion catalog-listing
+
+[DataFusion][df] is an extensible query execution framework, written in Rust, that uses Apache Arrow as its in-memory format.
+
+This crate is a submodule of DataFusion with [ListingTable], an implementation
+of [TableProvider] based on files in a directory (either locally or on remote
+object storage such as S3).
+
+[df]: https://crates.io/crates/datafusion
+[listingtable]: https://docs.rs/datafusion/latest/datafusion/datasource/listing/struct.ListingTable.html
+[tableprovider]: https://docs.rs/datafusion/latest/datafusion/datasource/trait.TableProvider.html

datafusion/catalog-listing/README.md

@@ -22,7 +22,7 @@ use std::sync::Arc;
 
 use super::ListingTableUrl;
 use super::PartitionedFile;
-use crate::execution::context::SessionState;
+use datafusion_catalog::Session;
 use datafusion_common::internal_err;
 use datafusion_common::{HashMap, Result, ScalarValue};
 use datafusion_expr::{BinaryExpr, Operator};
@@ -154,7 +154,7 @@ pub fn split_files(
     chunks
 }
 
-struct Partition {
+pub struct Partition {
     /// The path to the partition, including the table prefix
     path: Path,
     /// How many path segments below the table prefix `path` contains
@@ -183,7 +183,7 @@ impl Partition {
 }
 
 /// Returns a recursive list of the partitions in `table_path` up to `max_depth`
-async fn list_partitions(
+pub async fn list_partitions(
     store: &dyn ObjectStore,
     table_path: &ListingTableUrl,
     max_depth: usize,
@@ -364,7 +364,7 @@ fn populate_partition_values<'a>(
     }
 }
 
-fn evaluate_partition_prefix<'a>(
+pub fn evaluate_partition_prefix<'a>(
     partition_cols: &'a [(String, DataType)],
     filters: &'a [Expr],
 ) -> Option<Path> {
@@ -405,7 +405,7 @@ fn evaluate_partition_prefix<'a>(
 /// `filters` should only contain expressions that can be evaluated
 /// using only the partition columns.
 pub async fn pruned_partition_list<'a>(
-    ctx: &'a SessionState,
+    ctx: &'a dyn Session,
     store: &'a dyn ObjectStore,
     table_path: &'a ListingTableUrl,
     filters: &'a [Expr],
@@ -489,7 +489,7 @@ pub async fn pruned_partition_list<'a>(
 
 /// Extract the partition values for the given `file_path` (in the given `table_path`)
 /// associated to the partitions defined by `table_partition_cols`
-fn parse_partitions_for_path<'a, I>(
+pub fn parse_partitions_for_path<'a, I>(
     table_path: &ListingTableUrl,
     file_path: &'a Path,
     table_partition_cols: I,
@@ -517,17 +517,36 @@ where
     }
     Some(part_values)
 }
+/// Describe a partition as a (path, depth, files) tuple for easier assertions
+pub fn describe_partition(partition: &Partition) -> (&str, usize, Vec<&str>) {
+    (
+        partition.path.as_ref(),
+        partition.depth,
+        partition
+            .files
+            .as_ref()
+            .map(|f| f.iter().map(|f| f.location.filename().unwrap()).collect())
+            .unwrap_or_default(),
+    )
+}
 
 #[cfg(test)]
 mod tests {
+    use async_trait::async_trait;
+    use datafusion_execution::config::SessionConfig;
+    use datafusion_execution::runtime_env::RuntimeEnv;
+    use futures::FutureExt;
+    use object_store::memory::InMemory;
+    use std::any::Any;
     use std::ops::Not;
-
-    use futures::StreamExt;
-
-    use crate::test::object_store::make_test_store_and_state;
-    use datafusion_expr::{case, col, lit, Expr};
+    // use futures::StreamExt;
 
     use super::*;
+    use datafusion_expr::{
+        case, col, lit, AggregateUDF, Expr, LogicalPlan, ScalarUDF, WindowUDF,
+    };
+    use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
+    use datafusion_physical_plan::ExecutionPlan;
 
     #[test]
     fn test_split_files() {
@@ -578,7 +597,7 @@ mod tests {
         ]);
         let filter = Expr::eq(col("mypartition"), lit("val1"));
         let pruned = pruned_partition_list(
-            &state,
+            state.as_ref(),
             store.as_ref(),
             &ListingTableUrl::parse("file:///tablepath/").unwrap(),
             &[filter],
@@ -603,7 +622,7 @@ mod tests {
         ]);
         let filter = Expr::eq(col("mypartition"), lit("val1"));
         let pruned = pruned_partition_list(
-            &state,
+            state.as_ref(),
             store.as_ref(),
             &ListingTableUrl::parse("file:///tablepath/").unwrap(),
             &[filter],
@@ -643,7 +662,7 @@ mod tests {
         let filter1 = Expr::eq(col("part1"), lit("p1v2"));
         let filter2 = Expr::eq(col("part2"), lit("p2v1"));
         let pruned = pruned_partition_list(
-            &state,
+            state.as_ref(),
             store.as_ref(),
             &ListingTableUrl::parse("file:///tablepath/").unwrap(),
             &[filter1, filter2],
@@ -680,19 +699,6 @@ mod tests {
         );
     }
 
-    /// Describe a partition as a (path, depth, files) tuple for easier assertions
-    fn describe_partition(partition: &Partition) -> (&str, usize, Vec<&str>) {
-        (
-            partition.path.as_ref(),
-            partition.depth,
-            partition
-                .files
-                .as_ref()
-                .map(|f| f.iter().map(|f| f.location.filename().unwrap()).collect())
-                .unwrap_or_default(),
-        )
-    }
-
     #[tokio::test]
     async fn test_list_partition() {
         let (store, _) = make_test_store_and_state(&[
@@ -994,4 +1000,74 @@ mod tests {
             Some(Path::from("a=1970-01-05")),
         );
     }
+
+    pub fn make_test_store_and_state(
+        files: &[(&str, u64)],
+    ) -> (Arc<InMemory>, Arc<dyn Session>) {
+        let memory = InMemory::new();
+
+        for (name, size) in files {
+            memory
+                .put(&Path::from(*name), vec![0; *size as usize].into())
+                .now_or_never()
+                .unwrap()
+                .unwrap();
+        }
+
+        (Arc::new(memory), Arc::new(MockSession {}))
+    }
+
+    struct MockSession {}
+
+    #[async_trait]
+    impl Session for MockSession {
+        fn session_id(&self) -> &str {
+            unimplemented!()
+        }
+
+        fn config(&self) -> &SessionConfig {
+            unimplemented!()
+        }
+
+        async fn create_physical_plan(
+            &self,
+            _logical_plan: &LogicalPlan,
+        ) -> Result<Arc<dyn ExecutionPlan>> {
+            unimplemented!()
+        }
+
+        fn create_physical_expr(
+            &self,
+            _expr: Expr,
+            _df_schema: &DFSchema,
+        ) -> Result<Arc<dyn PhysicalExpr>> {
+            unimplemented!()
+        }
+
+        fn scalar_functions(&self) -> &std::collections::HashMap<String, Arc<ScalarUDF>> {
+            unimplemented!()
+        }
+
+        fn aggregate_functions(
+            &self,
+        ) -> &std::collections::HashMap<String, Arc<AggregateUDF>> {
+            unimplemented!()
+        }
+
+        fn window_functions(&self) -> &std::collections::HashMap<String, Arc<WindowUDF>> {
+            unimplemented!()
+        }
+
+        fn runtime_env(&self) -> &Arc<RuntimeEnv> {
+            unimplemented!()
+        }
+
+        fn execution_props(&self) -> &ExecutionProps {
+            unimplemented!()
+        }
+
+        fn as_any(&self) -> &dyn Any {
+            unimplemented!()
+        }
+    }
 }