From 67619a71276588b129b9688fa555789187319c00 Mon Sep 17 00:00:00 2001 From: Xin Huang Date: Mon, 6 Jul 2026 17:10:14 -0700 Subject: [PATCH] Core: Read and write geometry and geography values in Avro TypeToSchema mapped Iceberg types to Avro schema for every primitive except geometry and geography, so converting a schema with a geo column threw UnsupportedOperationException before any value could be read or written. Map both to an Avro bytes field carrying WKB, per the Avro type mapping in the spec (binary, geometry, and geography all use bytes). The value read/write paths dispatch on the Avro physical type, so once the schema is bytes the existing byteBuffers reader/writer handle geo unchanged in both the generic and internal object models (geo values are WKB ByteBuffers, like binary). Enable the shared DataTest geospatial coverage for the Avro object models (TestGenericAvro, TestInternalAvro, generic data TestGenericData) and add the GEOMETRY/GEOGRAPHY -> ByteBuffer handling to RandomAvroData and AvroTestHelpers that the generic Avro path needs, mirroring RandomInternalData/InternalTestHelpers. --- .../org/apache/iceberg/avro/TypeToSchema.java | 5 ++ .../apache/iceberg/avro/AvroTestHelpers.java | 2 + .../apache/iceberg/avro/RandomAvroData.java | 2 + .../iceberg/avro/TestAvroDataWriter.java | 62 +++++++++++++++++++ .../iceberg/avro/TestAvroEncoderUtil.java | 5 ++ .../apache/iceberg/avro/TestGenericAvro.java | 5 ++ .../apache/iceberg/avro/TestInternalAvro.java | 5 ++ .../iceberg/data/avro/TestGenericData.java | 5 ++ .../apache/iceberg/parquet/TestParquet.java | 33 ++++++++++ 9 files changed, 124 insertions(+) diff --git a/core/src/main/java/org/apache/iceberg/avro/TypeToSchema.java b/core/src/main/java/org/apache/iceberg/avro/TypeToSchema.java index 4442695b7a44..a57406175abb 100644 --- a/core/src/main/java/org/apache/iceberg/avro/TypeToSchema.java +++ b/core/src/main/java/org/apache/iceberg/avro/TypeToSchema.java @@ -264,6 +264,11 @@ public Schema primitive(Type.PrimitiveType primitive) { case BINARY: primitiveSchema = BINARY_SCHEMA; break; + case GEOMETRY: + case GEOGRAPHY: + // geometry and geography values are stored as WKB in an Avro bytes field + primitiveSchema = BINARY_SCHEMA; + break; case DECIMAL: Types.DecimalType decimal = (Types.DecimalType) primitive; primitiveSchema = diff --git a/core/src/test/java/org/apache/iceberg/avro/AvroTestHelpers.java b/core/src/test/java/org/apache/iceberg/avro/AvroTestHelpers.java index fd73706ce082..4a42ba677b82 100644 --- a/core/src/test/java/org/apache/iceberg/avro/AvroTestHelpers.java +++ b/core/src/test/java/org/apache/iceberg/avro/AvroTestHelpers.java @@ -140,6 +140,8 @@ private static void assertEquals(Type type, Object expected, Object actual) { case FIXED: case BINARY: case DECIMAL: + case GEOMETRY: + case GEOGRAPHY: assertThat(actual).as("Primitive value should be equal to expected").isEqualTo(expected); break; case VARIANT: diff --git a/core/src/test/java/org/apache/iceberg/avro/RandomAvroData.java b/core/src/test/java/org/apache/iceberg/avro/RandomAvroData.java index b3b9ad858b66..1367e2fee089 100644 --- a/core/src/test/java/org/apache/iceberg/avro/RandomAvroData.java +++ b/core/src/test/java/org/apache/iceberg/avro/RandomAvroData.java @@ -110,6 +110,8 @@ public Object primitive(Type.PrimitiveType primitive) { case FIXED: return new GenericData.Fixed(typeToSchema.get(primitive), (byte[]) result); case BINARY: + case GEOMETRY: + case GEOGRAPHY: return ByteBuffer.wrap((byte[]) result); case UUID: return UUID.nameUUIDFromBytes((byte[]) result); diff --git a/core/src/test/java/org/apache/iceberg/avro/TestAvroDataWriter.java b/core/src/test/java/org/apache/iceberg/avro/TestAvroDataWriter.java index a9b9e7a6a74b..12488f6726a7 100644 --- a/core/src/test/java/org/apache/iceberg/avro/TestAvroDataWriter.java +++ b/core/src/test/java/org/apache/iceberg/avro/TestAvroDataWriter.java @@ -21,6 +21,8 @@ import static org.assertj.core.api.Assertions.assertThat; import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; import java.nio.file.Path; import java.util.List; import org.apache.iceberg.DataFile; @@ -110,4 +112,64 @@ public void testDataWriter() throws IOException { assertThat(writtenRecords).as("Written records should match").isEqualTo(records); } + + @Test + public void testGeospatialWkbRoundTrip() throws IOException { + Schema schema = + new Schema( + Types.NestedField.required(1, "id", Types.LongType.get()), + Types.NestedField.optional(2, "geom", Types.GeometryType.crs84()), + Types.NestedField.optional(3, "geog", Types.GeographyType.crs84())); + + GenericRecord record = GenericRecord.create(schema); + List geoRecords = + ImmutableList.of( + record.copy( + ImmutableMap.of("id", 1L, "geom", wkbPoint(30, 10), "geog", wkbPoint(-5, 40))), + // geog is left null + record.copy(ImmutableMap.of("id", 2L, "geom", wkbPoint(0, 0))), + // both geo columns are left null + record.copy(ImmutableMap.of("id", 3L))); + + OutputFile file = Files.localOutput(temp.toFile()); + DataWriter dataWriter = + Avro.writeData(file) + .schema(schema) + .createWriterFunc(org.apache.iceberg.data.avro.DataWriter::create) + .overwrite() + .withSpec(PartitionSpec.unpartitioned()) + .build(); + try (dataWriter) { + for (Record geoRecord : geoRecords) { + dataWriter.write(geoRecord); + } + } + + assertThat(dataWriter.toDataFile().recordCount()).isEqualTo(geoRecords.size()); + + List writtenRecords; + try (AvroIterable reader = + Avro.read(file.toInputFile()) + .project(schema) + .createResolvingReader(PlannedDataReader::create) + .build()) { + writtenRecords = Lists.newArrayList(reader); + } + + assertThat(writtenRecords) + .as("Geometry and geography WKB should round-trip through Avro") + .isEqualTo(geoRecords); + } + + private static ByteBuffer wkbPoint(double xCoord, double yCoord) { + // little-endian WKB encoding of a point + return ByteBuffer.wrap( + ByteBuffer.allocate(21) + .order(ByteOrder.LITTLE_ENDIAN) + .put((byte) 1) // byte order: little endian + .putInt(1) // WKB geometry type: Point + .putDouble(xCoord) + .putDouble(yCoord) + .array()); + } } diff --git a/core/src/test/java/org/apache/iceberg/avro/TestAvroEncoderUtil.java b/core/src/test/java/org/apache/iceberg/avro/TestAvroEncoderUtil.java index 456093501b72..9bb3a1c9f126 100644 --- a/core/src/test/java/org/apache/iceberg/avro/TestAvroEncoderUtil.java +++ b/core/src/test/java/org/apache/iceberg/avro/TestAvroEncoderUtil.java @@ -44,6 +44,11 @@ protected boolean supportsVariant() { return true; } + @Override + protected boolean supportsGeospatial() { + return true; + } + @Override protected void writeAndValidate(org.apache.iceberg.Schema schema) throws IOException { List expected = RandomAvroData.generate(schema, 100, 1990L); diff --git a/core/src/test/java/org/apache/iceberg/avro/TestGenericAvro.java b/core/src/test/java/org/apache/iceberg/avro/TestGenericAvro.java index ba74d066a24d..2d48c495b3b8 100644 --- a/core/src/test/java/org/apache/iceberg/avro/TestGenericAvro.java +++ b/core/src/test/java/org/apache/iceberg/avro/TestGenericAvro.java @@ -44,6 +44,11 @@ protected boolean supportsVariant() { return true; } + @Override + protected boolean supportsGeospatial() { + return true; + } + @Override protected void writeAndValidate(Schema schema) throws IOException { List expected = RandomAvroData.generate(schema, 100, 0L); diff --git a/core/src/test/java/org/apache/iceberg/avro/TestInternalAvro.java b/core/src/test/java/org/apache/iceberg/avro/TestInternalAvro.java index 527a1b56fd66..8f4af6d09507 100644 --- a/core/src/test/java/org/apache/iceberg/avro/TestInternalAvro.java +++ b/core/src/test/java/org/apache/iceberg/avro/TestInternalAvro.java @@ -52,6 +52,11 @@ protected boolean supportsVariant() { return true; } + @Override + protected boolean supportsGeospatial() { + return true; + } + @Override protected void writeAndValidate(Schema schema) throws IOException { List expected = RandomInternalData.generate(schema, 100, 42L); diff --git a/data/src/test/java/org/apache/iceberg/data/avro/TestGenericData.java b/data/src/test/java/org/apache/iceberg/data/avro/TestGenericData.java index b63aab49bd81..b9bc76d4fd34 100644 --- a/data/src/test/java/org/apache/iceberg/data/avro/TestGenericData.java +++ b/data/src/test/java/org/apache/iceberg/data/avro/TestGenericData.java @@ -100,4 +100,9 @@ protected boolean supportsTimestampNanos() { protected boolean supportsVariant() { return true; } + + @Override + protected boolean supportsGeospatial() { + return true; + } } diff --git a/parquet/src/test/java/org/apache/iceberg/parquet/TestParquet.java b/parquet/src/test/java/org/apache/iceberg/parquet/TestParquet.java index c1aabab0b251..98fca650e97e 100644 --- a/parquet/src/test/java/org/apache/iceberg/parquet/TestParquet.java +++ b/parquet/src/test/java/org/apache/iceberg/parquet/TestParquet.java @@ -330,6 +330,39 @@ public void testGeospatialFooterMetricsSkipParquetBounds() throws IOException { } } + @Test + public void testGeospatialWkbRoundTrip() throws IOException { + Schema schema = + new Schema( + optional(1, "geom", Types.GeometryType.crs84()), + optional(2, "geog", Types.GeographyType.crs84())); + + ByteBuffer geomWkb = ByteBuffer.wrap(new byte[] {0x01, 0x02, 0x03}); + ByteBuffer geogWkb = ByteBuffer.wrap(new byte[] {0x04, 0x05, 0x06}); + + org.apache.avro.Schema avroSchema = AvroSchemaUtil.convert(schema.asStruct()); + GenericData.Record record = new GenericData.Record(avroSchema); + record.put("geom", geomWkb); + record.put("geog", geogWkb); + GenericData.Record nulls = new GenericData.Record(avroSchema); + + File file = createTempFile(temp); + write(file, schema, Collections.emptyMap(), ParquetAvroWriter::buildWriter, record, nulls); + + try (CloseableIterable reader = + Parquet.read(Files.localInput(file)) + .project(schema) + .createReaderFunc(fileSchema -> ParquetAvroValueReaders.buildReader(schema, fileSchema)) + .build()) { + List rows = Lists.newArrayList(reader); + assertThat(rows).hasSize(2); + assertThat(rows.get(0).get("geom")).isEqualTo(geomWkb); + assertThat(rows.get(0).get("geog")).isEqualTo(geogWkb); + assertThat(rows.get(1).get("geom")).isNull(); + assertThat(rows.get(1).get("geog")).isNull(); + } + } + @Test public void testPerColumnDictionaryEncoding() throws Exception { Schema schema =