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 2b01bf882e75..5d791e66f35c 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 @@ public Type map(GroupType map, Type keyType, Type valueType) { @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 9a81626827c6..a74fec8697c8 100644 --- a/parquet/src/main/java/org/apache/iceberg/parquet/ParquetSchemaUtil.java +++ b/parquet/src/main/java/org/apache/iceberg/parquet/ParquetSchemaUtil.java @@ -159,6 +159,12 @@ public static MessageType pruneColumnsFallback(MessageType fileSchema, Schema ex int ordinal = 1; for (Type type : fileSchema.getFields()) { if (selectedIds.contains(ordinal)) { + Types.NestedField expectedField = expectedSchema.findField(ordinal); + if (type.isPrimitive() && expectedField.type().isPrimitiveType()) { + PruneColumns.validatePrimitive( + expectedField.type().asPrimitiveType(), type.asPrimitiveType()); + } + 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 0647a09f53fe..3471374fce89 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 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; @@ -162,9 +164,22 @@ public Type variant( @Override public Type primitive( org.apache.iceberg.types.Type.PrimitiveType expected, PrimitiveType primitive) { + validatePrimitive(expected, primitive); return null; } + static void validatePrimitive( + org.apache.iceberg.types.Type.PrimitiveType expected, PrimitiveType primitive) { + 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", + primitive, + expected); + } + } + 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 619b2c5a3470..1224b2c1bfac 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.MapType; @@ -31,6 +35,7 @@ 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; @@ -305,6 +310,111 @@ public void testVariant() { assertThat(actual).as("Pruned schema should be matched").isEqualTo(expected); } + @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("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("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))