From 8a9becefe6b7a082b0d9df8849b6fd6d76b9eb07 Mon Sep 17 00:00:00 2001 From: Tom Newton Date: Fri, 21 Aug 2026 08:35:49 -0500 Subject: [PATCH 1/2] ClickHouseIO: Add Decimal(P, S) support for fixed-point columns --- CHANGES.md | 1 + .../beam/sdk/io/clickhouse/ClickHouseIO.java | 7 + .../sdk/io/clickhouse/ClickHouseWriter.java | 13 ++ .../beam/sdk/io/clickhouse/TableSchema.java | 61 +++++- .../src/main/javacc/ColumnTypeParser.jj | 43 ++++ .../sdk/io/clickhouse/ClickHouseIOIT.java | 203 ++++++++++++++++++ .../io/clickhouse/ClickHouseWriterTest.java | 78 +++++++ .../sdk/io/clickhouse/TableSchemaTest.java | 152 +++++++++++++ 8 files changed, 557 insertions(+), 1 deletion(-) diff --git a/CHANGES.md b/CHANGES.md index 3b349b3bc5d2..a66ed51f5979 100644 --- a/CHANGES.md +++ b/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 diff --git a/sdks/java/io/clickhouse/src/main/java/org/apache/beam/sdk/io/clickhouse/ClickHouseIO.java b/sdks/java/io/clickhouse/src/main/java/org/apache/beam/sdk/io/clickhouse/ClickHouseIO.java index 6798e2f2bd75..24c7ef3aeee9 100644 --- a/sdks/java/io/clickhouse/src/main/java/org/apache/beam/sdk/io/clickhouse/ClickHouseIO.java +++ b/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,6 +141,7 @@ * {@link TableSchema.TypeName#DATE} {@link Schema.TypeName#DATETIME} * {@link TableSchema.TypeName#DATETIME} {@link Schema.TypeName#DATETIME} * {@link TableSchema.TypeName#DATETIME64} {@link Schema.TypeName#DATETIME} (precision ≤ 3), {@link SqlTypes#TIMESTAMP} (4–6), or {@link NanosInstant} (≥ 7) + * {@link TableSchema.TypeName#DECIMAL} {@link FixedPrecisionNumeric} * {@link TableSchema.TypeName#ARRAY} {@link Schema.TypeName#ARRAY} * {@link TableSchema.TypeName#ENUM8} {@link Schema.TypeName#STRING} * {@link TableSchema.TypeName#ENUM16} {@link Schema.TypeName#STRING} @@ -147,6 +149,11 @@ * {@link TableSchema.TypeName#TUPLE} {@link Schema.TypeName#ROW} * * + *

{@code Decimal(P, S)} columns accept {@link java.math.BigDecimal} values. Fractional digits + * beyond the column scale are truncated toward zero, and values are range-checked against the + * storage width chosen from the declared precision (32/64/128/256 bits) rather than the precision + * itself. + * *

Nullable row columns are supported through Nullable type in * ClickHouse. 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..ddd66b3e7caf 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,7 @@ import com.clickhouse.data.ClickHousePipedOutputStream; import com.clickhouse.data.format.BinaryStreamUtils; import java.io.IOException; +import java.math.BigDecimal; import java.nio.charset.StandardCharsets; import java.util.Collection; import java.util.List; @@ -178,6 +179,18 @@ 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"); + // Picks the 32/64/128/256-bit little-endian width from the declared precision and writes + // the value scaled to 10^-scale ticks. Fractional digits beyond the column scale are + // truncated toward zero; the range check is against the storage width, not the declared + // precision. + BinaryStreamUtils.writeDecimal(stream, (BigDecimal) value, 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..671cf295ffd0 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,206 @@ 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 testDecimalBeyondDeclaredPrecisionWithinStorageWidth() throws Exception { + // ClickHouse checks the declared precision only on conversion from a string; a binary + // insert that fits the storage width is stored and read back verbatim. The mapped + // FixedPrecisionNumeric type rejects such values at Row construction, so reaching the + // wire requires a plain DECIMAL field. Pin the lenient server behavior so a future + // server or client change surfaces here instead of silently shifting semantics. + Schema schema = Schema.of(Schema.Field.of("d", FieldType.DECIMAL)); + Row row = Row.withSchema(schema).addValue(new BigDecimal("999999")).build(); + + executeSql("CREATE TABLE test_decimal_beyond_precision (d Decimal(5, 0)) ENGINE=Log"); + + pipeline + .apply(Create.of(row).withRowSchema(schema)) + .apply(write("test_decimal_beyond_precision")); + pipeline.run().waitUntilFinish(); + + assertEquals( + "999999", executeQueryAsString("SELECT toString(d) FROM test_decimal_beyond_precision")); + } + + @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..8cc3158dae7b 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,76 @@ 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 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..7a80a3850b0d 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 writer only range-checks against the storage width, so the mapped + // FixedPrecisionNumeric type is what guards the declared precision: building a Row with + // more digits than the column declares must fail loudly before anything hits the wire. + 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<>(); From cfb58d6218bceb1d5bcd1e72c173363c9e5bb7d6 Mon Sep 17 00:00:00 2001 From: Tom Newton Date: Fri, 21 Aug 2026 17:17:33 -0500 Subject: [PATCH 2/2] ClickHouseIO: enforce declared Decimal precision when writing BinaryStreamUtils.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. A value like 100000 was therefore written and stored in a Decimal(5, 0) column whose declared maximum is 99999. Truncate to the column scale toward zero, as before, then bound the resulting unscaled integer by the declared precision and throw IllegalArgumentException when it does not fit. The bound applies to the unscaled integer at the column scale, so Decimal(5, 2) admits values up to 999.99, and the check runs after truncation so 999.999 remains valid. Removes the integration test that pinned the previous lenient behavior; the rejection is now covered by unit tests. --- .../beam/sdk/io/clickhouse/ClickHouseIO.java | 6 +-- .../sdk/io/clickhouse/ClickHouseWriter.java | 44 ++++++++++++++++--- .../sdk/io/clickhouse/ClickHouseIOIT.java | 21 --------- .../io/clickhouse/ClickHouseWriterTest.java | 38 ++++++++++++++++ .../sdk/io/clickhouse/TableSchemaTest.java | 6 +-- 5 files changed, 83 insertions(+), 32 deletions(-) diff --git a/sdks/java/io/clickhouse/src/main/java/org/apache/beam/sdk/io/clickhouse/ClickHouseIO.java b/sdks/java/io/clickhouse/src/main/java/org/apache/beam/sdk/io/clickhouse/ClickHouseIO.java index 24c7ef3aeee9..b18c573cf514 100644 --- a/sdks/java/io/clickhouse/src/main/java/org/apache/beam/sdk/io/clickhouse/ClickHouseIO.java +++ b/sdks/java/io/clickhouse/src/main/java/org/apache/beam/sdk/io/clickhouse/ClickHouseIO.java @@ -150,9 +150,9 @@ * * *

{@code Decimal(P, S)} columns accept {@link java.math.BigDecimal} values. Fractional digits - * beyond the column scale are truncated toward zero, and values are range-checked against the - * storage width chosen from the declared precision (32/64/128/256 bits) rather than the precision - * itself. + * 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. * *

Nullable row columns are supported through Nullable type in 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 ddd66b3e7caf..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 @@ -22,6 +22,8 @@ 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; @@ -45,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. @@ -184,11 +217,12 @@ static void writeValue(ClickHouseOutputStream stream, ColumnType columnType, Obj Preconditions.checkNotNull(columnType.precision(), "Decimal column missing precision"); int decimalScale = Preconditions.checkNotNull(columnType.scale(), "Decimal column missing scale"); - // Picks the 32/64/128/256-bit little-endian width from the declared precision and writes - // the value scaled to 10^-scale ticks. Fractional digits beyond the column scale are - // truncated toward zero; the range check is against the storage width, not the declared - // precision. - BinaryStreamUtils.writeDecimal(stream, (BigDecimal) value, decimalPrecision, decimalScale); + // 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: 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 671cf295ffd0..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 @@ -726,27 +726,6 @@ public void testDecimalTruncatesExcessFractionOnWrite() throws Exception { assertEquals("-1.23", executeQueryAsString("SELECT toString(d) FROM test_decimal_truncation")); } - @Test - public void testDecimalBeyondDeclaredPrecisionWithinStorageWidth() throws Exception { - // ClickHouse checks the declared precision only on conversion from a string; a binary - // insert that fits the storage width is stored and read back verbatim. The mapped - // FixedPrecisionNumeric type rejects such values at Row construction, so reaching the - // wire requires a plain DECIMAL field. Pin the lenient server behavior so a future - // server or client change surfaces here instead of silently shifting semantics. - Schema schema = Schema.of(Schema.Field.of("d", FieldType.DECIMAL)); - Row row = Row.withSchema(schema).addValue(new BigDecimal("999999")).build(); - - executeSql("CREATE TABLE test_decimal_beyond_precision (d Decimal(5, 0)) ENGINE=Log"); - - pipeline - .apply(Create.of(row).withRowSchema(schema)) - .apply(write("test_decimal_beyond_precision")); - pipeline.run().waitUntilFinish(); - - assertEquals( - "999999", executeQueryAsString("SELECT toString(d) FROM test_decimal_beyond_precision")); - } - @Test public void testDecimalWithDefault() throws Exception { Schema schema = 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 8cc3158dae7b..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 @@ -194,6 +194,44 @@ public void writeDecimalTruncatesExcessFractionTowardZero() throws IOException { 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). 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 7a80a3850b0d..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 @@ -417,9 +417,9 @@ public void testEquivalentSchemaNullableDecimal() { @Test public void testMappedDecimalRejectsPrecisionOverflowAtRowConstruction() { - // The writer only range-checks against the storage width, so the mapped - // FixedPrecisionNumeric type is what guards the declared precision: building a Row with - // more digits than the column declares must fail loudly before anything hits the wire. + // 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))));