fix(parquet): accept Decimal32 and Decimal64 during schema projection

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<const DecimalType&>(expected_type);
         const auto& arrow_decimal =
-            internal::checked_cast<const ::arrow::Decimal128Type&>(*arrow_type);
+            internal::checked_cast<const ::arrow::DecimalType&>(*arrow_type);
         if (decimal_type.scale() == arrow_decimal.scale() &&
             decimal_type.precision() >= arrow_decimal.precision()) {
           return {};

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)),

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<Schema>(std::vector<SchemaField>{
+      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<arrow::ArrowFileSystemFileIO&>(*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();