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: [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]
