parthchandra commented on code in PR #5331:
URL: https://github.com/apache/datafusion-comet/pull/5331#discussion_r3876750572
##########
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##########
@@ -835,12 +835,38 @@ 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. reportableOrdering is the same gate the native
scan/serde use, so the
+ // decision here cannot diverge from what the native path would do.
+ val orderingHonored: Boolean = {
Review Comment:
Good point. Changes so we evaluate the gate once and save the result on
`nativeIcebergScanMetadata`; `outputOrdering` and the proto serde both read
that field.
##########
native/core/src/execution/spark_config.rs:
##########
@@ -26,6 +26,21 @@ pub(crate) const COMET_PARQUET_ROW_FILTER_PUSHDOWN_ENABLED:
&str =
"spark.comet.parquet.rowFilterPushdown.enabled";
pub(crate) const SPARK_EXECUTOR_CORES: &str = "spark.executor.cores";
+/// Above this many files in a single Spark partition, the native Iceberg scan
does not attempt a
+/// per-file-stream k-way merge (which would open one reader per file at
once); it reads the
+/// partition unordered and sorts with a spillable SortExec instead. Read from
the config below and
+/// carried to the physical planner as an [`IcebergSortMergeConfig`] session
extension.
+pub(crate) const COMET_ICEBERG_SORT_MERGE_MAX_FILES_PER_PARTITION: &str =
+ "spark.comet.scan.icebergNative.sortMerge.maxFilesPerPartition";
+pub(crate) const DEFAULT_ICEBERG_SORT_MERGE_MAX_FILES_PER_PARTITION: usize =
64;
Review Comment:
Moved `max_files_per_partition` onto `IcebergScanCommon` next to
`data_file_concurrency_limit`, so there's one default (in CometConf) and
removed the SessionConfig extension. Added
`iceberg_scan_honors_max_files_per_partition_from_proto`, which sets the proto
field to 2 and asserts a 3-file scan takes SortExec — so the value is actually
read from the plan now, not the compiled-in constant
##########
spark/src/main/scala/org/apache/comet/CometConf.scala:
##########
@@ -121,6 +121,30 @@ object CometConf extends ShimCometConf {
.booleanConf
.createWithDefault(true)
+ val COMET_ICEBERG_SORT_MERGE_ENABLED: ConfigEntry[Boolean] =
+ conf("spark.comet.scan.icebergNative.sortMerge.enabled")
+ .category(CATEGORY_SCAN)
+ .doc("Whether the native Iceberg scan reports the table sort order and
performs a " +
+ "per-partition streaming merge of already-sorted files. When enabled
and Iceberg reports " +
+ "an ordering (requires Iceberg's
spark.sql.iceberg.planning.preserve-data-ordering), " +
+ "each Spark partition reads its files as separate sorted streams
merged into one sorted " +
+ "output, and the ordering is surfaced to Spark so redundant sorts are
eliminated. When " +
+ "disabled, files are read unordered as before.")
Review Comment:
You're right, doing what you suggest: `enabled=false` now behaves like
`maxFilesPerPartition=0` — the scan stays native and still honors the reported
order via the spillable `SortExec`, it just skips the k-way merge. The two
disabled-fallback tests are now "stays native" tests.
##########
spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala:
##########
@@ -935,6 +935,32 @@ object IcebergReflection extends Logging {
}
}
+ /**
+ * Names of top-level columns whose Iceberg sort order can differ from
Spark's comparison of the
+ * Spark type they map to, so a native k-way merge keyed on the Spark value
could mis-order
+ * rows. Today that is UUID: Iceberg maps UUID to Spark StringType (see
TypeToSparkType) but
+ * sorts by its own UUID comparator, not by the canonical string, so the
file order and a string
+ * comparison can disagree. reportableOrdering refuses a sort key in this
set. v1 only reports
+ * identity, top-level sort keys, so top-level columns are enough.
+ */
+ def orderingUnsafeColumns(schema: Any): Set[String] = {
+ import scala.jdk.CollectionConverters._
+ try {
+ val columns = getMethod(schema.getClass, "columns")
+ .invoke(schema)
+ .asInstanceOf[java.util.List[_]]
+ columns.asScala.flatMap { column =>
+ val name = getMethod(column.getClass,
"name").invoke(column).asInstanceOf[String]
+ val typeStr = getMethod(column.getClass,
"type").invoke(column).toString
+ if (typeStr == "uuid") Some(name) else None
+ }.toSet
+ } catch {
+ case e: Exception =>
+ logWarning(s"Failed to inspect schema for ordering-unsafe columns:
${e.getMessage}")
Review Comment:
Agreed. `orderingUnsafeColumns` now returns `None` when the schema can't be
read, and the caller treats `None` as "refuse the ordering" — a reflection
failure makes it un-reportable instead of assuming no UUID. The
`nativeIcebergScanMetadata == null `branch is gone too; see the
single-evaluation change below.
##########
native/core/src/execution/operators/iceberg_scan.rs:
##########
@@ -86,6 +86,24 @@ pub struct IcebergScanExec {
tasks: Vec<FileScanTask>,
/// Number of data files to read concurrently
data_file_concurrency_limit: usize,
+ /// FileIO (and, for S3, the JVM credential bridge behind it) built once
at plan time and shared
+ /// across partitions. FileIO is cheap to clone (Arc-backed), so each
`execute` clones this
+ /// rather than rebuilding the storage factory + credential bridge. This
matters in the ordered
+ /// path, where the scan is one partition per file and `execute` is called
once per file.
+ file_io: FileIO,
+ /// Table sort order Iceberg reported, translated against `output_schema`.
`Some` makes this a
+ /// multi-partition scan: one sorted stream per task, which a
SortPreservingMergeExec above
+ /// merges back into one sorted partition. It is also advertised in
`plan_properties`. `None`
+ /// keeps the old single-partition unordered read (all tasks streamed
together).
+ ///
+ /// Concurrency note: in the ordered path each partition reads exactly one
task, so
+ /// `data_file_concurrency_limit` no longer bounds cross-file concurrency;
instead the wrapping
+ /// SortPreservingMergeExec drives one reader per file to merge them. That
fan-out (files per
+ /// Spark partition) is intrinsic to a k-way merge of per-file sorted
streams -- the files must
+ /// be read as separate streams to stay individually sorted -- and is the
natural granularity
+ /// for a sorted Iceberg table. `data_file_concurrency_limit` still bounds
delete-file stats and
Review Comment:
Comment updated
--
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]
