Spark: vectorized read fails on Parquet files with a shredded VARIANT when the variant column isn't projected #18430

Apache Iceberg version

1.12.0 (latest release)

Query engine

Spark

Please describe the bug 🐞

Spark 4.0. With 1.12.0 (#16292, vectorized Parquet reads for variant columns), any vectorized Spark read of a data file that contains a shredded VARIANT column fails if the query doesn't project that column. Vectorization is on by default, so a table written with write.parquet.shred-variants=true can no longer serve queries that select only its non-variant columns. Queries that do select the variant work, because they fall back to row reads.

Repro (Spark 4.0, Hadoop catalog):

CREATE TABLE db.t (id BIGINT, v VARIANT) USING iceberg
  TBLPROPERTIES ('format-version'='3', 'write.parquet.shred-variants'='true');
INSERT INTO db.t
  SELECT id, parse_json(concat('{"a":', id, ',"b":"x', id, '"}')) FROM range(1000);

SELECT sum(variant_get(v, '$.a', 'int')) FROM db.t;  -- OK (row-based read)
SELECT sum(id) FROM db.t;                            -- fails
SET spark.sql.iceberg.vectorization.enabled=false;
SELECT sum(id) FROM db.t;                            -- OK

java.lang.UnsupportedOperationException: Unsupported variant: shredded typed_value primitive
  at org.apache.iceberg.arrow.vectorized.VectorizedVariantVisitor.primitive(VectorizedVariantVisitor.java:97)
  at org.apache.iceberg.parquet.ParquetVariantVisitor.visitValue(ParquetVariantVisitor.java:207)
  at org.apache.iceberg.parquet.ParquetVariantVisitor.visitObjectFields(ParquetVariantVisitor.java:259)
  at org.apache.iceberg.parquet.ParquetVariantVisitor.visit(ParquetVariantVisitor.java:177)
  at org.apache.iceberg.parquet.TypeWithSchemaVisitor.visitVariant(TypeWithSchemaVisitor.java:218)
  at org.apache.iceberg.parquet.TypeWithSchemaVisitor.visitFields(TypeWithSchemaVisitor.java:208)
  at org.apache.iceberg.spark.data.vectorized.VectorizedSparkParquetReaders.buildReader(VectorizedSparkParquetReaders.java:61)
  at org.apache.iceberg.spark.source.BatchDataReader.open(BatchDataReader.java:128)

Analysis. The shredded-file fallback in #16292 only covers projected variants:

• SparkBatch.useParquetBatchReads() / supportsParquetBatchReads(...) check expectedSchema.columns(). A projected variant on a shredded table (by table property, or variant lower bounds in the file) disables batch reads. An unprojected variant is never considered, so batch reads are chosen.
• VectorizedSparkParquetReaders.buildReader then visits the file schema with TypeWithSchemaVisitor. For the unprojected variant group, visitVariant is reached with iType == null, but it still calls ParquetVariantVisitor.visit(variantGroup, visitor.variantVisitor()).
• VectorizedVariantVisitor throws on the shredded typed_value (primitive(...), and value(...) when typedResult != null) before anything checks whether the column is projected. findVariantField would return null for it anyway.

Spark 4.1 / 4.2 on main appear to share the same SparkBatch logic and reader path, but I've only reproduced on 4.0.

Related: #17736 (closed) proposed routing all projected variants to the row reader because plan-time shredding detection isn't sound, and noted that queries not projecting a variant "are unaffected and still vectorize". This issue is that unprojected case: it isn't unaffected, because the reader builder still walks the unprojected shredded group. #18121 (open) reworks the same SparkBatch check but still only over projected variants.

Possible fix: skip the variant sub-visit when the column isn't projected. Either have TypeWithSchemaVisitor.visitVariant call visitor.variant(null, group, null) when variant == null, or have VectorizedReaderBuilder return no variant visitor for unprojected columns. Either way the unprojected column is ignored, the same as for unprojected primitives.

Workarounds: project the variant (forces a row read), or disable vectorization (spark.sql.iceberg.vectorization.enabled=false, or the vectorization-enabled=false read option), at the cost of all vectorized reads on that scan.

cc @nssalian (author of #16292)

Willingness to contribute

• I can contribute a fix for this bug independently
• I would be willing to contribute a fix for this bug with guidance from the Iceberg community
• I cannot contribute a fix for this bug at this time