Prune parquet row groups using fully dictionary-encoded columns

Cargo.toml

@@ -209,6 +209,31 @@ url = "2.5.7"
 uuid = "1.23"
 zstd = { version = "0.13", default-features = false }
 
+# TEMPORARY patch: points at an unreleased arrow-rs branch that adds the
+# dictionary-page decode API this branch's Parquet dictionary row-group
+# pruning depends on (arrow-rs issue #9010 / PR #10420, which supersedes
+# the earlier, now-closed PR #9011). Remove once arrow-rs releases a
+# version with this API and the dependency above is bumped to it -- do
+# not merge this patch.
+[patch.crates-io]
+arrow = { git = "https://github.com/DarkWanderer/arrow-rs", branch = "get-dictionary" }
+arrow-arith = { git = "https://github.com/DarkWanderer/arrow-rs", branch = "get-dictionary" }
+arrow-array = { git = "https://github.com/DarkWanderer/arrow-rs", branch = "get-dictionary" }
+arrow-avro = { git = "https://github.com/DarkWanderer/arrow-rs", branch = "get-dictionary" }
+arrow-buffer = { git = "https://github.com/DarkWanderer/arrow-rs", branch = "get-dictionary" }
+arrow-cast = { git = "https://github.com/DarkWanderer/arrow-rs", branch = "get-dictionary" }
+arrow-csv = { git = "https://github.com/DarkWanderer/arrow-rs", branch = "get-dictionary" }
+arrow-data = { git = "https://github.com/DarkWanderer/arrow-rs", branch = "get-dictionary" }
+arrow-flight = { git = "https://github.com/DarkWanderer/arrow-rs", branch = "get-dictionary" }
+arrow-ipc = { git = "https://github.com/DarkWanderer/arrow-rs", branch = "get-dictionary" }
+arrow-json = { git = "https://github.com/DarkWanderer/arrow-rs", branch = "get-dictionary" }
+arrow-ord = { git = "https://github.com/DarkWanderer/arrow-rs", branch = "get-dictionary" }
+arrow-row = { git = "https://github.com/DarkWanderer/arrow-rs", branch = "get-dictionary" }
+arrow-schema = { git = "https://github.com/DarkWanderer/arrow-rs", branch = "get-dictionary" }
+arrow-select = { git = "https://github.com/DarkWanderer/arrow-rs", branch = "get-dictionary" }
+arrow-string = { git = "https://github.com/DarkWanderer/arrow-rs", branch = "get-dictionary" }
+parquet = { git = "https://github.com/DarkWanderer/arrow-rs", branch = "get-dictionary" }
+
 [workspace.lints.clippy]
 # Detects large stack-allocated futures that may cause stack overflow crashes (see threshold in clippy.toml)
 large_futures = "warn"

datafusion/common/src/config.rs

@@ -1181,6 +1181,16 @@ config_namespace! {
         /// (reading) Use any available bloom filters when reading parquet files
         pub bloom_filter_on_read: bool, default = true
 
+        /// (reading) Use fully dictionary-encoded `BYTE_ARRAY` (Utf8/Binary)
+        /// column chunks as an exact row-group membership index when reading
+        /// parquet files. Unlike bloom filters, a fully dictionary-encoded
+        /// column chunk's dictionary is the exact, complete set of the row
+        /// group's distinct values, so this can prune both `IN`/`=` and
+        /// `NOT IN`/`!=` predicates. Only column chunks whose page encoding
+        /// statistics prove every data page came from the dictionary are
+        /// used; chunks that fell back to `PLAIN` encoding are ignored.
+        pub dictionary_filter_on_read: bool, default = false
+
         /// (reading) The maximum predicate cache size, in bytes. When
         /// `pushdown_filters` is enabled, sets the maximum memory used to cache
         /// the results of predicate evaluation between filter evaluation and

datafusion/common/src/file_options/parquet_writer.rs

@@ -242,6 +242,7 @@ impl ParquetOptions {
             maximum_parallel_row_group_writers: _,
             maximum_buffered_record_batches_per_stream: _,
             bloom_filter_on_read: _, // reads not used for writer props
+            dictionary_filter_on_read: _, // reads not used for writer props
             schema_force_view_types: _,
             binary_as_string: _, // not used for writer props
             coerce_int96: _,     // not used for writer props
@@ -500,6 +501,7 @@ mod tests {
             maximum_buffered_record_batches_per_stream: defaults
                 .maximum_buffered_record_batches_per_stream,
             bloom_filter_on_read: defaults.bloom_filter_on_read,
+            dictionary_filter_on_read: defaults.dictionary_filter_on_read,
             schema_force_view_types: defaults.schema_force_view_types,
             binary_as_string: defaults.binary_as_string,
             skip_arrow_metadata: defaults.skip_arrow_metadata,
@@ -620,6 +622,8 @@ mod tests {
                 maximum_buffered_record_batches_per_stream: global_options_defaults
                     .maximum_buffered_record_batches_per_stream,
                 bloom_filter_on_read: global_options_defaults.bloom_filter_on_read,
+                dictionary_filter_on_read: global_options_defaults
+                    .dictionary_filter_on_read,
                 max_predicate_cache_size: global_options_defaults
                     .max_predicate_cache_size,
                 schema_force_view_types: global_options_defaults.schema_force_view_types,

datafusion/datasource-parquet/Cargo.toml

@@ -91,3 +91,7 @@ harness = false
 [[bench]]
 name = "parquet_metadata_statistics"
 harness = false
+
+[[bench]]
+name = "parquet_dictionary_pruning"
+harness = false