split datafusion-object-store module

Cargo.toml

@@ -25,6 +25,7 @@ members = [
     "datafusion-cli",
     "datafusion-examples",
     "datafusion-proto",
+    "datafusion-storage",
     "benchmarks",
     "ballista/rust/client",
     "ballista/rust/core",

ballista/rust/core/src/serde/logical_plan/mod.rs

@@ -885,14 +885,17 @@ mod roundtrip_tests {
     use crate::serde::{AsLogicalPlan, BallistaCodec};
     use async_trait::async_trait;
     use core::panic;
-    use datafusion::datasource::listing::ListingTable;
-    use datafusion::datasource::object_store::{
-        FileMetaStream, ListEntryStream, ObjectReader, ObjectStore, SizedFile,
+    use datafusion::datafusion_storage::{
+        object_store::{
+            local::LocalFileSystem, FileMetaStream, ListEntryStream, ObjectReader,
+            ObjectStore,
+        },
+        SizedFile,
     };
+    use datafusion::datasource::listing::ListingTable;
     use datafusion::error::DataFusionError;
     use datafusion::{
         arrow::datatypes::{DataType, Field, Schema},
-        datasource::object_store::local::LocalFileSystem,
         logical_plan::{
             col, CreateExternalTable, Expr, LogicalPlan, LogicalPlanBuilder, Repartition,
             ToDFSchema,

ballista/rust/core/src/serde/mod.rs

@@ -347,7 +347,7 @@ fn str_to_byte(s: &str) -> Result<u8, BallistaError> {
 mod tests {
     use async_trait::async_trait;
     use datafusion::arrow::datatypes::SchemaRef;
-    use datafusion::datasource::object_store::local::LocalFileSystem;
+    use datafusion::datafusion_storage::object_store::local::LocalFileSystem;
     use datafusion::error::DataFusionError;
     use datafusion::execution::context::{QueryPlanner, SessionState, TaskContext};
     use datafusion::execution::runtime_env::{RuntimeConfig, RuntimeEnv};

ballista/rust/core/src/serde/physical_plan/from_proto.rs

@@ -26,9 +26,9 @@ use crate::serde::{from_proto_binary_op, proto_error, protobuf};
 use crate::{convert_box_required, convert_required};
 use chrono::{TimeZone, Utc};
 
-use datafusion::datasource::object_store::local::LocalFileSystem;
-use datafusion::datasource::object_store::{FileMeta, SizedFile};
-use datafusion::datasource::PartitionedFile;
+use datafusion::datafusion_storage::{
+    object_store::local::LocalFileSystem, FileMeta, PartitionedFile, SizedFile,
+};
 use datafusion::execution::context::ExecutionProps;
 
 use datafusion::physical_plan::file_format::FileScanConfig;

ballista/rust/core/src/serde/physical_plan/mod.rs

@@ -33,8 +33,8 @@ use crate::serde::{
 use crate::{convert_box_required, convert_required, into_physical_plan, into_required};
 use datafusion::arrow::compute::SortOptions;
 use datafusion::arrow::datatypes::SchemaRef;
-use datafusion::datasource::object_store::local::LocalFileSystem;
-use datafusion::datasource::PartitionedFile;
+use datafusion::datafusion_storage::object_store::local::LocalFileSystem;
+use datafusion::datafusion_storage::PartitionedFile;
 use datafusion::logical_plan::window_frames::WindowFrame;
 use datafusion::physical_plan::aggregates::create_aggregate_expr;
 use datafusion::physical_plan::coalesce_batches::CoalesceBatchesExec;
@@ -941,8 +941,9 @@ mod roundtrip_tests {
     use std::sync::Arc;
 
     use crate::serde::{AsExecutionPlan, BallistaCodec};
-    use datafusion::datasource::object_store::local::LocalFileSystem;
-    use datafusion::datasource::PartitionedFile;
+    use datafusion::datafusion_storage::{
+        object_store::local::LocalFileSystem, PartitionedFile,
+    };
     use datafusion::physical_plan::sorts::sort::SortExec;
     use datafusion::prelude::SessionContext;
     use datafusion::{

ballista/rust/core/src/serde/physical_plan/to_proto.rs

@@ -35,7 +35,7 @@ use datafusion::physical_plan::{
     Statistics,
 };
 
-use datafusion::datasource::PartitionedFile;
+use datafusion::datafusion_storage::PartitionedFile;
 use datafusion::physical_plan::file_format::FileScanConfig;
 
 use datafusion::physical_plan::expressions::{Count, Literal};

ballista/rust/scheduler/src/scheduler_server/grpc.rs

@@ -33,9 +33,9 @@ use ballista_core::serde::scheduler::{
     ExecutorData, ExecutorDataChange, ExecutorMetadata,
 };
 use ballista_core::serde::{AsExecutionPlan, AsLogicalPlan};
+use datafusion::datafusion_storage::object_store::{local::LocalFileSystem, ObjectStore};
 use datafusion::datasource::file_format::parquet::ParquetFormat;
 use datafusion::datasource::file_format::FileFormat;
-use datafusion::datasource::object_store::{local::LocalFileSystem, ObjectStore};
 use futures::StreamExt;
 use log::{debug, error, info, trace, warn};
 use rand::{distributions::Alphanumeric, thread_rng, Rng};

benchmarks/src/bin/tpch.rs

@@ -50,10 +50,8 @@ use datafusion::{
 };
 use datafusion::{
     arrow::util::pretty,
-    datasource::{
-        listing::{ListingOptions, ListingTable, ListingTableConfig},
-        object_store::local::LocalFileSystem,
-    },
+    datafusion_storage::object_store::local::LocalFileSystem,
+    datasource::listing::{ListingOptions, ListingTable, ListingTableConfig},
 };
 
 use datafusion::datasource::file_format::csv::DEFAULT_CSV_EXTENSION;

datafusion-examples/examples/flight_server.rs

@@ -19,9 +19,9 @@ use std::pin::Pin;
 use std::sync::Arc;
 
 use arrow_flight::SchemaAsIpc;
+use datafusion::datafusion_storage::object_store::local::LocalFileSystem;
 use datafusion::datasource::file_format::parquet::ParquetFormat;
 use datafusion::datasource::listing::ListingOptions;
-use datafusion::datasource::object_store::local::LocalFileSystem;
 use futures::Stream;
 use tonic::transport::Server;
 use tonic::{Request, Response, Status, Streaming};

datafusion-storage/Cargo.toml

@@ -0,0 +1,42 @@
+# 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-storage"
+description = "Storage for DataFusion query engine"
+version = "7.0.0"
+homepage = "https://github.com/apache/arrow-datafusion"
+repository = "https://github.com/apache/arrow-datafusion"
+readme = "README.md"
+authors = ["Apache Arrow <[email protected]>"]
+license = "Apache-2.0"
+keywords = [ "arrow", "query", "sql" ]
+edition = "2021"
+rust-version = "1.59"
+
+[lib]
+name = "datafusion_storage"
+path = "src/lib.rs"
+
+[dependencies]
+async-trait = "0.1.41"
+chrono = { version = "0.4", default-features = false }
+datafusion-common = { path = "../datafusion-common", version = "7.0.0" }
+futures = "0.3"
+parking_lot = "0.12"
+tempfile = "3"
+tokio = { version = "1.0", features = ["macros", "rt", "rt-multi-thread", "sync", "fs", "parking_lot"] }

datafusion-storage/README.md

@@ -0,0 +1,24 @@
+<!---
+  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 Common
+
+This is an internal module for the most fundamental datasource storage of [DataFusion][df].
+
+[df]: https://crates.io/crates/datafusion

datafusion-storage/src/lib.rs

@@ -0,0 +1,104 @@
+// 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.
+
+pub mod object_store;
+
+use chrono::{DateTime, Utc};
+use datafusion_common::ScalarValue;
+
+/// Represents a specific file or a prefix (folder) that may
+/// require further resolution
+#[derive(Debug)]
+pub enum ListEntry {
+    /// Specific file with metadata
+    FileMeta(FileMeta),
+    /// Prefix to be further resolved during partition discovery
+    Prefix(String),
+}
+
+/// The path and size of the file.
+#[derive(Debug, Clone, PartialEq)]
+pub struct SizedFile {
+    /// Path of the file. It is relative to the current object
+    /// store (it does not specify the `xx://` scheme).
+    pub path: String,
+    /// File size in total
+    pub size: u64,
+}
+
+/// Description of a file as returned by the listing command of a
+/// given object store. The resulting path is relative to the
+/// object store that generated it.
+#[derive(Debug, Clone, PartialEq)]
+pub struct FileMeta {
+    /// The path and size of the file.
+    pub sized_file: SizedFile,
+    /// The last modification time of the file according to the
+    /// object store metadata. This information might be used by
+    /// catalog systems like Delta Lake for time travel (see
+    /// <https://github.com/delta-io/delta/issues/192>)
+    pub last_modified: Option<DateTime<Utc>>,
+}
+
+impl FileMeta {
+    /// The path that describes this file. It is relative to the
+    /// associated object store.
+    pub fn path(&self) -> &str {
+        &self.sized_file.path
+    }
+
+    /// The size of the file.
+    pub fn size(&self) -> u64 {
+        self.sized_file.size
+    }
+}
+
+impl std::fmt::Display for FileMeta {
+    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+        write!(f, "{} (size: {})", self.path(), self.size())
+    }
+}
+
+#[derive(Debug, Clone)]
+/// A single file that should be read, along with its schema, statistics
+/// and partition column values that need to be appended to each row.
+pub struct PartitionedFile {
+    /// Path for the file (e.g. URL, filesystem path, etc)
+    pub file_meta: FileMeta,
+    /// Values of partition columns to be appended to each row
+    pub partition_values: Vec<ScalarValue>,
+    // We may include row group range here for a more fine-grained parallel execution
+}
+
+impl PartitionedFile {
+    /// Create a simple file without metadata or partition
+    pub fn new(path: String, size: u64) -> Self {
+        Self {
+            file_meta: FileMeta {
+                sized_file: SizedFile { path, size },
+                last_modified: None,
+            },
+            partition_values: vec![],
+        }
+    }
+}
+
+impl std::fmt::Display for PartitionedFile {
+    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+        write!(f, "{}", self.file_meta)
+    }
+}

datafusion/src/datasource/object_store/local.rs → datafusion-storage/src/object_store/local.rs

@@ -24,13 +24,15 @@ use std::sync::Arc;
 use async_trait::async_trait;
 use futures::{stream, AsyncRead, StreamExt};
 
-use crate::datasource::object_store::{
-    FileMeta, FileMetaStream, ListEntryStream, ObjectReader, ObjectStore,
+use datafusion_common::{DataFusionError, Result};
+
+use crate::{FileMeta, PartitionedFile, SizedFile};
+
+use super::{
+    FileMetaStream, ListEntryStream, ObjectReader, ObjectReaderStream, ObjectStore,
 };
-use crate::datasource::PartitionedFile;
-use crate::error::{DataFusionError, Result};
 
-use super::{ObjectReaderStream, SizedFile};
+pub static LOCAL_SCHEME: &str = "file";
 
 #[derive(Debug)]
 /// Local File System as Object Store.