This is an automated email from the ASF dual-hosted git repository.
szehon-ho 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 7a74178db9 Parquet: Validate geospatial projection parameters (#17578)
7a74178db9 is described below
commit 7a74178db9487d934313f5e8430d14f7aa3769d9
Author: Manu Zhang <[email protected]>
AuthorDate: Thu Oct 22 06:50:20 2026 +0800
Parquet: Validate geospatial projection parameters (#17578)
---
.../apache/iceberg/parquet/MessageTypeToType.java | 4 +
.../apache/iceberg/parquet/ParquetSchemaUtil.java | 3 +
.../org/apache/iceberg/parquet/PruneColumns.java | 78 ++++++++
.../apache/iceberg/parquet/TestPruneColumns.java | 220 +++++++++++++++++++++
4 files changed, 305 insertions(+)
diff --git
a/parquet/src/main/java/org/apache/iceberg/parquet/MessageTypeToType.java
b/parquet/src/main/java/org/apache/iceberg/parquet/MessageTypeToType.java
index 2b01bf882e..5d791e66f3 100644
--- a/parquet/src/main/java/org/apache/iceberg/parquet/MessageTypeToType.java
+++ b/parquet/src/main/java/org/apache/iceberg/parquet/MessageTypeToType.java
@@ -149,6 +149,10 @@ class MessageTypeToType extends ParquetTypeVisitor<Type> {
@Override
public Type primitive(PrimitiveType primitive) {
+ return convertPrimitive(primitive);
+ }
+
+ static Type convertPrimitive(PrimitiveType primitive) {
// first, use the logical type annotation, if present
LogicalTypeAnnotation logicalType = primitive.getLogicalTypeAnnotation();
if (logicalType != null) {
diff --git
a/parquet/src/main/java/org/apache/iceberg/parquet/ParquetSchemaUtil.java
b/parquet/src/main/java/org/apache/iceberg/parquet/ParquetSchemaUtil.java
index 75e070381e..ccbf68ec9e 100644
--- a/parquet/src/main/java/org/apache/iceberg/parquet/ParquetSchemaUtil.java
+++ b/parquet/src/main/java/org/apache/iceberg/parquet/ParquetSchemaUtil.java
@@ -267,6 +267,9 @@ public class ParquetSchemaUtil {
int ordinal = 1;
for (Type type : fileSchema.getFields()) {
if (selectedIds.contains(ordinal)) {
+ Types.NestedField expectedField = expectedSchema.findField(ordinal);
+ PruneColumns.validateFallbackType(expectedField.type(), type,
type.getName());
+
builder.addField(type.withId(ordinal));
}
ordinal += 1;
diff --git a/parquet/src/main/java/org/apache/iceberg/parquet/PruneColumns.java
b/parquet/src/main/java/org/apache/iceberg/parquet/PruneColumns.java
index 1675325c78..0caa376fbb 100644
--- a/parquet/src/main/java/org/apache/iceberg/parquet/PruneColumns.java
+++ b/parquet/src/main/java/org/apache/iceberg/parquet/PruneColumns.java
@@ -24,6 +24,8 @@ import java.util.Set;
import org.apache.iceberg.relocated.com.google.common.base.Objects;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Type.TypeID;
+import org.apache.iceberg.types.TypeUtil;
import org.apache.iceberg.types.Types.ListType;
import org.apache.iceberg.types.Types.MapType;
import org.apache.iceberg.types.Types.NestedField;
@@ -170,9 +172,85 @@ class PruneColumns extends TypeWithSchemaVisitor<Type> {
@Override
public Type primitive(
org.apache.iceberg.types.Type.PrimitiveType expected, PrimitiveType
primitive) {
+ validatePrimitive(expected, primitive, String.join(".", currentPath()));
return null;
}
+ static void validatePrimitive(
+ org.apache.iceberg.types.Type.PrimitiveType expected, PrimitiveType
primitive, String path) {
+ if (expected != null
+ && (expected.typeId() == TypeID.GEOMETRY || expected.typeId() ==
TypeID.GEOGRAPHY)) {
+ Preconditions.checkArgument(
+
TypeUtil.isPromotionAllowed(MessageTypeToType.convertPrimitive(primitive),
expected),
+ "Cannot read Parquet type %s as Iceberg type %s for field %s",
+ primitive,
+ expected,
+ path);
+ }
+ }
+
+ static void validateFallbackType(org.apache.iceberg.types.Type expected,
Type type, String path) {
+ if (expected == null) {
+ return;
+ }
+
+ if (expected.isPrimitiveType() || type.isPrimitive()) {
+ if (expected.isPrimitiveType() && type.isPrimitive()) {
+ validatePrimitive(expected.asPrimitiveType(), type.asPrimitiveType(),
path);
+ }
+
+ return;
+ }
+
+ GroupType group = type.asGroupType();
+ if (expected.isStructType()) {
+ validateFallbackStruct(expected.asStructType(), group, path);
+ } else if (expected.isListType()) {
+ validateFallbackList(expected.asListType(), group, path);
+ } else if (expected.isMapType()) {
+ validateFallbackMap(expected.asMapType(), group, path);
+ }
+ }
+
+ private static void validateFallbackStruct(StructType expected, GroupType
struct, String path) {
+ for (NestedField expectedField : expected.fields()) {
+ // Projected structs omit unselected fields, so projected positions may
not match the file.
+ // Validate only when an expected name identifies the corresponding file
field.
+ Type field = ParquetSchemaUtil.fieldType(struct, expectedField.name());
+ if (field != null) {
+ validateFallbackType(expectedField.type(), field, path + "." +
field.getName());
+ }
+ }
+ }
+
+ private static void validateFallbackList(ListType expected, GroupType list,
String path) {
+ Type element = ParquetSchemaUtil.determineListElementType(list);
+ String elementPath = path;
+ if (!element.isRepetition(Type.Repetition.REPEATED)) {
+ elementPath += "." + list.getFieldName(0);
+ }
+
+ validateFallbackType(expected.elementType(), element, elementPath + "." +
element.getName());
+ }
+
+ private static void validateFallbackMap(MapType expected, GroupType map,
String path) {
+ GroupType repeated = map.getType(0).asGroupType();
+ String repeatedPath = path + "." + repeated.getName();
+ if (repeated.getFieldCount() == 2) {
+ Type key = repeated.getType(0);
+ Type value = repeated.getType(1);
+ validateFallbackType(expected.keyType(), key, repeatedPath + "." +
key.getName());
+ validateFallbackType(expected.valueType(), value, repeatedPath + "." +
value.getName());
+
+ } else if (repeated.getFieldCount() == 1) {
+ Type keyOrValue = repeated.getType(0);
+ org.apache.iceberg.types.Type expectedKeyOrValue =
+ keyOrValue.getName().equalsIgnoreCase("key") ? expected.keyType() :
expected.valueType();
+ validateFallbackType(
+ expectedKeyOrValue, keyOrValue, repeatedPath + "." +
keyOrValue.getName());
+ }
+ }
+
private Integer getId(Type type) {
return type.getId() == null ? null : type.getId().intValue();
}
diff --git
a/parquet/src/test/java/org/apache/iceberg/parquet/TestPruneColumns.java
b/parquet/src/test/java/org/apache/iceberg/parquet/TestPruneColumns.java
index 46b00585d0..8d2b3210ad 100644
--- a/parquet/src/test/java/org/apache/iceberg/parquet/TestPruneColumns.java
+++ b/parquet/src/test/java/org/apache/iceberg/parquet/TestPruneColumns.java
@@ -19,10 +19,14 @@
package org.apache.iceberg.parquet;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import org.apache.iceberg.Schema;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.types.EdgeAlgorithm;
import org.apache.iceberg.types.Types.DoubleType;
+import org.apache.iceberg.types.Types.GeographyType;
+import org.apache.iceberg.types.Types.GeometryType;
import org.apache.iceberg.types.Types.IntegerType;
import org.apache.iceberg.types.Types.ListType;
import org.apache.iceberg.types.Types.LongType;
@@ -32,6 +36,7 @@ import org.apache.iceberg.types.Types.StringType;
import org.apache.iceberg.types.Types.StructType;
import org.apache.iceberg.types.Types.VariantType;
import org.apache.iceberg.variants.Variant;
+import org.apache.parquet.column.schema.EdgeInterpolationAlgorithm;
import org.apache.parquet.schema.LogicalTypeAnnotation;
import org.apache.parquet.schema.MessageType;
import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName;
@@ -681,6 +686,221 @@ public class TestPruneColumns {
assertThat(wholeActual).as("Whole-map projection is
unchanged").isEqualTo(fileSchema);
}
+ @Test
+ public void acceptsMatchingGeospatialParameters() {
+ MessageType fileSchema =
+ Types.buildMessage()
+ .optional(PrimitiveTypeName.BINARY)
+ .as(LogicalTypeAnnotation.geometryType(null))
+ .id(1)
+ .named("geom_default")
+ .optional(PrimitiveTypeName.BINARY)
+ .as(LogicalTypeAnnotation.geometryType("EPSG:3857"))
+ .id(2)
+ .named("geom_projected")
+ .optional(PrimitiveTypeName.BINARY)
+ .as(LogicalTypeAnnotation.geographyType(null, null))
+ .id(3)
+ .named("geog_default")
+ .optional(PrimitiveTypeName.BINARY)
+ .as(
+ LogicalTypeAnnotation.geographyType(
+ "EPSG:4326", EdgeInterpolationAlgorithm.ANDOYER))
+ .id(4)
+ .named("geog_custom")
+ .named("table");
+
+ Schema projection =
+ new Schema(
+ NestedField.optional(1, "geom_default", GeometryType.crs84()),
+ NestedField.optional(2, "geom_projected",
GeometryType.of("epsg:3857")),
+ NestedField.optional(3, "geog_default", GeographyType.crs84()),
+ NestedField.optional(
+ 4, "geog_custom", GeographyType.of("epsg:4326",
EdgeAlgorithm.ANDOYER)));
+
+ assertThat(ParquetSchemaUtil.pruneColumns(fileSchema,
projection)).isEqualTo(fileSchema);
+ }
+
+ @Test
+ public void rejectsGeometryCrsMismatch() {
+ MessageType fileSchema =
+ Types.buildMessage()
+ .optional(PrimitiveTypeName.BINARY)
+ .as(LogicalTypeAnnotation.geometryType("EPSG:3857"))
+ .id(1)
+ .named("geom")
+ .named("table");
+ Schema projection = new Schema(NestedField.optional(1, "geom",
GeometryType.crs84()));
+
+ assertThatThrownBy(() -> ParquetSchemaUtil.pruneColumns(fileSchema,
projection))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Cannot read Parquet type")
+ .hasMessageContaining("field geom")
+ .hasMessageContaining("geometry(OGC:CRS84)");
+ }
+
+ @Test
+ public void rejectsGeometryCrsMismatchWithoutIds() {
+ MessageType fileSchema =
+ Types.buildMessage()
+ .optional(PrimitiveTypeName.BINARY)
+ .as(LogicalTypeAnnotation.geometryType("EPSG:3857"))
+ .named("geom")
+ .named("table");
+ Schema projection = new Schema(NestedField.optional(1, "geom",
GeometryType.crs84()));
+
+ assertThatThrownBy(() ->
ParquetSchemaUtil.pruneColumnsFallback(fileSchema, projection))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Cannot read Parquet type")
+ .hasMessageContaining("field geom")
+ .hasMessageContaining("geometry(OGC:CRS84)");
+ }
+
+ @Test
+ public void rejectsProjectedNestedStructGeometryCrsMismatchWithoutIds() {
+ MessageType fileSchema =
+ Types.buildMessage()
+ .addField(
+ Types.buildGroup(Type.Repetition.OPTIONAL)
+ .optional(PrimitiveTypeName.BINARY)
+ .as(LogicalTypeAnnotation.geometryType(null))
+ .named("other_geom")
+ .optional(PrimitiveTypeName.BINARY)
+ .as(LogicalTypeAnnotation.geometryType("EPSG:3857"))
+ .named("geom")
+ .named("location"))
+ .named("table");
+ Schema projection =
+ new Schema(
+ NestedField.optional(
+ 1,
+ "location",
+ StructType.of(NestedField.optional(3, "geom",
GeometryType.crs84()))));
+
+ assertThatThrownBy(() ->
ParquetSchemaUtil.pruneColumnsFallback(fileSchema, projection))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Cannot read Parquet type")
+ .hasMessageContaining("field location.geom")
+ .hasMessageContaining("geometry(OGC:CRS84)");
+ }
+
+ @Test
+ public void acceptsProjectedNestedStructGeometryParametersWithoutIds() {
+ MessageType fileSchema =
+ Types.buildMessage()
+ .addField(
+ Types.buildGroup(Type.Repetition.OPTIONAL)
+ .optional(PrimitiveTypeName.BINARY)
+ .as(LogicalTypeAnnotation.geometryType("EPSG:3857"))
+ .named("other_geom")
+ .optional(PrimitiveTypeName.BINARY)
+ .as(LogicalTypeAnnotation.geometryType(null))
+ .named("geom")
+ .named("location"))
+ .named("table");
+ Schema projection =
+ new Schema(
+ NestedField.optional(
+ 1,
+ "location",
+ StructType.of(NestedField.optional(3, "geom",
GeometryType.crs84()))));
+
+ assertThat(ParquetSchemaUtil.pruneColumnsFallback(fileSchema,
projection)).isNotNull();
+ }
+
+ @Test
+ public void rejectsListElementGeometryCrsMismatchWithoutIds() {
+ MessageType fileSchema =
+ Types.buildMessage()
+ .addField(
+ Types.buildGroup(Type.Repetition.OPTIONAL)
+ .addField(
+ Types.buildGroup(Type.Repetition.REPEATED)
+ .optional(PrimitiveTypeName.BINARY)
+
.as(LogicalTypeAnnotation.geometryType("EPSG:3857"))
+ .named("element")
+ .named("list"))
+ .as(LogicalTypeAnnotation.listType())
+ .named("locations"))
+ .named("table");
+ Schema projection =
+ new Schema(
+ NestedField.optional(1, "locations", ListType.ofOptional(2,
GeometryType.crs84())));
+
+ assertThatThrownBy(() ->
ParquetSchemaUtil.pruneColumnsFallback(fileSchema, projection))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Cannot read Parquet type")
+ .hasMessageContaining("field locations.list.element")
+ .hasMessageContaining("geometry(OGC:CRS84)");
+ }
+
+ @Test
+ public void rejectsMapValueGeometryCrsMismatchWithoutIds() {
+ MessageType fileSchema =
+ Types.buildMessage()
+ .addField(
+ Types.buildGroup(Type.Repetition.OPTIONAL)
+ .addField(
+ Types.buildGroup(Type.Repetition.REPEATED)
+ .required(PrimitiveTypeName.BINARY)
+ .as(LogicalTypeAnnotation.stringType())
+ .named("key")
+ .optional(PrimitiveTypeName.BINARY)
+
.as(LogicalTypeAnnotation.geometryType("EPSG:3857"))
+ .named("value")
+ .named("key_value"))
+ .as(LogicalTypeAnnotation.mapType())
+ .named("locations"))
+ .named("table");
+ Schema projection =
+ new Schema(
+ NestedField.optional(
+ 1, "locations", MapType.ofOptional(2, 3, StringType.get(),
GeometryType.crs84())));
+
+ assertThatThrownBy(() ->
ParquetSchemaUtil.pruneColumnsFallback(fileSchema, projection))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Cannot read Parquet type")
+ .hasMessageContaining("field locations.key_value.value")
+ .hasMessageContaining("geometry(OGC:CRS84)");
+ }
+
+ @Test
+ public void rejectsGeographyCrsMismatch() {
+ MessageType fileSchema =
+ Types.buildMessage()
+ .optional(PrimitiveTypeName.BINARY)
+ .as(
+ LogicalTypeAnnotation.geographyType(
+ "EPSG:4326", EdgeInterpolationAlgorithm.SPHERICAL))
+ .id(1)
+ .named("geog")
+ .named("table");
+ Schema projection = new Schema(NestedField.optional(1, "geog",
GeographyType.crs84()));
+
+ assertThatThrownBy(() -> ParquetSchemaUtil.pruneColumns(fileSchema,
projection))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Cannot read Parquet type")
+ .hasMessageContaining("geography(OGC:CRS84, spherical)");
+ }
+
+ @Test
+ public void rejectsGeographyAlgorithmMismatch() {
+ MessageType fileSchema =
+ Types.buildMessage()
+ .optional(PrimitiveTypeName.BINARY)
+ .as(LogicalTypeAnnotation.geographyType("OGC:CRS84",
EdgeInterpolationAlgorithm.KARNEY))
+ .id(1)
+ .named("geog")
+ .named("table");
+ Schema projection = new Schema(NestedField.optional(1, "geog",
GeographyType.crs84()));
+
+ assertThatThrownBy(() -> ParquetSchemaUtil.pruneColumns(fileSchema,
projection))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Cannot read Parquet type")
+ .hasMessageContaining("KARNEY")
+ .hasMessageContaining("geography(OGC:CRS84, spherical)");
+ }
+
private static Type buildVariantType(int id, String name) {
return Types.buildGroup(Type.Repetition.OPTIONAL)
.as(LogicalTypeAnnotation.variantType(Variant.VARIANT_SPEC_VERSION))
