Initial commit

Author: rok

Cargo.lock

@@ -89,20 +89,21 @@ ahash = { version = "0.8", default-features = false, features = [
     "runtime-rng",
 ] }
 apache-avro = { version = "0.17", default-features = false }
-arrow = { version = "55.2.0", features = [
+arrow = { git = "https://github.com/rok/arrow-rs.git", branch = "multi-threaded_encrypted_writing", features = [
     "prettyprint",
     "chrono-tz",
 ] }
-arrow-buffer = { version = "55.2.0", default-features = false }
-arrow-flight = { version = "55.2.0", features = [
+
+arrow-buffer = { git = "https://github.com/rok/arrow-rs.git", branch = "multi-threaded_encrypted_writing", default-features = false }
+arrow-flight = { git = "https://github.com/rok/arrow-rs.git", features = [
     "flight-sql-experimental",
 ] }
-arrow-ipc = { version = "55.2.0", default-features = false, features = [
+arrow-ipc = { git = "https://github.com/rok/arrow-rs.git", branch = "multi-threaded_encrypted_writing", default-features = false, features = [
     "lz4",
     "zstd",
 ] }
-arrow-ord = { version = "55.2.0", default-features = false }
-arrow-schema = { version = "55.2.0", default-features = false }
+arrow-ord = { git = "https://github.com/rok/arrow-rs.git", branch = "multi-threaded_encrypted_writing", default-features = false }
+arrow-schema = { git = "https://github.com/rok/arrow-rs.git", branch = "multi-threaded_encrypted_writing", default-features = false }
 async-trait = "0.1.88"
 bigdecimal = "0.4.8"
 bytes = "1.10"
@@ -155,7 +156,7 @@ itertools = "0.14"
 log = "^0.4"
 object_store = { version = "0.12.2", default-features = false }
 parking_lot = "0.12"
-parquet = { version = "55.2.0", default-features = false, features = [
+parquet = { git = "https://github.com/rok/arrow-rs.git", branch = "multi-threaded_encrypted_writing", default-features = false, features = [
     "arrow",
     "async",
     "object_store",

Cargo.toml

@@ -32,18 +32,6 @@ rust-version = { workspace = true }
 [lints]
 workspace = true
 
-[[example]]
-name = "flight_sql_server"
-path = "examples/flight/flight_sql_server.rs"
-
-[[example]]
-name = "flight_server"
-path = "examples/flight/flight_server.rs"
-
-[[example]]
-name = "flight_client"
-path = "examples/flight/flight_client.rs"
-
 [[example]]
 name = "dataframe_to_s3"
 path = "examples/external_dependency/dataframe-to-s3.rs"

datafusion-examples/Cargo.toml

@@ -65,7 +65,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.24.2", optional = true }
+pyo3 = { version = "0.25.1", optional = true }
 recursive = { workspace = true, optional = true }
 sqlparser = { workspace = true }
 tokio = { workspace = true }

datafusion/common/Cargo.toml

@@ -25,6 +25,8 @@ use crate::{
     DataFusionError, Result, _internal_datafusion_err,
 };
 
+pub const DEFAULT_MAX_STATISTICS_SIZE: usize = 4096;
+
 use arrow::datatypes::Schema;
 // TODO: handle once deprecated
 #[allow(deprecated)]
@@ -35,7 +37,7 @@ use parquet::{
         metadata::KeyValue,
         properties::{
             EnabledStatistics, WriterProperties, WriterPropertiesBuilder, WriterVersion,
-            DEFAULT_MAX_STATISTICS_SIZE, DEFAULT_STATISTICS_ENABLED,
+            DEFAULT_STATISTICS_ENABLED,
         },
     },
     schema::types::ColumnPath,
@@ -167,13 +169,13 @@ impl TryFrom<&TableParquetOptions> for WriterPropertiesBuilder {
 
             // 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)
-                }
-            }
+            // #[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)
@@ -222,7 +224,7 @@ impl ParquetOptions {
             dictionary_enabled,
             dictionary_page_size_limit,
             statistics_enabled,
-            max_statistics_size,
+            max_statistics_size: _max_statistics_size,
             max_row_group_size,
             created_by,
             column_index_truncate_length,
@@ -268,12 +270,12 @@ 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),
-            )
-        };
+        // 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);
@@ -557,7 +559,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)),
+            max_statistics_size: Some(DEFAULT_MAX_STATISTICS_SIZE),
         }
     }
 

datafusion/common/src/file_options/parquet_writer.rs

@@ -2134,7 +2134,9 @@ impl ScalarValue {
             | DataType::Time64(TimeUnit::Millisecond)
             | DataType::RunEndEncoded(_, _)
             | DataType::ListView(_)
-            | DataType::LargeListView(_) => {
+            | DataType::LargeListView(_)
+            | DataType::Decimal32(_, _)
+            | DataType::Decimal64(_, _) => {
                 return _not_impl_err!(
                     "Unsupported creation of {:?} array from ScalarValue {:?}",
                     data_type,

datafusion/common/src/scalar/mod.rs

@@ -407,7 +407,10 @@ impl From<DataType> for NativeType {
             DataType::Union(union_fields, _) => {
                 Union(LogicalUnionFields::from(&union_fields))
             }
-            DataType::Decimal128(p, s) | DataType::Decimal256(p, s) => Decimal(p, s),
+            DataType::Decimal32(p, s)
+            | DataType::Decimal64(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

@@ -277,6 +277,7 @@ mod tests {
         // Write encrypted parquet using write_parquet
         let mut options = TableParquetOptions::default();
         options.crypto.file_encryption = Some((&encrypt).into());
+        options.global.allow_single_file_parallelism = true;
 
         df.write_parquet(
             tempfile_str.as_str(),

datafusion/core/src/dataframe/parquet.rs

@@ -314,18 +314,12 @@ async fn execute_with_predicate(
 }
 
 async fn write_parquet_file(
-    truncation_length: Option<usize>,
+    _truncation_length: Option<usize>,
     schema: Arc<Schema>,
     row_groups: Vec<Vec<String>>,
 ) -> Bytes {
     let mut buf = BytesMut::new().writer();
     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();
     {

datafusion/core/tests/fuzz_cases/pruning.rs

@@ -237,6 +237,8 @@ fn default_field_name(dt: &DataType) -> &str {
         }
         DataType::Decimal128(_, _) => "decimal",
         DataType::Decimal256(_, _) => "decimal",
+        DataType::Decimal32(_, _) => "decimal",
+        DataType::Decimal64(_, _) => "decimal",
     }
 }