Revert arrow upgrade and related changes (#50)

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

This reverts commit fa1f8c1.

* Revert "refactor: use upstream inline_key_fast (apache#17044)"

This reverts commit 71b92bc.

* Revert "fix: respect inexact flags in row group metadata (apache#16412)"

This reverts commit afc90f7.

* Revert "Test grouping by FixedSizeList (apache#17415)"

This reverts commit 03f39e5.

* Spelling (got reverted)

* Also allow Byt from tests

* Adjust sqllogictests

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-testing

@@ -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.
@@ -615,7 +622,7 @@ config_namespace! {
 
         /// (writing) Sets statistics 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/Cargo.toml

@@ -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/config.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/file_options/parquet_writer.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/scalar/mod.rs

@@ -523,23 +523,11 @@ mod tests {
         let dic_array = DictionaryArray::<Int32Type>::try_new(keys, Arc::new(values))?;
         let c_dic: ArrayRef = Arc::new(dic_array);
 
-        // Data for column string_truncation: ["a".repeat(128), null, "b".repeat(128), null]
-        let string_truncation: ArrayRef = Arc::new(StringArray::from(vec![
-            Some("a".repeat(128)),
-            None,
-            Some("b".repeat(128)),
-            None,
-        ]));
-
-        let batch1 = RecordBatch::try_from_iter(vec![
-            ("c_dic", c_dic),
-            ("string_truncation", string_truncation),
-        ])?;
+        let batch1 = RecordBatch::try_from_iter(vec![("c_dic", c_dic)])?;
 
         // Use store_parquet to write each batch to its own file
         // . batch1 written into first file and includes:
         //    - column c_dic that has 4 rows with no null. Stats min and max of dictionary column is available.
-        //    - column string_truncation that has 4 rows with 2 nulls. Stats min and max of string column is available but not exact.
         let store = Arc::new(RequestCountingObjectStore::new(Arc::new(
             LocalFileSystem::new(),
         )));
@@ -575,19 +563,6 @@ mod tests {
             Precision::Exact(Utf8(Some("a".into())))
         );
 
-        // column string_truncation
-        let string_truncation_stats = &stats.column_statistics[1];
-
-        assert_eq!(string_truncation_stats.null_count, Precision::Exact(2));
-        assert_eq!(
-            string_truncation_stats.max_value,
-            Precision::Inexact(ScalarValue::Utf8View(Some("b".repeat(63) + "c")))
-        );
-        assert_eq!(
-            string_truncation_stats.min_value,
-            Precision::Inexact(ScalarValue::Utf8View(Some("a".repeat(64))))
-        );
-
         Ok(())
     }