From fb67ebc95ba8cb860fdfa1a1ff1391532364733c Mon Sep 17 00:00:00 2001 From: yinzhengsun Date: Fri, 9 Oct 2026 12:39:52 +0800 Subject: [PATCH 1/2] fix(parquet): accept Decimal32 and Decimal64 during schema projection --- src/iceberg/parquet/parquet_schema_util.cc | 6 ++- src/iceberg/test/parquet_schema_test.cc | 55 ++++++++++++++++++++++ src/iceberg/test/parquet_test.cc | 43 +++++++++++++++++ 3 files changed, 102 insertions(+), 2 deletions(-) diff --git a/src/iceberg/parquet/parquet_schema_util.cc b/src/iceberg/parquet/parquet_schema_util.cc index 0b6abf2185..05f038c133 100644 --- a/src/iceberg/parquet/parquet_schema_util.cc +++ b/src/iceberg/parquet/parquet_schema_util.cc @@ -418,11 +418,13 @@ Status ValidateParquetTypeCompatibility( } break; case TypeId::kDecimal: - if (arrow_type->id() == ::arrow::Type::DECIMAL128) { + if (arrow_type->id() == ::arrow::Type::DECIMAL32 || + arrow_type->id() == ::arrow::Type::DECIMAL64 || + arrow_type->id() == ::arrow::Type::DECIMAL128) { const auto& decimal_type = internal::checked_cast(expected_type); const auto& arrow_decimal = - internal::checked_cast(*arrow_type); + internal::checked_cast(*arrow_type); if (decimal_type.scale() == arrow_decimal.scale() && decimal_type.precision() >= arrow_decimal.precision()) { return {}; diff --git a/src/iceberg/test/parquet_schema_test.cc b/src/iceberg/test/parquet_schema_test.cc index c15c51b236..f3801dcbec 100644 --- a/src/iceberg/test/parquet_schema_test.cc +++ b/src/iceberg/test/parquet_schema_test.cc @@ -861,6 +861,61 @@ TEST(ParquetSchemaProjectionTest, ProjectDecimalType) { ASSERT_PROJECTED_FIELD(projection.fields[0], 0); } +TEST(ParquetSchemaProjectionTest, ProjectSmallDecimalTypes) { + struct TestCase { + ::parquet::Type::type physical_type; + int32_t precision; + ::arrow::Type::type arrow_type; + }; + const std::array cases = { + TestCase{::parquet::Type::INT32, 9, ::arrow::Type::DECIMAL32}, + TestCase{::parquet::Type::INT64, 18, ::arrow::Type::DECIMAL64}, + }; + + for (const auto& test_case : cases) { + SCOPED_TRACE(test_case.precision); + auto node = ::parquet::schema::PrimitiveNode::Make( + "value", ::parquet::Repetition::REQUIRED, + ::parquet::LogicalType::Decimal(test_case.precision, 2), test_case.physical_type, + /*primitive_length=*/-1, /*field_id=*/1); + ::parquet::SchemaDescriptor descriptor; + descriptor.Init(MakeGroupNode("iceberg_schema", {node})); + + for (bool smallest_decimal_enabled : {false, true}) { + SCOPED_TRACE(smallest_decimal_enabled); + auto properties = ::parquet::default_arrow_reader_properties(); + properties.set_smallest_decimal_enabled(smallest_decimal_enabled); + ::parquet::arrow::SchemaManifest manifest; + ASSERT_TRUE(::parquet::arrow::SchemaManifest::Make( + &descriptor, /*key_value_metadata=*/nullptr, properties, &manifest) + .ok()); + ASSERT_EQ( + manifest.schema_fields[0].field->type()->id(), + smallest_decimal_enabled ? test_case.arrow_type : ::arrow::Type::DECIMAL128); + + for (int32_t precision : + {test_case.precision - 1, test_case.precision, test_case.precision + 1}) { + for (int32_t scale : {2, 3}) { + SCOPED_TRACE(precision); + SCOPED_TRACE(scale); + Schema expected_schema({ + SchemaField::MakeRequired(1, "value", iceberg::decimal(precision, scale)), + }); + auto result = Project(expected_schema, manifest); + if (precision >= test_case.precision && scale == 2) { + ASSERT_THAT(result, IsOk()); + ASSERT_EQ(result->fields.size(), 1); + ASSERT_PROJECTED_FIELD(result->fields[0], 0); + } else { + ASSERT_THAT(result, IsError(ErrorKind::kInvalidSchema)); + ASSERT_THAT(result, HasErrorMessage("Cannot read")); + } + } + } + } + } +} + TEST(ParquetSchemaProjectionTest, ProjectDecimalIncompatible) { Schema expected_schema({ SchemaField::MakeRequired(/*field_id=*/1, "value", iceberg::decimal(18, 3)), diff --git a/src/iceberg/test/parquet_test.cc b/src/iceberg/test/parquet_test.cc index ec96167d1c..e929c90364 100644 --- a/src/iceberg/test/parquet_test.cc +++ b/src/iceberg/test/parquet_test.cc @@ -528,6 +528,49 @@ TEST_F(ParquetReaderTest, RoundTripWithGenericFileIO) { ASSERT_TRUE(out->Equals(*array)); } +TEST_F(ParquetReaderTest, ReadSmallDecimalsWithStoredArrowSchema) { + auto expected_schema = std::make_shared(std::vector{ + SchemaField::MakeRequired(1, "value", iceberg::decimal(20, 2)), + }); + auto expected_type = ::arrow::struct_({ + ::arrow::field("value", ::arrow::decimal128(20, 2), /*nullable=*/false), + }); + auto expected = ::arrow::json::ArrayFromJSONString(expected_type, + R"([["-1.23"], ["0.00"], ["4.56"]])") + .ValueOrDie(); + + for (const auto& type : {::arrow::decimal32(9, 2), ::arrow::decimal64(18, 2)}) { + SCOPED_TRACE(type->ToString()); + auto arrow_schema = ::arrow::schema( + {::arrow::field("value", type, /*nullable=*/false, + ::arrow::key_value_metadata({"PARQUET:field_id"}, {"1"}))}); + auto values = ::arrow::json::ArrayFromJSONString(type, R"(["-1.23", "0.00", "4.56"])") + .ValueOrDie(); + auto table = ::arrow::Table::Make(arrow_schema, {values}); + auto& io = internal::checked_cast(*file_io_); + auto output = io.fs()->OpenOutputStream(temp_parquet_file_).ValueOrDie(); + auto writer_properties = + ::parquet::WriterProperties::Builder().enable_store_decimal_as_integer()->build(); + auto arrow_properties = + ::parquet::ArrowWriterProperties::Builder().store_schema()->build(); + ASSERT_TRUE(::parquet::arrow::WriteTable(*table, ::arrow::default_memory_pool(), + output, table->num_rows(), writer_properties, + arrow_properties) + .ok()); + ASSERT_TRUE(output->Close().ok()); + + std::shared_ptr<::arrow::Array> actual; + ASSERT_THAT( + ReadArray( + actual, + {.path = temp_parquet_file_, .io = file_io_, .projection = expected_schema}, + nullptr), + IsOk()); + ASSERT_NE(actual, nullptr); + EXPECT_TRUE(actual->Equals(*expected)) << actual->ToString(); + } +} + TEST_F(ParquetReaderTest, ReadReorderedFieldsWithNulls) { CreateSimpleParquetFile(); From a796060e75fb3aa46f26e1e5d187a0b18c125fe9 Mon Sep 17 00:00:00 2001 From: yinzhengsun Date: Fri, 9 Oct 2026 14:17:16 +0800 Subject: [PATCH 2/2] test(parquet): simplify small decimal regression coverage --- src/iceberg/test/parquet_schema_test.cc | 60 +++++++++++-------------- src/iceberg/test/parquet_test.cc | 40 +++++++---------- 2 files changed, 40 insertions(+), 60 deletions(-) diff --git a/src/iceberg/test/parquet_schema_test.cc b/src/iceberg/test/parquet_schema_test.cc index f3801dcbec..4663b184d4 100644 --- a/src/iceberg/test/parquet_schema_test.cc +++ b/src/iceberg/test/parquet_schema_test.cc @@ -868,9 +868,15 @@ TEST(ParquetSchemaProjectionTest, ProjectSmallDecimalTypes) { ::arrow::Type::type arrow_type; }; const std::array cases = { - TestCase{::parquet::Type::INT32, 9, ::arrow::Type::DECIMAL32}, - TestCase{::parquet::Type::INT64, 18, ::arrow::Type::DECIMAL64}, + TestCase{.physical_type = ::parquet::Type::INT32, + .precision = 9, + .arrow_type = ::arrow::Type::DECIMAL32}, + TestCase{.physical_type = ::parquet::Type::INT64, + .precision = 18, + .arrow_type = ::arrow::Type::DECIMAL64}, }; + auto properties = ::parquet::default_arrow_reader_properties(); + properties.set_smallest_decimal_enabled(true); for (const auto& test_case : cases) { SCOPED_TRACE(test_case.precision); @@ -880,39 +886,23 @@ TEST(ParquetSchemaProjectionTest, ProjectSmallDecimalTypes) { /*primitive_length=*/-1, /*field_id=*/1); ::parquet::SchemaDescriptor descriptor; descriptor.Init(MakeGroupNode("iceberg_schema", {node})); - - for (bool smallest_decimal_enabled : {false, true}) { - SCOPED_TRACE(smallest_decimal_enabled); - auto properties = ::parquet::default_arrow_reader_properties(); - properties.set_smallest_decimal_enabled(smallest_decimal_enabled); - ::parquet::arrow::SchemaManifest manifest; - ASSERT_TRUE(::parquet::arrow::SchemaManifest::Make( - &descriptor, /*key_value_metadata=*/nullptr, properties, &manifest) - .ok()); - ASSERT_EQ( - manifest.schema_fields[0].field->type()->id(), - smallest_decimal_enabled ? test_case.arrow_type : ::arrow::Type::DECIMAL128); - - for (int32_t precision : - {test_case.precision - 1, test_case.precision, test_case.precision + 1}) { - for (int32_t scale : {2, 3}) { - SCOPED_TRACE(precision); - SCOPED_TRACE(scale); - Schema expected_schema({ - SchemaField::MakeRequired(1, "value", iceberg::decimal(precision, scale)), - }); - auto result = Project(expected_schema, manifest); - if (precision >= test_case.precision && scale == 2) { - ASSERT_THAT(result, IsOk()); - ASSERT_EQ(result->fields.size(), 1); - ASSERT_PROJECTED_FIELD(result->fields[0], 0); - } else { - ASSERT_THAT(result, IsError(ErrorKind::kInvalidSchema)); - ASSERT_THAT(result, HasErrorMessage("Cannot read")); - } - } - } - } + ::parquet::arrow::SchemaManifest manifest; + ASSERT_TRUE(::parquet::arrow::SchemaManifest::Make( + &descriptor, /*key_value_metadata=*/nullptr, properties, &manifest) + .ok()); + ASSERT_EQ(manifest.schema_fields[0].field->type()->id(), test_case.arrow_type); + + auto project_decimal = [&](int32_t precision, int32_t scale) { + return Project( + Schema({SchemaField::MakeRequired(1, "value", decimal(precision, scale))}), + manifest); + }; + EXPECT_THAT(project_decimal(test_case.precision, 2), IsOk()); + EXPECT_THAT(project_decimal(test_case.precision + 1, 2), IsOk()); + EXPECT_THAT(project_decimal(test_case.precision - 1, 2), + IsError(ErrorKind::kInvalidSchema)); + EXPECT_THAT(project_decimal(test_case.precision, 3), + IsError(ErrorKind::kInvalidSchema)); } } diff --git a/src/iceberg/test/parquet_test.cc b/src/iceberg/test/parquet_test.cc index e929c90364..ce43962c6a 100644 --- a/src/iceberg/test/parquet_test.cc +++ b/src/iceberg/test/parquet_test.cc @@ -529,16 +529,9 @@ TEST_F(ParquetReaderTest, RoundTripWithGenericFileIO) { } TEST_F(ParquetReaderTest, ReadSmallDecimalsWithStoredArrowSchema) { - auto expected_schema = std::make_shared(std::vector{ - SchemaField::MakeRequired(1, "value", iceberg::decimal(20, 2)), - }); - auto expected_type = ::arrow::struct_({ - ::arrow::field("value", ::arrow::decimal128(20, 2), /*nullable=*/false), + auto schema = std::make_shared(std::vector{ + SchemaField::MakeRequired(1, "value", decimal(20, 2)), }); - auto expected = ::arrow::json::ArrayFromJSONString(expected_type, - R"([["-1.23"], ["0.00"], ["4.56"]])") - .ValueOrDie(); - for (const auto& type : {::arrow::decimal32(9, 2), ::arrow::decimal64(18, 2)}) { SCOPED_TRACE(type->ToString()); auto arrow_schema = ::arrow::schema( @@ -549,25 +542,22 @@ TEST_F(ParquetReaderTest, ReadSmallDecimalsWithStoredArrowSchema) { auto table = ::arrow::Table::Make(arrow_schema, {values}); auto& io = internal::checked_cast(*file_io_); auto output = io.fs()->OpenOutputStream(temp_parquet_file_).ValueOrDie(); - auto writer_properties = - ::parquet::WriterProperties::Builder().enable_store_decimal_as_integer()->build(); - auto arrow_properties = - ::parquet::ArrowWriterProperties::Builder().store_schema()->build(); - ASSERT_TRUE(::parquet::arrow::WriteTable(*table, ::arrow::default_memory_pool(), - output, table->num_rows(), writer_properties, - arrow_properties) + ASSERT_TRUE(::parquet::arrow::WriteTable( + *table, ::arrow::default_memory_pool(), output, table->num_rows(), + ::parquet::WriterProperties::Builder() + .enable_store_decimal_as_integer() + ->build(), + ::parquet::ArrowWriterProperties::Builder().store_schema()->build()) .ok()); ASSERT_TRUE(output->Close().ok()); - std::shared_ptr<::arrow::Array> actual; - ASSERT_THAT( - ReadArray( - actual, - {.path = temp_parquet_file_, .io = file_io_, .projection = expected_schema}, - nullptr), - IsOk()); - ASSERT_NE(actual, nullptr); - EXPECT_TRUE(actual->Equals(*expected)) << actual->ToString(); + auto reader = ReaderFactoryRegistry::Open( + FileFormatType::kParquet, + {.path = temp_parquet_file_, .io = file_io_, .projection = schema}); + ASSERT_THAT(reader, IsOk()); + ASSERT_NO_FATAL_FAILURE( + VerifyNextBatch(**reader, R"([["-1.23"], ["0.00"], ["4.56"]])")); + ASSERT_NO_FATAL_FAILURE(VerifyExhausted(**reader)); } }