cornelcreanga opened a new issue, #7239:
URL: https://github.com/apache/datafusion-comet/issues/7239
### 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
```java
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.
--
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]
