feat: reuse range partitioning for date_trunc aggregation

.github/dependabot.yml

@@ -62,14 +62,6 @@ updates:
         update-types:
           - "minor"
           - "patch"
-  # dev/depcheck is a standalone crate excluded from the workspace, so it is
-  # not covered by the workspace entry above
-  - package-ecosystem: cargo
-    directory: "/dev/depcheck"
-    schedule:
-      interval: weekly
-    target-branch: main
-    labels: [auto-dependencies]
   # the wasm test app has its own npm lockfile
   - package-ecosystem: npm
     directory: "/datafusion/wasmtest/datafusion-wasm-app"

.github/workflows/rust.yml

@@ -980,21 +980,15 @@ jobs:
         working-directory: datafusion/proto
         run: cargo msrv --output-format json --log-target stdout verify
 
-  # Check for dependency cycles between DataFusion crates with `dev/depcheck`.
+  # Check for dependency cycles between DataFusion crates, which would block
+  # publishing them to crates.io, with a publishing dry run.
   depcheck:
     name: Circular Dependency Check
     if: ${{ !(github.event_name == 'push' && github.ref == 'refs/heads/main' && github.repository == 'apache/datafusion') && !(github.event_name == 'pull_request' && github.event.pull_request.head.repo.fork && github.base_ref == 'main') }}
-    runs-on: ubuntu-latest
-    container:
-      image: amd64/rust
+    runs-on: ubuntu-slim
     steps:
       - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1  # v7.0.1
-        with:
-          submodules: true
-          fetch-depth: 1
       - name: Setup Rust toolchain
-        uses: ./.github/actions/setup-builder
-        with:
-          rust-version: stable
+        uses: ./.github/actions/setup-rustup
       - name: Check dependencies
         run: bash ci/scripts/check_circular_dependencies.sh

Cargo.toml

@@ -69,7 +69,6 @@ members = [
     "datafusion/doc",
     "xtask",
 ]
-exclude = ["dev/depcheck"]
 resolver = "2"
 
 [workspace.package]

ci/scripts/check_circular_dependencies.sh

@@ -17,18 +17,34 @@
 # specific language governing permissions and limitations
 # under the License.
 
-# Checks for circular dependencies between DataFusion crates with
-# `dev/depcheck`, the same way the "Circular Dependency Check" job does.
+# Checks for circular dependencies between DataFusion crates, the same way the
+# "Circular Dependency Check" job does.
+#
+# A cycle prevents publishing the crates to crates.io. A crate can be
+# published only after every DataFusion crate it depends on, including through
+# dev-dependencies, optional dependencies, and dependencies for other targets,
+# so no crate in a cycle can be published first. Cargo refuses to build a
+# cycle of normal dependencies, but it allows one that passes through a
+# dev-dependency, because a dev-dependency is built only for the crate's own
+# tests, examples, and benchmarks. Such a cycle builds and passes tests, and
+# fails only at release time.
+#
+# The check is a publishing dry run. `cargo publish --workspace` packages every
+# crate that does not set `publish = false`, in an order where each crate comes
+# after the DataFusion crates it depends on, and fails if there is no such
+# order. `--dry-run` stops before uploading, and `--no-verify` skips building
+# the packages. The dry run also fails on any other problem that would stop the
+# crates from being published, such as a dependency without a version.
+# `--allow-dirty` lets the check run on a working tree with uncommitted
+# changes.
 
 set -euo pipefail
 
 SCRIPT_NAME="$(basename "${BASH_SOURCE[0]}")"
 SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
 ROOT_DIR="$(cd "${SCRIPT_DIR}/../.." && pwd)"
 
-# depcheck is outside the root workspace and finds the root manifest from its
-# own directory, so run Cargo from there.
-cd "${ROOT_DIR}/dev/depcheck"
+cd "${ROOT_DIR}"
 
-echo "[${SCRIPT_NAME}] \`cargo run --locked\` in dev/depcheck"
-cargo run --locked
+echo "[${SCRIPT_NAME}] \`cargo publish --workspace --dry-run --no-verify --allow-dirty\`"
+cargo publish --workspace --dry-run --no-verify --allow-dirty

datafusion/catalog-listing/src/table.rs

