parthchandra commented on code in PR #5331:
URL: https://github.com/apache/datafusion-comet/pull/5331#discussion_r3981951427
##########
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:
Can you point me to your patch, I'll merge it in. Thank you!
##########
native/core/src/execution/operators/iceberg_scan.rs:
##########
@@ -164,18 +224,14 @@ impl IcebergScanExec {
fn execute_with_tasks(
&self,
tasks: Vec<FileScanTask>,
+ partition: usize,
context: Arc<TaskContext>,
) -> DFResult<SendableRecordBatchStream> {
let output_schema = Arc::clone(&self.output_schema);
- let file_io = load_file_io(
- &self.catalog_properties,
- &self.metadata_location,
- &self.catalog_name,
- AccessMode::Read,
- )?;
+ let file_io = self.file_io.clone();
Review Comment:
Nice catch. Let me take care of this in #5343
##########
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##########
@@ -914,16 +915,69 @@ case class CometScanRule(session: SparkSession)
}
}
+ // If Iceberg reports an ordering, EnsureRequirements may have already
dropped the Sort
+ // above this scan (it decides that on the vanilla BatchScanExec,
before Comet converts the
+ // scan). If the native scan cannot guarantee that ordering, reading
unordered here would
+ // silently return wrong results, so stay on Spark -- its Iceberg
reader produces the sorted
+ // output it promised. Evaluate the gate exactly once here and stash
the result on the
+ // metadata; CometIcebergNativeScanExec.outputOrdering and the proto
serde both read that
+ // stashed value, so the reported order cannot diverge from what
native advertises.
+ val icebergReportsOrdering: Boolean =
scanExec.ordering.exists(_.nonEmpty)
+ val reportedOrdering: Seq[SortOrder] = {
+ if (!icebergReportsOrdering) {
+ Nil
+ } else {
+ // None means the schema could not be read, so we cannot rule out
an unsafe (UUID) sort
+ // key -- refuse the ordering rather than assume it is safe.
+ IcebergReflection.orderingUnsafeColumns(metadata.tableSchema)
match {
+ case Some(unsafe) =>
+ CometIcebergNativeScan
+ .reportableOrdering(scanExec.ordering, scanExec.output,
unsafe)
+ case None => Nil
+ }
+ }
+ }
+ // Bind the reported ordering to proto now, against the same output
the gate used, so the
+ // executor-side serde writes it directly. None means the binding
failed -- the gate below
+ // then keeps the scan on Spark instead of converting and hard-failing
at task start.
+ val reportedOrderingProto: Option[Seq[Expr]] =
Review Comment:
Done.
--
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]
