From 71588d46297bcc05f4c1dc02c634e8aaf8541e8d Mon Sep 17 00:00:00 2001 From: Manu Zhang Date: Thu, 22 Oct 2026 10:26:00 +0800 Subject: [PATCH 1/2] fix(parquet): validate geospatial projection parameters Reject incompatible geometry/geography CRS and geography edge algorithms during projection. Resolve omitted Parquet CRS to OGC:CRS84 and include the field path in mismatch errors. Co-authored-by: Codex --- src/iceberg/parquet/parquet_schema_util.cc | 19 ++++ src/iceberg/test/parquet_schema_test.cc | 122 +++++++++++++++++++-- 2 files changed, 133 insertions(+), 8 deletions(-) diff --git a/src/iceberg/parquet/parquet_schema_util.cc b/src/iceberg/parquet/parquet_schema_util.cc index 05f038c13..bf4156a84 100644 --- a/src/iceberg/parquet/parquet_schema_util.cc +++ b/src/iceberg/parquet/parquet_schema_util.cc @@ -310,11 +310,30 @@ Status ValidateGeospatialParquetType(const Type& expected_type, if (!descr.logical_type()->is_geometry()) { return InvalidSchema("Iceberg geometry requires Parquet Geometry logical type"); } + const auto& expected = internal::checked_cast(expected_type); + const auto& actual = internal::checked_cast( + *descr.logical_type()); + // Parquet represents an omitted CRS as an empty string. + const auto crs = actual.crs().empty() ? GeometryType::kDefaultCrs : actual.crs(); + if (!StringUtils::EqualsIgnoreCase(expected.crs(), crs)) { + return InvalidSchema("Cannot read Parquet type {} as Iceberg type {} for field {}", + actual.ToString(), expected_type, descr.path()->ToDotString()); + } return {}; } if (!descr.logical_type()->is_geography()) { return InvalidSchema("Iceberg geography requires Parquet Geography logical type"); } + const auto& expected = internal::checked_cast(expected_type); + const auto& actual = internal::checked_cast( + *descr.logical_type()); + ICEBERG_ASSIGN_OR_RAISE(auto algorithm, ToParquetAlgorithm(expected.algorithm())); + const auto crs = actual.crs().empty() ? GeographyType::kDefaultCrs : actual.crs(); + if (!StringUtils::EqualsIgnoreCase(expected.crs(), crs) || + algorithm != actual.algorithm()) { + return InvalidSchema("Cannot read Parquet type {} as Iceberg type {} for field {}", + actual.ToString(), expected_type, descr.path()->ToDotString()); + } return {}; } diff --git a/src/iceberg/test/parquet_schema_test.cc b/src/iceberg/test/parquet_schema_test.cc index 4663b184d..219d2bed0 100644 --- a/src/iceberg/test/parquet_schema_test.cc +++ b/src/iceberg/test/parquet_schema_test.cc @@ -414,27 +414,133 @@ TEST(ParquetGeospatialSchemaTest, ConvertsGeospatialTypes) { } } -TEST(ParquetGeospatialSchemaTest, ProjectsTypesWithDifferentParameters) { +TEST(ParquetGeospatialSchemaTest, ProjectsTypesWithMatchingParameters) { Schema expected_schema({ SchemaField::MakeOptional(/*field_id=*/1, "geom", iceberg::geometry()), SchemaField::MakeOptional( /*field_id=*/2, "geog", iceberg::geography("epsg:4326", EdgeAlgorithm::kKarney)), + SchemaField::MakeOptional(/*field_id=*/3, "geog_default", iceberg::geography()), }); auto parquet_schema = MakeGroupNode( "iceberg_schema", - {MakeBinaryNode("geom", ::parquet::LogicalType::Geometry("EPSG:3857"), + {MakeBinaryNode("geom", ::parquet::LogicalType::Geometry(), /*field_id=*/1), - MakeBinaryNode("geog", - ::parquet::LogicalType::Geography( - "OGC:CRS84", - ::parquet::LogicalType::EdgeInterpolationAlgorithm::SPHERICAL), - /*field_id=*/2)}); + MakeBinaryNode( + "geog", + ::parquet::LogicalType::Geography( + "EPSG:4326", ::parquet::LogicalType::EdgeInterpolationAlgorithm::KARNEY), + /*field_id=*/2), + MakeBinaryNode("geog_default", ::parquet::LogicalType::Geography(), + /*field_id=*/3)}); auto manifest = MakeSchemaManifest(parquet_schema); ICEBERG_UNWRAP_OR_FAIL(auto projection, Project(expected_schema, manifest)); - ASSERT_EQ(projection.fields.size(), 2); + ASSERT_EQ(projection.fields.size(), 3); ASSERT_PROJECTED_FIELD(projection.fields[0], 0); ASSERT_PROJECTED_FIELD(projection.fields[1], 1); + ASSERT_PROJECTED_FIELD(projection.fields[2], 2); +} + +TEST(ParquetGeospatialSchemaTest, ProjectsCaseInsensitiveCrsAndAllAlgorithms) { + using ParquetAlgorithm = ::parquet::LogicalType::EdgeInterpolationAlgorithm; + const std::array, 5> algorithms = {{ + {EdgeAlgorithm::kSpherical, ParquetAlgorithm::SPHERICAL}, + {EdgeAlgorithm::kVincenty, ParquetAlgorithm::VINCENTY}, + {EdgeAlgorithm::kThomas, ParquetAlgorithm::THOMAS}, + {EdgeAlgorithm::kAndoyer, ParquetAlgorithm::ANDOYER}, + {EdgeAlgorithm::kKarney, ParquetAlgorithm::KARNEY}, + }}; + for (const auto& [iceberg_algorithm, parquet_algorithm] : algorithms) { + Schema schema({ + SchemaField::MakeOptional(1, "geom", iceberg::geometry("epsg:3857")), + SchemaField::MakeOptional(2, "geog", + iceberg::geography("ogc:crs84", iceberg_algorithm)), + }); + auto parquet_schema = MakeGroupNode( + "iceberg_schema", + {MakeBinaryNode("geom", ::parquet::LogicalType::Geometry("EPSG:3857"), 1), + MakeBinaryNode("geog", + ::parquet::LogicalType::Geography("OGC:CRS84", parquet_algorithm), + 2)}); + auto manifest = MakeSchemaManifest(parquet_schema); + ICEBERG_UNWRAP_OR_FAIL(auto projection, Project(schema, manifest)); + ASSERT_EQ(projection.fields.size(), 2); + ASSERT_PROJECTED_FIELD(projection.fields[0], 0); + ASSERT_PROJECTED_FIELD(projection.fields[1], 1); + } +} + +TEST(ParquetGeospatialSchemaTest, RejectsMismatchedParameters) { + using ParquetAlgorithm = ::parquet::LogicalType::EdgeInterpolationAlgorithm; + const std::vector< + std::pair, std::shared_ptr>> + types = { + {iceberg::geometry(), ::parquet::LogicalType::Geometry("EPSG:3857")}, + {iceberg::geography(), ::parquet::LogicalType::Geography("EPSG:4326")}, + {iceberg::geography(), + ::parquet::LogicalType::Geography("OGC:CRS84", ParquetAlgorithm::KARNEY)}, + {iceberg::geography("OGC:CRS84", EdgeAlgorithm::kKarney), + ::parquet::LogicalType::Geography()}, + }; + for (const auto& [expected, actual] : types) { + SCOPED_TRACE(expected->ToString() + " from " + actual->ToString()); + Schema schema({SchemaField::MakeOptional(1, "location", expected)}); + auto parquet_schema = MakeGroupNode( + "iceberg_schema", {MakeBinaryNode("location", actual, /*field_id=*/1)}); + auto manifest = MakeSchemaManifest(parquet_schema); + auto result = Project(schema, manifest); + ASSERT_THAT(result, IsError(ErrorKind::kInvalidSchema)); + ASSERT_THAT(result, HasErrorMessage("Cannot read Parquet type")); + ASSERT_THAT(result, HasErrorMessage(expected->ToString())); + ASSERT_THAT(result, HasErrorMessage(actual->ToString())); + ASSERT_THAT(result, HasErrorMessage("for field location")); + } +} + +TEST(ParquetGeospatialSchemaTest, ValidatesNestedParameters) { + for (bool matching : {true, false}) { + SCOPED_TRACE(matching); + const auto actual = + ::parquet::LogicalType::Geometry(matching ? "ogc:crs84" : "EPSG:3857"); + const std::vector fields = { + SchemaField::MakeOptional( + 1, "struct_field", + std::make_shared(std::vector{ + SchemaField::MakeOptional(3, "geom", iceberg::geometry())})), + SchemaField::MakeOptional(4, "list_field", + std::make_shared(SchemaField::MakeOptional( + 5, "element", iceberg::geometry()))), + SchemaField::MakeOptional( + 6, "map_field", + std::make_shared( + SchemaField::MakeRequired(7, "key", iceberg::string()), + SchemaField::MakeOptional(8, "value", iceberg::geometry()))), + }; + const std::vector<::parquet::schema::NodePtr> nodes = { + MakeGroupNode("struct_field", + {MakeBinaryNode("unselected", + ::parquet::LogicalType::Geometry("EPSG:3857"), 2), + MakeBinaryNode("geom", actual, 3)}, + 1), + MakeListNode("list_field", MakeBinaryNode("element", actual, 5), 4), + MakeMapNode("map_field", MakeStringNode("key", 7, /*optional=*/false), + MakeBinaryNode("value", actual, 8), 6), + }; + const std::array paths = { + "struct_field.geom", "list_field.element.element", "map_field.key_value.value"}; + for (size_t i = 0; i < fields.size(); ++i) { + SCOPED_TRACE(paths[i]); + Schema schema({fields[i]}); + auto manifest = MakeSchemaManifest(MakeGroupNode("iceberg_schema", {nodes[i]})); + auto result = Project(schema, manifest); + if (matching) { + ASSERT_THAT(result, IsOk()); + } else { + ASSERT_THAT(result, IsError(ErrorKind::kInvalidSchema)); + ASSERT_THAT(result, HasErrorMessage(std::string(paths[i]))); + } + } + } } TEST(ParquetGeospatialSchemaTest, RejectsMissingLogicalType) { From 7ad41087284fc555a5ef31e8b1f558d77c3ddccf Mon Sep 17 00:00:00 2001 From: Manu Zhang Date: Thu, 22 Oct 2026 11:07:04 +0800 Subject: [PATCH 2/2] refactor(parquet): share geospatial parameter validation Extract common CRS normalization, parameter comparison, and mismatch reporting into a helper used by geometry and geography projection. Co-authored-by: Codex --- src/iceberg/parquet/parquet_schema_util.cc | 31 ++++++++++++---------- 1 file changed, 17 insertions(+), 14 deletions(-) diff --git a/src/iceberg/parquet/parquet_schema_util.cc b/src/iceberg/parquet/parquet_schema_util.cc index bf4156a84..77f551ba6 100644 --- a/src/iceberg/parquet/parquet_schema_util.cc +++ b/src/iceberg/parquet/parquet_schema_util.cc @@ -292,6 +292,20 @@ void SelectAnchorColumnIfEmpty( } } +template +Status ValidateGeospatialParameters(const IcebergType& expected, + const ParquetType& actual, + const ::parquet::ColumnDescriptor& descr, + bool algorithm_matches = true) { + // Parquet represents an omitted CRS as an empty string. + const auto crs = actual.crs().empty() ? IcebergType::kDefaultCrs : actual.crs(); + if (!StringUtils::EqualsIgnoreCase(expected.crs(), crs) || !algorithm_matches) { + return InvalidSchema("Cannot read Parquet type {} as Iceberg type {} for field {}", + actual.ToString(), expected, descr.path()->ToDotString()); + } + return {}; +} + Status ValidateGeospatialParquetType(const Type& expected_type, const ::parquet::arrow::SchemaField& parquet_field, const ::parquet::ColumnDescriptor& descr) { @@ -313,13 +327,7 @@ Status ValidateGeospatialParquetType(const Type& expected_type, const auto& expected = internal::checked_cast(expected_type); const auto& actual = internal::checked_cast( *descr.logical_type()); - // Parquet represents an omitted CRS as an empty string. - const auto crs = actual.crs().empty() ? GeometryType::kDefaultCrs : actual.crs(); - if (!StringUtils::EqualsIgnoreCase(expected.crs(), crs)) { - return InvalidSchema("Cannot read Parquet type {} as Iceberg type {} for field {}", - actual.ToString(), expected_type, descr.path()->ToDotString()); - } - return {}; + return ValidateGeospatialParameters(expected, actual, descr); } if (!descr.logical_type()->is_geography()) { return InvalidSchema("Iceberg geography requires Parquet Geography logical type"); @@ -328,13 +336,8 @@ Status ValidateGeospatialParquetType(const Type& expected_type, const auto& actual = internal::checked_cast( *descr.logical_type()); ICEBERG_ASSIGN_OR_RAISE(auto algorithm, ToParquetAlgorithm(expected.algorithm())); - const auto crs = actual.crs().empty() ? GeographyType::kDefaultCrs : actual.crs(); - if (!StringUtils::EqualsIgnoreCase(expected.crs(), crs) || - algorithm != actual.algorithm()) { - return InvalidSchema("Cannot read Parquet type {} as Iceberg type {} for field {}", - actual.ToString(), expected_type, descr.path()->ToDotString()); - } - return {}; + return ValidateGeospatialParameters(expected, actual, descr, + algorithm == actual.algorithm()); } } // namespace