Merge branch 'main' into fix/25941-preserve-timezone-cast

Author: mikamikasuki

SECURITY.md

@@ -0,0 +1,102 @@
+<!---
+  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.
+-->
+
+# Security Policy
+
+This document outlines the security model for Apache DataFusion and how to
+report vulnerabilities.
+
+This model also applies to the [datafusion-cli] command line tool, which is
+a thin wrapper around the DataFusion library.
+
+## Security Model
+
+DataFusion is a low level library, designed to be embedded in applications
+that have their own security model. This section describes DataFusion's own
+model: what counts as a bug versus a vulnerability.
+
+In general, crashes, panics, hangs, and excessive resource consumption
+(memory, CPU, or disk) are treated as **bugs**, not vulnerabilities, unless
+they are **exploitable** and could let an attacker:
+
+* Execute arbitrary code (Remote Code Execution), or
+* Exfiltrate sensitive information from process memory (Information Disclosure).
+
+If the exploitation path is unclear, please report the issue as a bug rather
+than a vulnerability. The sections below describe what DataFusion treats as
+trusted input in a few specific areas.
+
+### SQL and DataFrame Queries
+
+SQL and DataFrame queries are executable code, comparable to a scripting
+language: a query can legitimately read files, open network connections
+(e.g. via `CREATE EXTERNAL TABLE`), and consume significant CPU, memory, or
+disk. Running such a query is not, on its own, a vulnerability.
+
+It is the embedding application's responsibility to decide whether a query is
+safe to run (e.g. validating externally supplied SQL text or URLs) and to
+sandbox untrusted queries at the OS/container level if needed. APIs such as
+[`SQLOptions::with_allow_dml`] can help restrict what a query is allowed to
+do, but do not guarantee that all input is safe to execute.
+
+### Data Files
+
+Format readers (e.g. Parquet, CSV, JSON, Avro, and Arrow IPC) assume a
+well-formed file from a trusted writer; use the [arrow validation APIs] to
+validate an untrusted file's structure. Issues due to malformed files, and
+well-formed files that contain unexpected or adversarial *data*, are bugs rather
+than vulnerabilities, as explained above.
+
+### Serialized Plans
+
+Serialized plans, such as [Substrait] and [`datafusion-proto`], are treated
+as trusted input. If received from an untrusted source, they should be
+validated before being passed to DataFusion for execution.
+
+### Extensions and Other Uses of Public APIs
+
+Code that uses DataFusion's public extension APIs (e.g. user defined functions,
+`TableProvider`, `ExecutionPlan`) is trusted to uphold their contracts (for
+example, that any `ArrayRef` is a valid Arrow array). Issues arising from
+violating the API contracts are not considered a DataFusion vulnerability.
+
+## Reporting a Bug
+
+We treat all bugs seriously and welcome help fixing them. If you find a bug
+that does not meet the criteria for a security vulnerability, please report it
+in the [public issue tracker](https://github.com/apache/datafusion/issues/).
+
+## Reporting a Vulnerability
+
+For security vulnerabilities **do not file a public issue.** Follow the [ASF
+security reporting process] by emailing
+[security@apache.org](mailto:security@apache.org).
+
+Include in your report:
+
+- A clear description and minimal reproducer.
+- Affected crates and versions.
+- Potential impact.
+
+[datafusion-cli]: https://datafusion.apache.org/user-guide/cli/index.html
+[`sqloptions::with_allow_dml`]: https://docs.rs/datafusion/latest/datafusion/execution/context/struct.SQLOptions.html#method.with_allow_dml
+[arrow validation apis]: https://docs.rs/arrow/latest/arrow/array/struct.ArrayData.html#method.validate_full
+[substrait]: https://docs.rs/datafusion-substrait/latest/datafusion_substrait/
+[`datafusion-proto`]: https://docs.rs/datafusion-proto/latest/datafusion_proto/
+[asf security reporting process]: https://www.apache.org/security/#reporting-a-vulnerability

datafusion/common/src/config.rs

@@ -1426,6 +1426,17 @@ config_namespace! {
         /// Defaults to 20.
         pub max_in_list_size: usize, default = 20
 
+        /// (reading) If true, top-level string and binary Parquet columns with
+        /// dictionary pages are inferred and scanned as
+        /// `Dictionary<Int32, Utf8>` / `Dictionary<Int32, Binary>` instead of
+        /// their plain value type.
+        ///
+        /// This applies only when DataFusion infers the table schema. Tables with
+        /// a user-supplied schema are not promoted because the Parquet footer is
+        /// not read at DDL time, so dictionary pages cannot be detected per column.
+        /// See <https://github.com/apache/datafusion/issues/24112>
+        pub enable_rle_to_dictionary: bool, default = false
+
         /// (reading) Which byte range of a split file reads each row group.
         /// `start_offset` picks the range containing the row group's start.
         /// `midpoint` picks the range containing its midpoint, as Spark does,

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: _,
             row_group_range_assignment: _,
         } = self;
 
@@ -429,6 +430,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,
         }
     }
 
