andygrove commented on code in PR #5331:
URL: https://github.com/apache/datafusion-comet/pull/5331#discussion_r4233156707
##########
native/core/src/execution/operators/iceberg_scan.rs:
##########
@@ -707,6 +768,141 @@ mod tests {
.unwrap();
}
+ fn int_schema() -> arrow::datatypes::SchemaRef {
+ use arrow::datatypes::{DataType, Field, Schema};
+ Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, false)]))
+ }
+
+ fn single_col_ordering() -> Option<datafusion::physical_expr::LexOrdering>
{
+ use arrow::compute::SortOptions;
+ use datafusion::physical_expr::expressions::Column;
+ use datafusion::physical_expr::{LexOrdering, PhysicalSortExpr};
+ LexOrdering::new(vec![PhysicalSortExpr {
+ expr: Arc::new(Column::new("a", 0)),
+ options: SortOptions::default(),
+ }])
+ }
+
+ // Builds a scan over three empty-delete tasks with the given reported
ordering.
+ fn exec_with_ordering(
+ ordering: Option<datafusion::physical_expr::LexOrdering>,
+ ) -> IcebergScanExec {
+ use std::collections::HashMap;
+ let tasks = vec![
+ task_with_deletes(vec![]),
+ task_with_deletes(vec![]),
+ task_with_deletes(vec![]),
+ ];
+ IcebergScanExec::new(
+ "metadata.json".to_string(),
+ int_schema(),
+ HashMap::new(),
+ "cat".to_string(),
+ tasks,
+ 1,
+ ordering,
+ )
+ .unwrap()
+ }
+
+ // A reported ordering turns the scan into a multi-partition operator (one
partition per task)
+ // so a SortPreservingMergeExec above can k-way merge the per-file sorted
streams.
+ #[test]
+ fn reported_ordering_makes_scan_multi_partition() {
+ let exec = exec_with_ordering(single_col_ordering());
+ assert_eq!(exec.properties().partitioning.partition_count(), 3);
+ }
+
+ // Without a reported ordering the scan stays single-partition (Comet
drives only execute(0),
+ // which must read every task), preserving the legacy unordered behaviour.
+ #[test]
+ fn no_ordering_keeps_single_partition() {
+ let exec = exec_with_ordering(None);
+ assert_eq!(exec.properties().partitioning.partition_count(), 1);
+ }
+
+ // The ordered scan reads each file as its own sorted partition and relies
on
+ // SortPreservingMergeExec to k-way merge them into one globally sorted
stream. This feeds known
+ // sorted partitions (with duplicate keys across partitions, and both asc
and desc) into that
+ // merge with the same kind of LexOrdering the planner builds, and checks
the output is globally
+ // sorted and complete. It is deterministic coverage of the merge that
does not depend on an
+ // ordering-reporting Iceberg build (which is why the end-to-end suite's
merge assertions cancel
+ // on the published Iceberg used in CI).
+ async fn merge_ints(input: Vec<Vec<i32>>, descending: bool) -> Vec<i32> {
Review Comment:
Sorry for the slow turnaround. It's pushed to your branch as 987d3c6c54, on
top of 1c5f95b8b8. Three tests in iceberg_scan.rs write real Parquet files and
merge them through `IcebergScanExec`: ascending, descending, and with duplicate
keys across files. Three in planner.rs go through `create_plan`. They check
that the merge path and the `maxFilesPerPartition` sort fallback return the
same rows, and that all four direction and null-ordering combinations survive
the proto. Making every partition read every file fails all six, and the
existing tests stay green.
There's also an ignored test for #6524 that asserts the shared delete file
is read once. Once that's fixed it can become the regression guard. Feel free
to drop it if you'd rather keep this PR focused.
##########
native/core/src/execution/operators/iceberg_scan.rs:
##########
@@ -647,6 +704,141 @@ mod tests {
.unwrap();
}
+ fn int_schema() -> arrow::datatypes::SchemaRef {
+ use arrow::datatypes::{DataType, Field, Schema};
+ Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, false)]))
+ }
+
+ fn single_col_ordering() -> Option<datafusion::physical_expr::LexOrdering>
{
+ use arrow::compute::SortOptions;
+ use datafusion::physical_expr::expressions::Column;
+ use datafusion::physical_expr::{LexOrdering, PhysicalSortExpr};
+ LexOrdering::new(vec![PhysicalSortExpr {
+ expr: Arc::new(Column::new("a", 0)),
+ options: SortOptions::default(),
+ }])
+ }
+
+ // Builds a scan over three empty-delete tasks with the given reported
ordering.
+ fn exec_with_ordering(
+ ordering: Option<datafusion::physical_expr::LexOrdering>,
+ ) -> IcebergScanExec {
+ use std::collections::HashMap;
+ let tasks = vec![
+ task_with_deletes(vec![]),
+ task_with_deletes(vec![]),
+ task_with_deletes(vec![]),
+ ];
+ IcebergScanExec::new(
+ "metadata.json".to_string(),
+ int_schema(),
+ HashMap::new(),
+ "cat".to_string(),
+ tasks,
+ 1,
+ ordering,
+ )
+ .unwrap()
+ }
+
+ // A reported ordering turns the scan into a multi-partition operator (one
partition per task)
+ // so a SortPreservingMergeExec above can k-way merge the per-file sorted
streams.
+ #[test]
+ fn reported_ordering_makes_scan_multi_partition() {
+ let exec = exec_with_ordering(single_col_ordering());
+ assert_eq!(exec.properties().partitioning.partition_count(), 3);
+ }
+
+ // Without a reported ordering the scan stays single-partition (Comet
drives only execute(0),
+ // which must read every task), preserving the legacy unordered behaviour.
+ #[test]
+ fn no_ordering_keeps_single_partition() {
+ let exec = exec_with_ordering(None);
+ assert_eq!(exec.properties().partitioning.partition_count(), 1);
+ }
+
+ // The ordered scan reads each file as its own sorted partition and relies
on
+ // SortPreservingMergeExec to k-way merge them into one globally sorted
stream. This feeds known
+ // sorted partitions (with duplicate keys across partitions, and both asc
and desc) into that
+ // merge with the same kind of LexOrdering the planner builds, and checks
the output is globally
+ // sorted and complete. It is deterministic coverage of the merge that
does not depend on an
+ // ordering-reporting Iceberg build (which is why the end-to-end suite's
merge assertions cancel
+ // on the published Iceberg used in CI).
+ async fn merge_ints(input: Vec<Vec<i32>>, descending: bool) -> Vec<i32> {
Review Comment:
I've pushed this to your branch as 987d3c6c54 instead, as you asked in
https://github.com/apache/datafusion-comet/pull/5331#discussion_r4160401757, so
there's nothing to apply. The only differences from the patch above are the
test names and a couple of comments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
