andygrove commented on code in PR #5331:
URL: https://github.com/apache/datafusion-comet/pull/5331#discussion_r3962418707
##########
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:
`serializeReportedOrdering` opens with the same `if
(reportedOrdering.isEmpty) Some(Nil)` check that this branch does, so one of
the two is unreachable. It has a single caller, so dropping the guard here
reads a little better since the callee is the one that documents the contract.
##########
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:
`execute_with_tasks` builds a new `ArrowReaderBuilder(...).build()` on every
call, a few lines below this one, and that is where iceberg-rust constructs the
`CachingDeleteFileLoader`. Its `delete_filter` is the shared state the crate
documents as existing "to allow caching loaded deletes across multiple calls to
`load_deletes` (e.g., across multiple file scan tasks)". In the ordered path
`execute` runs once per file, so that cache is now per-file, and a delete file
shared across the partition gets downloaded and parsed once per data file
instead of once for the whole partition. `fill_delete_file_sizes` dedups within
a single call too, so it also issues one HEAD per data file for the same delete
file now. At the default cap that is up to 64x on a merge-on-read table, and
equality deletes are the worst case since they are always partition-scoped and
the expensive ones to parse.
I measured it with two data files sharing one positional delete file.
`bytes_scanned` goes from 2717 on the unordered path to 4256 on the ordered
path, and the 1539-byte delta is exactly the delete file being read a second
time. Rows are correct either way, so this is purely IO and CPU rather than a
correctness problem.
The fix looks like the same shape as the `file_io` hoist on this line: hold
one `ArrowReader` on `IcebergScanExec` and clone it per partition. It needs
`batch_size` off the `TaskContext`, so it has to be a `OnceLock` filled on
first `execute` rather than built in `new`, but `ArrowReader` is `Clone`, and
`read()` calls `ScanMetrics::new()` per invocation and re-bases the loader's
metrics through `with_scan_metrics`, so per-partition metrics stay separate
while the delete cache is shared. With that prototype the ordered path drops
back to 2717 with identical rows. Would you rather do it here or fold it into
#5343?
##########
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:
`merge_ints` and the two `spm_merges_*` tests below build a
`MemorySourceConfig` and merge it, so they never touch `IcebergScanExec`, the
planner or the proto. Every symbol they use predates this PR, which means they
compile and pass verbatim on `main`. I checked with a mutation: changing
`IcebergScanExec::execute` to `let tasks = self.tasks.clone()` so every
partition reads every file leaves both of them green.
The useful part is that the gap looks closable without waiting on Iceberg. I
tried writing probes that write real Parquet into a tempdir and drive the
ordered path through `IcebergScanExec` and through `create_plan`, and the whole
set runs in about 50ms. Merging three real files ascending, descending, and
with duplicate keys across them gives tests that the mutation above does kill.
Going through `create_plan` and asserting the merge path and the
`maxFilesPerPartition` sort-fallback path produce identical output on the same
data covers the proto to `create_sort_expr` to `LexOrdering` translation, which
nothing exercises today. Round-tripping the null ordering through the proto in
all four direction and null-ordering combinations covers ASC NULLS LAST and
DESC NULLS FIRST, which are not Iceberg's defaults but are legal `SortField`
values.
All of those pass on your branch, so this is a coverage gap rather than a
bug report. Happy to hand you the patch if it saves you time.
--
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]
