sunchao commented on code in PR #5377:
URL: https://github.com/apache/datafusion-comet/pull/5377#discussion_r3815336537
##########
spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala:
##########
@@ -5160,6 +5160,165 @@ class CometIcebergNativeSuite
}
}
+ test("unprojected variant columns do not disable native Iceberg scans") {
+ assume(isSpark40Plus, "VARIANT type requires Spark 4.0+")
+ assume(icebergAvailable, "Iceberg not available in classpath")
+ assume(icebergVersionAtLeast(1, 10), "VARIANT type requires Iceberg 1.10+")
+ withTempIcebergDir { warehouseDir =>
+ withSQLConf(
+ "spark.sql.catalog.test_cat" ->
"org.apache.iceberg.spark.SparkCatalog",
+ "spark.sql.catalog.test_cat.type" -> "hadoop",
+ "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath,
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+ val table = "test_cat.db.variant_projection"
+ try {
+ spark.sql(
+ s"CREATE TABLE $table (id BIGINT, label STRING, data VARIANT)
USING iceberg " +
+ "TBLPROPERTIES ('format-version' = '3')")
+ spark.sql(s"""
+ INSERT INTO $table VALUES
+ (1, 'object', parse_json('{"num": 25}')),
+ (2, NULL, parse_json('null')),
+ (NULL, 'sql-null', NULL),
+ (4, 'array', parse_json('[1, 2]'))
+ """)
+
+ // Iceberg 1.10's Spark reader dereferences a null requested type
when it encounters an
+ // unprojected, annotated Variant. Keep Variant projected for the
Spark reference read,
+ // then discard it from the expected rows before checking the native
pruned projection.
+ var sparkAllRows = Seq.empty[Row]
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ sparkAllRows = spark
+ .sql(s"SELECT id, label, data FROM $table ORDER BY id NULLS
FIRST")
+ .collect()
+ .map(row => Row(row.get(0), row.get(1)))
+ .toSeq
+ }
+ assert(
+ sparkAllRows ==
+ Seq(Row(null, "sql-null"), Row(1L, "object"), Row(2L, null),
Row(4L, "array")))
+ val allRows = spark.sql(s"SELECT id, label FROM $table ORDER BY id
NULLS FIRST")
+ checkCometAnswer(allRows, sparkAllRows)
+ assertSingleNativeScan(allRows.queryExecution.executedPlan)
+
+ var sparkFilteredRows = Seq.empty[Row]
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ sparkFilteredRows = spark
+ .sql(s"SELECT id, data FROM $table " +
+ "WHERE label IS NOT NULL ORDER BY id NULLS FIRST")
+ .collect()
+ .map(row => Row(row.get(0)))
+ .toSeq
+ }
+ val filteredRows =
+ spark.sql(s"SELECT id FROM $table WHERE label IS NOT NULL ORDER BY
id NULLS FIRST")
+ checkCometAnswer(filteredRows, sparkFilteredRows)
+ assertSingleNativeScan(filteredRows.queryExecution.executedPlan)
+
+ checkIcebergNativeScanFallback(
+ s"SELECT id FROM $table WHERE " +
+ "try_variant_get(data, '$.num', 'int') > 20 ORDER BY id",
+ "projected VARIANT columns remain unsupported")
+
+ withSQLConf("spark.sql.iceberg.aggregate-push-down.enabled" ->
"false") {
+ val emptyProjection = spark.sql(s"SELECT COUNT(*) FROM $table")
+ assert(
+
collectIcebergNativeScans(emptyProjection.queryExecution.executedPlan).isEmpty,
+ "An empty projection must not read every field from a
VARIANT-bearing table")
+ }
+
+ val metadataOnly = spark.sql(s"SELECT _file FROM $table")
+ assert(
+
collectIcebergNativeScans(metadataOnly.queryExecution.executedPlan).isEmpty,
+ "A metadata-only projection must not read a VARIANT-bearing data
schema")
+ } finally {
+ spark.sql(s"DROP TABLE IF EXISTS $table PURGE")
+ }
+ }
+ }
+ }
+
+ test("projecting nested variant structs, arrays, and maps still falls back")
{
+ assume(isSpark40Plus, "VARIANT type requires Spark 4.0+")
+ assume(icebergAvailable, "Iceberg not available in classpath")
+ assume(icebergVersionAtLeast(1, 10), "VARIANT type requires Iceberg 1.10+")
+ withTempIcebergDir { warehouseDir =>
+ withSQLConf(
+ "spark.sql.catalog.test_cat" ->
"org.apache.iceberg.spark.SparkCatalog",
+ "spark.sql.catalog.test_cat.type" -> "hadoop",
+ "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath,
+ CometConf.COMET_ENABLED.key -> "true",
+ CometConf.COMET_EXEC_ENABLED.key -> "true",
+ CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+ val table = "test_cat.db.nested_variant_projection"
+ try {
+ spark.sql(
+ s"CREATE TABLE $table " +
+ "(id BIGINT, nested STRUCT<label: STRING, data: VARIANT>, " +
+ "variants ARRAY<VARIANT>, variants_by_key MAP<STRING, VARIANT>)
USING iceberg " +
+ "TBLPROPERTIES ('format-version' = '3')")
+ spark.sql(s"""
+ INSERT INTO $table VALUES
+ (1, named_struct('label', 'first', 'data', parse_json('{"num":
1}')),
+ array(parse_json('{"num": 2}'), parse_json('null')),
+ map('first', parse_json('{"num": 3}'))),
+ (2, named_struct('label', NULL, 'data', NULL),
+ array(CAST(NULL AS VARIANT)), map('sql-null', CAST(NULL AS
VARIANT))),
+ (3, NULL, NULL, NULL)
+ """)
+
+ var sparkScalarRows = Seq.empty[Row]
+ // Iceberg 1.10 cannot materialize nested Variant values in arrays
or maps, but
+ // reading their sizes still projects every Variant-bearing root for
Spark parity.
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ sparkScalarRows = spark
+ .sql(s"SELECT id, nested, size(variants), size(variants_by_key)
" +
Review Comment:
Fixed by removing the unnecessary `s` prefix from the first SQL string. The
second string still interpolates `$table`.
##########
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##########
@@ -766,12 +794,13 @@ case class CometScanRule(session: SparkSession)
}
fieldInfo match {
case Some((fieldName, fieldType)) =>
- if (fieldType.contains("struct")) {
+ if (fieldType.contains("struct") ||
fieldType.equalsIgnoreCase(
+ "variant")) {
Review Comment:
Added `variant equality deletes fall back to Spark`, which writes a real
Iceberg equality-delete file keyed on a `VARIANT` column, verifies the same
scalar projection is native before the delete and falls back afterward, and
checks the exact fallback reason. The assertion is intentionally plan-only
because Iceberg itself lacks a Variant equality comparator. Variant values are
constructed reflectively so the Spark 3.4 / Iceberg 1.5 profile still compiles.
--
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]