@@ -558,6 +560,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/datasource-parquet/src/file_format.rs

@@ -356,14 +356,16 @@ impl FileFormat for ParquetFormat {
                     &object.location,
                 )
                 .await?;
-                let result = DFParquetMetadata::new(store.as_ref(), object)
+                let meta = DFParquetMetadata::new(store.as_ref(), object)
                     .with_metadata_size_hint(self.metadata_size_hint())
                     .with_decryption_properties(file_decryption_properties)
                     .with_file_metadata_cache(Some(Arc::clone(&file_metadata_cache)))
                     .with_coerce_int96(coerce_int96)
                     .with_coerce_int96_tz(coerce_int96_tz.clone())
-                    .fetch_schema_with_location()
-                    .await?;
+                    .with_enable_rle_to_dictionary(
+                        self.options.global.enable_rle_to_dictionary,
+                    );
+                let result = meta.fetch_schema_with_location().await?;
                 Ok::<_, DataFusionError>(result)
             })
             .boxed() // Workaround https://github.com/rust-lang/rust/issues/64552
@@ -398,7 +400,12 @@ impl FileFormat for ParquetFormat {
         }
         drop(seen);
 
-        let schemas = schemas.into_iter().map(|(_, schema)| schema);
+        // Normalize dict-promoted schemas before merging so mixed dict/plain files merge cleanly.
+        let mut schemas: Vec<Schema> =
+            schemas.into_iter().map(|(_, schema)| schema).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))
@@ -520,7 +527,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)?;
@@ -672,6 +679,7 @@ impl From<&ParquetFormatFactory> for protobuf::TableParquetOptions {
             compression_opt: global_options.global.compression.map(|compression| {
                 parquet_options::CompressionOpt::Compression(compression.to_string())
             }),
+            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

@@ -23,11 +23,11 @@ use crate::{Int96Coercer, apply_file_schema_type_coercions};
 use arrow::array::{Array, ArrayRef, BooleanArray};
 use arrow::compute::kernels::cmp::eq;
 use arrow::compute::{and, sum};
-use arrow::datatypes::{DataType, Schema, SchemaRef, TimeUnit};
+use arrow::datatypes::{DataType, Field, 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::{
@@ -152,6 +152,8 @@ pub struct DFParquetMetadata<'a> {
     pub coerce_int96: Option<TimeUnit>,
     /// Optional timezone applied to INT96-coerced timestamps.
     pub coerce_int96_tz: Option<Arc<str>>,
+    /// If true, promote string/binary columns with dictionary pages to `Dictionary(Int32, ...)`.
+    enable_rle_to_dictionary: bool,
 }
 
 impl<'a> DFParquetMetadata<'a> {
@@ -169,9 +171,16 @@ impl<'a> DFParquetMetadata<'a> {
             page_index_policy: None,
             coerce_int96: None,
             coerce_int96_tz: None,
+            enable_rle_to_dictionary: false,
         }
     }
 
+    /// Promote string/binary columns with dictionary pages to `Dictionary(Int32, ...)`.
+    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`].
@@ -449,6 +458,68 @@ impl<'a> DFParquetMetadata<'a> {
                     .coerce()
             })
             .unwrap_or(schema);
+
+        let schema = if self.enable_rle_to_dictionary {
+            let schema_descr = file_metadata.schema_descr();
+            // Top-level columns that have a dictionary page in at least one row group.
+            let dict_cols: HashSet<String> = metadata
+                .row_groups()
+                .iter()
+                .flat_map(|rg| {
+                    rg.columns()
+                        .iter()
+                        .enumerate()
+                        .filter_map(|(col_idx, col)| {
+                            col.dictionary_page_offset()?;
+                            let col_desc = schema_descr.column(col_idx);
+                            let parts = col_desc.path().parts();
+                            // Skip nested columns: their leaf name doesn't match the
+                            // Arrow top-level field name.
+                            (parts.len() == 1).then(|| parts[0].clone())
+                        })
+                })
+                .collect();
+            if dict_cols.is_empty() {
+                schema
+            } else {
+                let promoted: Vec<_> = schema
+                    .fields()
+                    .iter()
+                    .map(|field| {
+                        if !dict_cols.contains(field.name()) {
+                            return Arc::clone(field);
+                        }
+                        let dict_value_type = match field.data_type() {
+                            DataType::Utf8 => Some(DataType::Utf8),
+                            DataType::LargeUtf8 => Some(DataType::LargeUtf8),
+                            DataType::Binary => Some(DataType::Binary),
+                            DataType::LargeBinary => Some(DataType::LargeBinary),
+                            _ => None,
+                        };
+                        dict_value_type.map_or_else(
+                            || Arc::clone(field),
+                            |value_type| {
+                                Arc::new(
+                                    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)
     }