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 =