LinSimon-901101 commented on code in PR #7155:
URL: https://github.com/apache/datafusion-comet/pull/7155#discussion_r4511117198
##########
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:
I’ve kept both guards and documented error precedence, exception type, and
message preservation for the new row bridge only. Existing Arrow-native writes
are outside this guarantee.
##########
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:
I’ve added readback tests against the Java writer for nested arrays, struct
elements, and map values, including nulls and empty collections, using small
batches with AQE on and off.
##########
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:
I’ve replaced the stack-frame assertion with exception type and message
checks against the Java writer. The dictionary setting remains only to
reproduce the failure.
--
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]
