feat: introduce optional RLE-dictionary preservation for Parquet reads

Adds `datafusion.execution.parquet.enable_rle_to_dictionary` (default
false). When enabled, string and binary columns that are physically
RLE_DICTIONARY-encoded in the Parquet file footer are surfaced as
Arrow `Dictionary(Int32, Utf8/Binary)` arrays instead of being decoded
to plain `Utf8/Binary`. This preserves the compact encoding in memory
and can improve aggregation performance on low-cardinality columns.

Schema promotion happens at planning time in `fetch_schema` so that
physical operators (`FilterExec`, `AggregateExec`, etc.) are compiled
against the correct output types before any file I/O begins. At scan
time the same promotion is applied via `ArrowReaderOptions::with_schema`
so arrow-rs produces dictionary arrays directly.

`apply_file_schema_type_coercions` keeps its original two-parameter
signature (no breaking change); the dictionary-aware variant is exposed
as the crate-internal `apply_file_schema_type_coercions_with_rle`.

Author: Rich-T-kid

datafusion/common/src/config.rs

@@ -1293,6 +1293,19 @@ config_namespace! {
         /// Defaults to 20.
         pub max_in_list_size: usize, default = 20
 
+        /// (reading) If true, string and binary columns that are dictionary-encoded
+        /// (RLE_DICTIONARY) in the Parquet file are read as
+        /// `Dictionary<Int32, Utf8>` / `Dictionary<Int32, Binary>` instead of
+        /// their plain value type. This reduces memory usage for low-cardinality
+        /// columns and can improve aggregation performance.
+        ///
+        /// This applies only to **inferred-schema** tables (i.e. `CREATE EXTERNAL
+        /// TABLE` without an explicit column list). Tables with a user-supplied
+        /// schema are not promoted because the Parquet footer is not read at DDL
+        /// time, so RLE encoding cannot be detected per-column.
+        /// See <https://github.com/apache/datafusion/issues/24112>
+        pub enable_rle_to_dictionary: bool, default = false
+
         // The following options affect writing to parquet files
         // and map to parquet::file::properties::WriterProperties
 

datafusion/common/src/file_options/parquet_writer.rs

@@ -249,6 +249,7 @@ impl ParquetOptions {
             skip_arrow_metadata: _,
             max_predicate_cache_size: _,
             max_in_list_size: _,
+            enable_rle_to_dictionary: _,
         } = self;
 
         let mut builder = WriterProperties::builder()
@@ -509,6 +510,7 @@ mod tests {
             coerce_int96_tz: None,
             max_predicate_cache_size: defaults.max_predicate_cache_size,
             content_defined_chunking: defaults.content_defined_chunking.clone(),
+            enable_rle_to_dictionary: defaults.enable_rle_to_dictionary,
         }
     }
 
@@ -631,6 +633,8 @@ mod tests {
                 coerce_int96: None,
                 coerce_int96_tz: None,
                 content_defined_chunking: props.content_defined_chunking().into(),
+                enable_rle_to_dictionary: global_options_defaults
+                    .enable_rle_to_dictionary,
             },
             column_specific_options,
             key_value_metadata,

datafusion/core/tests/parquet/schema_coercion.rs

