peterxcli commented on code in PR #5377:
URL: https://github.com/apache/datafusion-comet/pull/5377#discussion_r3815692771
##########
spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala:
##########
@@ -5160,6 +5160,224 @@ 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("variant equality deletes fall back to Spark") {
+ 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 tableName = "variant_equality_delete"
+ val table = s"test_cat.db.$tableName"
+ try {
+ spark.sql(
+ s"CREATE TABLE $table (id BIGINT, data VARIANT) USING iceberg " +
+ "TBLPROPERTIES ('format-version' = '3')")
+ spark.sql(
+ s"INSERT INTO $table VALUES " +
+ "(1, parse_json('1')), (2, parse_json('2'))")
+
+ val nativePlan = spark.sql(s"SELECT id FROM $table ORDER BY id")
+ assertSingleNativeScan(nativePlan.queryExecution.executedPlan)
+
+ // Variant classes do not exist in the Iceberg versions used by
older Spark profiles.
+ val variantsClass =
Class.forName("org.apache.iceberg.variants.Variants")
+ val metadata = variantsClass.getMethod("emptyMetadata").invoke(null)
+ val value = variantsClass
+ .getMethod("of", java.lang.Integer.TYPE)
+ .invoke(null, Integer.valueOf(2))
+ val variant = Class
+ .forName("org.apache.iceberg.variants.Variant")
+ .getMethod(
+ "of",
+ Class.forName("org.apache.iceberg.variants.VariantMetadata"),
+ Class.forName("org.apache.iceberg.variants.VariantValue"))
+ .invoke(null, metadata, value)
Review Comment:
asked ai how to clean this up. elsewhere.
```suggestion
val records = IcebergGenerics
.read(Spark3Util.loadIcebergTable(spark, table))
.where(Expressions.equal("id", 2L))
.select("data")
.build()
val variant =
try {
val rows = records.iterator()
assert(rows.hasNext)
rows.next().getField("data")
} finally {
records.close()
}
```
--
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]
