fix(manifest): round-trip decimal partition values (#998)

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.

Co-authored-by: Codex <codex@openai.com>

Author: manuzhang

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<int64_t>(value.size()));
   ICEBERG_NANOARROW_RETURN_UNEXPECTED(ArrowArrayAppendString(array, view));

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

src/iceberg/avro/avro_data_util.cc

@@ -17,6 +17,8 @@
  * under the License.
  */
 
+#include <algorithm>
+#include <bit>
 #include <span>
 
 #include <arrow/array/builder_binary.h>
@@ -669,6 +671,27 @@ constexpr int64_t kValueBranch = 1;
 
 }  // namespace
 
+Status EncodeDecimalToFixed(const ::arrow::Decimal128& value, size_t fixed_size,
+                            std::vector<uint8_t>* 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<const ::arrow::Decimal128Array&>(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: {

src/iceberg/avro/avro_data_util_internal.h

@@ -19,10 +19,14 @@
 
 #pragma once
 
+#include <cstddef>
+#include <cstdint>
 #include <memory>
+#include <vector>
 
 #include <arrow/array/builder_base.h>
 #include <arrow/scalar.h>
+#include <arrow/util/decimal.h>
 #include <avro/GenericDatum.hh>
 
 #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<uint8_t>* bytes);
+
 /// \brief Extract an Avro datum from an Arrow array.
 ///
 /// \param array The Arrow array to extract from.

src/iceberg/avro/avro_direct_encoder.cc

@@ -25,6 +25,7 @@
 #include <arrow/type.h>
 #include <avro/Specific.hh>
 
+#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<const ::arrow::Decimal128Array&>(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 {};
       }

src/iceberg/manifest/manifest_adapter.cc

@@ -147,8 +147,8 @@ Status ManifestEntryAdapter::AppendPartitionValues(
             AppendInt(child_array, std::get<int64_t>(partition_value.value())));
         break;
       case TypeId::kDecimal:
-        ICEBERG_RETURN_UNEXPECTED(AppendBytes(
-            child_array, std::get<Decimal>(partition_value.value()).ToBytes()));
+        ICEBERG_RETURN_UNEXPECTED(
+            AppendDecimal(child_array, std::get<Decimal>(partition_value.value())));
         break;
       case TypeId::kUuid:
         ICEBERG_RETURN_UNEXPECTED(

src/iceberg/manifest/manifest_reader.cc

@@ -20,6 +20,7 @@
 #include "iceberg/manifest/manifest_reader.h"
 
 #include <algorithm>
+#include <cstring>
 #include <memory>
 #include <optional>
 #include <ranges>
@@ -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<const DecimalType&>(*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(

src/iceberg/test/avro_data_test.cc

@@ -21,6 +21,7 @@
 #include <ranges>
 #include <vector>
 
+#include <arrow/array/builder_decimal.h>
 #include <arrow/array/builder_nested.h>
 #include <arrow/array/builder_primitive.h>
 #include <arrow/c/bridge.h>
@@ -31,21 +32,159 @@
 #include <avro/Compiler.hh>
 #include <avro/Generic.hh>
 #include <avro/Node.hh>
+#include <avro/Specific.hh>
+#include <avro/Stream.hh>
 #include <avro/Types.hh>
 #include <gmock/gmock.h>
 #include <gtest/gtest.h>
 
 #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"
 #include "iceberg/schema_internal.h"
 #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<bool> {};
+
+TEST_P(AvroDecimalEncodingTest, FixedWidthBigEndianBytes) {
+  struct TestCase {
+    int32_t precision;
+    std::string unscaled;
+    std::string expected_hex;
+  };
+  const std::vector<TestCase> 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<uint8_t> bytes;
+  ASSERT_THAT(EncodeDecimalToFixed(::arrow::Decimal128(127), 1, &bytes), IsOk());
+  EXPECT_EQ(bytes, (std::vector<uint8_t>{0x7f}));
+  ASSERT_THAT(EncodeDecimalToFixed(::arrow::Decimal128(-128), 1, &bytes), IsOk());
+  EXPECT_EQ(bytes, (std::vector<uint8_t>{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<bool>& info) {
+                           return info.param ? "DirectEncoder" : "GenericDatum";
+                         });
+
 /// \brief Test case structure for parameterized primitive type tests
 struct AppendDatumParam {
   std::string name;

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<Schema>(
+      std::vector<SchemaField>{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<iceberg::Schema>(std::vector<SchemaField>{
       SchemaField::MakeRequired(1, "uuid_col", iceberg::uuid())});