This is an automated email from the ASF dual-hosted git repository.
mxm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 09c80c42e4 Data: Add generic File Format API TCK binding and
geometry/geography generators (#18537)
09c80c42e4 is described below
commit 09c80c42e40190774ccaabb2045784df9c43f16a
Author: hreddygayam <[email protected]>
AuthorDate: Mon Oct 19 00:23:49 2026 -0700
Data: Add generic File Format API TCK binding and geometry/geography
generators (#18537)
---
.../org/apache/iceberg/data/DataGenerators.java | 38 ++++++++++-
.../apache/iceberg/data/ReadFormatModelTests.java | 76 ++++++++++++++++++++--
.../iceberg/data/TestGenericFormatModel.java | 73 +++++++++++++++++++++
.../iceberg/flink/data/TestFlinkFormatModel.java | 7 +-
.../iceberg/flink/data/TestFlinkFormatModel.java | 3 +-
.../iceberg/flink/data/TestFlinkFormatModel.java | 3 +-
.../iceberg/spark/data/TestSparkFormatModel.java | 7 +-
.../iceberg/spark/data/TestSparkFormatModel.java | 6 +-
.../iceberg/spark/data/TestSparkFormatModel.java | 10 ++-
.../iceberg/spark/data/TestSparkFormatModel.java | 10 ++-
10 files changed, 220 insertions(+), 13 deletions(-)
diff --git a/data/src/test/java/org/apache/iceberg/data/DataGenerators.java
b/data/src/test/java/org/apache/iceberg/data/DataGenerators.java
index 8e423ecabb..cf2ab00d7c 100644
--- a/data/src/test/java/org/apache/iceberg/data/DataGenerators.java
+++ b/data/src/test/java/org/apache/iceberg/data/DataGenerators.java
@@ -45,7 +45,9 @@ class DataGenerators {
new ListOfPrimitive(),
new MapOfPrimitive(),
new TimestampNano(),
- new Variant()
+ new Variant(),
+ new Geometry(),
+ new Geography()
};
private DataGenerators() {}
@@ -117,6 +119,40 @@ class DataGenerators {
}
}
+ static class Geometry implements DataGenerator {
+ private final Schema schema =
+ new Schema(
+ Types.NestedField.required(1, "geometry_col",
Types.GeometryType.crs84()),
+ Types.NestedField.optional(2, "geometry_opt_col",
Types.GeometryType.crs84()));
+
+ @Override
+ public Schema schema() {
+ return schema;
+ }
+
+ @Override
+ public String toString() {
+ return "Geometry";
+ }
+ }
+
+ static class Geography implements DataGenerator {
+ private final Schema schema =
+ new Schema(
+ Types.NestedField.required(1, "geography_col",
Types.GeographyType.crs84()),
+ Types.NestedField.optional(2, "geography_opt_col",
Types.GeographyType.crs84()));
+
+ @Override
+ public Schema schema() {
+ return schema;
+ }
+
+ @Override
+ public String toString() {
+ return "Geography";
+ }
+ }
+
static class UUID implements DataGenerator {
private final Schema schema =
new Schema(Types.NestedField.required(1, "uuid_col",
Types.UUIDType.get()));
diff --git
a/data/src/test/java/org/apache/iceberg/data/ReadFormatModelTests.java
b/data/src/test/java/org/apache/iceberg/data/ReadFormatModelTests.java
index 34712be7e7..b8596c632b 100644
--- a/data/src/test/java/org/apache/iceberg/data/ReadFormatModelTests.java
+++ b/data/src/test/java/org/apache/iceberg/data/ReadFormatModelTests.java
@@ -69,6 +69,7 @@ import org.apache.iceberg.expressions.Literal;
import org.apache.iceberg.formats.FileWriterBuilder;
import org.apache.iceberg.formats.FormatModelRegistry;
import org.apache.iceberg.formats.ReadBuilder;
+import org.apache.iceberg.geospatial.GeospatialBound;
import org.apache.iceberg.inmemory.InMemoryFileIO;
import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.io.DataWriter;
@@ -86,6 +87,7 @@ import org.apache.iceberg.types.Conversions;
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.TypeUtil;
import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.RandomUtil;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.io.TempDir;
@@ -179,6 +181,11 @@ public abstract class ReadFormatModelTests<T> {
static final String FEATURE_NATIVE_ENCRYPTION = "nativeEncryption";
static final String FEATURE_AES_STREAM_ENCRYPTION = "aesStreamEncryption";
static final String FEATURE_VARIANT = "variant";
+ static final String FEATURE_GEOSPATIAL = "geospatial";
+
+ private static final Set<Type.TypeID> VARIANT_TYPE_IDS =
Set.of(Type.TypeID.VARIANT);
+ private static final Set<Type.TypeID> GEOSPATIAL_TYPE_IDS =
+ Set.of(Type.TypeID.GEOMETRY, Type.TypeID.GEOGRAPHY);
private static final Map<FileFormat, String[]> MISSING_FEATURES =
Map.of(
@@ -198,7 +205,8 @@ public abstract class ReadFormatModelTests<T> {
FEATURE_READER_DEFAULT,
FEATURE_AES_STREAM_ENCRYPTION,
FEATURE_NATIVE_ENCRYPTION,
- FEATURE_VARIANT
+ FEATURE_VARIANT,
+ FEATURE_GEOSPATIAL
},
FileFormat.PARQUET,
new String[] {FEATURE_AES_STREAM_ENCRYPTION});
@@ -1473,6 +1481,43 @@ public abstract class ReadFormatModelTests<T> {
assertColumnSize(fileFormat, dataFile);
}
+ @ParameterizedTest
+ @FieldSource("FILE_FORMATS")
+ void testDataWriterGeospatialBounds(FileFormat fileFormat) throws
IOException {
+ assumeSupports(fileFormat, FEATURE_GEOSPATIAL);
+ assumeSupports(fileFormat, FEATURE_COLUMN_LEVEL_METRICS);
+
+ Schema schema = new DataGenerators.Geometry().schema();
+
+ // The spatial lower/upper bounds are the corners of the bounding box over
all values, so no
+ // single value needs to sit at a corner. Points are chosen so the
expected corners differ from
+ // every input point, and the optional column exercises a skipped null.
+ List<Record> records =
+ List.of(
+ GenericRecord.create(schema)
+ .copy("geometry_col", wkbPoint(1.0, 4.0), "geometry_opt_col",
wkbPoint(-3.0, 7.0)),
+ GenericRecord.create(schema)
+ .copy("geometry_col", wkbPoint(5.0, 2.0), "geometry_opt_col",
null),
+ GenericRecord.create(schema)
+ .copy("geometry_col", wkbPoint(3.0, 6.0), "geometry_opt_col",
wkbPoint(2.0, -1.0)));
+
+ DataFile dataFile = writeGenericRecords(fileFormat, schema, records);
+
+ Type geometryType = schema.findType("geometry_col");
+
+ // Required column box: min corner (1, 2), max corner (5, 6).
+ assertGeospatialBound(
+ geometryType, dataFile.lowerBounds().get(1),
GeospatialBound.createXY(1.0, 2.0));
+ assertGeospatialBound(
+ geometryType, dataFile.upperBounds().get(1),
GeospatialBound.createXY(5.0, 6.0));
+
+ // Optional column box over the two non-null values: min corner (-3, -1),
max corner (2, 7).
+ assertGeospatialBound(
+ geometryType, dataFile.lowerBounds().get(2),
GeospatialBound.createXY(-3.0, -1.0));
+ assertGeospatialBound(
+ geometryType, dataFile.upperBounds().get(2),
GeospatialBound.createXY(2.0, 7.0));
+ }
+
@ParameterizedTest
@FieldSource("FILE_FORMATS")
void testDataWriterMetricsWithNoneMode(FileFormat fileFormat) throws
IOException {
@@ -2412,9 +2457,22 @@ public abstract class ReadFormatModelTests<T> {
}
private static boolean supportsGenerator(FileFormat fileFormat,
DataGenerator generator) {
- boolean hasVariant =
- TypeUtil.find(generator.schema(), type -> type.typeId() ==
Type.TypeID.VARIANT) != null;
- return !hasVariant || supportsFeature(fileFormat, FEATURE_VARIANT);
+ Schema schema = generator.schema();
+ return supportsTypes(fileFormat, schema, VARIANT_TYPE_IDS, FEATURE_VARIANT)
+ && supportsTypes(fileFormat, schema, GEOSPATIAL_TYPE_IDS,
FEATURE_GEOSPATIAL);
+ }
+
+ /**
+ * Returns whether the format can handle the gated types the schema uses:
either the schema uses
+ * none of them, or the format supports the feature that gates them.
+ */
+ private static boolean supportsTypes(
+ FileFormat fileFormat, Schema schema, Set<Type.TypeID> typeIds, String
feature) {
+ return !usesAnyType(schema.asStruct(), typeIds) ||
supportsFeature(fileFormat, feature);
+ }
+
+ private static boolean usesAnyType(Type type, Set<Type.TypeID> typeIds) {
+ return TypeUtil.find(type, nested -> typeIds.contains(nested.typeId())) !=
null;
}
/**
@@ -2576,6 +2634,14 @@ public abstract class ReadFormatModelTests<T> {
return new Object[] {min, max};
}
+ private static ByteBuffer wkbPoint(double xCoord, double yCoord) {
+ return ByteBuffer.wrap(RandomUtil.wkbPoint(xCoord, yCoord));
+ }
+
+ private static void assertGeospatialBound(Type type, ByteBuffer bound,
GeospatialBound expected) {
+ assertThat((Object) Conversions.fromByteBuffer(type,
bound)).isEqualTo(expected);
+ }
+
private static void assertBoundsNull(Schema schema, ContentFile<?> file) {
Map<Integer, ByteBuffer> lowerBounds = file.lowerBounds();
Map<Integer, ByteBuffer> upperBounds = file.upperBounds();
@@ -2754,7 +2820,7 @@ public abstract class ReadFormatModelTests<T> {
projectionSchema.findType(entry.getKey()),
entry.getValue())));
}
- private static Record structLikeToRecord(StructLike structLike,
Types.StructType structType) {
+ static Record structLikeToRecord(StructLike structLike, Types.StructType
structType) {
Record record = GenericRecord.create(structType);
int sourceSize = structLike.size();
for (int i = 0; i < structType.fields().size(); i++) {
diff --git
a/data/src/test/java/org/apache/iceberg/data/TestGenericFormatModel.java
b/data/src/test/java/org/apache/iceberg/data/TestGenericFormatModel.java
new file mode 100644
index 0000000000..8035034d2a
--- /dev/null
+++ b/data/src/test/java/org/apache/iceberg/data/TestGenericFormatModel.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.data;
+
+import java.util.List;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.types.Type;
+
+/**
+ * File Format API TCK binding for the generic {@link Record} object model.
+ *
+ * <p>Unlike the Spark and Flink bindings, the generic engine represents every
Iceberg type
+ * directly, so it exercises the full generator matrix (including geospatial
types) against the
+ * Avro, ORC and Parquet object models. Format-level gaps (for example ORC's
lack of geospatial
+ * support) are handled by {@code MISSING_FEATURES}/{@code supportsGenerator}
in the base class.
+ */
+class TestGenericFormatModel extends BaseFormatModelTests<Record> {
+
+ @Override
+ protected Class<Record> engineType() {
+ return Record.class;
+ }
+
+ @Override
+ protected Object engineSchema(Schema schema) {
+ // The generic object model derives its schema from the Iceberg schema; no
engine-native
+ // schema is required.
+ return null;
+ }
+
+ @Override
+ protected Record convertToEngine(Record record, Schema schema) {
+ return record;
+ }
+
+ @Override
+ protected void assertEquals(Schema schema, List<Record> expected,
List<Record> actual) {
+ DataTestHelpers.assertEquals(schema.asStruct(), expected, actual);
+ }
+
+ @Override
+ protected Object convertConstantToEngine(Type type, Object value) {
+ // Metadata-column constants such as the _partition struct arrive as a
StructLike (for example
+ // PartitionData) rather than a Record. Materialize them as a Record up
front, mirroring how the
+ // Flink and Spark bindings convert constants into their engine
representation, so reads expose
+ // the same type the generic comparison in DataTestHelpers expects.
+ if (value instanceof StructLike struct) {
+ return structLikeToRecord(struct, type.asStructType());
+ }
+
+ // Scalar constants arrive in Iceberg's internal representation (for
example dates as epoch-day
+ // ints and timestamps as micro longs); convert them to the generic object
model representation
+ // (LocalDate, LocalDateTime, ...) so reads match the generic comparison
in DataTestHelpers.
+ return GenericDataUtil.internalToGeneric(type, value);
+ }
+}
diff --git
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkFormatModel.java
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkFormatModel.java
index 6638b4d65b..5399ca306f 100644
---
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkFormatModel.java
+++
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkFormatModel.java
@@ -37,7 +37,12 @@ public class TestFlinkFormatModel extends
BaseFormatModelTests<RowData> {
private static final Set<Type.TypeID> UNSUPPORTED_TYPE_IDS =
Set.of(
- Type.TypeID.TIME, Type.TypeID.TIMESTAMP_NANO, Type.TypeID.VARIANT,
Type.TypeID.UNKNOWN);
+ Type.TypeID.TIME,
+ Type.TypeID.TIMESTAMP_NANO,
+ Type.TypeID.VARIANT,
+ Type.TypeID.UNKNOWN,
+ Type.TypeID.GEOMETRY,
+ Type.TypeID.GEOGRAPHY);
@Override
protected Collection<Type.TypeID> unsupportedTypeIds() {
diff --git
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkFormatModel.java
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkFormatModel.java
index 8df5c26d13..2bcc55975c 100644
---
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkFormatModel.java
+++
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkFormatModel.java
@@ -35,7 +35,8 @@ import org.apache.iceberg.types.Types;
public class TestFlinkFormatModel extends BaseFormatModelTests<RowData> {
- private static final Set<Type.TypeID> UNSUPPORTED_TYPE_IDS =
Set.of(Type.TypeID.TIME);
+ private static final Set<Type.TypeID> UNSUPPORTED_TYPE_IDS =
+ Set.of(Type.TypeID.TIME, Type.TypeID.GEOMETRY, Type.TypeID.GEOGRAPHY);
@Override
protected Collection<Type.TypeID> unsupportedTypeIds() {
diff --git
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkFormatModel.java
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkFormatModel.java
index 8df5c26d13..2bcc55975c 100644
---
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkFormatModel.java
+++
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkFormatModel.java
@@ -35,7 +35,8 @@ import org.apache.iceberg.types.Types;
public class TestFlinkFormatModel extends BaseFormatModelTests<RowData> {
- private static final Set<Type.TypeID> UNSUPPORTED_TYPE_IDS =
Set.of(Type.TypeID.TIME);
+ private static final Set<Type.TypeID> UNSUPPORTED_TYPE_IDS =
+ Set.of(Type.TypeID.TIME, Type.TypeID.GEOMETRY, Type.TypeID.GEOGRAPHY);
@Override
protected Collection<Type.TypeID> unsupportedTypeIds() {
diff --git
a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkFormatModel.java
b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkFormatModel.java
index 06654efcdb..968d71d1a5 100644
---
a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkFormatModel.java
+++
b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkFormatModel.java
@@ -34,7 +34,12 @@ import org.apache.spark.sql.catalyst.InternalRow;
public class TestSparkFormatModel extends BaseFormatModelTests<InternalRow> {
private static final Set<Type.TypeID> UNSUPPORTED_TYPE_IDS =
- Set.of(Type.TypeID.TIME, Type.TypeID.TIMESTAMP_NANO,
Type.TypeID.VARIANT);
+ Set.of(
+ Type.TypeID.TIME,
+ Type.TypeID.TIMESTAMP_NANO,
+ Type.TypeID.VARIANT,
+ Type.TypeID.GEOMETRY,
+ Type.TypeID.GEOGRAPHY);
@Override
protected Collection<Type.TypeID> unsupportedTypeIds() {
diff --git
a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkFormatModel.java
b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkFormatModel.java
index 4f964f8a74..8d830c3a6d 100644
---
a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkFormatModel.java
+++
b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkFormatModel.java
@@ -34,7 +34,11 @@ import org.apache.spark.sql.catalyst.InternalRow;
public class TestSparkFormatModel extends BaseFormatModelTests<InternalRow> {
private static final Set<Type.TypeID> UNSUPPORTED_TYPE_IDS =
- Set.of(Type.TypeID.TIME, Type.TypeID.TIMESTAMP_NANO);
+ Set.of(
+ Type.TypeID.TIME,
+ Type.TypeID.TIMESTAMP_NANO,
+ Type.TypeID.GEOMETRY,
+ Type.TypeID.GEOGRAPHY);
@Override
protected Collection<Type.TypeID> unsupportedTypeIds() {
diff --git
a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkFormatModel.java
b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkFormatModel.java
index 4f964f8a74..41b032b0e8 100644
---
a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkFormatModel.java
+++
b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkFormatModel.java
@@ -34,7 +34,15 @@ import org.apache.spark.sql.catalyst.InternalRow;
public class TestSparkFormatModel extends BaseFormatModelTests<InternalRow> {
private static final Set<Type.TypeID> UNSUPPORTED_TYPE_IDS =
- Set.of(Type.TypeID.TIME, Type.TypeID.TIMESTAMP_NANO);
+ Set.of(
+ Type.TypeID.TIME,
+ Type.TypeID.TIMESTAMP_NANO,
+ // TODO: Spark 4.1+ supports geometry/geography, but only on Parquet
- the Spark Avro
+ // reader/writer don't handle geo. The TCK gates geospatial by file
format rather than
+ // engine, so enabling these here would also run the unsupported
Avro cases. Enable once
+ // the harness geo-gating is engine-aware and
InternalRowConverter/TestHelpers handle geo.
+ Type.TypeID.GEOMETRY,
+ Type.TypeID.GEOGRAPHY);
@Override
protected Collection<Type.TypeID> unsupportedTypeIds() {
diff --git
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkFormatModel.java
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkFormatModel.java
index 4f964f8a74..41b032b0e8 100644
---
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkFormatModel.java
+++
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkFormatModel.java
@@ -34,7 +34,15 @@ import org.apache.spark.sql.catalyst.InternalRow;
public class TestSparkFormatModel extends BaseFormatModelTests<InternalRow> {
private static final Set<Type.TypeID> UNSUPPORTED_TYPE_IDS =
- Set.of(Type.TypeID.TIME, Type.TypeID.TIMESTAMP_NANO);
+ Set.of(
+ Type.TypeID.TIME,
+ Type.TypeID.TIMESTAMP_NANO,
+ // TODO: Spark 4.1+ supports geometry/geography, but only on Parquet
- the Spark Avro
+ // reader/writer don't handle geo. The TCK gates geospatial by file
format rather than
+ // engine, so enabling these here would also run the unsupported
Avro cases. Enable once
+ // the harness geo-gating is engine-aware and
InternalRowConverter/TestHelpers handle geo.
+ Type.TypeID.GEOMETRY,
+ Type.TypeID.GEOGRAPHY);
@Override
protected Collection<Type.TypeID> unsupportedTypeIds() {
