NikitaMatskevich commented on code in PR #5724:
URL: https://github.com/apache/datafusion-comet/pull/5724#discussion_r4372890719
##########
spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala:
##########
@@ -890,72 +1012,15 @@ class CometIcebergWriteDetectionSuite extends
CometTestBase with CometIcebergTes
}
}
- /**
- * Runs Spark's transition insertion followed by
[[EliminateRedundantTransitions]] over a
- * hand-built `CometIcebergWriteExec -> CometSparkToColumnarExec -> source`
plan and returns the
- * write's final child.
- *
- * Hand-built rather than driven through SQL because the shape depends on
- * `spark.comet.sparkToColumnar.enabled` admitting the write's source
operator, and the set of
- * admitted operators is itself configurable. What matters is the rule's
behaviour at that
- * boundary, which this pins directly.
- */
- private def writeChildAfterTransitionRules(source: SparkPlan): SparkPlan = {
- val write = CometIcebergWriteExec(
- Operator.newBuilder().build(),
- CometSparkToColumnarExec(source),
- batchWrite = null,
- table = null,
- partitionSpecId = 0)
- val withTransitions = ApplyColumnarRulesAndInsertTransitions(Seq.empty,
false).apply(write)
- // Spark must insert a columnar-to-row transition below the row-based
write; if it stops doing
- // so the rest of the assertion is vacuous.
- assert(
-
withTransitions.asInstanceOf[CometIcebergWriteExec].child.isInstanceOf[ColumnarToRowExec],
- s"expected an inserted ColumnarToRowExec below the write,
got:\n$withTransitions")
- EliminateRedundantTransitions(spark)
- .apply(withTransitions)
- .asInstanceOf[CometIcebergWriteExec]
- .child
- }
-
- // https://github.com/apache/datafusion-comet/issues/5689: the write's input
transition has to
- // be stripped before the generic
`ColumnarToRowExec(CometSparkToColumnarExec)` cancellation
- // consumes it, otherwise that arm removes the Arrow bridge the write's FFI
input depends on.
- // Both source representations are covered because the cancellation treats
them differently:
- // over a row source it drops the bridge outright, over a Spark-columnar
source it keeps a
- // transition but leaves the write reading Spark `ColumnarVector`s instead
of `CometVector`s.
- test("row source keeps its Arrow bridge under the native Iceberg write") {
- val source = TransitionProbeLeaf(columnar = false)
- val child = writeChildAfterTransitionRules(source)
- assert(
- child == CometSparkToColumnarExec(source),
- s"expected the write to sit directly on CometSparkToColumnarExec,
got:\n$child")
- }
-
- test("Spark-columnar source keeps its Arrow bridge under the native Iceberg
write") {
- val source = TransitionProbeLeaf(columnar = true)
- val child = writeChildAfterTransitionRules(source)
- assert(
- child == CometSparkToColumnarExec(source),
- s"expected the write to sit directly on CometSparkToColumnarExec,
got:\n$child")
+ private def assumeIcebergBloomShapeProperties(): Unit = {
+ assume(
+ IcebergReflection
+ .tablePropertyConstantOpt("PARQUET_BLOOM_FILTER_COLUMN_FPP_PREFIX")
Review Comment:
Thank you, addressed in
[#7226](https://github.com/apache/datafusion-comet/pull/7226), commit `test:
cover legacy Bloom caps without NDV support`. This PR keeps these older
runtimes on Java.
##########
spark/src/main/scala/org/apache/comet/serde/operator/IcebergWriteProtoTranslation.scala:
##########
@@ -112,6 +162,28 @@ object IcebergWriteProtoTranslation {
parseJavaInt(props, Keys.ParquetDictSizeBytes,
Defaults.DictSizeBytes.toInt).toLong
val pageRowLimit = parseJavaInt(props, Keys.ParquetPageRowLimit,
Defaults.PageRowLimit)
val compression = resolveCompression(props)
+ // Iceberg properties use logical schema paths, while Parquet writer
properties require the
+ // physical leaf path. Missing fields are skipped, matching Iceberg Java's
writer behavior.
+ val bloomFilterColumns = enabledBloomFilterColumnNames(props)
+ .flatMap { icebergName =>
+ parquetPathByIcebergColumnName.get(icebergName).map(icebergName -> _)
Review Comment:
Thank you, addressed in
[#7226](https://github.com/apache/datafusion-comet/pull/7226), commit `feat:
enable safeguarded native Bloom writes on legacy Iceberg`. This PR keeps these
older runtimes on Java.
--
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]
