anuragmantri commented on code in PR #5331:
URL: https://github.com/apache/datafusion-comet/pull/5331#discussion_r3777421395
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala:
##########
@@ -859,6 +859,49 @@ object CometIcebergNativeScan extends
CometOperatorSerde[CometBatchScanExec] wit
Some(builder.setIcebergScan(icebergScanBuilder).build())
}
+ /**
+ * The part of an Iceberg-reported sort order that the native per-partition
merge can honour, or
+ * Nil when the merge must stay off. Two callers use this one gate: the
proto serialization
+ * (which turns on the native SortPreservingMergeExec) and
+ * CometIcebergNativeScanExec.outputOrdering (which tells Spark the scan is
sorted). Sharing the
+ * gate means the two always agree.
+ *
+ * v1 accepts only identity sort fields on top-level columns that are in the
projection. Each
+ * SortOrder child must be an AttributeReference in `output`, and must
serialize to proto.
+ * Transform sort fields (bucket/truncate/...) are not AttributeReferences,
so they fall through
+ * to Nil and we read unordered. Checking exprToProto here, not just in the
proto path, keeps
+ * the two callers in step: outputOrdering never advertises an order the
proto path would drop.
+ *
+ * We trust Iceberg on file-level sortedness. If it reports an ordering,
SortOrderAnalyzer has
+ * already checked each file's sort_order_id matches the table order, so
every file is sorted.
+ *
+ * We read scanExec.ordering (the raw reported order), not
scanExec.outputOrdering. Spark blanks
+ * outputOrdering when a partition holds more than one file -- the case this
merge handles.
+ */
+ def reportableOrdering(
+ ordering: Option[Seq[SortOrder]],
+ output: Seq[Attribute]): Seq[SortOrder] = {
+ if (!CometConf.COMET_ICEBERG_SORT_MERGE_ENABLED.get()) {
+ Nil
+ } else {
+ ordering match {
+ case Some(orders) if orders.nonEmpty && orders.forall(isReportable(_,
output)) =>
+ orders
+ case _ =>
+ Nil
+ }
+ }
+ }
+
+ private def isReportable(order: SortOrder, output: Seq[Attribute]): Boolean =
+ isIdentityProjected(order, output) && exprToProto(order, output).isDefined
+
Review Comment:
As identified in the [Iceberg
PR](https://github.com/apache/iceberg/pull/14948#pullrequestreview-4932308014)
and the design doc https://github.com/apache/datafusion-comet/issues/5323, UUID
orders differently in Iceberg than in Spark's comparator, and the identity case
for UUID needs to follow Iceberg's byte ordering specifically. This gate
doesn't check the sort column's type at all. I believe we could rely on
upstream https://github.com/apache/iceberg/pull/16750 to not report ordering on
UUID. Is my understanding correct.
##########
spark/src/test/scala/org/apache/comet/CometIcebergSortMergeReadSuite.scala:
##########
@@ -0,0 +1,638 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.comet
+
+import java.util.concurrent.atomic.AtomicInteger
+
+import org.apache.spark.sql.CometTestBase
+import org.apache.spark.sql.catalyst.expressions.{Add, Ascending,
AttributeReference, Literal, SortOrder}
+import org.apache.spark.sql.comet.{CometIcebergNativeScanExec, CometSortExec}
+import org.apache.spark.sql.comet.execution.shuffle.CometShuffleExchangeExec
+import org.apache.spark.sql.execution.{SortExec, SparkPlan}
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+import org.apache.spark.sql.execution.exchange.ShuffleExchangeExec
+import org.apache.spark.sql.types.IntegerType
+
+import org.apache.comet.serde.operator.CometIcebergNativeScan
+
+/**
+ * Tests for the sort-aware native Iceberg scan (branch `stream-merge`): the
scan reports the
+ * Iceberg table sort order to Spark and does a per-partition streaming k-way
merge of the
+ * already-sorted files, so Catalyst can drop the Sort and (with
storage-partitioned join) the
+ * Exchange that a sort-merge join / grouped aggregate / window would
otherwise need.
+ *
+ * The suite has two parts:
+ * - Unit tests of the [[CometIcebergNativeScan.reportableOrdering]] gate
(no SparkSession
+ * required) -- the single decision shared by the proto serialization
(which turns on the
+ * native SortPreservingMergeExec) and
CometIcebergNativeScanExec.outputOrdering (which tells
+ * Spark the scan is sorted).
+ * - End-to-end tests over real Iceberg tables that exercise every Spark 4.0
mechanism which
+ * exploits already-sorted input to avoid a Sort or a shuffle (all keyed
off
+ * `SortOrder.orderingSatisfies`, prefix semantics):
`SupportsReportOrdering` ->
+ * `BatchScanExec.outputOrdering` and `EnsureRequirements` eliding the
required-child Sort;
+ * `EliminateSorts` / `RemoveRedundantSorts`;
`SortMergeJoinExec.requiredChildOrdering` +
+ * storage-partitioned join (`KeyGroupedPartitioning`,
+ * `spark.sql.sources.v2.bucketing.enabled`); `ReplaceHashWithSortAgg` /
`SortAggregateExec`;
+ * `WindowExec`; `TakeOrderedAndProjectExec`.
+ *
+ * Two invariants determine the end-to-end assertions:
+ * 1. Correctness is checked unconditionally via `checkSparkAnswer` (Comet
vs vanilla Spark).
+ * This is the primary guarantee: any k-way-merge defect (dropped,
duplicated or mis-ordered
+ * rows, or an outputOrdering/outputPartitioning that does not match the
rows the native
+ * operator actually produces) shows up as a result mismatch. It holds on
any Iceberg build.
+ * 2. The strict "no Sort / no Exchange" plan assertions are the *target*
of this feature.
+ * They only hold where the Iceberg build actually reports the ordering
+ * (`SupportsReportOrdering`, today an Iceberg fork feature -- the
published/upstream Iceberg
+ * used in CI does not report it) and where the native scan reports
`KeyGroupedPartitioning`.
+ * Each such test therefore runs the correctness check first, then
`assume`s the reporting is
+ * active before asserting the plan shape, so it enforces the contract on
a reporting build
+ * and is skipped (not failed) elsewhere. `sort = 0` is asserted only for
*operator-required*
+ * orderings (SMJ / aggregate / window), never for a global `ORDER BY`,
which Spark keeps
+ * regardless (the per-partition merge is not a global order).
+ *
+ * These cannot be SQL-file fixtures: setting an Iceberg sort order needs the
Iceberg Java API
+ * (the Comet test session registers no Iceberg SQL extensions, so `WRITE
ORDERED BY` will not
+ * parse), and the plan-shape assertions need access to the executed SparkPlan.
+ *
+ * Each test gets its own catalog name and temp warehouse (the Hadoop
`SparkCatalog` instance is
+ * cached per catalog name, so a shared name would bind every test to the
first warehouse), and
+ * its tables are dropped in a `finally` so a failing test cannot leak a table
into a later one.
+ */
+class CometIcebergSortMergeReadSuite
Review Comment:
I reviewed the tests, very nice coverage already. I would also a test that
deletes some rows from one file in a multi-file sorted partition, to cover the
merge alongside MOR deletes.
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala:
##########
@@ -859,6 +859,49 @@ object CometIcebergNativeScan extends
CometOperatorSerde[CometBatchScanExec] wit
Some(builder.setIcebergScan(icebergScanBuilder).build())
}
+ /**
+ * The part of an Iceberg-reported sort order that the native per-partition
merge can honour, or
+ * Nil when the merge must stay off. Two callers use this one gate: the
proto serialization
+ * (which turns on the native SortPreservingMergeExec) and
+ * CometIcebergNativeScanExec.outputOrdering (which tells Spark the scan is
sorted). Sharing the
+ * gate means the two always agree.
+ *
+ * v1 accepts only identity sort fields on top-level columns that are in the
projection. Each
+ * SortOrder child must be an AttributeReference in `output`, and must
serialize to proto.
+ * Transform sort fields (bucket/truncate/...) are not AttributeReferences,
so they fall through
+ * to Nil and we read unordered. Checking exprToProto here, not just in the
proto path, keeps
+ * the two callers in step: outputOrdering never advertises an order the
proto path would drop.
+ *
+ * We trust Iceberg on file-level sortedness. If it reports an ordering,
SortOrderAnalyzer has
+ * already checked each file's sort_order_id matches the table order, so
every file is sorted.
+ *
+ * We read scanExec.ordering (the raw reported order), not
scanExec.outputOrdering. Spark blanks
+ * outputOrdering when a partition holds more than one file -- the case this
merge handles.
Review Comment:
I believe using `scanExec.ordering` is to bypass the check in Spark 3.4 and
3.5 which was fixed in Spark 4.2 by
[SPARK-55715](https://github.com/apache/spark/pull/55116) and this is needed to
support all these Spark versions. @peter-toth, would you please also take a
look at this to see if this is safe?
##########
native/core/src/execution/operators/iceberg_scan.rs:
##########
Review Comment:
Claude flagged this:
`BaselineMetrics::new(metrics, 0)` hardcodes partition 0, but this operator
is now multi-partition in the ordered path. Should this thread through the
actual partition index from `execute()`?
--
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]
