andygrove commented on code in PR #5331:
URL: https://github.com/apache/datafusion-comet/pull/5331#discussion_r3872675077
##########
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:
This returns `Set.empty` when the reflection fails, which the callers read
as "no unsafe columns", so they go on to report the ordering. That's the
fail-open direction on a decision where being wrong means silently mis-ordered
rows, because Spark has already dropped the `Sort` by the time we get here.
The premise of this method is that a UUID sort key is invisible at the Spark
type level. If we couldn't read the schema, we can't rule one out. Could the
failure path make the ordering unreportable instead? Returning something like
`None` and having `reportableOrdering` treat that as "refuse" would keep the
safe direction the default.
The `nativeIcebergScanMetadata == null` branch in
`CometIcebergNativeScanExec.outputOrdering` has the same shape.
##########
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:
Thanks for collapsing the proto and `outputOrdering` down to one evaluation.
I think this guard reintroduces the same hazard one level up, though.
This is a second evaluation of `reportableOrdering`, and it's the one that
decides whether Comet converts the scan at all. It reads
`COMET_ICEBERG_SORT_MERGE_ENABLED` out of the active `SQLConf` and runs
`orderingUnsafeColumns` reflection independently of the exec's lazy val. If
this one says honorable, Comet converts and Spark drops the `Sort`. If the
later one returns `Nil`, native reads unordered and we return wrong results
with no error. The reverse mismatch just wastes some work.
You already have `honorable` right here. Could we stash it on
`nativeIcebergScanMetadata` and have
`CometIcebergNativeScanExec.outputOrdering` read that instead of re-running the
gate? Then there's one evaluation for the whole decision, and no conf read
after the point where Spark has committed to dropping the sort.
##########
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:
`data_file_concurrency_limit` is the same kind of setting and it rides on
`IcebergScanCommon`, set from `CometConf` in the serde, so there's exactly one
default. This one goes through the Spark config map plus a `SessionConfig`
extension with `64` written on both sides of JNI, and nothing catches it if the
two drift.
Is there a reason not to put `max_files_per_partition` on
`IcebergScanCommon` next to `data_file_concurrency_limit`? That removes the
duplicated default and the extension plumbing in one go.
Related: both `iceberg_scan_merges_when_files_within_limit` and
`iceberg_scan_falls_back_to_sort_above_file_limit` use
`PhysicalPlanner::default()`, so they only ever test the compiled-in constant.
If the extension weren't reaching the planner both would still pass, and the
Scala test that sets the conf to `1000` would quietly exercise the sort
fallback instead of the 70-way merge it's written to cover. Worth one test that
builds the planner from a `SessionConfig` carrying `IcebergSortMergeConfig {
max_files_per_partition: 2 }` and asserts a 3-file scan takes the `SortExec`
path.
##########
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:
The doc says "When disabled, files are read unordered as before", but with
the new `orderingHonored` guard that isn't what happens. Disabling this makes
`reportableOrdering` return `Nil`, which makes `orderingHonored` false, which
keeps the scan on Spark entirely. Your own two `fallback: sort-merge disabled
...` tests assert exactly that. So the escape hatch for a merge bug also costs
all Comet acceleration on every sorted Iceberg table, which is a steep price
for the knob people would reach for when something looks wrong.
I don't think it has to work that way. `maxFilesPerPartition = 0` already
gives us "report the ordering, never merge, honor it with the native
`SortExec`", which is correct and keeps Comet on the scan. Could
`sortMerge.enabled=false` mean that instead? Either way the doc needs to
describe the actual behavior.
##########
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:
This is the paragraph @mbutrovich and I were arguing with, and the code has
moved on. It still says the fan-out "is intrinsic to a k-way merge" and "is the
natural granularity for a sorted Iceberg table", with no mention that the
planner now caps it at `sortMerge.maxFilesPerPartition` and drops to a
spillable `SortExec` above that. Could you point it at the cap, and at #5343
for the bound-driven admission scheme? Otherwise the next person to read this
concludes the fan-out is unbounded by design.
--
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]
