Revert "Upgrade arrow/parquet to 56.0.0 (apache#16690)"

This reverts commit fa1f8c1.

Author: LiaCastaneda

Cargo.lock

@@ -90,20 +90,20 @@ ahash = { version = "0.8", default-features = false, features = [
     "runtime-rng",
 ] }
 apache-avro = { version = "0.20", default-features = false }
-arrow = { version = "56.0.0", features = [
+arrow = { version = "55.2.0", features = [
     "prettyprint",
     "chrono-tz",
 ] }
-arrow-buffer = { version = "56.0.0", default-features = false }
-arrow-flight = { version = "56.0.0", features = [
+arrow-buffer = { version = "55.2.0", default-features = false }
+arrow-flight = { version = "55.2.0", features = [
     "flight-sql-experimental",
 ] }
-arrow-ipc = { version = "56.0.0", default-features = false, features = [
+arrow-ipc = { version = "55.2.0", default-features = false, features = [
     "lz4",
 ] }
-arrow-ord = { version = "56.0.0", default-features = false }
-arrow-schema = { version = "56.0.0", default-features = false }
-async-trait = "0.1.89"
+arrow-ord = { version = "55.2.0", default-features = false }
+arrow-schema = { version = "55.2.0", default-features = false }
+async-trait = "0.1.88"
 bigdecimal = "0.4.8"
 bytes = "1.10"
 chrono = { version = "0.4.41", default-features = false }
@@ -157,7 +157,7 @@ itertools = "0.14"
 log = "^0.4"
 object_store = { version = "0.12.3", default-features = false }
 parking_lot = "0.12"
-parquet = { version = "56.0.0", default-features = false, features = [
+parquet = { version = "55.2.0", default-features = false, features = [
     "arrow",
     "async",
     "object_store",

Cargo.toml

@@ -81,7 +81,7 @@ serde_json = { workspace = true }
 tempfile = { workspace = true }
 test-utils = { path = "../test-utils" }
 tokio = { workspace = true, features = ["rt-multi-thread", "parking_lot"] }
-tonic = "0.13.1"
+tonic = "0.12.1"
 tracing = { version = "0.1" }
 tracing-subscriber = { version = "0.3" }
 url = { workspace = true }

datafusion-examples/Cargo.toml

@@ -71,7 +71,7 @@ log = { workspace = true }
 object_store = { workspace = true, optional = true }
 parquet = { workspace = true, optional = true, default-features = true }
 paste = "1.0.15"
-pyo3 = { version = "0.25", optional = true }
+pyo3 = { version = "0.24.2", optional = true }
 recursive = { workspace = true, optional = true }
 sqlparser = { workspace = true }
 tokio = { workspace = true }

datafusion/common/Cargo.toml

@@ -602,6 +602,13 @@ config_namespace! {
         /// default parquet writer setting
         pub statistics_enabled: Option<String>, transform = str::to_lowercase, default = Some("page".into())
 
+        /// (writing) Sets max statistics size for any column. If NULL, uses
+        /// default parquet writer setting
+        /// max_statistics_size is deprecated, currently it is not being used
+        // TODO: remove once deprecated
+        #[deprecated(since = "45.0.0", note = "Setting does not do anything")]
+        pub max_statistics_size: Option<usize>, default = Some(4096)
+
         /// (writing) Target maximum number of rows in each row group (defaults to 1M
         /// rows). Writing larger row groups requires more memory to write, but
         /// can get better compression and be faster to read.
@@ -613,9 +620,9 @@ config_namespace! {
         /// (writing) Sets column index truncate length
         pub column_index_truncate_length: Option<usize>, default = Some(64)
 
-        /// (writing) Sets statistics truncate length. If NULL, uses
+        /// (writing) Sets statictics truncate length. If NULL, uses
         /// default parquet writer setting
-        pub statistics_truncate_length: Option<usize>, default = Some(64)
+        pub statistics_truncate_length: Option<usize>, default = None
 
         /// (writing) Sets best effort maximum number of rows in data page
         pub data_page_row_count_limit: usize, default = 20_000
@@ -2134,6 +2141,13 @@ config_namespace_with_hashmap! {
         /// Sets bloom filter number of distinct values. If NULL, uses
         /// default parquet options
         pub bloom_filter_ndv: Option<u64>, default = None
+
+        /// Sets max statistics size for the column path. If NULL, uses
+        /// default parquet options
+        /// max_statistics_size is deprecated, currently it is not being used
+        // TODO: remove once deprecated
+        #[deprecated(since = "45.0.0", note = "Setting does not do anything")]
+        pub max_statistics_size: Option<usize>, default = None
     }
 }
 

datafusion/common/src/config.rs

@@ -35,7 +35,7 @@ use parquet::{
         metadata::KeyValue,
         properties::{
             EnabledStatistics, WriterProperties, WriterPropertiesBuilder, WriterVersion,
-            DEFAULT_STATISTICS_ENABLED,
+            DEFAULT_MAX_STATISTICS_SIZE, DEFAULT_STATISTICS_ENABLED,
         },
     },
     schema::types::ColumnPath,
@@ -160,6 +160,16 @@ impl TryFrom<&TableParquetOptions> for WriterPropertiesBuilder {
                 builder =
                     builder.set_column_bloom_filter_ndv(path.clone(), bloom_filter_ndv);
             }
+
+            // max_statistics_size is deprecated, currently it is not being used
+            // TODO: remove once deprecated
+            #[allow(deprecated)]
+            if let Some(max_statistics_size) = options.max_statistics_size {
+                builder = {
+                    #[allow(deprecated)]
+                    builder.set_column_max_statistics_size(path, max_statistics_size)
+                }
+            }
         }
 
         Ok(builder)
@@ -208,6 +218,7 @@ impl ParquetOptions {
             dictionary_enabled,
             dictionary_page_size_limit,
             statistics_enabled,
+            max_statistics_size,
             max_row_group_size,
             created_by,
             column_index_truncate_length,
@@ -253,6 +264,13 @@ impl ParquetOptions {
             .set_data_page_row_count_limit(*data_page_row_count_limit)
             .set_bloom_filter_enabled(*bloom_filter_on_write);
 
+        builder = {
+            #[allow(deprecated)]
+            builder.set_max_statistics_size(
+                max_statistics_size.unwrap_or(DEFAULT_MAX_STATISTICS_SIZE),
+            )
+        };
+
         if let Some(bloom_filter_fpp) = bloom_filter_fpp {
             builder = builder.set_bloom_filter_fpp(*bloom_filter_fpp);
         };
@@ -445,10 +463,12 @@ mod tests {
     fn column_options_with_non_defaults(
         src_col_defaults: &ParquetOptions,
     ) -> ParquetColumnOptions {
+        #[allow(deprecated)] // max_statistics_size
         ParquetColumnOptions {
             compression: Some("zstd(22)".into()),
             dictionary_enabled: src_col_defaults.dictionary_enabled.map(|v| !v),
             statistics_enabled: Some("none".into()),
+            max_statistics_size: Some(72),
             encoding: Some("RLE".into()),
             bloom_filter_enabled: Some(true),
             bloom_filter_fpp: Some(0.72),
@@ -473,6 +493,7 @@ mod tests {
             dictionary_enabled: Some(!defaults.dictionary_enabled.unwrap_or(false)),
             dictionary_page_size_limit: 42,
             statistics_enabled: Some("chunk".into()),
+            max_statistics_size: Some(42),
             max_row_group_size: 42,
             created_by: "wordy".into(),
             column_index_truncate_length: Some(42),
@@ -530,6 +551,7 @@ mod tests {
             ),
             bloom_filter_fpp: bloom_filter_default_props.map(|p| p.fpp),
             bloom_filter_ndv: bloom_filter_default_props.map(|p| p.ndv),
+            max_statistics_size: Some(props.max_statistics_size(&col)),
         }
     }
 
@@ -586,6 +608,7 @@ mod tests {
                 compression: default_col_props.compression,
                 dictionary_enabled: default_col_props.dictionary_enabled,
                 statistics_enabled: default_col_props.statistics_enabled,
+                max_statistics_size: default_col_props.max_statistics_size,
                 bloom_filter_on_write: default_col_props
                     .bloom_filter_enabled
                     .unwrap_or_default(),

datafusion/common/src/file_options/parquet_writer.rs

@@ -904,10 +904,11 @@ pub fn dict_from_values<K: ArrowDictionaryKeyType>(
         .map(|index| {
             if values_array.is_valid(index) {
                 let native_index = K::Native::from_usize(index).ok_or_else(|| {
-                    _internal_datafusion_err!(
-                        "Can not create index of type {} from value {index}",
-                        K::DATA_TYPE
-                    )
+                    DataFusionError::Internal(format!(
+                        "Can not create index of type {} from value {}",
+                        K::DATA_TYPE,
+                        index
+                    ))
                 })?;
                 Ok(Some(native_index))
             } else {
@@ -2202,16 +2203,6 @@ impl ScalarValue {
         }
 
         let array: ArrayRef = match &data_type {
-            DataType::Decimal32(_precision, _scale) => {
-                return _not_impl_err!(
-                    "Decimal32 not supported in ScalarValue::iter_to_array"
-                );
-            }
-            DataType::Decimal64(_precision, _scale) => {
-                return _not_impl_err!(
-                    "Decimal64 not supported in ScalarValue::iter_to_array"
-                );
-            }
             DataType::Decimal128(precision, scale) => {
                 let decimal_array =
                     ScalarValue::iter_to_decimal_array(scalars, *precision, *scale)?;

datafusion/common/src/scalar/mod.rs

@@ -407,10 +407,7 @@ impl From<DataType> for NativeType {
             DataType::Union(union_fields, _) => {
                 Union(LogicalUnionFields::from(&union_fields))
             }
-            DataType::Decimal32(p, s)
-            | DataType::Decimal64(p, s)
-            | DataType::Decimal128(p, s)
-            | DataType::Decimal256(p, s) => Decimal(p, s),
+            DataType::Decimal128(p, s) | DataType::Decimal256(p, s) => Decimal(p, s),
             DataType::Map(field, _) => Map(Arc::new(field.as_ref().into())),
             DataType::Dictionary(_, data_type) => data_type.as_ref().clone().into(),
             DataType::RunEndEncoded(_, field) => field.data_type().clone().into(),

datafusion/common/src/types/native.rs

@@ -319,9 +319,14 @@ async fn write_parquet_file(
     row_groups: Vec<Vec<String>>,
 ) -> Bytes {
     let mut buf = BytesMut::new().writer();
-    let props = WriterProperties::builder()
-        .set_statistics_enabled(EnabledStatistics::Chunk) // row group level
-        .set_statistics_truncate_length(truncation_length);
+    let mut props = WriterProperties::builder();
+    if let Some(truncation_length) = truncation_length {
+        props = {
+            #[allow(deprecated)]
+            props.set_max_statistics_size(truncation_length)
+        }
+    }
+    props = props.set_statistics_enabled(EnabledStatistics::Chunk); // row group level
     let props = props.build();
     {
         let mut writer =

datafusion/core/tests/fuzz_cases/pruning.rs

@@ -110,11 +110,11 @@ struct ContextWithParquet {
 
 /// The output of running one of the test cases
 struct TestOutput {
-    /// The input query SQL
+    /// The input string
     sql: String,
     /// Execution metrics for the Parquet Scan
     parquet_metrics: MetricsSet,
-    /// number of actual rows in results
+    /// number of rows in results
     result_rows: usize,
     /// the contents of the input, as a string
     pretty_input: String,