ClickHouseIO: Add Decimal(P, S) support for fixed-point columns

CHANGES.md

@@ -66,6 +66,7 @@
 
 * Support for X source added (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
 * BigQueryIO now supports reading BigQuery Lakehouse runtime catalog (BigLake metastore) Iceberg tables with the Storage Read API, using 4-part `project.catalog.namespace.table` identifiers (or a `TableReference` with a composite `catalog.namespace` dataset id). Previously such references were silently mis-parsed (Java) ([#39597](https://github.com/apache/beam/issues/39597)) .
+* ClickHouseIO: support writing `Decimal(P, S)` / `Decimal32/64/128/256` columns (Java) ([#39840](https://github.com/apache/beam/issues/39840)).
 
 ## New Features / Improvements
 

sdks/java/io/clickhouse/src/main/java/org/apache/beam/sdk/io/clickhouse/ClickHouseIO.java

@@ -38,6 +38,7 @@
 import org.apache.beam.sdk.schemas.FieldAccessDescriptor;
 import org.apache.beam.sdk.schemas.Schema;
 import org.apache.beam.sdk.schemas.logicaltypes.FixedBytes;
+import org.apache.beam.sdk.schemas.logicaltypes.FixedPrecisionNumeric;
 import org.apache.beam.sdk.schemas.logicaltypes.NanosInstant;
 import org.apache.beam.sdk.schemas.logicaltypes.SqlTypes;
 import org.apache.beam.sdk.schemas.transforms.Select;
@@ -140,13 +141,19 @@
  * <tr><td>{@link TableSchema.TypeName#DATE}</td> <td>{@link Schema.TypeName#DATETIME}</td></tr>
  * <tr><td>{@link TableSchema.TypeName#DATETIME}</td> <td>{@link Schema.TypeName#DATETIME}</td></tr>
  * <tr><td>{@link TableSchema.TypeName#DATETIME64}</td> <td>{@link Schema.TypeName#DATETIME} (precision &le; 3), {@link SqlTypes#TIMESTAMP} (4&ndash;6), or {@link NanosInstant} (&ge; 7)</td></tr>
+ * <tr><td>{@link TableSchema.TypeName#DECIMAL}</td> <td>{@link FixedPrecisionNumeric}</td></tr>
  * <tr><td>{@link TableSchema.TypeName#ARRAY}</td> <td>{@link Schema.TypeName#ARRAY}</td></tr>
  * <tr><td>{@link TableSchema.TypeName#ENUM8}</td> <td>{@link Schema.TypeName#STRING}</td></tr>
  * <tr><td>{@link TableSchema.TypeName#ENUM16}</td> <td>{@link Schema.TypeName#STRING}</td></tr>
  * <tr><td>{@link TableSchema.TypeName#BOOL}</td> <td>{@link Schema.TypeName#BOOLEAN}</td></tr>
  * <tr><td>{@link TableSchema.TypeName#TUPLE}</td> <td>{@link Schema.TypeName#ROW}</td></tr>
  * </table>
  *
+ * <p>{@code Decimal(P, S)} columns accept {@link java.math.BigDecimal} values. Fractional digits
+ * beyond the column scale are truncated toward zero, matching ClickHouse's own handling of excess
+ * fraction. A value whose truncated result still exceeds the column's declared range is rejected
+ * with an {@link IllegalArgumentException} rather than written.
+ *
  * <p>Nullable row columns are supported through <a
  * href="https://clickhouse.com/docs/sql-reference/data-types/nullable">Nullable type</a> in
  * ClickHouse. <a href="https://clickhouse.com/docs/sql-reference/data-types/LowCardinality">Low

sdks/java/io/clickhouse/src/main/java/org/apache/beam/sdk/io/clickhouse/ClickHouseWriter.java

@@ -21,6 +21,9 @@
 import com.clickhouse.data.ClickHousePipedOutputStream;
 import com.clickhouse.data.format.BinaryStreamUtils;
 import java.io.IOException;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.math.RoundingMode;
 import java.nio.charset.StandardCharsets;
 import java.util.Collection;
 import java.util.List;
@@ -44,6 +47,37 @@ public class ClickHouseWriter {
     1L, 10L, 100L, 1_000L, 10_000L, 100_000L, 1_000_000L, 10_000_000L, 100_000_000L, 1_000_000_000L
   };
 
+  // 10^0 through 10^76 — the exclusive bound on a Decimal's unscaled value at each precision.
+  // Precision is validated in [1, 76] by ColumnType.decimal.
+  private static final BigInteger[] DECIMAL_BOUNDS = new BigInteger[77];
+
+  static {
+    for (int i = 0; i < DECIMAL_BOUNDS.length; i++) {
+      DECIMAL_BOUNDS[i] = BigInteger.TEN.pow(i);
+    }
+  }
+
+  /**
+   * Truncates a value to a {@code Decimal(precision, scale)} column's scale and checks it against
+   * the column's declared range.
+   *
+   * <p>Excess fractional digits are discarded toward zero, matching ClickHouse's own behavior and
+   * the truncation {@link BinaryStreamUtils#writeDecimal} would otherwise apply. The result is then
+   * bounded by the declared precision: {@code writeDecimal} only range-checks against the backing
+   * storage width (32/64/128/256 bits, selected from the precision), which is wider than the
+   * declared type, and ClickHouse's RowBinary reader does not re-check. Without this, a value such
+   * as {@code 100000} would be stored in a {@code Decimal(5, 0)} column whose declared maximum is
+   * {@code 99999}.
+   */
+  static BigDecimal truncateAndCheckDecimal(BigDecimal value, int precision, int scale) {
+    BigDecimal truncated = value.setScale(scale, RoundingMode.DOWN);
+    if (truncated.unscaledValue().abs().compareTo(DECIMAL_BOUNDS[precision]) >= 0) {
+      throw new IllegalArgumentException(
+          "value " + value + " is out of range for Decimal(" + precision + ", " + scale + ")");
+    }
+    return truncated;
+  }
+
   /**
    * Encodes a timestamp into ClickHouse's {@code DateTime64(precision)} representation: a signed
    * 64-bit integer counting ticks of size 10<sup>-precision</sup> seconds since the Unix epoch.
@@ -178,6 +212,19 @@ static void writeValue(ClickHouseOutputStream stream, ColumnType columnType, Obj
         BinaryStreamUtils.writeInt64(stream, encodeDateTime64(value, precision));
         break;
 
+      case DECIMAL:
+        int decimalPrecision =
+            Preconditions.checkNotNull(columnType.precision(), "Decimal column missing precision");
+        int decimalScale =
+            Preconditions.checkNotNull(columnType.scale(), "Decimal column missing scale");
+        // Truncate to the column scale and bound by the declared precision first; writeDecimal
+        // then picks the 32/64/128/256-bit little-endian storage width from the precision and
+        // writes the value as 10^-scale ticks.
+        BigDecimal decimalValue =
+            truncateAndCheckDecimal((BigDecimal) value, decimalPrecision, decimalScale);
+        BinaryStreamUtils.writeDecimal(stream, decimalValue, decimalPrecision, decimalScale);
+        break;
+
       case ARRAY:
         List<Object> values = (List<Object>) value;
         BinaryStreamUtils.writeVarInt(stream, values.size());

sdks/java/io/clickhouse/src/main/java/org/apache/beam/sdk/io/clickhouse/TableSchema.java

@@ -20,15 +20,18 @@
 import com.google.auto.value.AutoValue;
 import java.io.Serializable;
 import java.io.StringReader;
+import java.math.BigDecimal;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.stream.Collectors;
 import org.apache.beam.sdk.schemas.Schema;
 import org.apache.beam.sdk.schemas.logicaltypes.FixedBytes;
+import org.apache.beam.sdk.schemas.logicaltypes.FixedPrecisionNumeric;
 import org.apache.beam.sdk.schemas.logicaltypes.NanosInstant;
 import org.apache.beam.sdk.schemas.logicaltypes.SqlTypes;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions;
 import org.checkerframework.checker.nullness.qual.Nullable;
 
 /**
@@ -97,6 +100,14 @@ public static Schema.FieldType getEquivalentFieldType(ColumnType columnType) {
           return NANOS_INSTANT_TYPE;
         }
 
+      case DECIMAL:
+        int decimalPrecision =
+            Preconditions.checkNotNull(columnType.precision(), "Decimal column missing precision");
+        int decimalScale =
+            Preconditions.checkNotNull(columnType.scale(), "Decimal column missing scale");
+        return Schema.FieldType.logicalType(
+            FixedPrecisionNumeric.of(decimalPrecision, decimalScale));
+
       case STRING:
         return Schema.FieldType.STRING;
 
@@ -185,6 +196,7 @@ public enum TypeName {
     DATE,
     DATETIME,
     DATETIME64,
+    DECIMAL,
     ENUM8,
     ENUM16,
     FIXEDSTRING,
@@ -260,9 +272,18 @@ public abstract static class ColumnType implements Serializable {
 
     public abstract @Nullable Map<String, ColumnType> tupleTypes();
 
-    /** Sub-second precision (0–9) of {@code DateTime64}. {@code null} for other types. */
+    /**
+     * Sub-second precision (0–9) of {@code DateTime64}, or total number of decimal digits (1–76) of
+     * {@code Decimal}. {@code null} for other types.
+     */
     public abstract @Nullable Integer precision();
 
+    /**
+     * Number of fractional decimal digits (0–precision) of {@code Decimal}. {@code null} for other
+     * types.
+     */
+    public abstract @Nullable Integer scale();
+
     public ColumnType withNullable(boolean nullable) {
       return toBuilder().nullable(nullable).build();
     }
@@ -303,6 +324,40 @@ public static ColumnType dateTime64(int precision) {
           .build();
     }
 
+    /** Default {@code Decimal} precision in ClickHouse when none is specified. */
+    public static final int DEFAULT_DECIMAL_PRECISION = 10;
+
+    /** Default {@code Decimal} scale in ClickHouse when none is specified. */
+    public static final int DEFAULT_DECIMAL_SCALE = 0;
+
+    /**
+     * Returns a {@code Decimal(precision, scale)} type.
+     *
+     * <p>ClickHouse stores {@code Decimal} values as integers of a width chosen from the declared
+     * precision: 32 bits for precision 1–9, 64 for 10–18, 128 for 19–38 and 256 for 39–76. The
+     * width aliases {@code Decimal32(S)}, {@code Decimal64(S)}, {@code Decimal128(S)} and {@code
+     * Decimal256(S)} correspond to precisions 9, 18, 38 and 76.
+     *
+     * @param precision total number of decimal digits, in {@code [1, 76]}
+     * @param scale number of fractional decimal digits, in {@code [0, precision]}
+     */
+    public static ColumnType decimal(int precision, int scale) {
+      if (precision < 1 || precision > 76) {
+        throw new IllegalArgumentException(
+            "Decimal precision must be in [1, 76], got " + precision);
+      }
+      if (scale < 0 || scale > precision) {
+        throw new IllegalArgumentException(
+            "Decimal scale must be in [0, " + precision + "], got " + scale);
+      }
+      return ColumnType.builder()
+          .typeName(TypeName.DECIMAL)
+          .nullable(false)
+          .precision(precision)
+          .scale(scale)
+          .build();
+    }
+
     public static ColumnType enum8(Map<String, Integer> enumValues) {
       return ColumnType.builder()
           .typeName(TypeName.ENUM8)
@@ -390,6 +445,8 @@ public static Object parseDefaultExpression(ColumnType columnType, String value)
           return Long.valueOf(value);
         case BOOL:
           return Boolean.valueOf(value);
+        case DECIMAL:
+          return new BigDecimal(value);
         default:
           throw new UnsupportedOperationException("Unsupported type: " + columnType);
       }
@@ -418,6 +475,8 @@ abstract static class Builder {
 
       public abstract Builder precision(@Nullable Integer precision);
 
+      public abstract Builder scale(@Nullable Integer scale);
+
       public abstract ColumnType build();
     }
   }

sdks/java/io/clickhouse/src/main/javacc/ColumnTypeParser.jj

@@ -81,6 +81,11 @@ TOKEN :
   | < DATE           : "DATE" >
   | < DATETIME64     : "DATETIME64" >
   | < DATETIME       : "DATETIME" >
+  | < DECIMAL256     : "DECIMAL256" >
+  | < DECIMAL128     : "DECIMAL128" >
+  | < DECIMAL64      : "DECIMAL64" >
+  | < DECIMAL32      : "DECIMAL32" >
+  | < DECIMAL        : "DECIMAL" >
   | < ENUM8          : "ENUM8" >
   | < ENUM16         : "ENUM16" >
   | < FIXEDSTRING    : "FIXEDSTRING" >
@@ -218,6 +223,8 @@ private ColumnType primitive() :
       }
     |
       (ct = dateTime64()) { return ct; }
+    |
+      (ct = decimal()) { return ct; }
     )
 
 }
@@ -245,6 +252,42 @@ private ColumnType dateTime64() :
     }
 }
 
+private ColumnType decimal() :
+{
+    String precision = null;
+    String scale = null;
+}
+{
+    (
+        <DECIMAL>
+        (
+            <LPAREN> ( precision = integer() )
+            ( <COMMA> ( scale = integer() ) )?
+            <RPAREN>
+        )?
+        {
+            // Bare Decimal is Decimal(10, 0) and Decimal(P) is Decimal(P, 0), matching ClickHouse.
+            int p = precision == null
+                ? ColumnType.DEFAULT_DECIMAL_PRECISION : Integer.parseInt(precision);
+            int s = scale == null ? ColumnType.DEFAULT_DECIMAL_SCALE : Integer.parseInt(scale);
+            return ColumnType.decimal(p, s);
+        }
+    |
+        // The width aliases pin the precision to the maximum of their storage width.
+        (<DECIMAL32> <LPAREN> ( scale = integer() ) <RPAREN>)
+        { return ColumnType.decimal(9, Integer.parseInt(scale)); }
+    |
+        (<DECIMAL64> <LPAREN> ( scale = integer() ) <RPAREN>)
+        { return ColumnType.decimal(18, Integer.parseInt(scale)); }
+    |
+        (<DECIMAL128> <LPAREN> ( scale = integer() ) <RPAREN>)
+        { return ColumnType.decimal(38, Integer.parseInt(scale)); }
+    |
+        (<DECIMAL256> <LPAREN> ( scale = integer() ) <RPAREN>)
+        { return ColumnType.decimal(76, Integer.parseInt(scale)); }
+    )
+}
+
 private ColumnType nullable() :
 {
     ColumnType ct;