fix

Author: zhuqi-lucas

datafusion/core/tests/physical_optimizer/pushdown_sort.rs

@@ -240,7 +240,7 @@ fn test_prefix_match_through_transparent_nodes() {
     OptimizationTest:
       input:
         - SortExec: expr=[a@0 ASC], preserve_partitioning=[false]
-        -   RepartitionExec: partitioning=RoundRobinBatch(10), input_partitions=1, maintains_sort_order=true
+        -   RepartitionExec: partitioning=RoundRobinBatch(10), input_partitions=1
         -     CoalesceBatchesExec: target_batch_size=1024
         -       DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 DESC NULLS LAST, b@1 ASC, c@2 DESC NULLS LAST], file_type=parquet
       output:
@@ -365,7 +365,7 @@ fn test_sort_through_repartition() {
     OptimizationTest:
       input:
         - SortExec: expr=[a@0 DESC NULLS LAST], preserve_partitioning=[false]
-        -   RepartitionExec: partitioning=RoundRobinBatch(10), input_partitions=1, maintains_sort_order=true
+        -   RepartitionExec: partitioning=RoundRobinBatch(10), input_partitions=1
         -     DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet
       output:
         Ok:
@@ -459,7 +459,7 @@ fn test_sort_through_coalesce_partitions() {
       input:
         - SortExec: expr=[a@0 DESC NULLS LAST], preserve_partitioning=[false]
         -   CoalescePartitionsExec
-        -     RepartitionExec: partitioning=RoundRobinBatch(10), input_partitions=1, maintains_sort_order=true
+        -     RepartitionExec: partitioning=RoundRobinBatch(10), input_partitions=1
         -       DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet
       output:
         Ok:
@@ -492,7 +492,7 @@ fn test_complex_plan_with_multiple_operators() {
       input:
         - SortExec: expr=[a@0 DESC NULLS LAST], preserve_partitioning=[false]
         -   CoalescePartitionsExec
-        -     RepartitionExec: partitioning=RoundRobinBatch(10), input_partitions=1, maintains_sort_order=true
+        -     RepartitionExec: partitioning=RoundRobinBatch(10), input_partitions=1
         -       CoalesceBatchesExec: target_batch_size=1024
         -         DataSourceExec: file_groups={1 group: [[x]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet
       output:

datafusion/datasource-parquet/src/opener.rs

@@ -823,8 +823,7 @@ fn should_enable_page_index(
 
 #[cfg(test)]
 mod test {
-    use std::sync::Arc;
-    use crate::{DefaultParquetFileReaderFactory, opener::ParquetOpener};
+    use crate::{opener::ParquetOpener, DefaultParquetFileReaderFactory};
     use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
     use bytes::{BufMut, BytesMut};
     use datafusion_common::{
@@ -840,6 +839,8 @@ mod test {
         PartitionedFile,
     };
     use datafusion_expr::{col, lit};
+    use datafusion_physical_expr::expressions::{Column, Literal};
+    use datafusion_physical_expr::projection::ProjectionExprs;
     use datafusion_physical_expr::{
         expressions::DynamicFilterPhysicalExpr, planner::logical2physical, PhysicalExpr,
     };
@@ -849,8 +850,7 @@ mod test {
     use object_store::{memory::InMemory, path::Path, ObjectStore};
     use parquet::arrow::ArrowWriter;
     use parquet::file::properties::WriterProperties;
-    use datafusion_physical_expr::expressions::{Column, Literal};
-    use datafusion_physical_expr::projection::ProjectionExprs;
+    use std::sync::Arc;
 
     fn constant_int_stats() -> (Statistics, SchemaRef) {
         let schema = Arc::new(Schema::new(vec![
@@ -1398,88 +1398,6 @@ mod test {
         }
     }
 
-        #[tokio::test]
-    async fn test_reverse_scan_row_groups() {
-        use parquet::file::properties::WriterProperties;
-
-        let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
-
-        // Create multiple batches to ensure multiple row groups
-        let batch1 =
-            record_batch!(("a", Int32, vec![Some(1), Some(2), Some(3)])).unwrap();
-        let batch2 =
-            record_batch!(("a", Int32, vec![Some(4), Some(5), Some(6)])).unwrap();
-        let batch3 =
-            record_batch!(("a", Int32, vec![Some(7), Some(8), Some(9)])).unwrap();
-
-        // Write parquet file with multiple row groups
-        // Force small row groups by setting max_row_group_size
-        let props = WriterProperties::builder()
-            .set_max_row_group_size(3) // Force each batch into its own row group
-            .build();
-
-        let data_len = write_parquet_batches(
-            Arc::clone(&store),
-            "test.parquet",
-            vec![batch1.clone(), batch2, batch3],
-            Some(props),
-        )
-        .await;
-
-        let schema = batch1.schema();
-        let file = PartitionedFile::new(
-            "test.parquet".to_string(),
-            u64::try_from(data_len).unwrap(),
-        );
-
-        let make_opener = |reverse_scan: bool| ParquetOpener {
-            projection: Arc::new([0]),
-            partition_index: 0,
-            batch_size: 1024,
-            limit: None,
-            predicate: None,
-            metadata_size_hint: None,
-            metrics: ExecutionPlanMetricsSet::new(),
-            parquet_file_reader_factory: Arc::new(DefaultParquetFileReaderFactory::new(
-                Arc::clone(&store),
-            )),
-            partition_fields: vec![],
-            pushdown_filters: true,
-            reorder_filters: false,
-            enable_page_index: false,
-            enable_bloom_filter: false,
-            enable_limit_pruning: false,
-            schema_adapter_factory: Arc::new(DefaultSchemaAdapterFactory),
-            enable_row_group_stats_pruning: false,
-            coerce_int96: None,
-            #[cfg(feature = "parquet_encryption")]
-            file_decryption_properties: None,
-            expr_adapter_factory: Some(Arc::new(DefaultPhysicalExprAdapterFactory)),
-            #[cfg(feature = "parquet_encryption")]
-            encryption_factory: None,
-            max_predicate_cache_size: None,
-            reverse_row_groups: reverse_scan,
-            logical_file_schema: schema.clone(),
-        };
-
-        // Test normal scan (forward)
-        let opener = make_opener(false);
-        let stream = opener.open(file.clone()).unwrap().await.unwrap();
-        let forward_values = collect_int32_values(stream).await;
-
-        // Test reverse scan
-        let opener = make_opener(true);
-        let stream = opener.open(file.clone()).unwrap().await.unwrap();
-        let reverse_values = collect_int32_values(stream).await;
-
-        // The forward scan should return data in the order written
-        assert_eq!(forward_values, vec![1, 2, 3, 4, 5, 6, 7, 8, 9]);
-
-        // With reverse scan, row groups are reversed, so we expect:
-        // Row group 3 (7,8,9), then row group 2 (4,5,6), then row group 1 (1,2,3)
-        assert_eq!(reverse_values, vec![7, 8, 9, 4, 5, 6, 1, 2, 3]);
-    }
-
     #[tokio::test]
     async fn test_reverse_scan_row_groups() {
         use parquet::file::properties::WriterProperties;
@@ -1506,7 +1424,7 @@ mod test {
             vec![batch1.clone(), batch2, batch3],
             Some(props),
         )
-            .await;
+        .await;
 
         let schema = batch1.schema();
         let file = PartitionedFile::new(

datafusion/datasource-parquet/src/source.rs

@@ -44,14 +44,14 @@ use datafusion_datasource::TableSchema;
 use datafusion_physical_expr::conjunction;
 use datafusion_physical_expr_adapter::DefaultPhysicalExprAdapterFactory;
 use datafusion_physical_expr_common::physical_expr::{fmt_sql, PhysicalExpr};
-use datafusion_physical_plan::SortOrderPushdownResult;
 use datafusion_physical_plan::filter_pushdown::PushedDown;
 use datafusion_physical_plan::filter_pushdown::{
     FilterPushdownPropagation, PushedDownPredicate,
 };
 use datafusion_physical_plan::metrics::Count;
 use datafusion_physical_plan::metrics::ExecutionPlanMetricsSet;
 use datafusion_physical_plan::DisplayFormatType;
+use datafusion_physical_plan::SortOrderPushdownResult;
 
 #[cfg(feature = "parquet_encryption")]
 use datafusion_execution::parquet_encryption::EncryptionFactory;
@@ -60,7 +60,6 @@ use itertools::Itertools;
 use object_store::ObjectStore;
 #[cfg(feature = "parquet_encryption")]
 use parquet::encryption::decrypt::FileDecryptionProperties;
-use datafusion_physical_expr::projection::ProjectionExprs;
 
 /// Execution plan for reading one or more Parquet files.
 ///
@@ -308,7 +307,7 @@ impl ParquetSource {
     pub fn new(table_parquet_options: TableParquetOptions) -> Self {
         Self {
             table_schema: None,
-            table_parquet_options: TableParquetOptions::default(),
+            table_parquet_options,
             metrics: ExecutionPlanMetricsSet::new(),
             predicate: None,
             parquet_file_reader_factory: None,

datafusion/datasource/src/file.rs

@@ -30,14 +30,14 @@ use crate::TableSchema;
 use datafusion_common::config::ConfigOptions;
 use datafusion_common::{not_impl_err, Result, Statistics};
 use datafusion_physical_expr::{LexOrdering, PhysicalExpr};
-use datafusion_physical_plan::SortOrderPushdownResult;
 use datafusion_physical_plan::filter_pushdown::{FilterPushdownPropagation, PushedDown};
 use datafusion_physical_plan::metrics::ExecutionPlanMetricsSet;
 use datafusion_physical_plan::DisplayFormatType;
+use datafusion_physical_plan::SortOrderPushdownResult;
 
+use datafusion_physical_expr::projection::ProjectionExprs;
 use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr;
 use object_store::ObjectStore;
-use datafusion_physical_expr::projection::ProjectionExprs;
 
 /// Helper function to convert any type implementing FileSource to Arc<dyn FileSource>
 pub fn as_file_source<T: FileSource + 'static>(source: T) -> Arc<dyn FileSource> {

datafusion/datasource/src/file_scan_config.rs

@@ -36,7 +36,10 @@ use arrow::{
     datatypes::{ArrowNativeType, DataType, Field, Schema, SchemaRef, UInt16Type},
 };
 use datafusion_common::config::ConfigOptions;
-use datafusion_common::{exec_datafusion_err, exec_err, internal_datafusion_err, ColumnStatistics, Constraints, Result, ScalarValue, Statistics};
+use datafusion_common::{
+    exec_datafusion_err, exec_err, internal_datafusion_err, ColumnStatistics,
+    Constraints, Result, ScalarValue, Statistics,
+};
 use datafusion_execution::{
     object_store::ObjectStoreUrl, SendableRecordBatchStream, TaskContext,
 };
@@ -60,12 +63,14 @@ use datafusion_physical_plan::{
 use datafusion_physical_expr::equivalence::project_orderings;
 use datafusion_physical_plan::coop::cooperative;
 use datafusion_physical_plan::execution_plan::SchedulingType;
+use datafusion_physical_plan::projection::{
+    all_alias_free_columns, new_projections_for_columns,
+};
 use log::{debug, warn};
-use std::{any::Any, fmt::Debug, fmt::Formatter, fmt::Result as FmtResult, sync::Arc};
 use std::borrow::Cow;
 use std::collections::HashMap;
 use std::marker::PhantomData;
-use datafusion_physical_plan::projection::{all_alias_free_columns, new_projections_for_columns};
+use std::{any::Any, fmt::Debug, fmt::Formatter, fmt::Result as FmtResult, sync::Arc};
 
 /// The base configurations for a [`DataSourceExec`], the a physical plan for
 /// any given file format.

datafusion/datasource/src/source.rs

@@ -40,10 +40,10 @@ use datafusion_common::{Constraints, Result, Statistics};
 use datafusion_execution::{SendableRecordBatchStream, TaskContext};
 use datafusion_physical_expr::{EquivalenceProperties, Partitioning, PhysicalExpr};
 use datafusion_physical_expr_common::sort_expr::{LexOrdering, PhysicalSortExpr};
-use datafusion_physical_plan::SortOrderPushdownResult;
 use datafusion_physical_plan::filter_pushdown::{
     ChildPushdownResult, FilterPushdownPhase, FilterPushdownPropagation, PushedDown,
 };
+use datafusion_physical_plan::SortOrderPushdownResult;
 
 /// A source of data, typically a list of files or memory
 ///

datafusion/physical-optimizer/src/pushdown_sort.rs

@@ -50,12 +50,12 @@
 //! - Complete Sort elimination when ordering is guaranteed
 
 use crate::PhysicalOptimizerRule;
-use datafusion_common::Result;
 use datafusion_common::config::ConfigOptions;
 use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
+use datafusion_common::Result;
+use datafusion_physical_plan::sorts::sort::SortExec;
 use datafusion_physical_plan::ExecutionPlan;
 use datafusion_physical_plan::SortOrderPushdownResult;
-use datafusion_physical_plan::sorts::sort::SortExec;
 use std::sync::Arc;
 
 /// A PhysicalOptimizerRule that attempts to push down sort requirements to data sources.

datafusion/physical-plan/src/coalesce_partitions.rs

@@ -30,7 +30,7 @@ use super::{
 use crate::execution_plan::{CardinalityEffect, EvaluationType, SchedulingType};
 use crate::filter_pushdown::{FilterDescription, FilterPushdownPhase};
 
-use crate::projection::{ProjectionExec, make_with_child};
+use crate::projection::{make_with_child, ProjectionExec};
 use crate::sort_pushdown::SortOrderPushdownResult;
 use crate::{DisplayFormatType, ExecutionPlan, Partitioning};
 use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr;

datafusion/physical-plan/src/repartition/mod.rs

@@ -27,7 +27,10 @@ use std::{any::Any, vec};
 
 use super::common::SharedMemoryReservation;
 use super::metrics::{self, ExecutionPlanMetricsSet, MetricBuilder, MetricsSet};
-use super::{DisplayAs, ExecutionPlanProperties, RecordBatchStream, SendableRecordBatchStream, SortOrderPushdownResult};
+use super::{
+    DisplayAs, ExecutionPlanProperties, RecordBatchStream, SendableRecordBatchStream,
+    SortOrderPushdownResult,
+};
 use crate::execution_plan::{CardinalityEffect, EvaluationType, SchedulingType};
 use crate::hash_utils::create_hashes;
 use crate::metrics::{BaselineMetrics, SpillMetrics};

datafusion/proto-common/src/generated/pbjson.rs

@@ -1107,9 +1107,6 @@ impl serde::Serialize for ColumnStats {
         if let Some(v) = self.distinct_count.as_ref() {
             struct_ser.serialize_field("distinctCount", v)?;
         }
-        if let Some(v) = self.byte_size.as_ref() {
-            struct_ser.serialize_field("byteSize", v)?;
-        }
         struct_ser.end()
     }
 }
@@ -1130,8 +1127,6 @@ impl<'de> serde::Deserialize<'de> for ColumnStats {
             "nullCount",
             "distinct_count",
             "distinctCount",
-            "byte_size",
-            "byteSize",
         ];
 
         #[allow(clippy::enum_variant_names)]
@@ -1141,7 +1136,6 @@ impl<'de> serde::Deserialize<'de> for ColumnStats {
             SumValue,
             NullCount,
             DistinctCount,
-            ByteSize,
         }
         impl<'de> serde::Deserialize<'de> for GeneratedField {
             fn deserialize<D>(deserializer: D) -> std::result::Result<GeneratedField, D::Error>
@@ -1168,7 +1162,6 @@ impl<'de> serde::Deserialize<'de> for ColumnStats {
                             "sumValue" | "sum_value" => Ok(GeneratedField::SumValue),
                             "nullCount" | "null_count" => Ok(GeneratedField::NullCount),
                             "distinctCount" | "distinct_count" => Ok(GeneratedField::DistinctCount),
-                            "byteSize" | "byte_size" => Ok(GeneratedField::ByteSize),
                             _ => Err(serde::de::Error::unknown_field(value, FIELDS)),
                         }
                     }
@@ -1225,12 +1218,6 @@ impl<'de> serde::Deserialize<'de> for ColumnStats {
                             }
                             distinct_count__ = map_.next_value()?;
                         }
-                        GeneratedField::ByteSize => {
-                            if byte_size__.is_some() {
-                                return Err(serde::de::Error::duplicate_field("byteSize"));
-                            }
-                            byte_size__ = map_.next_value()?;
-                        }
                     }
                 }
                 Ok(ColumnStats {