mbutrovich commented on code in PR #5853:
URL: https://github.com/apache/datafusion-comet/pull/5853#discussion_r3992932032
##########
spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala:
##########
@@ -598,6 +598,182 @@ class CometIcebergNativeSuite
}
}
+ // V3 tables store positional deletes as deletion vectors (Puffin blobs)
instead of position
+ // delete files. This verifies Comet reads a V3 table's deletion vectors
natively and applies
+ // them, matching Spark.
+ test("MOR V3 table with DELETION VECTORS - verify deletes are applied") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+ // Format version 3 (and its deletion vectors) requires Iceberg 1.11+.
Older Iceberg
+ // rejects `format-version=3` at table creation.
+ assume(icebergVersionAtLeast(1, 11), "Iceberg V3 tables require Iceberg
1.11+")
+
+ 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") {
+
+ spark.sql("""
+ CREATE TABLE test_cat.db.dv_delete_test (
+ id INT,
+ name STRING,
+ value DOUBLE
+ ) USING iceberg
+ TBLPROPERTIES (
+ 'format-version' = '3',
+ 'write.delete.mode' = 'merge-on-read',
+ 'write.merge.mode' = 'merge-on-read'
+ )
+ """)
+
+ spark.sql("""
+ INSERT INTO test_cat.db.dv_delete_test
+ VALUES
+ (1, 'Alice', 10.5), (2, 'Bob', 20.3), (3, 'Charlie', 30.7),
+ (4, 'Diana', 15.2), (5, 'Eve', 25.8), (6, 'Frank', 35.0),
+ (7, 'Grace', 12.1), (8, 'Hank', 22.5)
+ """)
+
+ spark.sql("DELETE FROM test_cat.db.dv_delete_test WHERE id IN (2, 4,
6)")
+
+ // Confirm the delete produced a deletion vector (a Puffin delete
file), not a parquet
+ // position-delete file or a copy-on-write rewrite, so this test
actually exercises the
+ // deletion-vector read path.
+ val deleteFormats = spark
+ .sql("SELECT file_format FROM test_cat.db.dv_delete_test.files WHERE
content = 1")
+ .collect()
+ .map(_.getString(0))
+ .toSet
+ assert(
+ deleteFormats.contains("PUFFIN"),
+ s"expected a deletion vector (PUFFIN delete file) but found:
$deleteFormats")
+
+ checkIcebergNativeScan("SELECT * FROM test_cat.db.dv_delete_test ORDER
BY id")
+
+ spark.sql("DROP TABLE test_cat.db.dv_delete_test")
+ }
+ }
+ }
+
+ // A DELETE writes one deletion vector per data file it touches, and
Iceberg's DV writer packs
+ // every vector produced by one write task into a single Puffin file.
Vectors that apply to
+ // different data files therefore share a delete-file path and differ only
by their content
+ // offset, which is what makes the delete-file pool's identity key
load-bearing: keyed on the
+ // path alone (as the pool was before deletion vectors), the second and
later vectors dedup into
+ // the first and their data files come back with the deleted rows still in
them. Keyed on the
+ // whole message they stay distinct, and the path they share is interned
separately so it is
+ // still serialized once.
+ test("V3 deletion vectors sharing one Puffin file are interned but not
collapsed") {
Review Comment:
Hm, yeah migrations and encrypted tables would be good followups. I'll open
an issue.
--
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]
