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/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/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/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())}); 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();