@@ -25,13 +25,13 @@ use async_trait::async_trait;
 use datafusion_catalog::{ScanArgs, ScanResult, Session, TableProvider};
 use datafusion_common::stats::{Precision, is_known_empty};
 use datafusion_common::{
-    Constraints, DFSchema, SchemaExt, Statistics, internal_datafusion_err, plan_err,
-    project_schema,
+    Constraints, DFSchema, SchemaExt, SplitPoint, Statistics, internal_datafusion_err,
+    plan_err, project_schema,
 };
 use datafusion_datasource::file::FileSource;
 use datafusion_datasource::file_groups::FileGroup;
 use datafusion_datasource::file_scan_config::{
-    FileScanConfig, FileScanConfigBuilder, output_partitioning_from_partition_fields,
+    FileScanConfig, FileScanConfigBuilder, output_range_partitioning_from_split_points,
 };
 use datafusion_datasource::file_sink_config::{FileOutputMode, FileSinkConfig};
 #[expect(deprecated)]
@@ -69,6 +69,8 @@ pub struct ListFilesResult {
     pub statistics: Statistics,
     /// Whether files are grouped by partition values.
     pub grouped_by_partition: bool,
+    /// Boundaries between file groups, empty unless `grouped_by_partition` is true.
+    pub partition_split_points: Vec<SplitPoint>,
 }
 
 /// Built in [`TableProvider`] that reads data from one or more files as a single table.
@@ -668,6 +670,7 @@ impl ListingTable {
             file_groups: mut partitioned_file_lists,
             statistics,
             grouped_by_partition: partitioned_by_file_group,
+            partition_split_points,
         } = self
             .list_files_for_scan(state, &partition_filters, statistic_file_limit)
             .await?;
