Skip to content

Commit b409051

Browse files
authored
fix(parquet): accept Decimal32 and Decimal64 during schema projection (#992)
1 parent 66d7ef5 commit b409051

3 files changed

Lines changed: 82 additions & 2 deletions

File tree

‎src/iceberg/parquet/parquet_schema_util.cc‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -418,11 +418,13 @@ Status ValidateParquetTypeCompatibility(
418418
}
419419
break;
420420
case TypeId::kDecimal:
421-
if (arrow_type->id() == ::arrow::Type::DECIMAL128) {
421+
if (arrow_type->id() == ::arrow::Type::DECIMAL32 ||
422+
arrow_type->id() == ::arrow::Type::DECIMAL64 ||
423+
arrow_type->id() == ::arrow::Type::DECIMAL128) {
422424
const auto& decimal_type =
423425
internal::checked_cast<const DecimalType&>(expected_type);
424426
const auto& arrow_decimal =
425-
internal::checked_cast<const ::arrow::Decimal128Type&>(*arrow_type);
427+
internal::checked_cast<const ::arrow::DecimalType&>(*arrow_type);
426428
if (decimal_type.scale() == arrow_decimal.scale() &&
427429
decimal_type.precision() >= arrow_decimal.precision()) {
428430
return {};

‎src/iceberg/test/parquet_schema_test.cc‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -861,6 +861,51 @@ TEST(ParquetSchemaProjectionTest, ProjectDecimalType) {
861861
ASSERT_PROJECTED_FIELD(projection.fields[0], 0);
862862
}
863863

864+
TEST(ParquetSchemaProjectionTest, ProjectSmallDecimalTypes) {
865+
struct TestCase {
866+
::parquet::Type::type physical_type;
867+
int32_t precision;
868+
::arrow::Type::type arrow_type;
869+
};
870+
const std::array cases = {
871+
TestCase{.physical_type = ::parquet::Type::INT32,
872+
.precision = 9,
873+
.arrow_type = ::arrow::Type::DECIMAL32},
874+
TestCase{.physical_type = ::parquet::Type::INT64,
875+
.precision = 18,
876+
.arrow_type = ::arrow::Type::DECIMAL64},
877+
};
878+
auto properties = ::parquet::default_arrow_reader_properties();
879+
properties.set_smallest_decimal_enabled(true);
880+
881+
for (const auto& test_case : cases) {
882+
SCOPED_TRACE(test_case.precision);
883+
auto node = ::parquet::schema::PrimitiveNode::Make(
884+
"value", ::parquet::Repetition::REQUIRED,
885+
::parquet::LogicalType::Decimal(test_case.precision, 2), test_case.physical_type,
886+
/*primitive_length=*/-1, /*field_id=*/1);
887+
::parquet::SchemaDescriptor descriptor;
888+
descriptor.Init(MakeGroupNode("iceberg_schema", {node}));
889+
::parquet::arrow::SchemaManifest manifest;
890+
ASSERT_TRUE(::parquet::arrow::SchemaManifest::Make(
891+
&descriptor, /*key_value_metadata=*/nullptr, properties, &manifest)
892+
.ok());
893+
ASSERT_EQ(manifest.schema_fields[0].field->type()->id(), test_case.arrow_type);
894+
895+
auto project_decimal = [&](int32_t precision, int32_t scale) {
896+
return Project(
897+
Schema({SchemaField::MakeRequired(1, "value", decimal(precision, scale))}),
898+
manifest);
899+
};
900+
EXPECT_THAT(project_decimal(test_case.precision, 2), IsOk());
901+
EXPECT_THAT(project_decimal(test_case.precision + 1, 2), IsOk());
902+
EXPECT_THAT(project_decimal(test_case.precision - 1, 2),
903+
IsError(ErrorKind::kInvalidSchema));
904+
EXPECT_THAT(project_decimal(test_case.precision, 3),
905+
IsError(ErrorKind::kInvalidSchema));
906+
}
907+
}
908+
864909
TEST(ParquetSchemaProjectionTest, ProjectDecimalIncompatible) {
865910
Schema expected_schema({
866911
SchemaField::MakeRequired(/*field_id=*/1, "value", iceberg::decimal(18, 3)),

‎src/iceberg/test/parquet_test.cc‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -528,6 +528,39 @@ TEST_F(ParquetReaderTest, RoundTripWithGenericFileIO) {
528528
ASSERT_TRUE(out->Equals(*array));
529529
}
530530

531+
TEST_F(ParquetReaderTest, ReadSmallDecimalsWithStoredArrowSchema) {
532+
auto schema = std::make_shared<Schema>(std::vector<SchemaField>{
533+
SchemaField::MakeRequired(1, "value", decimal(20, 2)),
534+
});
535+
for (const auto& type : {::arrow::decimal32(9, 2), ::arrow::decimal64(18, 2)}) {
536+
SCOPED_TRACE(type->ToString());
537+
auto arrow_schema = ::arrow::schema(
538+
{::arrow::field("value", type, /*nullable=*/false,
539+
::arrow::key_value_metadata({"PARQUET:field_id"}, {"1"}))});
540+
auto values = ::arrow::json::ArrayFromJSONString(type, R"(["-1.23", "0.00", "4.56"])")
541+
.ValueOrDie();
542+
auto table = ::arrow::Table::Make(arrow_schema, {values});
543+
auto& io = internal::checked_cast<arrow::ArrowFileSystemFileIO&>(*file_io_);
544+
auto output = io.fs()->OpenOutputStream(temp_parquet_file_).ValueOrDie();
545+
ASSERT_TRUE(::parquet::arrow::WriteTable(
546+
*table, ::arrow::default_memory_pool(), output, table->num_rows(),
547+
::parquet::WriterProperties::Builder()
548+
.enable_store_decimal_as_integer()
549+
->build(),
550+
::parquet::ArrowWriterProperties::Builder().store_schema()->build())
551+
.ok());
552+
ASSERT_TRUE(output->Close().ok());
553+
554+
auto reader = ReaderFactoryRegistry::Open(
555+
FileFormatType::kParquet,
556+
{.path = temp_parquet_file_, .io = file_io_, .projection = schema});
557+
ASSERT_THAT(reader, IsOk());
558+
ASSERT_NO_FATAL_FAILURE(
559+
VerifyNextBatch(**reader, R"([["-1.23"], ["0.00"], ["4.56"]])"));
560+
ASSERT_NO_FATAL_FAILURE(VerifyExhausted(**reader));
561+
}
562+
}
563+
531564
TEST_F(ParquetReaderTest, ReadReorderedFieldsWithNulls) {
532565
CreateSimpleParquetFile();
533566

0 commit comments

Comments
 (0)