You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "huan233usc (via GitHub)" <gi...@apache.org> on 2026/07/09 20:44:08 UTC
[PR] Spark 4.1: Test geometry and geography DML and fall back from vectorized reads [iceberg]
huan233usc opened a new pull request, #17149:
URL: https://github.com/apache/iceberg/pull/17149
Adds end-to-end Spark SQL DML tests for `geometry` and `geography` columns, and
fixes a vectorized-read bug those tests surfaced. This completes the Phase-1
geo end-to-end item (row-level DML with deletion vectors / nested geo / nulls)
and builds on #17073 (Spark geo value path) and #16982 (Parquet geo WKB values).
## Tests
`TestSparkGeospatial` previously covered only `INSERT` → `SELECT` read-back. This
adds row-level DML on a geo table under **merge-on-read + format version 3**
(deletion vectors):
- `testDeleteGeospatialMergeOnRead` — `DELETE` leaves a survivor in a single data
file, and the snapshot summary confirms a deletion vector was written
(`added-dvs = 1`); the surviving null-geo row round-trips.
- `testUpdateGeospatialMergeOnRead` — `UPDATE` swaps a row's geometry; others
unchanged.
- `testMergeGeospatialMergeOnRead` — `MERGE` updates a matched row and inserts a
not-matched row, both carrying geo values.
- `testDeleteNestedGeometryMergeOnRead` — geometry nested in a `STRUCT`; delete by
a nested non-geo field, surviving nested geometry (and a null nested geometry)
still round-trip.
Geo values are asserted via `hex(st_asbinary(...))` round-trips and DML filters on
non-geo columns, since Spark 4.1 has no spatial predicates.
## Vectorization fallback (production fix)
`geometry`/`geography` are primitive types, so they passed
`SparkBatch.supportsParquetBatchReads(NestedField)` and were routed to the
vectorized Parquet reader — but there is no Arrow geo vector yet, so scanning a geo
column with vectorization enabled threw
`UnsupportedOperationException: Unsupported primitive type: geometry`. This excludes
the two geo types from vectorized reads so they fall back to the row-based reader.
With this fix, `testGeospatialWkbReadBack` now runs (no longer skipped) with
vectorization both **off and on**, proving the fallback. A dedicated Arrow geo
vector remains future work.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Test geometry and geography DML and fall back from vectorized reads [iceberg]
Posted by "huan233usc (via GitHub)" <gi...@apache.org>.
huan233usc commented on code in PR #17149:
URL: https://github.com/apache/iceberg/pull/17149#discussion_r3566679002
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatch.java:
##########
@@ -169,7 +170,18 @@ private boolean supportsParquetBatchReads(ScanTask task) {
}
private boolean supportsParquetBatchReads(Types.NestedField field) {
- return field.type().isPrimitiveType() || MetadataColumns.isMetadataColumn(field.fieldId());
+ if (MetadataColumns.isMetadataColumn(field.fieldId())) {
+ return true;
+ }
+
+ Type type = field.type();
+ // Geometry and geography are primitive types but have no Arrow vector yet, so they must be
+ // read through the non-vectorized reader.
+ if (type.typeId() == Type.TypeID.GEOMETRY || type.typeId() == Type.TypeID.GEOGRAPHY) {
Review Comment:
Done in 3698f8c73 — updated the comment above `useParquetBatchReads()` to note geometry and geography are primitives that are intentionally excluded.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Test geometry and geography DML and fall back from vectorized reads [iceberg]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on code in PR #17149:
URL: https://github.com/apache/iceberg/pull/17149#discussion_r3562728548
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatch.java:
##########
@@ -169,7 +170,18 @@ private boolean supportsParquetBatchReads(ScanTask task) {
}
private boolean supportsParquetBatchReads(Types.NestedField field) {
- return field.type().isPrimitiveType() || MetadataColumns.isMetadataColumn(field.fieldId());
+ if (MetadataColumns.isMetadataColumn(field.fieldId())) {
+ return true;
+ }
+
+ Type type = field.type();
+ // Geometry and geography are primitive types but have no Arrow vector yet, so they must be
+ // read through the non-vectorized reader.
+ if (type.typeId() == Type.TypeID.GEOMETRY || type.typeId() == Type.TypeID.GEOGRAPHY) {
Review Comment:
Correct fix and the right layer. Nit: the method-level comment above `useParquetBatchReads()` still reads `// - only primitives or metadata columns are projected`, which is now slightly stale since geometry/geography are primitives that are intentionally excluded here. A one-line tweak there would keep the two in sync.
##########
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:
Nit: the DV assertion queries the `snapshots` metadata view via SQL. That works, and the `operation = 'delete'` filter nicely avoids the `committed_at` tie. Elsewhere in Spark 4.1 tests (e.g. `TestDeleteFrom.truncateWithDVs`) the convention is to load the table via the catalog and assert on `snapshot().summary()` with `SnapshotSummary.ADDED_DVS_PROP`, which is a bit more direct and avoids depending on the snapshots view. Either is acceptable.
##########
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:
Nit: these DML tests rely on the default `read.parquet.vectorization.enabled=true` to exercise the fallback path. That's fine today, but parametrizing vectorization on/off (like `testGeospatialWkbReadBack` does) would make the intent explicit and guard against a future default change. Optional.
##########
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:
Question: this nested case covers `GEOMETRY` inside a struct but not nested `GEOGRAPHY`. Intentional scope for Phase 1, or worth a symmetric geography case?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Test geometry and geography DML and fall back from vectorized reads [iceberg]
Posted by "huan233usc (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Test geometry and geography DML and fall back from vectorized reads [iceberg]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho commented on PR #17149:
URL: https://github.com/apache/iceberg/pull/17149#issuecomment-4973904570
Merged, thanks @huan233usc !
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Test geometry and geography DML and fall back from vectorized reads [iceberg]
Posted by "huan233usc (via GitHub)" <gi...@apache.org>.
huan233usc commented on PR #17149:
URL: https://github.com/apache/iceberg/pull/17149#issuecomment-4951921887
Thanks for the review @szehon-ho! Addressed in 3698f8c73:
- Synced the `useParquetBatchReads` comment to note geometry/geography are excluded.
- Parametrized DELETE/UPDATE/MERGE/nested over `(mode, vectorized)` so the fallback runs with vectorization both on and off.
- Extended the nested-delete test to a struct with both a geometry and a geography field, asserting a non-null nested geometry and a null nested geography round-trip.
On the DV assertion: kept the SQL form since this test extends `TestBase` (no `validationCatalog`), with the `operation = 'delete'` filter for determinism — replied inline, happy to revisit.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Test geometry and geography DML and fall back from vectorized reads [iceberg]
Posted by "huan233usc (via GitHub)" <gi...@apache.org>.
huan233usc commented on code in PR #17149:
URL: https://github.com/apache/iceberg/pull/17149#discussion_r3566679543
##########
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:
Thanks, that's a fair point. I looked at how other tests in this package handle it: `TestSelect` (same `spark/.../sql/` package) also reads the `.snapshots` metadata view via SQL, whereas the `currentSnapshot().summary()` + `SnapshotSummary.ADDED_DVS_PROP` convention you mention is used by the `CatalogTestBase` tests (e.g. `TestDeleteFrom`), which have `validationCatalog`/`tableIdent` available. This test extends `TestBase` and creates its table via SQL in a hadoop catalog (Hive doesn't support geo yet), so there's no `validationCatalog` to load from, and the SQL form keeps it consistent with the sibling tests here. I've kept the SQL assertion with the `operation = 'delete'` filter for determinism — but happy to switch to the catalog form if you'd prefer it.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Spark 4.1: Test geometry and geography DML and fall back from vectorized reads [iceberg]
Posted by "szehon-ho (via GitHub)" <gi...@apache.org>.
szehon-ho merged PR #17149:
URL: https://github.com/apache/iceberg/pull/17149
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org