Move SQL Query Planner into new datafusion-sql crate

Cargo.toml

@@ -25,6 +25,7 @@ members = [
     "datafusion/physical-expr",
     "datafusion/proto",
     "datafusion/row",
+    "datafusion/sql",
     "datafusion-examples",
     "benchmarks",
 ]

datafusion/core/Cargo.toml

@@ -51,7 +51,7 @@ regex_expressions = ["datafusion-physical-expr/regex_expressions"]
 # Used to enable scheduler
 scheduler = ["rayon"]
 simd = ["arrow/simd"]
-unicode_expressions = ["datafusion-physical-expr/regex_expressions"]
+unicode_expressions = ["datafusion-physical-expr/regex_expressions", "datafusion-sql/unicode_expressions"]
 
 [dependencies]
 ahash = { version = "0.7", default-features = false }
@@ -65,6 +65,7 @@ datafusion-expr = { path = "../expr", version = "8.0.0" }
 datafusion-jit = { path = "../jit", version = "8.0.0", optional = true }
 datafusion-physical-expr = { path = "../physical-expr", version = "8.0.0" }
 datafusion-row = { path = "../row", version = "8.0.0" }
+datafusion-sql = { path = "../sql", version = "8.0.0" }
 futures = "0.3"
 hashbrown = { version = "0.12", features = ["raw"] }
 lazy_static = { version = "^1.4.0" }

datafusion/core/src/catalog/mod.rs

@@ -23,109 +23,4 @@ pub mod catalog;
 pub mod information_schema;
 pub mod schema;
 
-/// Represents a resolved path to a table of the form "catalog.schema.table"
-#[derive(Clone, Copy)]
-pub struct ResolvedTableReference<'a> {
-    /// The catalog (aka database) containing the table
-    pub catalog: &'a str,
-    /// The schema containing the table
-    pub schema: &'a str,
-    /// The table name
-    pub table: &'a str,
-}
-
-/// Represents a path to a table that may require further resolution
-#[derive(Clone, Copy)]
-pub enum TableReference<'a> {
-    /// An unqualified table reference, e.g. "table"
-    Bare {
-        /// The table name
-        table: &'a str,
-    },
-    /// A partially resolved table reference, e.g. "schema.table"
-    Partial {
-        /// The schema containing the table
-        schema: &'a str,
-        /// The table name
-        table: &'a str,
-    },
-    /// A fully resolved table reference, e.g. "catalog.schema.table"
-    Full {
-        /// The catalog (aka database) containing the table
-        catalog: &'a str,
-        /// The schema containing the table
-        schema: &'a str,
-        /// The table name
-        table: &'a str,
-    },
-}
-
-impl<'a> TableReference<'a> {
-    /// Retrieve the actual table name, regardless of qualification
-    pub fn table(&self) -> &str {
-        match self {
-            Self::Full { table, .. }
-            | Self::Partial { table, .. }
-            | Self::Bare { table } => table,
-        }
-    }
-
-    /// Given a default catalog and schema, ensure this table reference is fully resolved
-    pub fn resolve(
-        self,
-        default_catalog: &'a str,
-        default_schema: &'a str,
-    ) -> ResolvedTableReference<'a> {
-        match self {
-            Self::Full {
-                catalog,
-                schema,
-                table,
-            } => ResolvedTableReference {
-                catalog,
-                schema,
-                table,
-            },
-            Self::Partial { schema, table } => ResolvedTableReference {
-                catalog: default_catalog,
-                schema,
-                table,
-            },
-            Self::Bare { table } => ResolvedTableReference {
-                catalog: default_catalog,
-                schema: default_schema,
-                table,
-            },
-        }
-    }
-}
-
-impl<'a> From<&'a str> for TableReference<'a> {
-    fn from(s: &'a str) -> Self {
-        let parts: Vec<&str> = s.split('.').collect();
-
-        match parts.len() {
-            1 => Self::Bare { table: s },
-            2 => Self::Partial {
-                schema: parts[0],
-                table: parts[1],
-            },
-            3 => Self::Full {
-                catalog: parts[0],
-                schema: parts[1],
-                table: parts[2],
-            },
-            _ => Self::Bare { table: s },
-        }
-    }
-}
-
-impl<'a> From<ResolvedTableReference<'a>> for TableReference<'a> {
-    fn from(resolved: ResolvedTableReference<'a>) -> Self {
-        Self::Full {
-            catalog: resolved.catalog,
-            schema: resolved.schema,
-            table: resolved.table,
-        }
-    }
-}
+pub use datafusion_sql::{ResolvedTableReference, TableReference};

datafusion/core/src/execution/context.rs

@@ -54,7 +54,6 @@ use arrow::datatypes::{DataType, SchemaRef};
 use crate::catalog::{
     catalog::{CatalogProvider, MemoryCatalogProvider},
     schema::{MemorySchemaProvider, SchemaProvider},
-    ResolvedTableReference, TableReference,
 };
 use crate::dataframe::DataFrame;
 use crate::datasource::listing::ListingTableConfig;
@@ -73,6 +72,7 @@ use crate::optimizer::projection_push_down::ProjectionPushDown;
 use crate::optimizer::simplify_expressions::SimplifyExpressions;
 use crate::optimizer::single_distinct_to_groupby::SingleDistinctToGroupBy;
 use crate::optimizer::subquery_filter_to_join::SubqueryFilterToJoin;
+use datafusion_sql::{ResolvedTableReference, TableReference};
 
 use crate::physical_optimizer::coalesce_batches::CoalesceBatches;
 use crate::physical_optimizer::merge_exec::AddCoalescePartitionsExec;
