huan233usc commented on code in PR #17149:
URL: https://github.com/apache/iceberg/pull/17149#discussion_r3566679504
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestSparkGeospatial.java:
##########
@@ -92,6 +94,159 @@ public void testGeospatialWkbReadBack(boolean vectorized) {
TABLE));
}
+ @ParameterizedTest
+ @ValueSource(strings = {"copy-on-write", "merge-on-read"})
+ public void testDeleteGeospatial(String mode) {
Review Comment:
Good call — done in 3698f8c73. The DELETE/UPDATE/MERGE/nested tests now
parametrize over `(mode, vectorized)` via `@CsvSource`, setting
`read.parquet.vectorization.enabled` per table, so the fallback is exercised
explicitly with vectorization both on and off rather than relying on the
default.
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestSparkGeospatial.java:
##########
@@ -92,6 +94,159 @@ public void testGeospatialWkbReadBack(boolean vectorized) {
TABLE));
}
+ @ParameterizedTest
+ @ValueSource(strings = {"copy-on-write", "merge-on-read"})
+ public void testDeleteGeospatial(String mode) {
+ // Exercise both row-level modes: copy-on-write (the default) rewrites the
surviving rows into a
+ // new data file, while merge-on-read writes a deletion vector. Both must
read the geo values
+ // back out correctly. Write both rows into one data file (COALESCE(1)) so
the merge-on-read
+ // delete leaves a survivor in that file, forcing a deletion vector rather
than a whole-file
+ // removal. Filter on id since Spark has no spatial predicate.
+ String deleteTable = CATALOG + ".default.geo_delete";
+ sql("DROP TABLE IF EXISTS %s", deleteTable);
+ sql(
+ "CREATE TABLE %s (id BIGINT, geom GEOMETRY(4326), geog
GEOGRAPHY(4326)) USING iceberg "
+ + "TBLPROPERTIES ('format-version'='3', 'write.delete.mode'='%s')",
+ deleteTable, mode);
+ sql(
+ "INSERT INTO %s SELECT /*+ COALESCE(1) */ id, "
+ + "CASE WHEN geom_wkb IS NULL THEN NULL "
+ + "ELSE st_setsrid(st_geomfromwkb(geom_wkb), 4326) END, "
+ + "CASE WHEN geog_wkb IS NULL THEN NULL "
+ + "ELSE st_setsrid(st_geogfromwkb(geog_wkb), 4326) END "
+ + "FROM VALUES (1, X'%s', X'%s'), (2, CAST(NULL AS BINARY),
CAST(NULL AS BINARY)) "
+ + "AS v(id, geom_wkb, geog_wkb)",
+ deleteTable, GEOM_WKB, GEOG_WKB);
+
+ sql("DELETE FROM %s WHERE id = 1", deleteTable);
+
+ assertEquals(
+ "Only the null-geo row should remain after the delete",
+ ImmutableList.of(row(2L, null, null)),
+ sql(
+ "SELECT id, hex(st_asbinary(geom)), hex(st_asbinary(geog)) FROM %s
ORDER BY id",
+ deleteTable));
+
+ // A merge-on-read delete of a row from a multi-row file writes a deletion
vector (format v3),
+ // recorded on the 'delete' snapshot; copy-on-write instead rewrites the
file under an
+ // 'overwrite' snapshot and adds no deletion vector. Select the
merge-on-read delete snapshot by
+ // operation so the assertion does not depend on committed_at ordering,
whose millisecond
+ // granularity could tie with the insert snapshot.
+ if (mode.equals("merge-on-read")) {
+ assertThat(
+ scalarSql(
+ "SELECT summary['added-dvs'] FROM %s.snapshots WHERE
operation = 'delete'",
Review Comment:
I'd like to keep the SQL form here: this test extends `TestBase` (not
`TestBaseWithCatalog`), so `validationCatalog`/`tableIdent` aren't available,
and loading the table through the catalog by name would add setup that the
pure-SQL assertion avoids. The `operation = 'delete'` filter keeps it
deterministic. Happy to switch if you'd prefer the class move to
`TestBaseWithCatalog`.
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestSparkGeospatial.java:
##########
@@ -92,6 +94,159 @@ public void testGeospatialWkbReadBack(boolean vectorized) {
TABLE));
}
+ @ParameterizedTest
+ @ValueSource(strings = {"copy-on-write", "merge-on-read"})
+ public void testDeleteGeospatial(String mode) {
+ // Exercise both row-level modes: copy-on-write (the default) rewrites the
surviving rows into a
+ // new data file, while merge-on-read writes a deletion vector. Both must
read the geo values
+ // back out correctly. Write both rows into one data file (COALESCE(1)) so
the merge-on-read
+ // delete leaves a survivor in that file, forcing a deletion vector rather
than a whole-file
+ // removal. Filter on id since Spark has no spatial predicate.
+ String deleteTable = CATALOG + ".default.geo_delete";
+ sql("DROP TABLE IF EXISTS %s", deleteTable);
+ sql(
+ "CREATE TABLE %s (id BIGINT, geom GEOMETRY(4326), geog
GEOGRAPHY(4326)) USING iceberg "
+ + "TBLPROPERTIES ('format-version'='3', 'write.delete.mode'='%s')",
+ deleteTable, mode);
+ sql(
+ "INSERT INTO %s SELECT /*+ COALESCE(1) */ id, "
+ + "CASE WHEN geom_wkb IS NULL THEN NULL "
+ + "ELSE st_setsrid(st_geomfromwkb(geom_wkb), 4326) END, "
+ + "CASE WHEN geog_wkb IS NULL THEN NULL "
+ + "ELSE st_setsrid(st_geogfromwkb(geog_wkb), 4326) END "
+ + "FROM VALUES (1, X'%s', X'%s'), (2, CAST(NULL AS BINARY),
CAST(NULL AS BINARY)) "
+ + "AS v(id, geom_wkb, geog_wkb)",
+ deleteTable, GEOM_WKB, GEOG_WKB);
+
+ sql("DELETE FROM %s WHERE id = 1", deleteTable);
+
+ assertEquals(
+ "Only the null-geo row should remain after the delete",
+ ImmutableList.of(row(2L, null, null)),
+ sql(
+ "SELECT id, hex(st_asbinary(geom)), hex(st_asbinary(geog)) FROM %s
ORDER BY id",
+ deleteTable));
+
+ // A merge-on-read delete of a row from a multi-row file writes a deletion
vector (format v3),
+ // recorded on the 'delete' snapshot; copy-on-write instead rewrites the
file under an
+ // 'overwrite' snapshot and adds no deletion vector. Select the
merge-on-read delete snapshot by
+ // operation so the assertion does not depend on committed_at ordering,
whose millisecond
+ // granularity could tie with the insert snapshot.
+ if (mode.equals("merge-on-read")) {
+ assertThat(
+ scalarSql(
+ "SELECT summary['added-dvs'] FROM %s.snapshots WHERE
operation = 'delete'",
+ deleteTable))
+ .as("Merge-on-read delete should add a deletion vector")
+ .isEqualTo("1");
+ }
+
+ sql("DROP TABLE IF EXISTS %s", deleteTable);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"copy-on-write", "merge-on-read"})
+ public void testUpdateGeospatial(String mode) {
+ // Update the first row's geometry; the untouched row must round-trip
unchanged. Copy-on-write
+ // rewrites the untouched row, merge-on-read writes a deletion vector plus
the new row.
+ String updateTable = CATALOG + ".default.geo_update";
+ sql("DROP TABLE IF EXISTS %s", updateTable);
+ sql(
+ "CREATE TABLE %s (id BIGINT, geom GEOMETRY(4326), geog
GEOGRAPHY(4326)) USING iceberg "
+ + "TBLPROPERTIES ('format-version'='3', 'write.update.mode'='%s')",
+ updateTable, mode);
+ sql(
+ "INSERT INTO %s VALUES "
+ + "(1, st_setsrid(st_geomfromwkb(X'%s'), 4326),
st_setsrid(st_geogfromwkb(X'%s'), 4326)), "
+ + "(2, st_setsrid(st_geomfromwkb(X'%s'), 4326), NULL)",
+ updateTable, GEOM_WKB, GEOG_WKB, GEOM_WKB);
+
+ sql(
+ "UPDATE %s SET geom = st_setsrid(st_geomfromwkb(X'%s'), 4326) WHERE id
= 1",
+ updateTable, OTHER_WKB);
+
+ assertEquals(
+ "The updated row changes and the untouched row round-trips unchanged",
+ ImmutableList.of(
+ row(1L, OTHER_WKB.toUpperCase(Locale.ROOT),
GEOG_WKB.toUpperCase(Locale.ROOT)),
+ row(2L, GEOM_WKB.toUpperCase(Locale.ROOT), null)),
+ sql(
+ "SELECT id, hex(st_asbinary(geom)), hex(st_asbinary(geog)) FROM %s
ORDER BY id",
+ updateTable));
+
+ sql("DROP TABLE IF EXISTS %s", updateTable);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"copy-on-write", "merge-on-read"})
+ public void testMergeGeospatial(String mode) {
+ // Source updates id=1 (matched) and inserts id=3 (not matched); geo
values flow through both.
+ String mergeTable = CATALOG + ".default.geo_merge";
+ sql("DROP TABLE IF EXISTS %s", mergeTable);
+ sql(
+ "CREATE TABLE %s (id BIGINT, geom GEOMETRY(4326), geog
GEOGRAPHY(4326)) USING iceberg "
+ + "TBLPROPERTIES ('format-version'='3', 'write.merge.mode'='%s')",
+ mergeTable, mode);
+ sql(
+ "INSERT INTO %s VALUES "
+ + "(1, st_setsrid(st_geomfromwkb(X'%s'), 4326),
st_setsrid(st_geogfromwkb(X'%s'), 4326)), "
+ + "(2, NULL, NULL)",
+ mergeTable, GEOM_WKB, GEOG_WKB);
+
+ sql(
+ "MERGE INTO %s AS t USING ("
+ + "SELECT 1 AS id, st_setsrid(st_geomfromwkb(X'%s'), 4326) AS
geom, "
+ + "st_setsrid(st_geogfromwkb(X'%s'), 4326) AS geog "
+ + "UNION ALL "
+ + "SELECT 3 AS id, st_setsrid(st_geomfromwkb(X'%s'), 4326) AS
geom, "
+ + "st_setsrid(st_geogfromwkb(X'%s'), 4326) AS geog) AS s "
+ + "ON t.id = s.id "
+ + "WHEN MATCHED THEN UPDATE SET t.geom = s.geom, t.geog = s.geog "
+ + "WHEN NOT MATCHED THEN INSERT *",
+ mergeTable, OTHER_WKB, GEOG_WKB, GEOM_WKB, GEOG_WKB);
+
+ assertEquals(
+ "Matched row is updated and unmatched row is inserted",
+ ImmutableList.of(
+ row(1L, OTHER_WKB.toUpperCase(Locale.ROOT),
GEOG_WKB.toUpperCase(Locale.ROOT)),
+ row(2L, null, null),
+ row(3L, GEOM_WKB.toUpperCase(Locale.ROOT),
GEOG_WKB.toUpperCase(Locale.ROOT))),
+ sql(
+ "SELECT id, hex(st_asbinary(geom)), hex(st_asbinary(geog)) FROM %s
ORDER BY id",
+ mergeTable));
+
+ sql("DROP TABLE IF EXISTS %s", mergeTable);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"copy-on-write", "merge-on-read"})
+ public void testDeleteNestedGeometry(String mode) {
Review Comment:
Good catch — added in 3698f8c73. The nested test (now
`testDeleteNestedGeospatial`) uses `STRUCT<c1, geom, geog>` and asserts that
after the delete the surviving row's non-null nested geometry and null nested
geography both round-trip.
--
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]
