andygrove commented on code in PR #5361:
URL: https://github.com/apache/datafusion-comet/pull/5361#discussion_r3916282742
##########
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:
This checks that the level parses as a Java int, but the native side is
stricter than that. parquet-rs enforces zstd `1..=22`, gzip `0..=9` and brotli
`0..=11`, and `build_writer_properties` returns an error outside those, which
surfaces as a task failure rather than a fallback.
iceberg-java does not validate the level at all. I pulled
`Parquet$WriteBuilder$Context` out of the 1.10.0 runtime jar to check, and
unlike the four size properties, `compressionLevel` is kept as a raw `String`
and dropped straight into `parquet.compression.codec.zstd.level` /
`zlib.compress.level` / `compression.brotli.quality`. No `propertyAsInt`, no
`checkArgument`. So `write.parquet.compression-level=0` with zstd, or a
negative zstd fast level, writes fine on the stock path and dies mid-task on
this one.
Could this rule resolve the codec (`resolveCompression` is right there in
`IcebergWriteProtoTranslation`) and decline anything the corresponding
parquet-rs level type would reject? That is the same treatment
`requirePositiveIntParquetSizes` just gave the size properties. Boundary cases
worth pinning in the detection suite next to the existing `fast` test: zstd
`0`, zstd `-3`, zstd `23`, gzip `-1`, brotli `12`.
##########
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:
Now that `catalog_properties` picks up the translated `fs.s3a.*` settings,
could this override `stringArgs` the way `CometIcebergNativeScanExec` does?
`nativeOp` is the first field of the case class and there is no `stringArgs`
override, so `TreeNode.argString` falls through to its `case other => other ::
Nil` branch and calls `toString` on the protobuf, which gives the whole
TextFormat dump. That lands in `explain()`, in the SQL UI node description, and
in `SparkListenerSQLExecutionStart` in the event log. With the merge you just
added in `buildIcebergWriteProto`, the dump now includes `catalog_properties {
key: "s3.secret-access-key" ... }` for an S3A table whose keys live in the
Hadoop configuration.
Spark's own redaction will not catch this. `argString` redacts through
`redactMapString` only for args that are Scala `Map`s, and a protobuf message
falls past that to the plain `toString`.
`CometIcebergNativeScanExec.stringArgs` sidesteps exactly this by emitting
`output` plus a short descriptor instead of the proto, and it carries the same
property bag. Something like `Iterator(output, s"$dataLocation, $writerMode")`
here would keep the node readable and keep the secret out of the log.
Your new "write proto forwards fs.s3a.*" test looks like a good place to pin
it, by asserting the exec's `simpleString` does not contain the secret value.
--
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]
