Introduce DecimalUtil to convert BigDecimal to reused fix length bytes

core/src/main/java/org/apache/iceberg/avro/ValueWriters.java

@@ -35,6 +35,7 @@
 import org.apache.avro.util.Utf8;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.types.TypeUtil;
+import org.apache.iceberg.util.DecimalUtil;
 
 public class ValueWriters {
   private ValueWriters() {
@@ -338,37 +339,17 @@ public void write(ByteBuffer bytes, Encoder encoder) throws IOException {
   private static class DecimalWriter implements ValueWriter<BigDecimal> {
     private final int precision;
     private final int scale;
-    private final int length;
     private final ThreadLocal<byte[]> bytes;
 
     private DecimalWriter(int precision, int scale) {
       this.precision = precision;
       this.scale = scale;
-      this.length = TypeUtil.decimalRequiredBytes(precision);
-      this.bytes = ThreadLocal.withInitial(() -> new byte[length]);
+      this.bytes = ThreadLocal.withInitial(() -> new byte[TypeUtil.decimalRequiredBytes(precision)]);
     }
 
     @Override
     public void write(BigDecimal decimal, Encoder encoder) throws IOException {
-      Preconditions.checkArgument(decimal.scale() == scale,
-          "Cannot write value as decimal(%s,%s), wrong scale: %s", precision, scale, decimal);
-      Preconditions.checkArgument(decimal.precision() <= precision,
-          "Cannot write value as decimal(%s,%s), too large: %s", precision, scale, decimal);
-
-      byte fillByte = (byte) (decimal.signum() < 0 ? 0xFF : 0x00);
-      byte[] unscaled = decimal.unscaledValue().toByteArray();
-      byte[] buf = bytes.get();
-      int offset = length - unscaled.length;
-
-      for (int i = 0; i < length; i += 1) {
-        if (i < offset) {
-          buf[i] = fillByte;
-        } else {
-          buf[i] = unscaled[i - offset];
-        }
-      }
-
-      encoder.writeFixed(buf);
+      encoder.writeFixed(DecimalUtil.toReusedFixLengthBytes(precision, scale, decimal, bytes.get()));
     }
   }
 

core/src/main/java/org/apache/iceberg/util/DecimalUtil.java

@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.util;
+
+import java.math.BigDecimal;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+public class DecimalUtil {
+  private DecimalUtil() {
+  }
+
+  /**
+   * Convert a {@link BigDecimal} to reused fix length bytes, the extra bytes are filled according to the signum.
+   */
+  public static byte[] toReusedFixLengthBytes(int precision, int scale, BigDecimal decimal, byte[] reuseBuf) {
+    Preconditions.checkArgument(decimal.scale() == scale,
+        "Cannot write value as decimal(%s,%s), wrong scale: %s", precision, scale, decimal);
+    Preconditions.checkArgument(decimal.precision() <= precision,
+        "Cannot write value as decimal(%s,%s), too large: %s", precision, scale, decimal);
+
+    byte[] unscaled = decimal.unscaledValue().toByteArray();
+    if (unscaled.length == reuseBuf.length) {
+      return unscaled;
+    }
+
+    byte fillByte = (byte) (decimal.signum() < 0 ? 0xFF : 0x00);
+    int offset = reuseBuf.length - unscaled.length;
+
+    for (int i = 0; i < reuseBuf.length; i += 1) {
+      if (i < offset) {
+        reuseBuf[i] = fillByte;
+      } else {
+        reuseBuf[i] = unscaled[i - offset];
+      }
+    }
+
+    return reuseBuf;
+  }
+}

parquet/src/main/java/org/apache/iceberg/parquet/ParquetValueWriters.java

@@ -30,6 +30,7 @@
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 import org.apache.iceberg.types.TypeUtil;
+import org.apache.iceberg.util.DecimalUtil;
 import org.apache.parquet.column.ColumnDescriptor;
 import org.apache.parquet.column.ColumnWriteStore;
 import org.apache.parquet.io.api.Binary;
@@ -229,38 +230,19 @@ public void write(int repetitionLevel, BigDecimal decimal) {
   private static class FixedDecimalWriter extends PrimitiveWriter<BigDecimal> {
     private final int precision;
     private final int scale;
-    private final int length;
     private final ThreadLocal<byte[]> bytes;
 
     private FixedDecimalWriter(ColumnDescriptor desc, int precision, int scale) {
       super(desc);
       this.precision = precision;
       this.scale = scale;
-      this.length = TypeUtil.decimalRequiredBytes(precision);
-      this.bytes = ThreadLocal.withInitial(() -> new byte[length]);
+      this.bytes = ThreadLocal.withInitial(() -> new byte[TypeUtil.decimalRequiredBytes(precision)]);
     }
 
     @Override
     public void write(int repetitionLevel, BigDecimal decimal) {
-      Preconditions.checkArgument(decimal.scale() == scale,
-          "Cannot write value as decimal(%s,%s), wrong scale: %s", precision, scale, decimal);
-      Preconditions.checkArgument(decimal.precision() <= precision,
-          "Cannot write value as decimal(%s,%s), too large: %s", precision, scale, decimal);
-
-      byte fillByte = (byte) (decimal.signum() < 0 ? 0xFF : 0x00);
-      byte[] unscaled = decimal.unscaledValue().toByteArray();
-      byte[] buf = bytes.get();
-      int offset = length - unscaled.length;
-
-      for (int i = 0; i < length; i += 1) {
-        if (i < offset) {
-          buf[i] = fillByte;
-        } else {
-          buf[i] = unscaled[i - offset];
-        }
-      }
-
-      column.writeBinary(repetitionLevel, Binary.fromReusedByteArray(buf));
+      byte[] binary = DecimalUtil.toReusedFixLengthBytes(precision, scale, decimal, bytes.get());
+      column.writeBinary(repetitionLevel, Binary.fromReusedByteArray(binary));
     }
   }
 

spark/src/main/java/org/apache/iceberg/spark/data/SparkParquetWriters.java

@@ -19,7 +19,6 @@
 
 package org.apache.iceberg.spark.data;
 
-import java.math.BigDecimal;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -33,6 +32,7 @@
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.types.TypeUtil;
+import org.apache.iceberg.util.DecimalUtil;
 import org.apache.parquet.column.ColumnDescriptor;
 import org.apache.parquet.io.api.Binary;
 import org.apache.parquet.schema.DecimalMetadata;
@@ -272,40 +272,19 @@ public void write(int repetitionLevel, Decimal decimal) {
   private static class FixedDecimalWriter extends PrimitiveWriter<Decimal> {
     private final int precision;
     private final int scale;
-    private final int length;
     private final ThreadLocal<byte[]> bytes;
 
     private FixedDecimalWriter(ColumnDescriptor desc, int precision, int scale) {
       super(desc);
       this.precision = precision;
       this.scale = scale;
-      this.length = TypeUtil.decimalRequiredBytes(precision);
-      this.bytes = ThreadLocal.withInitial(() -> new byte[length]);
+      this.bytes = ThreadLocal.withInitial(() -> new byte[TypeUtil.decimalRequiredBytes(precision)]);
     }
 
     @Override
     public void write(int repetitionLevel, Decimal decimal) {
-      Preconditions.checkArgument(decimal.scale() == scale,
-          "Cannot write value as decimal(%s,%s), wrong scale: %s", precision, scale, decimal);
-      Preconditions.checkArgument(decimal.precision() <= precision,
-          "Cannot write value as decimal(%s,%s), too large: %s", precision, scale, decimal);
-
-      BigDecimal bigDecimal = decimal.toJavaBigDecimal();
-
-      byte fillByte = (byte) (bigDecimal.signum() < 0 ? 0xFF : 0x00);
-      byte[] unscaled = bigDecimal.unscaledValue().toByteArray();
-      byte[] buf = bytes.get();
-      int offset = length - unscaled.length;
-
-      for (int i = 0; i < length; i += 1) {
-        if (i < offset) {
-          buf[i] = fillByte;
-        } else {
-          buf[i] = unscaled[i - offset];
-        }
-      }
-
-      column.writeBinary(repetitionLevel, Binary.fromReusedByteArray(buf));
+      byte[] binary = DecimalUtil.toReusedFixLengthBytes(precision, scale, decimal.toJavaBigDecimal(), bytes.get());
+      column.writeBinary(repetitionLevel, Binary.fromReusedByteArray(binary));
     }
   }
 

spark/src/main/java/org/apache/iceberg/spark/data/SparkValueWriters.java

@@ -21,16 +21,15 @@
 
 import java.io.IOException;
 import java.lang.reflect.Array;
-import java.math.BigDecimal;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
 import java.util.List;
 import java.util.UUID;
 import org.apache.avro.io.Encoder;
 import org.apache.avro.util.Utf8;
 import org.apache.iceberg.avro.ValueWriter;
-import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.types.TypeUtil;
+import org.apache.iceberg.util.DecimalUtil;
 import org.apache.spark.sql.catalyst.InternalRow;
 import org.apache.spark.sql.catalyst.util.ArrayData;
 import org.apache.spark.sql.catalyst.util.MapData;
@@ -114,39 +113,17 @@ public void write(UTF8String s, Encoder encoder) throws IOException {
   private static class DecimalWriter implements ValueWriter<Decimal> {
     private final int precision;
     private final int scale;
-    private final int length;
     private final ThreadLocal<byte[]> bytes;
 
     private DecimalWriter(int precision, int scale) {
       this.precision = precision;
       this.scale = scale;
-      this.length = TypeUtil.decimalRequiredBytes(precision);
-      this.bytes = ThreadLocal.withInitial(() -> new byte[length]);
+      this.bytes = ThreadLocal.withInitial(() -> new byte[TypeUtil.decimalRequiredBytes(precision)]);
     }
 
     @Override
     public void write(Decimal d, Encoder encoder) throws IOException {
-      Preconditions.checkArgument(d.scale() == scale,
-          "Cannot write value as decimal(%s,%s), wrong scale: %s", precision, scale, d);
-      Preconditions.checkArgument(d.precision() <= precision,
-          "Cannot write value as decimal(%s,%s), too large: %s", precision, scale, d);
-
-      BigDecimal decimal = d.toJavaBigDecimal();
-
-      byte fillByte = (byte) (decimal.signum() < 0 ? 0xFF : 0x00);
-      byte[] unscaled = decimal.unscaledValue().toByteArray();
-      byte[] buf = bytes.get();
-      int offset = length - unscaled.length;
-
-      for (int i = 0; i < length; i += 1) {
-        if (i < offset) {
-          buf[i] = fillByte;
-        } else {
-          buf[i] = unscaled[i - offset];
-        }
-      }
-
-      encoder.writeFixed(buf);
+      encoder.writeFixed(DecimalUtil.toReusedFixLengthBytes(precision, scale, d.toJavaBigDecimal(), bytes.get()));
     }
   }