@@ -24,9 +24,10 @@ use arrow::array::{
 use arrow::datatypes::{DataType, Field, Schema};
 use datafusion::datasource::physical_plan::ParquetSource;
 use datafusion::physical_plan::collect;
-use datafusion::prelude::SessionContext;
+use datafusion::prelude::{ParquetReadOptions, SessionConfig, SessionContext};
 use datafusion::test::object_store::local_unpartitioned_file;
 use datafusion_common::Result;
+use datafusion_common::ScalarValue;
 use datafusion_common::test_util::batches_to_sort_string;
 use datafusion_execution::object_store::ObjectStoreUrl;
 
@@ -145,6 +146,154 @@ async fn multi_parquet_coercion_projection() {
     ");
 }
 
+/// Writes `batch` to a temp parquet file where `rle_cols` use RLE_DICTIONARY encoding and all other columns use plain.
+fn store_mixed_encoding_parquet(
+    batch: &RecordBatch,
+    rle_cols: &[&str],
+) -> (ObjectMeta, NamedTempFile) {
+    use parquet::schema::types::ColumnPath;
+    let mut output = tempfile::Builder::new()
+        .suffix(".parquet")
+        .tempfile()
+        .expect("creating temp file");
+    let mut builder = WriterProperties::builder().set_dictionary_enabled(false);
+    for col in rle_cols {
+        builder = builder
+            .set_column_dictionary_enabled(ColumnPath::new(vec![col.to_string()]), true);
+    }
+    let mut writer =
+        ArrowWriter::try_new(&mut output, batch.schema(), Some(builder.build()))
+            .expect("creating writer");
+    writer.write(batch).expect("Writing batch");
+    writer.close().unwrap();
+    let meta = local_unpartitioned_file(&output);
+    (meta, output)
+}
+
+/// Promotion happens at logical planning time: after `register_parquet` with the
+/// flag on, `ctx.table()` already shows RLE columns as `Dictionary(Int32, Utf8)`
+/// before any query executes.
+#[tokio::test]
+async fn rle_dictionary_logical_schema_promotion() {
+    let schema = Arc::new(Schema::new(vec![
+        Field::new("env", DataType::Utf8, true),
+        Field::new("version", DataType::Utf8, true),
+    ]));
+    let batch = RecordBatch::try_new(
+        Arc::clone(&schema),
+        vec![
+            Arc::new(StringArray::from(vec!["prod", "staging"])) as ArrayRef,
+            Arc::new(StringArray::from(vec!["1.0", "1.1"])) as ArrayRef,
+        ],
+    )
+    .unwrap();
+    let (_meta, file) = store_mixed_encoding_parquet(&batch, &["env"]);
+
+    let config = SessionConfig::new().set(
+        "datafusion.execution.parquet.enable_rle_to_dictionary",
+        &ScalarValue::Boolean(Some(true)),
+    );
+    let ctx = SessionContext::new_with_config(config);
+    ctx.register_parquet(
+        "t",
+        file.path().to_str().unwrap(),
+        ParquetReadOptions::default(),
+    )
+    .await
+    .unwrap();
+
+    let logical_schema = ctx.table("t").await.unwrap().schema().clone();
+    let dict_type =
+        DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8));
+    assert_eq!(
+        logical_schema
+            .field_with_unqualified_name("env")
+            .unwrap()
+            .data_type(),
+        &dict_type,
+    );
+    assert_eq!(
+        logical_schema
+            .field_with_unqualified_name("version")
+            .unwrap()
+            .data_type(),
+        &DataType::Utf8View,
+    );
+}
+
+/// Five string columns, three RLE_DICTIONARY encoded and two plain.
+/// Flag on: only the three RLE columns become `Dictionary(Int32, Utf8)`; plain
+/// columns stay `Utf8View`.  Flag off: all columns stay `Utf8View`.
+#[tokio::test]
+async fn rle_dictionary_selective_promotion() {
+    let schema = Arc::new(Schema::new(vec![
+        Field::new("env", DataType::Utf8, true),
+        Field::new("region", DataType::Utf8, true),
+        Field::new("tier", DataType::Utf8, true),
+        Field::new("version", DataType::Utf8, true),
+        Field::new("build", DataType::Utf8, true),
+    ]));
+    let make_col = |vals: Vec<&str>| -> ArrayRef { Arc::new(StringArray::from(vals)) };
+    let batch = RecordBatch::try_new(
+        Arc::clone(&schema),
+        vec![
+            make_col(vec!["prod", "staging", "prod", "canary", "staging", "prod"]),
+            make_col(vec![
+                "us-east", "us-west", "us-east", "eu", "us-west", "us-east",
+            ]),
+            make_col(vec!["free", "pro", "free", "enterprise", "pro", "free"]),
+            make_col(vec!["1.0", "1.1", "1.2", "1.0", "1.1", "1.2"]),
+            make_col(vec![
+                "debug", "release", "debug", "release", "debug", "release",
+            ]),
+        ],
+    )
+    .unwrap();
+
+    let (_meta, file) = store_mixed_encoding_parquet(&batch, &["env", "region", "tier"]);
+    let path = file.path().to_str().unwrap();
+    let sql = "SELECT env, region, tier, version, build FROM t LIMIT 1";
+    let dict_type =
+        DataType::Dictionary(Box::new(DataType::Int32), Box::new(DataType::Utf8));
+
+    // Flag off: all string columns stay Utf8View.
+    let ctx = SessionContext::new();
+    ctx.register_parquet("t", path, ParquetReadOptions::default())
+        .await
+        .unwrap();
+    let result = ctx.sql(sql).await.unwrap().collect().await.unwrap();
+    assert!(!result.is_empty());
+    let s = result[0].schema();
+    for col in ["env", "region", "tier", "version", "build"] {
+        assert_eq!(
+            s.field_with_name(col).unwrap().data_type(),
+            &DataType::Utf8View
+        );
+    }
+
+    // Flag on: only RLE-encoded columns become Dict; plain columns stay Utf8View.
+    let config = SessionConfig::new().set(
+        "datafusion.execution.parquet.enable_rle_to_dictionary",
+        &ScalarValue::Boolean(Some(true)),
+    );
+    let ctx = SessionContext::new_with_config(config);
+    ctx.register_parquet("t", path, ParquetReadOptions::default())
+        .await
+        .unwrap();
+    let result = ctx.sql(sql).await.unwrap().collect().await.unwrap();
+    assert!(!result.is_empty());
+    let s = result[0].schema();
+    for col in ["env", "region", "tier"] {
+        assert_eq!(s.field_with_name(col).unwrap().data_type(), &dict_type);
+    }
+    for col in ["version", "build"] {
+        assert_eq!(
+            s.field_with_name(col).unwrap().data_type(),
+            &DataType::Utf8View
+        );
+    }
+}
+
 /// Writes `batches` to a temporary parquet file
 pub fn store_parquet(
     batches: Vec<RecordBatch>,
@@ -154,14 +303,10 @@ pub fn store_parquet(
         .into_iter()
         .map(|batch| {
             let mut output = NamedTempFile::new().expect("creating temp file");
-
-            let builder = WriterProperties::builder();
-            let props = builder.build();
-
+            let props = WriterProperties::builder().build();
             let mut writer =
                 ArrowWriter::try_new(&mut output, batch.schema(), Some(props))
                     .expect("creating writer");
-
             writer.write(&batch).expect("Writing batch");
             writer.close().unwrap();
             output

datafusion/datasource-parquet/src/file_format.rs

@@ -363,6 +363,9 @@ impl FileFormat for ParquetFormat {
                     .with_file_metadata_cache(Some(Arc::clone(&file_metadata_cache)))
                     .with_coerce_int96(coerce_int96)
                     .with_coerce_int96_tz(coerce_int96_tz.clone())
+                    .with_enable_rle_to_dictionary(
+                        self.options.global.enable_rle_to_dictionary,
+                    )
                     .fetch_schema_with_location()
                     .await?;
                 Ok::<_, DataFusionError>(result)
@@ -388,7 +391,14 @@ impl FileFormat for ParquetFormat {
         schemas
             .sort_unstable_by(|(location1, _), (location2, _)| location1.cmp(location2));
 
-        let schemas = schemas.into_iter().map(|(_, schema)| schema);
+        // Strip paths; when rle-to-dictionary is on, a column may be
+        // RLE-encoded in one file but plain in another.  Normalise to the
+        // Dictionary type before merging so Schema::try_merge does not see a
+        // type mismatch.
+        let mut schemas: Vec<Schema> = schemas.into_iter().map(|(_, s)| s).collect();
+        if self.options.global.enable_rle_to_dictionary {
+            schemas = crate::schema_coercion::uniform_dict_schemas(schemas);
+        }
 
         let schema = if self.skip_metadata() {
             Schema::try_merge(clear_metadata(schemas))
@@ -510,7 +520,7 @@ impl FileFormat for ParquetFormat {
         source = source.with_parquet_file_reader_factory(cached_parquet_read_factory);
 
         if let Some(metadata_size_hint) = metadata_size_hint {
-            source = source.with_metadata_size_hint(metadata_size_hint)
+            source = source.with_metadata_size_hint(metadata_size_hint);
         }
 
         source = self.set_source_encryption_factory(source, state)?;
@@ -724,6 +734,7 @@ impl From<&ParquetFormatFactory> for protobuf::TableParquetOptions {
             compression_opt: global_options.global.compression.map(|compression| {
                 parquet_options::CompressionOpt::Compression(compression)
             }),
+            enable_rle_to_dictionary: global_options.global.enable_rle_to_dictionary,
             dictionary_enabled_opt: global_options.global.dictionary_enabled.map(|enabled| {
                 parquet_options::DictionaryEnabledOpt::DictionaryEnabled(enabled)
             }),

datafusion/datasource-parquet/src/metadata.rs

@@ -27,7 +27,7 @@ use arrow::datatypes::{DataType, Schema, SchemaRef, TimeUnit};
 use datafusion_common::encryption::FileDecryptionProperties;
 use datafusion_common::stats::Precision;
 use datafusion_common::{
-    ColumnStatistics, DataFusionError, HashMap, Result, ScalarValue, Statistics,
+    ColumnStatistics, DataFusionError, HashMap, HashSet, Result, ScalarValue, Statistics,
     internal_datafusion_err,
 };
 use datafusion_execution::cache::cache_manager::{
@@ -91,6 +91,9 @@ pub struct DFParquetMetadata<'a> {
     pub coerce_int96: Option<TimeUnit>,
     /// Optional timezone applied to INT96-coerced timestamps.
     pub coerce_int96_tz: Option<Arc<str>>,
+    /// When true, columns that are physically RLE_DICTIONARY encoded in the parquet file
+    /// are promoted to `Dictionary(Int32, …)` in the inferred schema.
+    enable_rle_to_dictionary: bool,
 }
 
 impl<'a> DFParquetMetadata<'a> {
@@ -108,9 +111,17 @@ impl<'a> DFParquetMetadata<'a> {
             page_index_policy: None,
             coerce_int96: None,
             coerce_int96_tz: None,
+            enable_rle_to_dictionary: false,
         }
     }
 
+    /// When set, columns that are physically RLE_DICTIONARY encoded are promoted to
+    /// `Dictionary(Int32, …)` in the schema returned by [`Self::fetch_schema`].
+    pub fn with_enable_rle_to_dictionary(mut self, enable: bool) -> Self {
+        self.enable_rle_to_dictionary = enable;
+        self
+    }
+
     /// Set a hint for the number of trailing bytes to prefetch from the end
     /// of the file, equivalent to
     /// [`ParquetMetaDataReader::with_prefetch_hint`].
@@ -384,6 +395,74 @@ impl<'a> DFParquetMetadata<'a> {
                     .coerce()
             })
             .unwrap_or(schema);
+
+        // Promote columns that are physically RLE_DICTIONARY encoded to Dictionary(Int32, …).
+        // This happens before force_view_types so Dict columns survive the Utf8→Utf8View pass.
+        let schema = if self.enable_rle_to_dictionary {
+            let rle_cols: HashSet<String> = {
+                let schema_descr = file_metadata.schema_descr();
+                metadata
+                    .row_groups()
+                    .iter()
+                    .flat_map(|rg| {
+                        rg.columns()
+                            .iter()
+                            .enumerate()
+                            .filter_map(|(col_idx, col)| {
+                                col.dictionary_page_offset()?;
+                                let desc = schema_descr.column(col_idx);
+                                // Only promote top-level columns.  For nested columns
+                                // the Parquet leaf name (e.g. "element" inside a List)
+                                // does not match the Arrow top-level field name (e.g.
+                                // "tags"), so using the leaf name here would miss the
+                                // target field or promote an unrelated one.
+                                let parts = desc.path().parts();
+                                (parts.len() == 1).then(|| parts[0].clone())
+                            })
+                    })
+                    .collect()
+            };
+            if rle_cols.is_empty() {
+                schema
+            } else {
+                let promoted: Vec<_> = schema
+                    .fields()
+                    .iter()
+                    .map(|field| {
+                        if !rle_cols.contains(field.name()) {
+                            return Arc::clone(field);
+                        }
+                        let dict_value_type = match field.data_type() {
+                            DataType::Utf8 | DataType::LargeUtf8 => Some(DataType::Utf8),
+                            DataType::Binary | DataType::LargeBinary => {
+                                Some(DataType::Binary)
+                            }
+                            _ => None,
+                        };
+                        dict_value_type.map_or_else(
+                            || Arc::clone(field),
+                            |value_type| {
+                                Arc::new(
+                                    arrow::datatypes::Field::new(
+                                        field.name(),
+                                        DataType::Dictionary(
+                                            Box::new(DataType::Int32),
+                                            Box::new(value_type),
+                                        ),
+                                        field.is_nullable(),
+                                    )
+                                    .with_metadata(field.metadata().clone()),
+                                )
+                            },
+                        )
+                    })
+                    .collect();
+                Schema::new_with_metadata(promoted, schema.metadata().clone())
+            }
+        } else {
+            schema
+        };
+
         Ok(schema)
     }