Low
diff --git a/sdks/java/io/clickhouse/src/main/java/org/apache/beam/sdk/io/clickhouse/ClickHouseWriter.java b/sdks/java/io/clickhouse/src/main/java/org/apache/beam/sdk/io/clickhouse/ClickHouseWriter.java
index 4d9f072e598f..bf2f53944a0b 100644
--- a/sdks/java/io/clickhouse/src/main/java/org/apache/beam/sdk/io/clickhouse/ClickHouseWriter.java
+++ b/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.
+ *
+ * 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-precision 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 values = (List) value;
BinaryStreamUtils.writeVarInt(stream, values.size());
diff --git a/sdks/java/io/clickhouse/src/main/java/org/apache/beam/sdk/io/clickhouse/TableSchema.java b/sdks/java/io/clickhouse/src/main/java/org/apache/beam/sdk/io/clickhouse/TableSchema.java
index 1b9fdffd4c86..65b4910773d7 100644
--- a/sdks/java/io/clickhouse/src/main/java/org/apache/beam/sdk/io/clickhouse/TableSchema.java
+++ b/sdks/java/io/clickhouse/src/main/java/org/apache/beam/sdk/io/clickhouse/TableSchema.java
@@ -20,6 +20,7 @@
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;
@@ -27,8 +28,10 @@
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 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.
+ *
+ * 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 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();
}
}
diff --git a/sdks/java/io/clickhouse/src/main/javacc/ColumnTypeParser.jj b/sdks/java/io/clickhouse/src/main/javacc/ColumnTypeParser.jj
index 53ad991b67c3..c3e08f8acbaf 100644
--- a/sdks/java/io/clickhouse/src/main/javacc/ColumnTypeParser.jj
+++ b/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;
+}
+{
+ (
+
+ (
+ ( precision = integer() )
+ ( ( scale = integer() ) )?
+
+ )?
+ {
+ // 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.
+ ( ( scale = integer() ) )
+ { return ColumnType.decimal(9, Integer.parseInt(scale)); }
+ |
+ ( ( scale = integer() ) )
+ { return ColumnType.decimal(18, Integer.parseInt(scale)); }
+ |
+ ( ( scale = integer() ) )
+ { return ColumnType.decimal(38, Integer.parseInt(scale)); }
+ |
+ ( ( scale = integer() ) )
+ { return ColumnType.decimal(76, Integer.parseInt(scale)); }
+ )
+}
+
private ColumnType nullable() :
{
ColumnType ct;
diff --git a/sdks/java/io/clickhouse/src/test/java/org/apache/beam/sdk/io/clickhouse/ClickHouseIOIT.java b/sdks/java/io/clickhouse/src/test/java/org/apache/beam/sdk/io/clickhouse/ClickHouseIOIT.java
index da435f206842..7a5a39ae83d6 100644
--- a/sdks/java/io/clickhouse/src/test/java/org/apache/beam/sdk/io/clickhouse/ClickHouseIOIT.java
+++ b/sdks/java/io/clickhouse/src/test/java/org/apache/beam/sdk/io/clickhouse/ClickHouseIOIT.java
@@ -23,15 +23,18 @@
import com.clickhouse.client.api.query.GenericRecord;
import com.clickhouse.client.api.query.Records;
+import java.math.BigDecimal;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Objects;
import java.util.Properties;
+import org.apache.beam.sdk.io.clickhouse.TableSchema.ColumnType;
import org.apache.beam.sdk.schemas.JavaFieldSchema;
import org.apache.beam.sdk.schemas.Schema;
import org.apache.beam.sdk.schemas.Schema.FieldType;
import org.apache.beam.sdk.schemas.annotations.DefaultSchema;
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.testing.TestPipeline;
@@ -599,6 +602,185 @@ public void testNullableDateTime64Nanos() throws Exception {
assertEquals(TEST_EPOCH_SECONDS * NANOS_PER_SECOND + TEST_NANOS_OF_SECOND, ticks);
}
+ @Test
+ public void testDecimal32() throws Exception {
+ Schema schema =
+ Schema.of(Schema.Field.of("d", FieldType.logicalType(FixedPrecisionNumeric.of(9, 2))));
+ Row row = Row.withSchema(schema).addValue(new BigDecimal("-12345.67")).build();
+
+ executeSql("CREATE TABLE test_decimal32 (d Decimal(9, 2)) ENGINE=Log");
+
+ pipeline.apply(Create.of(row).withRowSchema(schema)).apply(write("test_decimal32"));
+ pipeline.run().waitUntilFinish();
+
+ assertEquals("-12345.67", executeQueryAsString("SELECT toString(d) FROM test_decimal32"));
+ }
+
+ @Test
+ public void testDecimal64() throws Exception {
+ Schema schema =
+ Schema.of(Schema.Field.of("d", FieldType.logicalType(FixedPrecisionNumeric.of(18, 4))));
+ Row row = Row.withSchema(schema).addValue(new BigDecimal("12345678901234.5678")).build();
+
+ executeSql("CREATE TABLE test_decimal64 (d Decimal(18, 4)) ENGINE=Log");
+
+ pipeline.apply(Create.of(row).withRowSchema(schema)).apply(write("test_decimal64"));
+ pipeline.run().waitUntilFinish();
+
+ assertEquals(
+ "12345678901234.5678", executeQueryAsString("SELECT toString(d) FROM test_decimal64"));
+ }
+
+ @Test
+ public void testDecimal128() throws Exception {
+ Schema schema =
+ Schema.of(Schema.Field.of("d", FieldType.logicalType(FixedPrecisionNumeric.of(38, 10))));
+ Row row =
+ Row.withSchema(schema)
+ .addValue(new BigDecimal("1234567890123456789012345678.0123456789"))
+ .build();
+
+ executeSql("CREATE TABLE test_decimal128 (d Decimal(38, 10)) ENGINE=Log");
+
+ pipeline.apply(Create.of(row).withRowSchema(schema)).apply(write("test_decimal128"));
+ pipeline.run().waitUntilFinish();
+
+ assertEquals(
+ "1234567890123456789012345678.0123456789",
+ executeQueryAsString("SELECT toString(d) FROM test_decimal128"));
+ }
+
+ @Test
+ public void testDecimal256() throws Exception {
+ Schema schema =
+ Schema.of(Schema.Field.of("d", FieldType.logicalType(FixedPrecisionNumeric.of(76, 20))));
+ String value = "-123456789012345678901234567890123456789012345678901234.12345678901234567891";
+ Row row = Row.withSchema(schema).addValue(new BigDecimal(value)).build();
+
+ executeSql("CREATE TABLE test_decimal256 (d Decimal(76, 20)) ENGINE=Log");
+
+ pipeline.apply(Create.of(row).withRowSchema(schema)).apply(write("test_decimal256"));
+ pipeline.run().waitUntilFinish();
+
+ assertEquals(value, executeQueryAsString("SELECT toString(d) FROM test_decimal256"));
+ }
+
+ @Test
+ public void testNullableDecimal() throws Exception {
+ Schema schema =
+ Schema.of(
+ Schema.Field.nullable("d", FieldType.logicalType(FixedPrecisionNumeric.of(10, 2))));
+ Row row1 = Row.withSchema(schema).addValue(new BigDecimal("3.14")).build();
+ Row row2 = Row.withSchema(schema).addValue(null).build();
+
+ executeSql("CREATE TABLE test_nullable_decimal (d Nullable(Decimal(10, 2))) ENGINE=Log");
+
+ pipeline
+ .apply(Create.of(row1, row2).withRowSchema(schema))
+ .apply(write("test_nullable_decimal"));
+ pipeline.run().waitUntilFinish();
+
+ long total = executeQueryAsLong("SELECT COUNT(*) FROM test_nullable_decimal");
+ long nonNull = executeQueryAsLong("SELECT COUNT(d) FROM test_nullable_decimal");
+ String value =
+ executeQueryAsString("SELECT toString(d) FROM test_nullable_decimal WHERE d IS NOT NULL");
+ assertEquals(2L, total);
+ assertEquals(1L, nonNull);
+ assertEquals("3.14", value);
+ }
+
+ @Test
+ public void testArrayOfDecimal() throws Exception {
+ Schema schema =
+ Schema.of(
+ Schema.Field.of(
+ "d", FieldType.array(FieldType.logicalType(FixedPrecisionNumeric.of(9, 2)))));
+ Row row =
+ Row.withSchema(schema)
+ .addValue(Arrays.asList(new BigDecimal("1.23"), new BigDecimal("-4.56")))
+ .build();
+
+ executeSql("CREATE TABLE test_array_of_decimal (d Array(Decimal(9, 2))) ENGINE=Log");
+
+ pipeline.apply(Create.of(row).withRowSchema(schema)).apply(write("test_array_of_decimal"));
+ pipeline.run().waitUntilFinish();
+
+ assertEquals("1.23", executeQueryAsString("SELECT toString(d[1]) FROM test_array_of_decimal"));
+ assertEquals("-4.56", executeQueryAsString("SELECT toString(d[2]) FROM test_array_of_decimal"));
+ }
+
+ @Test
+ public void testDecimalTruncatesExcessFractionOnWrite() throws Exception {
+ // Fractional digits beyond the column scale are truncated toward zero by the client
+ // before anything hits the wire; the server just receives the tick count. Pin the
+ // user-visible result of writing an over-scaled value.
+ Schema schema =
+ Schema.of(Schema.Field.of("d", FieldType.logicalType(FixedPrecisionNumeric.of(9, 2))));
+ Row row = Row.withSchema(schema).addValue(new BigDecimal("-1.239")).build();
+
+ executeSql("CREATE TABLE test_decimal_truncation (d Decimal(9, 2)) ENGINE=Log");
+
+ pipeline.apply(Create.of(row).withRowSchema(schema)).apply(write("test_decimal_truncation"));
+ pipeline.run().waitUntilFinish();
+
+ assertEquals("-1.23", executeQueryAsString("SELECT toString(d) FROM test_decimal_truncation"));
+ }
+
+ @Test
+ public void testDecimalWithDefault() throws Exception {
+ Schema schema =
+ Schema.of(
+ Schema.Field.nullable("d", FieldType.logicalType(FixedPrecisionNumeric.of(9, 2))));
+ Row row1 = Row.withSchema(schema).addValue(new BigDecimal("1.25")).build();
+ Row row2 = Row.withSchema(schema).addValue(null).build();
+ Row row3 = Row.withSchema(schema).addValue(new BigDecimal("3.25")).build();
+
+ executeSql("CREATE TABLE test_decimal_with_default (d Decimal(9, 2) DEFAULT 2.25) ENGINE=Log");
+
+ pipeline
+ .apply(Create.of(row1, row2, row3).withRowSchema(schema))
+ .apply(write("test_decimal_with_default"));
+ pipeline.run().waitUntilFinish();
+
+ // The null row takes the DEFAULT value: 1.25 + 2.25 + 3.25 = 6.75.
+ assertEquals(
+ "6.75", executeQueryAsString("SELECT toString(SUM(d)) FROM test_decimal_with_default"));
+ }
+
+ @Test
+ public void testDecimalTableSchema() throws Exception {
+ // DESCRIBE TABLE canonicalizes the width aliases to Decimal(P, S); getTableSchema must
+ // parse whatever the server actually emits.
+ executeSql(
+ "CREATE TABLE test_decimal_schema ("
+ + "d0 Decimal(10, 2),"
+ + "d1 Decimal32(2),"
+ + "d2 Decimal64(4),"
+ + "d3 Decimal128(10),"
+ + "d4 Decimal256(20),"
+ + "d5 Nullable(Decimal(10, 2)),"
+ + "d6 Array(Decimal(38, 10))"
+ + ") ENGINE=Log");
+
+ Properties properties = new Properties();
+ properties.setProperty("user", clickHouse.getUsername());
+ properties.setProperty("password", clickHouse.getPassword());
+
+ TableSchema schema =
+ ClickHouseIO.getTableSchema(clickHouseUrl, database, "test_decimal_schema", properties);
+
+ assertEquals(
+ TableSchema.of(
+ TableSchema.Column.of("d0", ColumnType.decimal(10, 2)),
+ TableSchema.Column.of("d1", ColumnType.decimal(9, 2)),
+ TableSchema.Column.of("d2", ColumnType.decimal(18, 4)),
+ TableSchema.Column.of("d3", ColumnType.decimal(38, 10)),
+ TableSchema.Column.of("d4", ColumnType.decimal(76, 20)),
+ TableSchema.Column.of("d5", ColumnType.decimal(10, 2).withNullable(true)),
+ TableSchema.Column.of("d6", ColumnType.array(ColumnType.decimal(38, 10)))),
+ schema);
+ }
+
@Test
public void testUserAgentInQueryLog() throws Exception {
Schema schema = Schema.of(Schema.Field.of("f0", FieldType.INT64));
diff --git a/sdks/java/io/clickhouse/src/test/java/org/apache/beam/sdk/io/clickhouse/ClickHouseWriterTest.java b/sdks/java/io/clickhouse/src/test/java/org/apache/beam/sdk/io/clickhouse/ClickHouseWriterTest.java
index 89f5c8b7c85f..20b11c3a4624 100644
--- a/sdks/java/io/clickhouse/src/test/java/org/apache/beam/sdk/io/clickhouse/ClickHouseWriterTest.java
+++ b/sdks/java/io/clickhouse/src/test/java/org/apache/beam/sdk/io/clickhouse/ClickHouseWriterTest.java
@@ -17,9 +17,15 @@
*/
package org.apache.beam.sdk.io.clickhouse;
+import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThrows;
+import com.clickhouse.data.ClickHouseOutputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.math.BigDecimal;
+import org.apache.beam.sdk.io.clickhouse.TableSchema.ColumnType;
import org.joda.time.DateTime;
import org.joda.time.DateTimeZone;
import org.junit.Test;
@@ -138,4 +144,114 @@ public void encodeDateTime64RejectsNull() {
"DateTime64 requires a Joda ReadableInstant or java.time.Instant, got null",
e.getMessage());
}
+
+ private static byte[] writtenBytes(ColumnType columnType, Object value) throws IOException {
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ try (ClickHouseOutputStream stream = ClickHouseOutputStream.of(bytes)) {
+ ClickHouseWriter.writeValue(stream, columnType, value);
+ }
+ return bytes.toByteArray();
+ }
+
+ @Test
+ public void writeDecimal32AsLittleEndianUnscaledInt32() throws IOException {
+ // Decimal(9, 2) is stored as Int32; 1.23 has unscaled value 123 at scale 2.
+ assertArrayEquals(
+ new byte[] {123, 0, 0, 0}, writtenBytes(ColumnType.decimal(9, 2), new BigDecimal("1.23")));
+ }
+
+ @Test
+ public void writeDecimal64ScalesValueBelowColumnScale() throws IOException {
+ // Decimal(18, 4) is stored as Int64; 2 becomes 20000 = 0x4E20 ticks.
+ assertArrayEquals(
+ new byte[] {0x20, 0x4E, 0, 0, 0, 0, 0, 0},
+ writtenBytes(ColumnType.decimal(18, 4), new BigDecimal("2")));
+ }
+
+ @Test
+ public void writeDecimal128NegativeOneIsSignExtended() throws IOException {
+ // Decimal(38, 0) is stored as Int128; -1 is sixteen 0xFF bytes in two's complement.
+ byte[] expected = new byte[16];
+ java.util.Arrays.fill(expected, (byte) 0xFF);
+ assertArrayEquals(expected, writtenBytes(ColumnType.decimal(38, 0), new BigDecimal("-1")));
+ }
+
+ @Test
+ public void writeDecimal256OneAtScaleTwenty() throws IOException {
+ // Decimal(76, 20) is stored as Int256; 1 becomes 10^20 = 0x056BC75E2D63100000 ticks,
+ // little-endian in 32 bytes.
+ byte[] expected = new byte[32];
+ byte[] littleEndianTicks = {0x00, 0x00, 0x10, 0x63, 0x2D, 0x5E, (byte) 0xC7, 0x6B, 0x05};
+ System.arraycopy(littleEndianTicks, 0, expected, 0, littleEndianTicks.length);
+ assertArrayEquals(expected, writtenBytes(ColumnType.decimal(76, 20), new BigDecimal("1")));
+ }
+
+ @Test
+ public void writeDecimalTruncatesExcessFractionTowardZero() throws IOException {
+ // -1.239 at scale 2 is -123.9 ticks, truncated toward zero to -123 = 0xFFFFFF85.
+ assertArrayEquals(
+ new byte[] {(byte) 0x85, (byte) 0xFF, (byte) 0xFF, (byte) 0xFF},
+ writtenBytes(ColumnType.decimal(9, 2), new BigDecimal("-1.239")));
+ }
+
+ @Test
+ public void writeDecimalRejectsValueBeyondDeclaredPrecision() throws IOException {
+ // Decimal(5, 0) tops out at 99999; 100000 fits the 32-bit storage width that
+ // BinaryStreamUtils checks, so the declared-precision check is the only thing rejecting it.
+ IllegalArgumentException e =
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> writtenBytes(ColumnType.decimal(5, 0), new BigDecimal("100000")));
+ assertEquals("value 100000 is out of range for Decimal(5, 0)", e.getMessage());
+
+ // The largest in-range value still writes: 99999 = 0x0001869F.
+ assertArrayEquals(
+ new byte[] {(byte) 0x9F, (byte) 0x86, 0x01, 0x00},
+ writtenBytes(ColumnType.decimal(5, 0), new BigDecimal("99999")));
+ }
+
+ @Test
+ public void writeDecimalBoundsScaledValueByDeclaredPrecision() throws IOException {
+ // The bound applies to the unscaled integer at the column scale, so Decimal(5, 2) tops out
+ // at 999.99 (unscaled 99999), not at 99999.
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> writtenBytes(ColumnType.decimal(5, 2), new BigDecimal("1000.00")));
+
+ assertArrayEquals(
+ new byte[] {(byte) 0x9F, (byte) 0x86, 0x01, 0x00},
+ writtenBytes(ColumnType.decimal(5, 2), new BigDecimal("999.99")));
+ }
+
+ @Test
+ public void writeDecimalChecksRangeAfterTruncatingToScale() throws IOException {
+ // 999.999 has six digits, but truncating to scale 2 brings it to 999.99 — within
+ // Decimal(5, 2). The check must run on the truncated value, not the input.
+ assertArrayEquals(
+ new byte[] {(byte) 0x9F, (byte) 0x86, 0x01, 0x00},
+ writtenBytes(ColumnType.decimal(5, 2), new BigDecimal("999.999")));
+ }
+
+ @Test
+ public void writeDecimalRejectsValueBeyondStorageWidth() {
+ // 10^7 at scale 2 is 10^9 ticks, outside Int32's Decimal range of ±(10^9 - 1).
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> writtenBytes(ColumnType.decimal(9, 2), new BigDecimal("10000000.00")));
+ }
+
+ @Test
+ public void writeNullableDecimal() throws IOException {
+ ByteArrayOutputStream nullBytes = new ByteArrayOutputStream();
+ try (ClickHouseOutputStream stream = ClickHouseOutputStream.of(nullBytes)) {
+ ClickHouseWriter.writeNullableValue(stream, ColumnType.decimal(9, 2), null);
+ }
+ assertArrayEquals(new byte[] {1}, nullBytes.toByteArray());
+
+ ByteArrayOutputStream valueBytes = new ByteArrayOutputStream();
+ try (ClickHouseOutputStream stream = ClickHouseOutputStream.of(valueBytes)) {
+ ClickHouseWriter.writeNullableValue(stream, ColumnType.decimal(9, 2), new BigDecimal("1.23"));
+ }
+ assertArrayEquals(new byte[] {0, 123, 0, 0, 0}, valueBytes.toByteArray());
+ }
}
diff --git a/sdks/java/io/clickhouse/src/test/java/org/apache/beam/sdk/io/clickhouse/TableSchemaTest.java b/sdks/java/io/clickhouse/src/test/java/org/apache/beam/sdk/io/clickhouse/TableSchemaTest.java
index 2ce9c27d02b1..48f9f95377b2 100644
--- a/sdks/java/io/clickhouse/src/test/java/org/apache/beam/sdk/io/clickhouse/TableSchemaTest.java
+++ b/sdks/java/io/clickhouse/src/test/java/org/apache/beam/sdk/io/clickhouse/TableSchemaTest.java
@@ -20,10 +20,13 @@
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThrows;
+import java.math.BigDecimal;
import java.util.HashMap;
import java.util.Map;
import org.apache.beam.sdk.io.clickhouse.TableSchema.ColumnType;
import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.schemas.logicaltypes.FixedPrecisionNumeric;
+import org.apache.beam.sdk.values.Row;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
import org.junit.Test;
@@ -94,6 +97,88 @@ public void testParseDateTime64GarbagePrecisionFailsToParse() {
assertEquals("failed to parse", e.getMessage());
}
+ @Test
+ public void testParseDecimal() {
+ assertEquals(ColumnType.decimal(10, 2), ColumnType.parse("Decimal(10, 2)"));
+ }
+
+ @Test
+ public void testParseDecimalPrecisionOnlyDefaultsToScale0() {
+ assertEquals(ColumnType.decimal(5, 0), ColumnType.parse("Decimal(5)"));
+ }
+
+ @Test
+ public void testParseBareDecimalDefaultsToPrecision10Scale0() {
+ assertEquals(ColumnType.decimal(10, 0), ColumnType.parse("Decimal"));
+ }
+
+ @Test
+ public void testParseDecimal32() {
+ assertEquals(ColumnType.decimal(9, 2), ColumnType.parse("Decimal32(2)"));
+ }
+
+ @Test
+ public void testParseDecimal64() {
+ assertEquals(ColumnType.decimal(18, 4), ColumnType.parse("Decimal64(4)"));
+ }
+
+ @Test
+ public void testParseDecimal128() {
+ assertEquals(ColumnType.decimal(38, 20), ColumnType.parse("Decimal128(20)"));
+ }
+
+ @Test
+ public void testParseDecimal256() {
+ assertEquals(ColumnType.decimal(76, 40), ColumnType.parse("Decimal256(40)"));
+ }
+
+ @Test
+ public void testParseNullableDecimal() {
+ assertEquals(
+ ColumnType.decimal(10, 2).withNullable(true), ColumnType.parse("Nullable(Decimal(10, 2))"));
+ }
+
+ @Test
+ public void testParseArrayOfDecimal() {
+ assertEquals(
+ ColumnType.array(ColumnType.decimal(38, 10)), ColumnType.parse("Array(Decimal(38, 10))"));
+ }
+
+ @Test
+ public void testParseDecimalPrecisionAboveMaxFailsToParse() {
+ IllegalArgumentException e =
+ assertThrows(IllegalArgumentException.class, () -> ColumnType.parse("Decimal(77, 2)"));
+ assertEquals("failed to parse", e.getMessage());
+ }
+
+ @Test
+ public void testParseDecimalZeroPrecisionFailsToParse() {
+ IllegalArgumentException e =
+ assertThrows(IllegalArgumentException.class, () -> ColumnType.parse("Decimal(0)"));
+ assertEquals("failed to parse", e.getMessage());
+ }
+
+ @Test
+ public void testParseDecimalScaleAbovePrecisionFailsToParse() {
+ IllegalArgumentException e =
+ assertThrows(IllegalArgumentException.class, () -> ColumnType.parse("Decimal(9, 10)"));
+ assertEquals("failed to parse", e.getMessage());
+ }
+
+ @Test
+ public void testParseDecimalNegativeScaleFailsToParse() {
+ IllegalArgumentException e =
+ assertThrows(IllegalArgumentException.class, () -> ColumnType.parse("Decimal(9, -1)"));
+ assertEquals("failed to parse", e.getMessage());
+ }
+
+ @Test
+ public void testParseDecimalGarbagePrecisionFailsToParse() {
+ IllegalArgumentException e =
+ assertThrows(IllegalArgumentException.class, () -> ColumnType.parse("Decimal(abc)"));
+ assertEquals("failed to parse", e.getMessage());
+ }
+
@Test
public void testParseFloat32() {
assertEquals(ColumnType.FLOAT32, ColumnType.parse("Float32"));
@@ -238,6 +323,16 @@ public void testParseDefaultExpressionInt64() {
assertEquals(-1L, ColumnType.parseDefaultExpression(ColumnType.INT64, "-1"));
}
+ @Test
+ public void testParseDefaultExpressionDecimal() {
+ assertEquals(
+ new BigDecimal("1.23"),
+ ColumnType.parseDefaultExpression(ColumnType.decimal(9, 2), "1.23"));
+ assertEquals(
+ new BigDecimal("-1.23"),
+ ColumnType.parseDefaultExpression(ColumnType.decimal(9, 2), "-1.23"));
+ }
+
@Test
public void testEquivalentSchema() {
TableSchema tableSchema =
@@ -300,6 +395,63 @@ public void testDateTime64RejectsPrecisionAboveNine() {
ColumnType.dateTime64(10);
}
+ @Test
+ public void testEquivalentSchemaDecimal() {
+ TableSchema tableSchema = TableSchema.of(TableSchema.Column.of("d", ColumnType.decimal(10, 2)));
+ Schema expected =
+ Schema.of(
+ Schema.Field.of("d", Schema.FieldType.logicalType(FixedPrecisionNumeric.of(10, 2))));
+ assertEquals(expected, TableSchema.getEquivalentSchema(tableSchema));
+ }
+
+ @Test
+ public void testEquivalentSchemaNullableDecimal() {
+ TableSchema tableSchema =
+ TableSchema.of(TableSchema.Column.of("d", ColumnType.decimal(38, 10).withNullable(true)));
+ Schema expected =
+ Schema.of(
+ Schema.Field.nullable(
+ "d", Schema.FieldType.logicalType(FixedPrecisionNumeric.of(38, 10))));
+ assertEquals(expected, TableSchema.getEquivalentSchema(tableSchema));
+ }
+
+ @Test
+ public void testMappedDecimalRejectsPrecisionOverflowAtRowConstruction() {
+ // The mapped FixedPrecisionNumeric type rejects most over-precision values at Row
+ // construction, well before the writer's own range check: building a Row with more digits
+ // than the column declares must fail loudly.
+ Schema schema =
+ TableSchema.getEquivalentSchema(
+ TableSchema.of(TableSchema.Column.of("d", ColumnType.decimal(5, 0))));
+
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> Row.withSchema(schema).addValue(new BigDecimal("999999")).build());
+
+ Row row = Row.withSchema(schema).addValue(new BigDecimal("99999")).build();
+ assertEquals(new BigDecimal("99999"), row.getValue("d"));
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testDecimalRejectsZeroPrecision() {
+ ColumnType.decimal(0, 0);
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testDecimalRejectsPrecisionAboveSeventySix() {
+ ColumnType.decimal(77, 0);
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testDecimalRejectsNegativeScale() {
+ ColumnType.decimal(10, -1);
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testDecimalRejectsScaleAbovePrecision() {
+ ColumnType.decimal(10, 11);
+ }
+
@Test
public void testParseTupleSingle() {
Map m1 = new HashMap<>();