jordepic commented on code in PR #5361:
URL: https://github.com/apache/datafusion-comet/pull/5361#discussion_r3917002498
##########
spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala:
##########
@@ -263,6 +275,15 @@ object CometIcebergNativeWrite extends
CometOperatorSerde[IcebergWriteExec] with
.map(_ => s"$key=true (variant shredding changes the parquet schema)")
}
+ // iceberg-java throws NumberFormatException at write time for a non-integer
level, while the
+ // native translation would silently substitute the codec default. Fall back
so the failure
+ // behaviour matches the stock path.
+ private val requireParseableCompressionLevel: TriggerRule = ctx =>
+ ctx.properties
+ .get(PropertyKeys.ParquetCompressionLevel)
+ .filter(v => scala.util.Try(java.lang.Integer.parseInt(v)).isFailure)
Review Comment:
Confirmed against the pinned parquet-rs 58 source: `CompressionLevel` bounds
are exactly zstd `1..=22`, gzip `0..=9`, brotli `0..=11`, and
`build_writer_properties` errors outside them — while, as you dug out,
iceberg-java keeps the level as a raw string with no validation at all. The
rule is now `requireNativeSupportedCompressionLevel`, delegating to a new
`IcebergWriteProtoTranslation.compressionLevelRejection` that resolves the
effective codec via `resolveCompression` and declines any level the
corresponding parquet-rs type would reject (codecs without a native level
concept — snappy/lz4/none — ignore the property on both sides, so any int
passes). All five of your boundary cases are pinned in the detection suite next
to the existing `fast` test, plus zstd `22` staying Compatible so the gate is
not over-broad, and the range matrix (rejected and accepted boundaries per
codec) is unit-tested in `IcebergWriteProtoTranslationSuite`.
##########
spark/src/main/scala/org/apache/spark/sql/comet/CometIcebergWriteExec.scala:
##########
@@ -0,0 +1,280 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.comet
+
+import org.apache.spark.TaskContext
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute,
AttributeReference}
+import org.apache.spark.sql.catalyst.expressions.UnsafeProjection
+import org.apache.spark.sql.comet.execution.arrow.CometArrowStream
+import org.apache.spark.sql.comet.util.{Utils => CometUtils}
+import org.apache.spark.sql.connector.write.{BatchWrite, WriterCommitMessage}
+import org.apache.spark.sql.execution.{ColumnarToRowTransition, SparkPlan,
UnaryExecNode}
+import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
+import org.apache.spark.sql.types.BinaryType
+import org.apache.spark.sql.vectorized.ColumnarBatch
+
+import com.google.protobuf.CodedOutputStream
+
+import org.apache.comet.CometExecIterator
+import org.apache.comet.iceberg.IcebergReflection
+import org.apache.comet.serde.OperatorOuterClass.Operator
+
+/**
+ * Native variant of [[IcebergWriteExec]]. Drives the iceberg-rust writer
stack via Comet's native
+ * execution pipeline; the JVM side decodes the per-task Avro-encoded
`DataFile` blob the native
+ * operator emits and packages it as a [[WriterCommitMessage]] so the outer
[[IcebergCommitExec]]
+ * consumes it unchanged.
+ *
+ * Selected by [[org.apache.comet.serde.operator.CometIcebergNativeWrite]]
when the table's
+ * properties allow it (parquet, V2, no encryption, ...) and the child plan is
fully Comet-native;
+ * otherwise the JVM path's [[IcebergWriteExec]] runs instead.
+ *
+ * @param nativeOp
+ * Template operator carrying the `IcebergWrite` proto. Per-task
`partition_id` /
+ * `task_attempt_id` get stamped on a copy at execution time.
+ * @param child
+ * Comet native child (must be a [[CometNativeExec]] so columnar batches
flow through FFI).
+ * @param batchWrite
+ * Shared with the outer [[IcebergCommitExec]] -- the same instance the
strategy materialised
+ * via `write.toBatch`. Used here only to provide the `dataLocation` /
partition spec needed by
+ * the native side; never invoked for commit.
+ * @param partitionSpecId
+ * Output partition spec id (from `SparkWrite.outputSpecId`). Decoded
`DataFile`s are stamped
+ * with this spec id; required because iceberg-rust's `DataFile` is
spec-agnostic at the wire.
+ */
+case class CometIcebergWriteExec(
+ nativeOp: Operator,
+ child: SparkPlan,
+ @transient batchWrite: BatchWrite,
+ @transient table: AnyRef,
+ partitionSpecId: Int)
+ extends CometNativeExec
+ with UnaryExecNode
+ // We consume Arrow batches (via FFI) and emit row-shaped commit messages,
so we are a
+ // columnar-to-row transition. Without this trait Spark's
+ // `ApplyColumnarRulesAndInsertTransitions` wedges a
`CometNativeColumnarToRowExec` between
+ // us and the Comet-native child, which would then fail
`child.executeColumnar()` in
+ // `doExecuteColumnar`.
+ with ColumnarToRowTransition {
+
+ override def originalPlan: SparkPlan = child
+
+ // Same output schema as IcebergWriteExec so the outer IcebergCommitExec
consumes the
+ // commit messages identically regardless of which inner exec emitted them.
+ override def output: Seq[Attribute] = Seq(
+ AttributeReference(IcebergWriteExec.CommitMessageColumn, BinaryType,
nullable = false)())
+
+ // Native exec emits a single Binary column; the surrounding command
framework expects rows, so
+ // the outer commit exec calls executeCollect on us. supportsColumnar =
false keeps Spark from
+ // inserting a ColumnarToRow that would clash with our (Nil-output-like) row
contract.
+ override def supportsColumnar: Boolean = false
+
+ override def executeCollect(): Array[InternalRow] = {
+ val rdd = doExecute()
+ // SparkPlan.executeCollect defaults to byteArrayRdd which goes through
UnsafeRow encoding;
+ // doExecute already projects each row through UnsafeProjection (see the
per-task closure
+ // below) so a plain `collect()` is safe.
+ rdd.collect()
+ }
+
+ override def serializedPlanOpt: SerializedPlan = {
+ val size = nativeOp.getSerializedSize
+ val bytes = new Array[Byte](size)
+ val codedOutput = CodedOutputStream.newInstance(bytes)
+ nativeOp.writeTo(codedOutput)
+ codedOutput.checkNoSpaceLeft()
+ SerializedPlan(Some(bytes))
+ }
+
+ override def withNewChildInternal(newChild: SparkPlan): SparkPlan =
copy(child = newChild)
+
+ override def nodeName: String = "CometIcebergWrite"
Review Comment:
Yikes — right, and this became load-bearing the moment the fs.s3a merge
landed. Added the `stringArgs` override modeled on
`CometIcebergNativeScanExec`: `Iterator(output, s"$dataLocation,
$writerMode")`, so explain()/UI/event-log render a short descriptor instead of
the protobuf TextFormat dump. Pinned exactly where you suggested: the fs.s3a
forwarding test now also asserts `simpleString(Int.MaxValue)` does not contain
the injected access key (and does contain the data location, so the node stays
readable).
--
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]