@@ -86,13 +86,14 @@ use crate::physical_plan::udaf::AggregateUDF;
 use crate::physical_plan::udf::ScalarUDF;
 use crate::physical_plan::ExecutionPlan;
 use crate::physical_plan::PhysicalPlanner;
-use crate::sql::{
-    parser::DFParser,
-    planner::{ContextProvider, SqlToRel},
-};
 use crate::variable::{VarProvider, VarType};
 use async_trait::async_trait;
 use chrono::{DateTime, Utc};
+use datafusion_expr::TableSource;
+use datafusion_sql::{
+    parser::DFParser,
+    planner::{ContextProvider, SqlToRel},
+};
 use parquet::file::properties::WriterProperties;
 use uuid::Uuid;
 
@@ -1423,15 +1424,18 @@ impl SessionState {
 }
 
 impl ContextProvider for SessionState {
-    fn get_table_provider(&self, name: TableReference) -> Result<Arc<dyn TableProvider>> {
+    fn get_table_provider(&self, name: TableReference) -> Result<Arc<dyn TableSource>> {
         let resolved_ref = self.resolve_table_ref(name);
         match self.schema_for_ref(resolved_ref) {
-            Ok(schema) => schema.table(resolved_ref.table).ok_or_else(|| {
-                DataFusionError::Plan(format!(
-                    "'{}.{}.{}' not found",
-                    resolved_ref.catalog, resolved_ref.schema, resolved_ref.table
-                ))
-            }),
+            Ok(schema) => {
+                let provider = schema.table(resolved_ref.table).ok_or_else(|| {
+                    DataFusionError::Plan(format!(
+                        "'{}.{}.{}' not found",
+                        resolved_ref.catalog, resolved_ref.schema, resolved_ref.table
+                    ))
+                })?;
+                Ok(provider_as_source(provider))
+            }
             Err(e) => Err(e),
         }
     }

datafusion/core/src/lib.rs

@@ -220,7 +220,6 @@ pub mod prelude;
 pub mod scalar;
 #[cfg(feature = "scheduler")]
 pub mod scheduler;
-pub mod sql;
 pub mod variable;
 
 // re-export dependencies from arrow-rs to minimise version maintenance for crate users
@@ -232,6 +231,7 @@ pub use datafusion_common as common;
 pub use datafusion_data_access;
 pub use datafusion_expr as logical_expr;
 pub use datafusion_physical_expr as physical_expr;
+pub use datafusion_sql as sql;
 
 pub use datafusion_row as row;
 

datafusion/core/src/physical_plan/planner.rs

@@ -53,7 +53,6 @@ use crate::physical_plan::windows::WindowAggExec;
 use crate::physical_plan::{join_utils, Partitioning};
 use crate::physical_plan::{AggregateExpr, ExecutionPlan, PhysicalExpr, WindowExpr};
 use crate::scalar::ScalarValue;
-use crate::sql::utils::window_expr_common_partition_keys;
 use crate::variable::VarType;
 use crate::{
     error::{DataFusionError, Result},
@@ -65,6 +64,7 @@ use arrow::{compute::can_cast_types, datatypes::DataType};
 use async_trait::async_trait;
 use datafusion_expr::expr::GroupingSet;
 use datafusion_physical_expr::expressions::DateIntervalExpr;
+use datafusion_sql::utils::window_expr_common_partition_keys;
 use futures::future::BoxFuture;
 use futures::{FutureExt, StreamExt, TryStreamExt};
 use log::{debug, trace};

datafusion/sql/Cargo.toml

@@ -0,0 +1,46 @@
+# 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-sql"
+description = "DataFusion SQL Query Planner"
+version = "8.0.0"
+homepage = "https://github.com/apache/arrow-datafusion"
+repository = "https://github.com/apache/arrow-datafusion"
+readme = "README.md"
+authors = ["Apache Arrow <dev@arrow.apache.org>"]
+license = "Apache-2.0"
+keywords = [ "datafusion", "sql", "parser", "planner" ]
+edition = "2021"
+rust-version = "1.59"
+
+[lib]
+name = "datafusion_sql"
+path = "src/lib.rs"
+
+[features]
+default = ["unicode_expressions"]
+unicode_expressions = []
+
+[dependencies]
+ahash = { version = "0.7", default-features = false }
+arrow = { version = "14.0.0", features = ["prettyprint"] }
+datafusion-common = { path = "../common", version = "8.0.0" }
+datafusion-expr = { path = "../expr", version = "8.0.0" }
+hashbrown = "0.12"
+sqlparser = "0.17"
+tokio = { version = "1.0", features = ["macros", "rt", "rt-multi-thread", "sync", "fs", "parking_lot"] }

datafusion/sql/README.md

@@ -0,0 +1,26 @@
+<!---
+  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 SQL Query Planner
+
+[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 that provides a SQL query planner.
+
+[df]: https://crates.io/crates/datafusion

datafusion/core/src/sql/mod.rs → datafusion/sql/src/lib.rs

@@ -20,4 +20,7 @@
 
 pub mod parser;
 pub mod planner;
-pub(crate) mod utils;
+mod table_reference;
+pub mod utils;
+
+pub use table_reference::{ResolvedTableReference, TableReference};

datafusion/core/src/sql/parser.rs → datafusion/sql/src/parser.rs

@@ -19,7 +19,7 @@
 //!
 //! Declares a SQL parser based on sqlparser that handles custom formats that we need.
 
-use crate::logical_plan::FileType;
+use datafusion_expr::logical_plan::FileType;
 use sqlparser::{
     ast::{ColumnDef, ColumnOptionDef, Statement as SQLStatement, TableConstraint},
     dialect::{keywords::Keyword, Dialect, GenericDialect},