comphead commented on code in PR #7155:
URL: https://github.com/apache/datafusion-comet/pull/7155#discussion_r4486201716
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala:
##########
@@ -297,6 +299,64 @@ object CometIcebergNativeWrite extends
CometOperatorSerde[IcebergWriteExec] {
Unsupported(Some(s"Iceberg native write detection failed:
${e.getMessage}"))
}
+ /**
+ * Row-to-Arrow conversion reads ahead before the writer consumes the batch.
Keep fixed fields
+ * on the JVM because Spark's BinaryType does not enforce their length. For
a clustered writer,
+ * materialize the writer partition so a later Spark expression failure
cannot overtake an
+ * earlier partition-order error.
+ */
+ def isRowInputEvaluationSupported(
+ op: IcebergWriteExec,
+ fallbackReasons: ListBuffer[String]): Boolean = {
+ val rowInputConstraints =
+ try {
+ for {
+ sparkWrite <- IcebergReflection.getOuterSparkWrite(op.batchWrite)
+ table <- IcebergReflection.getTableFromSparkWrite(sparkWrite)
+ specId <- IcebergReflection.getOutputSpecIdFromSparkWrite(sparkWrite)
+ spec <- IcebergReflection.getPartitionSpecById(table, specId)
+ fanout <-
IcebergReflection.getUseFanoutWriterFromSparkWrite(sparkWrite)
+ writeSchema <-
IcebergReflection.getWriteSchemaFromSparkWrite(sparkWrite)
+ } yield {
+ val writerMode = IcebergWriteProtoTranslation.resolveWriterMode(
+ specIsUnpartitioned = isUnpartitionedSpec(spec),
+ useFanoutWriter = fanout)
+ val hasFixedFields =
+ IcebergReflection.findFieldWithTypeIds(writeSchema,
Set("FIXED")).nonEmpty
+ (writerMode, hasFixedFields)
+ }
+ } catch {
+ case NonFatal(_) => None
+ }
+
+ rowInputConstraints match {
+ case None =>
+ fallbackReasons +=
+ "Iceberg writer row conversion could not resolve the effective
writer constraints"
+ false
+ case Some((_, true)) =>
+ fallbackReasons += "Iceberg writer row conversion keeps fixed fields
on the JVM to " +
+ "preserve fixed-width validation failures"
+ false
+ case Some((IcebergWriterMode.ICEBERG_WRITER_CLUSTERED, false))
+ if !hasMaterializedRowInput(op.child) =>
+ fallbackReasons += "Iceberg clustered writer row conversion requires
materialized Spark " +
+ "input so batch read-ahead cannot change expression and
partition-order error precedence"
+ false
+ case _ => true
Review Comment:
As far as I can tell, these two declines only decide which of two errors a
failing write reports, and I could not find a case where committed data
differs. Native input with `fixed(N)` already stays eligible and reports
Arrow's cast error where iceberg-java reports Parquet's. If error identity is
part of the contract here, could the description say so? Otherwise dropping
both would remove the reflection chain in `isRowInputEvaluationSupported`
(which repeats lines 1122 to 1174), `hasMaterializedRowInput` and two of the
longest tests.
##########
spark/src/main/scala/org/apache/spark/sql/comet/CometSparkToColumnarExec.scala:
##########
@@ -132,6 +135,73 @@ case class CometSparkToColumnarExec(child: SparkPlan)
object CometSparkToColumnarExec extends CometSink[SparkPlan] with
DataTypeSupport {
+ private object RowTypeSupport extends DataTypeSupport with CometTypeShim
+
+ // RowArrowReader recursively writes the collection types supported by
DataTypeSupport.
+ def isRowSchemaSupported(schema: StructType, fallbackReasons:
ListBuffer[String]): Boolean =
+ RowTypeSupport.isSchemaSupported(schema, fallbackReasons)
Review Comment:
`isRowSchemaSupported` admits every array and map by going around the
`isTypeSupported` override below, which #6726 removes for every caller. Could
this build on #6726, or add reads of the deeper shapes it now admits (struct
elements, nested arrays, null elements)? The only end-to-end coverage is
`ARRAY<INT>` and `MAP<STRING, INT>`.
##########
spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala:
##########
@@ -3980,6 +4544,135 @@ class CometIcebergWriteActionSuite
}
}
+ for {
+ adaptive <- Seq(false, true)
+ laterNull <- Seq(false, true)
+ } {
+ test(
+ "native Iceberg writer: fixed-length validation preserves the first
error, " +
+ s"laterNull=$laterNull, AQE=$adaptive") {
+ assume(icebergAvailable && isSpark35Plus)
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> adaptive.toString,
+ CometConf.COMET_BATCH_SIZE.key -> "8") {
+ withIcebergCatalog { _ =>
+ withTempPath { path =>
+ val schema = StructType(
+ Seq(
+ StructField("id", IntegerType, nullable = true),
+ StructField("fx", BinaryType, nullable = true)))
+ val invalidRows = Seq(Row(1, Array[Byte](1, 2, 3))) ++
+ (if (laterNull) Seq(Row(null, Array[Byte](5, 6, 7, 8))) else
Seq.empty)
+ val invalidPath = new File(path, "invalid").toString
+ val validPath = new File(path, "valid").toString
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ spark
+ .createDataFrame(spark.sparkContext.parallelize(invalidRows,
1), schema)
+ .write
+ .parquet(invalidPath)
+ spark
+ .createDataFrame(
+ spark.sparkContext.parallelize(
+ Seq(Row(1, Array[Byte](1, 2, 3, 4)), Row(2, Array[Byte](5,
6, 7, 8))),
+ 1),
+ schema)
+ .write
+ .parquet(validPath)
+ val expectedInput = Seq(Row(1, "010203")) ++
+ (if (laterNull) Seq(Row(null, "05060708")) else Seq.empty)
+ assert(
+ spark.read.parquet(invalidPath).selectExpr("id",
"hex(fx)").collect().toSeq ==
+ expectedInput)
+ }
+ val suffix = s"${laterNull}_$adaptive"
+ val candidate = s"required_fixed_native_$suffix"
+ val reference = s"required_fixed_jvm_$suffix"
+ val nativeControl = s"required_fixed_control_native_$suffix"
+ val jvmControl = s"required_fixed_control_jvm_$suffix"
+ Seq(candidate, reference, nativeControl, jvmControl).foreach {
table =>
+ sql(s"""CREATE TABLE $catalog.$ns.$table (id INT NOT NULL) USING
iceberg
+ |TBLPROPERTIES ('write.distribution-mode'='none',
+ |'write.parquet.dict-size-bytes'='1')""".stripMargin)
+ addIcebergColumn(
+ loadIcebergTable(spark, catalog, ns, table),
+ "fx",
+ icebergFixedType(4))
+ sql(s"REFRESH TABLE $catalog.$ns.$table")
+ }
+ def assertJvmRequiredWrite(plans: Seq[SparkPlan]): Unit = {
+ val writers =
+ plans.flatMap(p => collectWithSubqueries(p) { case w:
IcebergWriteExec => w })
+ assert(writers.nonEmpty, plans.mkString("\n"))
+ assert(writers.forall(hasRequiredProjection),
plans.mkString("\n"))
+ assert(
+ plans.forall(p =>
+ collectWithSubqueries(p) { case w: CometIcebergWriteExec =>
w }.isEmpty),
+ plans.mkString("\n"))
+ plans.foreach(assertColumnarContract)
+ }
+ val valid = spark.read.parquet(validPath)
+ val successful = withNativeEnabled {
+ captureWrite(nativeControl) {
+ valid.writeTo(s"$catalog.$ns.$nativeControl").append()
+ }
+ }
+ assert(successful.snapshotDelta == 1)
+ assertJvmRequiredWrite(successful.plans)
+ assert(
+ successful.plans.forall(p =>
+ collectWithSubqueries(p) {
+ case s: SortExec => s
+ case e: ShuffleExchangeLike => e
+ }.isEmpty),
+ successful.plans.mkString("\n"))
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ valid.writeTo(s"$catalog.$ns.$jvmControl").append()
+ assert(countSnapshots(jvmControl) == 1)
+ Seq(nativeControl, jvmControl).foreach { table =>
+ checkAnswer(
+ sql(s"SELECT id, hex(fx) FROM $catalog.$ns.$table"),
+ Seq(Row(1, "01020304"), Row(2, "05060708")))
+ }
+ }
+
+ // A one-byte dictionary limit forces the first fixed(4) value
through Parquet's
+ // plain writer, which rejects its three bytes before any later
null assertion.
+ def appendInvalid(table: String): Unit =
+
spark.read.parquet(invalidPath).writeTo(s"$catalog.$ns.$table").append()
+ var candidateFailure: Throwable = null
+ val candidatePlans = withNativeEnabled {
+ capturePlans(spark, includeFailures = true) {
+ candidateFailure =
intercept[Exception](appendInvalid(candidate))
+ }
+ }
+ var referenceFailure: Throwable = null
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ referenceFailure = intercept[Exception](appendInvalid(reference))
+ }
+ assertJvmRequiredWrite(candidatePlans)
+ Seq(candidate, reference).foreach { table =>
+ assert(countSnapshots(table) == 0)
+ assert(parquetFiles(dataDir(table)).isEmpty)
+ checkAnswer(sql(s"SELECT * FROM $catalog.$ns.$table"),
Seq.empty[Row])
+ }
+ val expected = causeChain(referenceFailure).last
+ assert(expected.isInstanceOf[IllegalArgumentException],
expected.toString)
+ assert(
+ expected.getStackTrace.exists(frame =>
+
frame.getClassName.endsWith(".FixedLenByteArrayPlainValuesWriter") &&
+ frame.getMethodName == "writeBytes"),
+ expected.toString)
Review Comment:
This pins the Parquet internal frame
`FixedLenByteArrayPlainValuesWriter.writeBytes` and relies on
`write.parquet.dict-size-bytes=1`, so a parquet-mr refactor in one of the
tested Iceberg versions would fail it without any Comet change. Could it assert
on the error message instead?
##########
spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala:
##########
@@ -600,7 +601,30 @@ case class CometExecRule(session: SparkSession,
queryStagePrep: Boolean = false)
// A shuffle directly under the native write stays in the child stage
and is not wrapped
// again.
case op: IcebergWriteExec if
CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.get(op.conf) =>
- convertToComet(op, CometIcebergNativeWrite).getOrElse(op)
+ // Spark inserts AssertNotNull projections for nullable input to
required Iceberg fields.
+ // Keep their row-wise evaluation in Spark, including any sort or
limit above them, and
+ // convert the writer's input to Arrow only after Spark has evaluated
those operators.
+ val needsRowConversion = !op.child.isInstanceOf[CometNativeExec] &&
+ op.child.exists {
+ case project: ProjectExec =>
+
project.projectList.exists(_.exists(_.isInstanceOf[AssertNotNull]))
+ case _ => false
+ }
+ val rowConversionReasons = ListBuffer.empty[String]
+ val prepared =
+ if (needsRowConversion &&
+ CometSparkToColumnarExec.isRowConversionSupported(op.child,
rowConversionReasons) &&
+ CometIcebergNativeWrite.isRowInputEvaluationSupported(op,
rowConversionReasons)) {
+ val rows = ApplyColumnarRulesAndInsertTransitions(Seq.empty,
outputsColumnar = false)
+ .apply(op.child)
+ convertToComet(rows, CometSparkToColumnarExec)
+ .map(child => op.withNewChildren(Seq(child)))
+ .getOrElse(op)
+ } else {
+ rowConversionReasons.foreach(reason => withFallbackReason(op,
reason))
+ op
+ }
+ convertToComet(prepared, CometIcebergNativeWrite).getOrElse(op)
Review Comment:
`prepared` is a copy of `op`, so a fallback reason that
`convertToComet(prepared, CometIcebergNativeWrite)` records is tagged on
`prepared` and dropped when `getOrElse(op)` returns the original node. Before
this change those reasons landed on `op`. I expect `EXPLAIN` for a declined
write with a required-field projection to show no Comet reason, but I have not
run it. Could this copy the reasons back as the broadcast arm below does, or
call `CometIcebergNativeWrite.getSupportLevel(op)` before building the bridge,
which would also skip the bridge for ineligible writes?
--
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]
