andygrove commented on code in PR #5853:
URL: https://github.com/apache/datafusion-comet/pull/5853#discussion_r3991434033
##########
native/proto/src/proto/operator.proto:
##########
@@ -407,14 +414,24 @@ message IcebergDeleteFile {
// Equality field IDs (empty for positional deletes)
repeated int32 equality_ids = 4;
- // Fields 5-7 are reserved for the deletion-vector coordinates on the
dv-read branch
- // (referenced_data_file / content_offset / content_size_in_bytes) so the
two features can merge
- // in either order without a tag collision.
- reserved 5, 6, 7;
+ // Deletion vector coordinates, set only for V3 deletion vectors.
referenced_data_file is the
+ // data file the vector applies to; content_offset and content_size_in_bytes
locate the
+ // deletion-vector-v1 blob within its Puffin file.
+ optional string referenced_data_file = 5;
Review Comment:
This comment (and the matching ones in `CometIcebergNativeScan.scala` and
`planner.rs`) says `referenced_data_file` is set only for V3 deletion vectors,
but Iceberg also populates `DeleteFile.referencedDataFile()` for file-scoped
Parquet position deletes. The javadoc calls this out, and `BaseFile` stores it
from the manifest entry on 1.7+. So V2 tables will carry it on Parquet delete
files too.
iceberg-rust ignores it on the Parquet path so nothing breaks, but could we
either clear it when the format is not Puffin, or reword the comments so nobody
later keys DV detection on this field? The format field is the discriminator
and it would be good for the comments to say so. Fine as a follow-up.
##########
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:
Two coverage ideas for a follow-up:
1. A test where one scan sees both delete formats: create a V2 table with
file-scoped position deletes, `ALTER TABLE ... SET TBLPROPERTIES
('format-version'='3')`, then DELETE from a data file that had no prior
deletes. That exercises the per-delete-file `file_format` in one plan and
confirms Java's planning hands Comet the DV rather than both for a file that
gets rewritten.
2. An encrypted V3 table DV case. iceberg-rust has a separate decrypt branch
in `load_deletion_vector` that reads `content_offset` over the plaintext file
view, and forwarding `key_metadata` on a Puffin entry is the one thing the
existing encryption tests do not cover.
--
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]
