teach row group pruning about struct fields

Author: friendlymatthew

Cargo.lock

@@ -91,19 +91,19 @@ ahash = { version = "0.8", default-features = false, features = [
     "runtime-rng",
 ] }
 apache-avro = { version = "0.21", default-features = false }
-arrow = { version = "58.0.0", features = [
+arrow = { git = "https://github.com/pydantic/arrow-rs.git", branch = "friendlymatthew/statistics-converter-from-col-index", features = [
     "prettyprint",
     "chrono-tz",
 ] }
-arrow-buffer = { version = "58.0.0", default-features = false }
-arrow-flight = { version = "58.0.0", features = [
+arrow-buffer = { git = "https://github.com/pydantic/arrow-rs.git", branch = "friendlymatthew/statistics-converter-from-col-index", default-features = false }
+arrow-flight = { git = "https://github.com/pydantic/arrow-rs.git", branch = "friendlymatthew/statistics-converter-from-col-index", features = [
     "flight-sql-experimental",
 ] }
-arrow-ipc = { version = "58.0.0", default-features = false, features = [
+arrow-ipc = { git = "https://github.com/pydantic/arrow-rs.git", branch = "friendlymatthew/statistics-converter-from-col-index", default-features = false, features = [
     "lz4",
 ] }
-arrow-ord = { version = "58.0.0", default-features = false }
-arrow-schema = { version = "58.0.0", default-features = false }
+arrow-ord = { git = "https://github.com/pydantic/arrow-rs.git", branch = "friendlymatthew/statistics-converter-from-col-index", default-features = false }
+arrow-schema = { git = "https://github.com/pydantic/arrow-rs.git", branch = "friendlymatthew/statistics-converter-from-col-index", default-features = false }
 async-trait = "0.1.89"
 bigdecimal = "0.4.8"
 bytes = "1.11"
@@ -168,7 +168,7 @@ memchr = "2.8.0"
 num-traits = { version = "0.2" }
 object_store = { version = "0.13.1", default-features = false }
 parking_lot = "0.12"
-parquet = { version = "58.0.0", default-features = false, features = [
+parquet = { git = "https://github.com/pydantic/arrow-rs.git", branch = "friendlymatthew/statistics-converter-from-col-index", default-features = false, features = [
     "arrow",
     "async",
     "object_store",

Cargo.toml

@@ -25,7 +25,7 @@ use arrow::datatypes::{Int32Type, SchemaRef};
 use arrow::util::pretty::pretty_format_batches;
 use async_trait::async_trait;
 use datafusion::catalog::Session;
-use datafusion::common::pruning::PruningStatistics;
+use datafusion::common::pruning::{PruningColumn, PruningStatistics};
 use datafusion::common::{
     DFSchema, DataFusionError, Result, ScalarValue, internal_datafusion_err,
 };
@@ -432,21 +432,19 @@ impl ParquetMetadataIndex {
 /// the required statistics via the [`PruningStatistics`] trait
 impl PruningStatistics for ParquetMetadataIndex {
     /// return the minimum values for the value column
-    fn min_values(&self, column: &Column) -> Option<ArrayRef> {
-        if column.name.eq("value") {
-            Some(self.value_column_mins().clone())
-        } else {
-            None
-        }
+    fn min_values(&self, column: &PruningColumn) -> Option<ArrayRef> {
+        column
+            .name()
+            .eq("value")
+            .then_some(self.value_column_mins().clone())
     }
 
     /// return the maximum values for the value column
-    fn max_values(&self, column: &Column) -> Option<ArrayRef> {
-        if column.name.eq("value") {
-            Some(self.value_column_maxes().clone())
-        } else {
-            None
-        }
+    fn max_values(&self, column: &PruningColumn) -> Option<ArrayRef> {
+        column
+            .name()
+            .eq("value")
+            .then_some(self.value_column_maxes().clone())
     }
 
     /// return the number of "containers". In this example, each "container" is
@@ -457,20 +455,20 @@ impl PruningStatistics for ParquetMetadataIndex {
 
     /// Return `None` to signal we don't have any information about null
     /// counts in the index,
-    fn null_counts(&self, _column: &Column) -> Option<ArrayRef> {
+    fn null_counts(&self, _column: &PruningColumn) -> Option<ArrayRef> {
         None
     }
 
     /// return the row counts for each file
-    fn row_counts(&self, _column: &Column) -> Option<ArrayRef> {
+    fn row_counts(&self, _column: &PruningColumn) -> Option<ArrayRef> {
         Some(self.row_counts_ref().clone())
     }
 
     /// The `contained` API can be used with structures such as Bloom filters,
     /// but is not used in this example, so return `None`
     fn contained(
         &self,
-        _column: &Column,
+        _column: &PruningColumn,
         _values: &HashSet<ScalarValue>,
     ) -> Option<BooleanArray> {
         None

datafusion-examples/examples/data_io/parquet_index.rs

@@ -22,7 +22,7 @@ use std::sync::Arc;
 
 use arrow::array::{ArrayRef, BooleanArray, Int32Array};
 use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
-use datafusion::common::pruning::PruningStatistics;
+use datafusion::common::pruning::{PruningColumn, PruningStatistics};
 use datafusion::common::{DFSchema, ScalarValue};
 use datafusion::error::Result;
 use datafusion::execution::context::ExecutionProps;
@@ -148,40 +148,40 @@ impl PruningStatistics for MyCatalog {
         3
     }
 
-    fn min_values(&self, column: &Column) -> Option<ArrayRef> {
+    fn min_values(&self, column: &PruningColumn) -> Option<ArrayRef> {
         // The pruning predicate evaluates the bounds for multiple expressions
         // at once, so  return an array with an element for the minimum value in
         // each file
-        match column.name.as_str() {
+        match column.name() {
             "x" => Some(i32_array(self.x_values.iter().map(|(min, _)| min))),
             "y" => Some(i32_array(self.y_values.iter().map(|(min, _)| min))),
             name => panic!("unknown column name: {name}"),
         }
     }
 
-    fn max_values(&self, column: &Column) -> Option<ArrayRef> {
+    fn max_values(&self, column: &PruningColumn) -> Option<ArrayRef> {
         // similarly to min_values, return an array with an element for the
         // maximum value in each file
-        match column.name.as_str() {
+        match column.name() {
             "x" => Some(i32_array(self.x_values.iter().map(|(_, max)| max))),
             "y" => Some(i32_array(self.y_values.iter().map(|(_, max)| max))),
             name => panic!("unknown column name: {name}"),
         }
     }
 
-    fn null_counts(&self, _column: &Column) -> Option<ArrayRef> {
+    fn null_counts(&self, _column: &PruningColumn) -> Option<ArrayRef> {
         // In this example, we know nothing about the number of nulls
         None
     }
 
-    fn row_counts(&self, _column: &Column) -> Option<ArrayRef> {
+    fn row_counts(&self, _column: &PruningColumn) -> Option<ArrayRef> {
         // In this example, we know nothing about the number of rows in each file
         None
     }
 
     fn contained(
         &self,
-        _column: &Column,
+        _column: &PruningColumn,
         _values: &HashSet<ScalarValue>,
     ) -> Option<BooleanArray> {
         // this method can be used to implement Bloom filter like filtering