Merge remote-tracking branch 'apache/main' into alamb/test_upstream_coalesce

Author: alamb

.github/workflows/rust.yml

@@ -39,14 +39,6 @@ on:
   workflow_dispatch:
 
 jobs:
-  # Check license header
-  license-header-check:
-    runs-on: ubuntu-latest
-    name: Check License Header
-    steps:
-      - uses: actions/checkout@v4
-      - uses: korandoru/hawkeye@v6
-
   # Check crate compiles and base cargo check passes
   linux-build-lib:
     name: linux build test

Cargo.lock

@@ -167,7 +167,7 @@ recursive = "0.1.1"
 regex = "1.8"
 rstest = "0.25.0"
 serde_json = "1"
-sqlparser = { version = "0.55.0", features = ["visitor"] }
+sqlparser = { version = "0.55.0", default-features = false, features = ["std", "visitor"] }
 tempfile = "3"
 tokio = { version = "1.45", features = ["macros", "rt", "sync"] }
 url = "2.5.4"

Cargo.toml

@@ -518,7 +518,6 @@ In addition, topk_tpch is available from the bench.sh script:
 ./bench.sh run topk_tpch
 ```
 
-
 ## IMDB
 
 Run Join Order Benchmark (JOB) on IMDB dataset.

benchmarks/README.md

@@ -103,12 +103,14 @@ impl InformationSchemaConfig {
                     // schema name may not exist in the catalog, so we need to check
                     if let Some(schema) = catalog.schema(&schema_name) {
                         for table_name in schema.table_names() {
-                            if let Some(table) = schema.table(&table_name).await? {
+                            if let Some(table_type) =
+                                schema.table_type(&table_name).await?
+                            {
                                 builder.add_table(
                                     &catalog_name,
                                     &schema_name,
                                     &table_name,
-                                    table.table_type(),
+                                    table_type,
                                 );
                             }
                         }
@@ -1359,3 +1361,92 @@ impl PartitionStream for InformationSchemaParameters {
         ))
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::CatalogProvider;
+
+    #[tokio::test]
+    async fn make_tables_uses_table_type() {
+        let config = InformationSchemaConfig {
+            catalog_list: Arc::new(Fixture),
+        };
+        let mut builder = InformationSchemaTablesBuilder {
+            catalog_names: StringBuilder::new(),
+            schema_names: StringBuilder::new(),
+            table_names: StringBuilder::new(),
+            table_types: StringBuilder::new(),
+            schema: Arc::new(Schema::empty()),
+        };
+
+        assert!(config.make_tables(&mut builder).await.is_ok());
+
+        assert_eq!("BASE TABLE", builder.table_types.finish().value(0));
+    }
+
+    #[derive(Debug)]
+    struct Fixture;
+
+    #[async_trait]
+    impl SchemaProvider for Fixture {
+        // InformationSchemaConfig::make_tables should use this.
+        async fn table_type(&self, _: &str) -> Result<Option<TableType>> {
+            Ok(Some(TableType::Base))
+        }
+
+        // InformationSchemaConfig::make_tables used this before `table_type`
+        // existed but should not, as it may be expensive.
+        async fn table(&self, _: &str) -> Result<Option<Arc<dyn TableProvider>>> {
+            panic!("InformationSchemaConfig::make_tables called SchemaProvider::table instead of table_type")
+        }
+
+        fn as_any(&self) -> &dyn Any {
+            unimplemented!("not required for these tests")
+        }
+
+        fn table_names(&self) -> Vec<String> {
+            vec!["atable".to_string()]
+        }
+
+        fn table_exist(&self, _: &str) -> bool {
+            unimplemented!("not required for these tests")
+        }
+    }
+
+    impl CatalogProviderList for Fixture {
+        fn as_any(&self) -> &dyn Any {
+            unimplemented!("not required for these tests")
+        }
+
+        fn register_catalog(
+            &self,
+            _: String,
+            _: Arc<dyn CatalogProvider>,
+        ) -> Option<Arc<dyn CatalogProvider>> {
+            unimplemented!("not required for these tests")
+        }
+
+        fn catalog_names(&self) -> Vec<String> {
+            vec!["acatalog".to_string()]
+        }
+
+        fn catalog(&self, _: &str) -> Option<Arc<dyn CatalogProvider>> {
+            Some(Arc::new(Self))
+        }
+    }
+
+    impl CatalogProvider for Fixture {
+        fn as_any(&self) -> &dyn Any {
+            unimplemented!("not required for these tests")
+        }
+
+        fn schema_names(&self) -> Vec<String> {
+            vec!["aschema".to_string()]
+        }
+
+        fn schema(&self, _: &str) -> Option<Arc<dyn SchemaProvider>> {
+            Some(Arc::new(Self))
+        }
+    }
+}

datafusion/catalog/src/information_schema.rs

@@ -26,6 +26,7 @@ use std::sync::Arc;
 
 use crate::table::TableProvider;
 use datafusion_common::Result;
+use datafusion_expr::TableType;
 
 /// Represents a schema, comprising a number of named tables.
 ///
@@ -54,6 +55,14 @@ pub trait SchemaProvider: Debug + Sync + Send {
         name: &str,
     ) -> Result<Option<Arc<dyn TableProvider>>, DataFusionError>;
 
+    /// Retrieves the type of a specific table from the schema by name, if it exists, otherwise
+    /// returns `None`.  Implementations for which this operation is cheap but [Self::table] is
+    /// expensive can override this to improve operations that only need the type, e.g.
+    /// `SELECT * FROM information_schema.tables`.
+    async fn table_type(&self, name: &str) -> Result<Option<TableType>> {
+        self.table(name).await.map(|o| o.map(|t| t.table_type()))
+    }
+
     /// If supported by the implementation, adds a new table named `name` to
     /// this schema.
     ///

datafusion/catalog/src/schema.rs

@@ -58,7 +58,7 @@ base64 = "0.22.1"
 half = { workspace = true }
 hashbrown = { workspace = true }
 indexmap = { workspace = true }
-libc = "0.2.173"
+libc = "0.2.174"
 log = { workspace = true }
 object_store = { workspace = true, optional = true }
 parquet = { workspace = true, optional = true, default-features = true }

datafusion/common/Cargo.toml

@@ -614,6 +614,13 @@ config_namespace! {
         /// during aggregations, if possible
         pub enable_topk_aggregation: bool, default = true
 
+        /// When set to true attempts to push down dynamic filters generated by operators into the file scan phase.
+        /// For example, for a query such as `SELECT * FROM t ORDER BY timestamp DESC LIMIT 10`, the optimizer
+        /// will attempt to push down the current top 10 timestamps that the TopK operator references into the file scans.
+        /// This means that if we already have 10 timestamps in the year 2025
+        /// any files that only have timestamps in the year 2024 can be skipped / pruned at various stages in the scan.
+        pub enable_dynamic_filter_pushdown: bool, default = true
+
         /// When set to true, the optimizer will insert filters before a join between
         /// a nullable and non-nullable column to filter out nulls on the nullable side. This
         /// filter can add additional overhead when the file format does not fully support

datafusion/common/src/config.rs

@@ -46,6 +46,7 @@ pub mod file_options;
 pub mod format;
 pub mod hash_utils;
 pub mod instant;
+mod null_equality;
 pub mod parsers;
 pub mod pruning;
 pub mod rounding;
@@ -79,6 +80,7 @@ pub use functional_dependencies::{
 };
 use hashbrown::hash_map::DefaultHashBuilder;
 pub use join_type::{JoinConstraint, JoinSide, JoinType};
+pub use null_equality::NullEquality;
 pub use param_value::ParamValues;
 pub use scalar::{ScalarType, ScalarValue};
 pub use schema_reference::SchemaReference;

datafusion/common/src/lib.rs

@@ -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.
+
+/// Represents the behavior for null values when evaluating equality. Currently, its primary use
+/// case is to define the behavior of joins for null values.
+///
+/// # Examples
+///
+/// The following table shows the expected equality behavior for `NullEquality`.
+///
+/// | A    | B    | NullEqualsNothing | NullEqualsNull |
+/// |------|------|-------------------|----------------|
+/// | NULL | NULL | false             | true           |
+/// | NULL | 'b'  | false             | false          |
+/// | 'a'  | NULL | false             | false          |
+/// | 'a'  | 'b'  | false             | false          |
+///
+/// # Order
+///
+/// The order on this type represents the "restrictiveness" of the behavior. The more restrictive
+/// a behavior is, the fewer elements are considered to be equal to null.
+/// [NullEquality::NullEqualsNothing] represents the most restrictive behavior.
+///
+/// This mirrors the old order with `null_equals_null` booleans, as `false` indicated that
+/// `null != null`.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Hash)]
+pub enum NullEquality {
+    /// Null is *not* equal to anything (`null != null`)
+    NullEqualsNothing,
+    /// Null is equal to null (`null == null`)
+    NullEqualsNull,
+}