Repository navigation
Native Iceberg write fails with "no more field nodes" on a struct column when its input is a Spark-to-Arrow conversion #7239
Description
Describe the bug
With native Iceberg writes enabled, a write whose input is a row-to-columnar conversion (e.g. a
DataFrame built from an RDD) fails on a nullable struct column from the second batch of a
partition. The task fails rather than falling back:
org.apache.comet.CometNativeException: C Data interface error: java.lang.IllegalArgumentException: no more field nodes for field a: Int(32, true) and vector []
at org.apache.comet.shaded.arrow.util.Preconditions.checkArgument(Preconditions.java:365)
at org.apache.comet.shaded.arrow.vector.VectorLoader.loadBuffers(VectorLoader.java:109)
at org.apache.comet.shaded.arrow.vector.VectorLoader.loadBuffers(VectorLoader.java:161)
at org.apache.comet.shaded.arrow.vector.VectorLoader.load(VectorLoader.java:88)
at org.apache.comet.shaded.arrow.vector.ipc.ArrowReader.loadRecordBatch(ArrowReader.java:213)
at org.apache.spark.sql.comet.execution.arrow.ColumnarBatchArrowReader.loadNextBatch(ColumnarBatchArrowReader.scala:63)
at org.apache.arrow.c.ArrayStreamExporter$ExportedArrayStreamPrivateData.getNext(ArrayStreamExporter.java:66)
at org.apache.comet.Native.executePlan(Native Method)
...
at org.apache.comet.CometExecIterator.getNextBatch(CometExecIterator.scala:237)
at org.apache.comet.CometExecIterator.hasNext(CometExecIterator.scala:281)
at org.apache.spark.sql.comet.CometIcebergWriteExec.drainNativePayload(CometIcebergWriteExec.scala:298)
at org.apache.spark.sql.comet.CometIcebergWriteExec.$anonfun$doExecute$9(CometIcebergWriteExec.scala:196)
This looks like the same root cause as #6685, fixed for native shuffle in #6689: the conversion
writes every batch into the same vectors, the batch is wrapped in ColumnarBatchArrowReader and
closed once native has it, and closing a struct vector drops its children, so the next batch has
none. #6689 changed only the shuffle input; CometIcebergWriteExec still reads a non-native child
the old way. Flat columns and map<string,string> are not affected.
Steps to reproduce
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructType;
public class StructIcebergWriteRepro {
public static void main(String[] args) throws Exception {
SparkSession spark = SparkSession.builder()
.master("local[1]")
.config("spark.plugins", "org.apache.spark.CometPlugin")
.config("spark.shuffle.manager",
"org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager")
.config("spark.memory.offHeap.enabled", "true")
.config("spark.memory.offHeap.size", "2g")
.config("spark.sql.extensions",
"org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions")
.config("spark.sql.catalog.local", "org.apache.iceberg.spark.SparkCatalog")
.config("spark.sql.catalog.local.type", "hadoop")
.config("spark.sql.catalog.local.warehouse", "/tmp/comet-struct-repro")
.config("spark.comet.iceberg.write.enabled", "true")
.config("spark.comet.write.iceberg.splitOperator.enabled", "true")
.config("spark.comet.sparkToColumnar.enabled", "true")
.config("spark.comet.batchSize", "2")
.getOrCreate();
spark.sql("CREATE TABLE local.db.t (id BIGINT, s STRUCT<a: INT>) USING iceberg");
StructType schema = new StructType()
.add("id", DataTypes.LongType)
.add("s", new StructType().add("a", DataTypes.IntegerType), true);
List<Integer> ids = IntStream.rangeClosed(1, 10).boxed().collect(Collectors.toList());
JavaRDD<Row> rows = JavaSparkContext.fromSparkContext(spark.sparkContext())
.parallelize(ids, 1)
.map(i -> RowFactory.create((long) i, RowFactory.create(i)));
spark.createDataFrame(rows, schema).writeTo("local.db.t").append();
}
}Verified on Comet 1.1.0 / Spark 3.5.6 / Iceberg 1.11.0 (Linux, JDK 17). The write succeeds with
spark.comet.sparkToColumnar.enabled=false or spark.comet.iceberg.write.enabled=false.
Expected behavior
The write succeeds, or the operator falls back to Spark's writer. Today, turning on native Iceberg
writes breaks writes to any table with a struct column fed by a row source.
Additional context
Suggested fix: read a CometNativeArrowSource child of CometIcebergWriteExec through
doExecuteAsArrowStream(), as #6689 does for shuffle. Related: #6685, #6689, #6607, #5649, #7051.
Activity