From bb00e0aa6d99de4182dbeea44e640add876da0a7 Mon Sep 17 00:00:00 2001 From: Manu Zhang Date: Sat, 10 Oct 2026 23:34:31 +0800 Subject: [PATCH 1/2] fix(manifest): round-trip decimal partition values Co-authored-by: Codex --- src/iceberg/arrow_row_builder.cc | 10 +++++++ src/iceberg/arrow_row_builder_internal.h | 3 ++ src/iceberg/manifest/manifest_adapter.cc | 4 +-- src/iceberg/manifest/manifest_reader.cc | 11 +++++++ src/iceberg/test/manifest_reader_test.cc | 37 ++++++++++++++++++++++++ 5 files changed, 63 insertions(+), 2 deletions(-) diff --git a/src/iceberg/arrow_row_builder.cc b/src/iceberg/arrow_row_builder.cc index 91e3cd35ef..017c7cc2e3 100644 --- a/src/iceberg/arrow_row_builder.cc +++ b/src/iceberg/arrow_row_builder.cc @@ -28,6 +28,7 @@ #include "iceberg/nanoarrow_status_internal.h" #include "iceberg/schema.h" #include "iceberg/schema_internal.h" +#include "iceberg/util/decimal.h" namespace iceberg { @@ -123,6 +124,15 @@ Status AppendDouble(ArrowArray* array, double value) { return {}; } +Status AppendDecimal(ArrowArray* array, const Decimal& value) { + // Append the unscaled integer; the array schema supplies precision and scale. + ArrowDecimal decimal; + ArrowDecimalInit(&decimal, Decimal::kBitWidth, Decimal::kMaxPrecision, 0); + ArrowDecimalSetBytes(&decimal, value.native_endian_bytes()); + ICEBERG_NANOARROW_RETURN_UNEXPECTED(ArrowArrayAppendDecimal(array, &decimal)); + return {}; +} + Status AppendString(ArrowArray* array, std::string_view value) { ArrowStringView view(value.data(), static_cast(value.size())); ICEBERG_NANOARROW_RETURN_UNEXPECTED(ArrowArrayAppendString(array, view)); diff --git a/src/iceberg/arrow_row_builder_internal.h b/src/iceberg/arrow_row_builder_internal.h index e9c55f07af..819f5339ef 100644 --- a/src/iceberg/arrow_row_builder_internal.h +++ b/src/iceberg/arrow_row_builder_internal.h @@ -125,6 +125,9 @@ ICEBERG_EXPORT Status AppendUInt(ArrowArray* array, uint64_t value); /// \brief Append a double value to a nanoarrow array builder. ICEBERG_EXPORT Status AppendDouble(ArrowArray* array, double value); +/// \brief Append an unscaled decimal value to a decimal128 array builder. +ICEBERG_EXPORT Status AppendDecimal(ArrowArray* array, const Decimal& value); + /// \brief Append a string value to a nanoarrow array builder. ICEBERG_EXPORT Status AppendString(ArrowArray* array, std::string_view value); diff --git a/src/iceberg/manifest/manifest_adapter.cc b/src/iceberg/manifest/manifest_adapter.cc index b26249a6c5..766a27ef71 100644 --- a/src/iceberg/manifest/manifest_adapter.cc +++ b/src/iceberg/manifest/manifest_adapter.cc @@ -147,8 +147,8 @@ Status ManifestEntryAdapter::AppendPartitionValues( AppendInt(child_array, std::get(partition_value.value()))); break; case TypeId::kDecimal: - ICEBERG_RETURN_UNEXPECTED(AppendBytes( - child_array, std::get(partition_value.value()).ToBytes())); + ICEBERG_RETURN_UNEXPECTED( + AppendDecimal(child_array, std::get(partition_value.value()))); break; case TypeId::kUuid: ICEBERG_RETURN_UNEXPECTED( diff --git a/src/iceberg/manifest/manifest_reader.cc b/src/iceberg/manifest/manifest_reader.cc index 7c920b0484..b1b6535d91 100644 --- a/src/iceberg/manifest/manifest_reader.cc +++ b/src/iceberg/manifest/manifest_reader.cc @@ -20,6 +20,7 @@ #include "iceberg/manifest/manifest_reader.h" #include +#include #include #include #include @@ -416,6 +417,16 @@ Status ParsePartitionValues(ArrowArrayView* view, int64_t row_idx, case ArrowType::NANOARROW_TYPE_DOUBLE: partition.AddValue(Literal::Double(ArrowArrayViewGetDoubleUnsafe(view, row_idx))); break; + case ArrowType::NANOARROW_TYPE_DECIMAL128: { + const auto& type = static_cast(*field_type); + ArrowDecimal value; + ArrowDecimalInit(&value, Decimal::kBitWidth, type.precision(), type.scale()); + ArrowArrayViewGetDecimalUnsafe(view, row_idx, &value); + int128_t unscaled; + std::memcpy(&unscaled, value.words, sizeof(unscaled)); + partition.AddValue(Literal::Decimal(unscaled, type.precision(), type.scale())); + break; + } case ArrowType::NANOARROW_TYPE_STRING: { auto str_value = ArrowArrayViewGetStringUnsafe(view, row_idx); partition.AddValue( diff --git a/src/iceberg/test/manifest_reader_test.cc b/src/iceberg/test/manifest_reader_test.cc index ad9a22eaa8..acfd199ea6 100644 --- a/src/iceberg/test/manifest_reader_test.cc +++ b/src/iceberg/test/manifest_reader_test.cc @@ -319,6 +319,43 @@ TEST_P(TestManifestReader, TestManifestReaderWithPartitionMetadata) { EXPECT_EQ(read_entry.data_file->partition.values()[0], Literal::Int(0)); } +TEST_P(TestManifestReader, DecimalPartitionValuesRoundTrip) { + schema_ = std::make_shared( + std::vector{SchemaField::MakeRequired(3, "amount", decimal(38, 2))}); + ICEBERG_UNWRAP_OR_FAIL( + spec_, + PartitionSpec::Make(0, {PartitionField(3, 1000, "amount", Transform::Identity())})); + + const int128_t large = (static_cast(1) << 80) + 123; + const std::vector values{ + Literal::Decimal(123, 38, 2), Literal::Decimal(-456, 38, 2), + Literal::Decimal(large, 38, 2), Literal::Decimal(-large, 38, 2), + Literal::Decimal(0, 38, 2), Literal::Null(decimal(38, 2))}; + std::vector entries; + for (size_t index = 0; index < values.size(); ++index) { + entries.push_back(MakeEntry(ManifestStatus::kAdded, 1000L, + MakeDataFile(std::format("decimal-{}.parquet", index), + PartitionValues({values[index]})))); + } + auto manifest = WriteManifest(GetParam(), 1000L, entries); + ICEBERG_UNWRAP_OR_FAIL(auto reader, + ManifestReader::Make(manifest, file_io_, schema_, spec_)); + ICEBERG_UNWRAP_OR_FAIL(auto read_entries, reader->Entries()); + + ASSERT_EQ(read_entries.size(), values.size()); + for (size_t index = 0; index < values.size(); ++index) { + ASSERT_NE(read_entries[index].data_file, nullptr); + const auto& partition = read_entries[index].data_file->partition; + ASSERT_EQ(partition.num_fields(), 1); + if (values[index].IsNull()) { + EXPECT_TRUE(partition.values()[0].IsNull()); + } else { + EXPECT_EQ(partition.values()[0], values[index]); + } + EXPECT_EQ(*partition.values()[0].type(), *decimal(38, 2)); + } +} + TEST_P(TestManifestReader, NullPartitionValuePreservedPositionallyAcrossRows) { auto version = GetParam(); From 38ef53be111ed6f2ae0d786ba48881554ef0783c Mon Sep 17 00:00:00 2001 From: Manu Zhang Date: Tue, 13 Oct 2026 01:41:49 +0800 Subject: [PATCH 2/2] fix(avro): honor decimal fixed widths and native byte order Share decimal encoding across direct and generic writers, preserve signed values at the schema fixed width, and reject representation overflow. Cover exact wire bytes and decimal file round trips in both backends. --- src/iceberg/avro/avro_data_util.cc | 31 ++++- src/iceberg/avro/avro_data_util_internal.h | 9 ++ src/iceberg/avro/avro_direct_encoder.cc | 8 +- src/iceberg/test/avro_data_test.cc | 139 +++++++++++++++++++++ src/iceberg/test/avro_test.cc | 17 +++ 5 files changed, 195 insertions(+), 9 deletions(-) diff --git a/src/iceberg/avro/avro_data_util.cc b/src/iceberg/avro/avro_data_util.cc index 297e46b523..39f2948b9f 100644 --- a/src/iceberg/avro/avro_data_util.cc +++ b/src/iceberg/avro/avro_data_util.cc @@ -17,6 +17,8 @@ * under the License. */ +#include +#include #include #include @@ -669,6 +671,27 @@ constexpr int64_t kValueBranch = 1; } // namespace +Status EncodeDecimalToFixed(const ::arrow::Decimal128& value, size_t fixed_size, + std::vector* bytes) { + ICEBERG_PRECHECK(fixed_size > 0, "Avro decimal fixed size must be positive"); + // Arrow's bytes are native-endian; Avro fixed decimals are big-endian. + auto big_endian = value.ToBytes(); + if constexpr (std::endian::native == std::endian::little) { + std::ranges::reverse(big_endian); + } + const uint8_t padding = (big_endian.front() & 0x80) == 0 ? 0x00 : 0xff; + const auto value_size = std::min(fixed_size, big_endian.size()); + const auto first = big_endian.end() - value_size; + // Discard only sign extension, without changing the retained sign bit. + ICEBERG_PRECHECK(std::all_of(big_endian.begin(), first, + [padding](uint8_t byte) { return byte == padding; }) && + ((*first ^ padding) & 0x80) == 0, + "Decimal does not fit Avro fixed size {}", fixed_size); + bytes->assign(fixed_size - value_size, padding); + bytes->insert(bytes->end(), first, big_endian.end()); + return {}; +} + Status ExtractDatumFromArray(const ::arrow::Array& array, int64_t index, ::avro::GenericDatum* datum) { if (index < 0 || index >= array.length()) { @@ -756,12 +779,10 @@ Status ExtractDatumFromArray(const ::arrow::Array& array, int64_t index, case ::arrow::Type::DECIMAL128: { const auto& decimal_array = internal::checked_cast(array); - std::string_view decimal_value = decimal_array.GetView(index); auto& fixed_datum = datum->value<::avro::GenericFixed>(); - auto& bytes = fixed_datum.value(); - bytes.assign(decimal_value.begin(), decimal_value.end()); - std::ranges::reverse(bytes); - return {}; + return EncodeDecimalToFixed(::arrow::Decimal128(decimal_array.GetValue(index)), + fixed_datum.schema()->fixedSize(), + &fixed_datum.value()); } case ::arrow::Type::DATE32: { diff --git a/src/iceberg/avro/avro_data_util_internal.h b/src/iceberg/avro/avro_data_util_internal.h index 9bb86c13f6..b654ce41dd 100644 --- a/src/iceberg/avro/avro_data_util_internal.h +++ b/src/iceberg/avro/avro_data_util_internal.h @@ -19,10 +19,14 @@ #pragma once +#include +#include #include +#include #include #include +#include #include #include "iceberg/arrow/metadata_column_util_internal.h" @@ -83,6 +87,11 @@ Status AppendDatumToBuilder(const ::avro::NodePtr& avro_node, const arrow::MetadataColumnContext& metadata_context, ::arrow::ArrayBuilder* array_builder); +/// \brief Encode a decimal as exactly `fixed_size` big-endian signed bytes. +/// Sign-extends smaller values and rejects values that cannot fit the fixed width. +Status EncodeDecimalToFixed(const ::arrow::Decimal128& value, size_t fixed_size, + std::vector* bytes); + /// \brief Extract an Avro datum from an Arrow array. /// /// \param array The Arrow array to extract from. diff --git a/src/iceberg/avro/avro_direct_encoder.cc b/src/iceberg/avro/avro_direct_encoder.cc index 5dcfd2511d..0ac9e8fd47 100644 --- a/src/iceberg/avro/avro_direct_encoder.cc +++ b/src/iceberg/avro/avro_direct_encoder.cc @@ -25,6 +25,7 @@ #include #include +#include "iceberg/avro/avro_data_util_internal.h" #include "iceberg/avro/avro_direct_encoder_internal.h" #include "iceberg/type.h" #include "iceberg/util/checked_cast.h" @@ -194,10 +195,9 @@ Status EncodeArrowToAvro(const ::avro::NodePtr& avro_node, ::avro::Encoder& enco if (avro_node->logicalType().type() == ::avro::LogicalType::DECIMAL) { const auto& decimal_array = internal::checked_cast(array); - std::string_view decimal_value = decimal_array.GetView(row_index); - ctx.bytes_scratch.assign(decimal_value.begin(), decimal_value.end()); - // Arrow Decimal128 bytes are in little-endian order, Avro requires big-endian - std::ranges::reverse(ctx.bytes_scratch); + ICEBERG_RETURN_UNEXPECTED( + EncodeDecimalToFixed(::arrow::Decimal128(decimal_array.GetValue(row_index)), + avro_node->fixedSize(), &ctx.bytes_scratch)); encoder.encodeFixed(ctx.bytes_scratch.data(), ctx.bytes_scratch.size()); return {}; } diff --git a/src/iceberg/test/avro_data_test.cc b/src/iceberg/test/avro_data_test.cc index 4ea7855ce2..90ed4d0370 100644 --- a/src/iceberg/test/avro_data_test.cc +++ b/src/iceberg/test/avro_data_test.cc @@ -21,6 +21,7 @@ #include #include +#include #include #include #include @@ -31,11 +32,14 @@ #include #include #include +#include +#include #include #include #include #include "iceberg/avro/avro_data_util_internal.h" +#include "iceberg/avro/avro_direct_encoder_internal.h" #include "iceberg/avro/avro_schema_util_internal.h" #include "iceberg/expression/literal.h" #include "iceberg/schema.h" @@ -43,9 +47,144 @@ #include "iceberg/schema_util.h" #include "iceberg/test/matchers.h" #include "iceberg/type.h" +#include "iceberg/util/string_util.h" namespace iceberg::avro { +class AvroDecimalEncodingTest : public ::testing::TestWithParam {}; + +TEST_P(AvroDecimalEncodingTest, FixedWidthBigEndianBytes) { + struct TestCase { + int32_t precision; + std::string unscaled; + std::string expected_hex; + }; + const std::vector cases{ + {.precision = 1, .unscaled = "0", .expected_hex = "00"}, + {.precision = 1, .unscaled = "1", .expected_hex = "01"}, + {.precision = 1, .unscaled = "-1", .expected_hex = "ff"}, + {.precision = 2, .unscaled = "99", .expected_hex = "63"}, + {.precision = 2, .unscaled = "-99", .expected_hex = "9d"}, + {.precision = 3, .unscaled = "127", .expected_hex = "007f"}, + {.precision = 3, .unscaled = "128", .expected_hex = "0080"}, + {.precision = 3, .unscaled = "-128", .expected_hex = "ff80"}, + {.precision = 3, .unscaled = "-129", .expected_hex = "ff7f"}, + {.precision = 5, .unscaled = "32767", .expected_hex = "007fff"}, + {.precision = 5, .unscaled = "32768", .expected_hex = "008000"}, + {.precision = 5, .unscaled = "-32768", .expected_hex = "ff8000"}, + {.precision = 5, .unscaled = "-32769", .expected_hex = "ff7fff"}, + {.precision = 10, .unscaled = "1001", .expected_hex = "00000003e9"}, + {.precision = 10, .unscaled = "-1001", .expected_hex = "fffffffc17"}, + {.precision = 38, + .unscaled = "0", + .expected_hex = "00000000000000000000000000000000"}, + {.precision = 38, + .unscaled = "1", + .expected_hex = "00000000000000000000000000000001"}, + {.precision = 38, + .unscaled = "-1", + .expected_hex = "ffffffffffffffffffffffffffffffff"}, + {.precision = 38, + .unscaled = "128", + .expected_hex = "00000000000000000000000000000080"}, + {.precision = 38, + .unscaled = "-129", + .expected_hex = "ffffffffffffffffffffffffffffff7f"}, + {.precision = 38, + .unscaled = "1208925819614629174706299", + .expected_hex = "0000000000010000000000000000007b"}, + {.precision = 38, + .unscaled = "-1208925819614629174706299", + .expected_hex = "fffffffffffeffffffffffffffffff85"}, + {.precision = 38, + .unscaled = "99999999999999999999999999999999999999", + .expected_hex = "4b3b4ca85a86c47a098a223fffffffff"}, + {.precision = 38, + .unscaled = "-99999999999999999999999999999999999999", + .expected_hex = "b4c4b357a5793b85f675ddc000000001"}, + }; + for (const auto& test : cases) { + SCOPED_TRACE(test.unscaled); + auto type = decimal(test.precision, 0); + ::avro::NodePtr node; + ASSERT_THAT(ToAvroNodeVisitor{}.Visit(*type, &node), IsOk()); + ICEBERG_UNWRAP_OR_FAIL(auto expected, + StringUtils::HexStringToBytes(test.expected_hex)); + ASSERT_EQ(node->fixedSize(), expected.size()); + ::arrow::Decimal128Builder builder(::arrow::decimal128(test.precision, 0), + ::arrow::default_memory_pool()); + ASSERT_TRUE(builder.Append(::arrow::Decimal128(test.unscaled)).ok()); + auto array = builder.Finish().ValueOrDie(); + auto stream = ::avro::memoryOutputStream(); + auto encoder = ::avro::binaryEncoder(); + encoder->init(*stream); + if (GetParam()) { + EncodeContext context; + ASSERT_THAT(EncodeArrowToAvro(node, *encoder, *type, *array, 0, context), IsOk()); + } else { + ::avro::GenericDatum datum(node); + ASSERT_THAT(ExtractDatumFromArray(*array, 0, &datum), IsOk()); + EXPECT_EQ(datum.value<::avro::GenericFixed>().value(), expected); + ::avro::encode(*encoder, datum); + } + encoder->flush(); + EXPECT_EQ(*::avro::snapshot(*stream), expected); + } +} + +TEST_P(AvroDecimalEncodingTest, RejectsFixedWidthOverflow) { + auto type = decimal(2, 0); + ::avro::NodePtr node; + ASSERT_THAT(ToAvroNodeVisitor{}.Visit(*type, &node), IsOk()); + ASSERT_EQ(node->fixedSize(), 1); + // Arrow's builder accepts these invalid values; do not silently truncate them. + for (const auto& unscaled : {"128", "-129", "256", "-257"}) { + SCOPED_TRACE(unscaled); + ::arrow::Decimal128Builder builder(::arrow::decimal128(2, 0), + ::arrow::default_memory_pool()); + ASSERT_TRUE(builder.Append(::arrow::Decimal128(std::string(unscaled))).ok()); + auto array = builder.Finish().ValueOrDie(); + auto stream = ::avro::memoryOutputStream(); + auto encoder = ::avro::binaryEncoder(); + encoder->init(*stream); + if (GetParam()) { + EncodeContext context; + EXPECT_THAT(EncodeArrowToAvro(node, *encoder, *type, *array, 0, context), + IsError(ErrorKind::kInvalidArgument)); + } else { + ::avro::GenericDatum datum(node); + EXPECT_THAT(ExtractDatumFromArray(*array, 0, &datum), + IsError(ErrorKind::kInvalidArgument)); + } + encoder->flush(); + EXPECT_TRUE(::avro::snapshot(*stream)->empty()); + } +} + +TEST(AvroDecimalFixedTest, SignedBoundariesAndPadding) { + std::vector bytes; + ASSERT_THAT(EncodeDecimalToFixed(::arrow::Decimal128(127), 1, &bytes), IsOk()); + EXPECT_EQ(bytes, (std::vector{0x7f})); + ASSERT_THAT(EncodeDecimalToFixed(::arrow::Decimal128(-128), 1, &bytes), IsOk()); + EXPECT_EQ(bytes, (std::vector{0x80})); + ASSERT_THAT(EncodeDecimalToFixed(::arrow::Decimal128(1), 17, &bytes), IsOk()); + ICEBERG_UNWRAP_OR_FAIL( + auto positive, StringUtils::HexStringToBytes("0000000000000000000000000000000001")); + EXPECT_EQ(bytes, positive); + ASSERT_THAT(EncodeDecimalToFixed(::arrow::Decimal128(-1), 17, &bytes), IsOk()); + ICEBERG_UNWRAP_OR_FAIL( + auto negative, StringUtils::HexStringToBytes("ffffffffffffffffffffffffffffffffff")); + EXPECT_EQ(bytes, negative); + EXPECT_THAT(EncodeDecimalToFixed(::arrow::Decimal128(0), 0, &bytes), + IsError(ErrorKind::kInvalidArgument)); +} + +INSTANTIATE_TEST_SUITE_P(WriterBackends, AvroDecimalEncodingTest, + ::testing::Values(true, false), + [](const ::testing::TestParamInfo& info) { + return info.param ? "DirectEncoder" : "GenericDatum"; + }); + /// \brief Test case structure for parameterized primitive type tests struct AppendDatumParam { std::string name; diff --git a/src/iceberg/test/avro_test.cc b/src/iceberg/test/avro_test.cc index aaee666df1..a631b4a103 100644 --- a/src/iceberg/test/avro_test.cc +++ b/src/iceberg/test/avro_test.cc @@ -1042,6 +1042,23 @@ TEST_P(AvroWriterTest, WritePrimitiveTypes) { VerifyWrittenData(test_data); } +TEST_P(AvroWriterTest, WriteDecimalFixedWidths) { + auto schema = std::make_shared( + std::vector{SchemaField::MakeRequired(1, "tiny", decimal(1, 0)), + SchemaField::MakeRequired(2, "narrow", decimal(3, 0)), + SchemaField::MakeRequired(3, "wide", decimal(38, 0)), + SchemaField::MakeRequired(4, "sentinel", int32()), + SchemaField::MakeOptional(5, "optional", decimal(10, 2))}); + const std::string data = R"([ + ["0", "127", "99999999999999999999999999999999999999", 7, "10.01"], + ["1", "128", "0", 8, null], + ["-1", "-128", "-99999999999999999999999999999999999999", 9, "-20.02"], + ["0", "-129", "1", 10, "0.00"] + ])"; + ASSERT_NO_FATAL_FAILURE(WriteAvroFile(schema, data)); + ASSERT_NO_FATAL_FAILURE(VerifyWrittenData(data)); +} + TEST_P(AvroWriterTest, WriteUuidType) { auto schema = std::make_shared(std::vector{ SchemaField::MakeRequired(1, "uuid_col", iceberg::uuid())});