@@ -687,6 +690,7 @@ impl ListingTable {
                 .config_options()
                 .execution
                 .split_file_groups_by_statistics;
+        let mut regrouped_by_statistics = false;
         match split_file_groups_by_statistics
             .then(|| {
                 output_ordering.first().map(|output_ordering| {
@@ -704,6 +708,7 @@ impl ListingTable {
             Some(Ok(new_groups)) => {
                 if new_groups.len() <= file_group_count {
                     partitioned_file_lists = new_groups;
+                    regrouped_by_statistics = true;
                 } else {
                     log::debug!(
                         "attempted to split file groups by statistics, but there were more file groups than target_partitions; falling back to unordered"
@@ -737,26 +742,25 @@ impl ListingTable {
                     )?
                 }
             };
+            Some(output_partitioning)
+        } else if partitioned_by_file_group && !regrouped_by_statistics {
+            output_range_partitioning_from_split_points(
+                &self.table_schema,
+                &table_partition_cols.into(),
+                partition_split_points,
+            )?
+        } else {
+            None
+        };
+        if let Some(output_partitioning) = &output_partitioning {
             let partition_count = output_partitioning.partition_count();
             if partitioned_file_lists.len() != partition_count {
                 return plan_err!(
                     "ListingTable output_partitioning has {partition_count} partitions, but the scan has {} file groups",
                     partitioned_file_lists.len()
                 );
             }
-            Some(output_partitioning)
-        } else if partitioned_by_file_group {
-            // Files are grouped by partition column values: declare output
-            // partitioning on those columns so the optimizer can skip
-            // repartitioning for aggregates and joins on the partition columns.
-            output_partitioning_from_partition_fields(
-                &self.table_schema,
-                &table_partition_cols.clone().into(),
-                partitioned_file_lists.len(),
-            )
-        } else {
-            None
-        };
+        }
 
         let Some(object_store_url) =
             self.table_paths.first().map(ListingTableUrl::object_store)
@@ -1005,6 +1009,7 @@ impl ListingTable {
                 file_groups: vec![],
                 statistics: Statistics::new_unknown(&self.file_schema),
                 grouped_by_partition: false,
+                partition_split_points: vec![],
             });
         };
         let (file_group, inexact_stats) = self
@@ -1018,28 +1023,31 @@ impl ListingTable {
         // skip repartitioning for aggregates and joins on partition columns.
         let threshold = ctx.config_options().optimizer.preserve_file_partitions;
 
-        let (file_groups, grouped_by_partition) =
+        let (file_groups, grouped_by_partition, partition_split_points) =
             if threshold > 0 && !self.options.table_partition_cols.is_empty() {
-                let grouped = file_group.group_by_partition_values(file_group_count);
+                let (grouped, split_points) = file_group
+                    .group_by_partition_values_with_split_points(file_group_count);
                 if grouped.len() >= threshold {
-                    (grouped, true)
+                    (grouped, true, split_points)
                 } else {
                     let all_files: Vec<_> =
                         grouped.into_iter().flat_map(|g| g.into_inner()).collect();
                     (
                         FileGroup::new(all_files).split_files(file_group_count),
                         false,
+                        vec![],
                     )
                 }
             } else {
-                (file_group.split_files(file_group_count), false)
+                (file_group.split_files(file_group_count), false, vec![])
             };
 
         self.list_files_result_from_groups(
             ctx,
             file_groups,
             inexact_stats,
             grouped_by_partition,
+            partition_split_points,
         )
     }
 
@@ -1067,6 +1075,7 @@ impl ListingTable {
                 file_groups: vec![],
                 statistics: Statistics::new_unknown(&self.file_schema),
                 grouped_by_partition: false,
+                partition_split_points: vec![],
             });
         };
         let (file_group, inexact_stats) =
@@ -1078,7 +1087,7 @@ impl ListingTable {
         let file_groups =
             self.filter_declared_file_groups_by_partition_filters(file_groups, filters)?;
 
-        self.list_files_result_from_groups(ctx, file_groups, inexact_stats, false)
+        self.list_files_result_from_groups(ctx, file_groups, inexact_stats, false, vec![])
     }
 
     fn filter_declared_file_groups_by_partition_filters(
@@ -1113,6 +1122,7 @@ impl ListingTable {
         file_groups: Vec<FileGroup>,
         inexact_stats: bool,
         grouped_by_partition: bool,
+        partition_split_points: Vec<SplitPoint>,
     ) -> datafusion_common::Result<ListFilesResult> {
         let (file_groups, stats) = compute_all_files_statistics(
             file_groups,
@@ -1129,6 +1139,7 @@ impl ListingTable {
             file_groups,
             statistics: stats,
             grouped_by_partition,
+            partition_split_points,
         })
     }
 

datafusion/common/src/nested_struct.rs

@@ -903,7 +903,7 @@ fn validate_map_key_data_type(
             }
             Ok(())
         }
-        _ if is_injective_map_key_cast(source_type, target_type) => Ok(()),
+        _ if is_injective_cast(source_type, target_type) => Ok(()),
         _ => _plan_err!(
             "Cannot safely evolve Map key type from {} to {}",
             source_type,
@@ -912,15 +912,15 @@ fn validate_map_key_data_type(
     }
 }
 
-/// Returns true when casting a Map key from `source_type` to `target_type` is
-/// known to preserve key identity.
+/// Returns true when casting from `source_type` to `target_type` is known to
+/// preserve value identity, so distinct source values stay distinct.
 ///
-/// Keep this list conservative. Map keys define equality/lookup identity, so
-/// only representation widenings that cannot merge distinct source keys belong
-/// here. Other Arrow-supported casts, including decimal/timestamp widening and
-/// dictionaries, should stay rejected until their identity semantics are
-/// deliberately reviewed.
-fn is_injective_map_key_cast(source_type: &DataType, target_type: &DataType) -> bool {
+/// Keep this list conservative. Map keys and decorrelated equality keys depend
+/// on identity, so only representation widenings that cannot merge distinct
+/// source values belong here. Other Arrow-supported casts, including
+/// decimal/timestamp widening and dictionaries, should stay rejected until
+/// their identity semantics are deliberately reviewed.
+pub fn is_injective_cast(source_type: &DataType, target_type: &DataType) -> bool {
     matches!(
         (source_type, target_type),
         (
@@ -2283,14 +2283,14 @@ mod tests {
             (DataType::UInt16, DataType::Int64),
             (DataType::UInt32, DataType::Int64),
         ] {
-            assert!(is_injective_map_key_cast(&source_type, &target_type));
+            assert!(is_injective_cast(&source_type, &target_type));
         }
         for (source_type, target_type) in [
             (DataType::UInt8, DataType::Int8),
             (DataType::UInt16, DataType::Int16),
             (DataType::UInt32, DataType::Int32),
         ] {
-            assert!(!is_injective_map_key_cast(&source_type, &target_type));
+            assert!(!is_injective_cast(&source_type, &target_type));
         }
     }
 

datafusion/core/Cargo.toml

@@ -245,6 +245,10 @@ name = "math_query_sql"
 harness = false
 name = "filter_query_sql"
 
+[[bench]]
+harness = false
+name = "in_list_rewrite"
+
 [[bench]]
 harness = false
 name = "struct_query_sql"