fix(manifest): round-trip decimal partition values

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