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..4663b184d4 100644 --- a/src/iceberg/test/parquet_schema_test.cc +++ b/src/iceberg/test/parquet_schema_test.cc @@ -861,6 +861,51 @@ 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{.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); + 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})); + ::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)); + } +} + 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..ce43962c6a 100644 --- a/src/iceberg/test/parquet_test.cc +++ b/src/iceberg/test/parquet_test.cc @@ -528,6 +528,39 @@ TEST_F(ParquetReaderTest, RoundTripWithGenericFileIO) { ASSERT_TRUE(out->Equals(*array)); } +TEST_F(ParquetReaderTest, ReadSmallDecimalsWithStoredArrowSchema) { + auto schema = std::make_shared(std::vector{ + SchemaField::MakeRequired(1, "value", decimal(20, 2)), + }); + 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(); + 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()); + + 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)); + } +} + TEST_F(ParquetReaderTest, ReadReorderedFieldsWithNulls) { CreateSimpleParquetFile();