Implement VariantVisitor (parquet) to support MERGE INTO operations

[enriquh]

Feature Request / Improvement

Feature description

Implement schema visitor for variant datatype in parquet. This is currently not implemented on the code

Steps to reproduce

1. Create a table with a variant field
2. Perform a MERGE INTO operation to update records on the target table using a variant property on the condition
   Result
3. Issue also happens when sub-variant extraction are not included in the condition.

UnsupportedOperationException: Not implemented for variant at org.apache.iceberg.parquet.TypeWithSchemaVisitor.variant(TypeWithSchemaVisitor.java:242)

spark.sql(f"""
CREATE TABLE {table_name} (
id BIGINT,
variant_data VARIANT
) USING iceberg
TBLPROPERTIES ('format-version' = '3')
""")

merge_sql_2 = f"""
MERGE INTO {table_name} AS target
USING merge_source AS source
ON variant_get(target.variant_data, '$.name', 'string') = variant_get(source.variant_data, '$.name', 'string')
AND target.id = source.id
WHEN MATCHED THEN
UPDATE SET target.variant_data = source.variant_data
WHEN NOT MATCHED THEN
INSERT (id, variant_data) VALUES (source.id, source.variant_data)
"""

Expected results

Ability to execute MERGE operation on tables with variant fields being able to use variant_get in Spark to reduce scanned data.

Environment details

• Spark 4.0
• Iceberg 1.11 (build from main)

Query engine

Spark

Willingness to contribute

• I can contribute this improvement/feature independently
• I would be willing to contribute this improvement/feature with guidance from the Iceberg community
• I cannot contribute this improvement/feature at this time

[geruh]

Hey @enriquh,

I tried running this scenario but haven't ran into the exception. I'm running something like:

CREATE TABLE test_table (
  id BIGINT, 
  data VARIANT
) USING iceberg 
TBLPROPERTIES ('format-version' = '3');

INSERT INTO test_table VALUES 
  (1, parse_json('{"field": "foo", "num": 25}')),
  (2, parse_json('{"field": "bar", "num": 30}')),
  (3, parse_json('{"field": "baz", "num": 35}'));

CREATE OR REPLACE TEMP VIEW merge_source AS 
SELECT id, data FROM VALUES 
  (1, parse_json('{"field": "foo", "num": 26}')),
  (2, parse_json('{"field": "bar", "num": 31}')),
  (4, parse_json('{"field": "qux", "num": 40}'))
AS source_data(id, data);

-- MERGE with variant_get in condition
MERGE INTO test_table AS target
USING merge_source AS source
ON variant_get(target.data, '$.field', 'string') = 
   variant_get(source.data, '$.field', 'string')
AND target.id = source.id
WHEN MATCHED THEN 
  UPDATE SET target.data = source.data
WHEN NOT MATCHED THEN 
  INSERT (id, data) VALUES (source.id, source.data);

-- Verify Merge wsa success
SELECT id, 
  variant_get(data, '$.field', 'string') as field,
  variant_get(data, '$.num', 'int') as num
FROM test_table 
ORDER BY id;

+---+------------------------+
|id |data                    |
+---+------------------------+
|5  |{"field":"qux","num":40}|
|3  |{"field":"baz","num":35}|
|4  |NULL                    |
|2  |{"field":"bar","num":31}|
|1  |{"field":"foo","num":26}|
+---+------------------------+

This query for me is working fine on current main with Spark 4.0 and format v3.

I had a hunch one of these PRs fixed but I reverted both PR #14261 and PR #14588 and it still works. So I can't actually reproduce the UnsupportedOperationException you're seeing.

Maybe it's something about your environment or maybe the test isn't hitting the same code path you are. Or maybe you were on a different version when you hit this.

What Iceberg commit were you using when you saw this? And can you paste the full stacktrace? That would help me figure out what's different.

[Neuw84]

Hi @geruh ,

I was testing this with @enriquh. The commit I use to compile the project is specifically this: 3d9feaa

The line with the Exception is this:

iceberg/parquet/src/main/java/org/apache/iceberg/parquet/TypeWithSchemaVisitor.java

Line 242 in cf27769

throw new UnsupportedOperationException("Not implemented for variant");

I checked that on the Main branch the implementation is also missing.

[lyogev]

Hi! happening to me as well.
I see @rdblue wrote something about this here

#12139 (comment)

I wonder why it's getting to this code:

Caused by: java.lang.UnsupportedOperationException: Not implemented for variant
	at org.apache.iceberg.parquet.TypeWithSchemaVisitor.variant(TypeWithSchemaVisitor.java:242)
	at org.apache.iceberg.parquet.TypeWithSchemaVisitor.visitVariant(TypeWithSchemaVisitor.java:221)
	at org.apache.iceberg.parquet.TypeWithSchemaVisitor.visit(TypeWithSchemaVisitor.java:67)
	at org.apache.iceberg.parquet.TypeWithSchemaVisitor.visitField(TypeWithSchemaVisitor.java:193)
	at org.apache.iceberg.parquet.TypeWithSchemaVisitor.visitFields(TypeWithSchemaVisitor.java:208)
	at org.apache.iceberg.parquet.TypeWithSchemaVisitor.visit(TypeWithSchemaVisitor.java:49)
	at org.apache.iceberg.spark.data.vectorized.VectorizedSparkParquetReaders.buildReader(VectorizedSparkParquetReaders.java:64)
	at org.apache.iceberg.spark.data.vectorized.VectorizedSparkParquetReaders.buildReader(VectorizedSparkParquetReaders.java:82)

This specifically happens in a MERGE operation

Setting
spark.sql.iceberg.vectorization.enabled=false works

[vakarisbk]

This reliably reproduces once a data file exists on disk. The key is that on a fresh/empty table, the MERGE scan has no files so the error doesn't trigger.

Minimal repro (Spark 4.0, Iceberg built from main, format-version 3):

CREATE TABLE test_variant (id BIGINT, data VARIANT)
USING iceberg TBLPROPERTIES ('format-version' = '3');

INSERT INTO test_variant VALUES (1, parse_json('{"a": 1}'));

MERGE INTO test_variant AS t
USING (SELECT 1 AS id, parse_json('{"a": 2}') AS data) AS s
ON t.id = s.id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;

spark.sql.iceberg.vectorization.enabled=false avoids the crash.

[qlong]

This issue is needed for vectorized read on variant. Is related to #10392

[nssalian]

@qlong I'm happy to work on this. Could someone assign this to me? I'll push up a fix for this