This is an automated email from the ASF dual-hosted git repository.
stankiewicz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 4748eb678a9 ClickHouseIO: Add Decimal(P, S) support for fixed-point
columns (#39846)
4748eb678a9 is described below
commit 4748eb678a95e1594c9be10e60a89e8785783da9
Author: Tom Newton <[email protected]>
AuthorDate: Mon Aug 31 03:41:03 2026 -0500
ClickHouseIO: Add Decimal(P, S) support for fixed-point columns (#39846)
* ClickHouseIO: Add Decimal(P, S) support for fixed-point columns
* 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.
---
CHANGES.md | 1 +
.../beam/sdk/io/clickhouse/ClickHouseIO.java | 7 +
.../beam/sdk/io/clickhouse/ClickHouseWriter.java | 47 ++++++
.../apache/beam/sdk/io/clickhouse/TableSchema.java | 61 ++++++-
.../clickhouse/src/main/javacc/ColumnTypeParser.jj | 43 +++++
.../beam/sdk/io/clickhouse/ClickHouseIOIT.java | 182 +++++++++++++++++++++
.../sdk/io/clickhouse/ClickHouseWriterTest.java | 116 +++++++++++++
.../beam/sdk/io/clickhouse/TableSchemaTest.java | 152 +++++++++++++++++
8 files changed, 608 insertions(+), 1 deletion(-)
diff --git a/CHANGES.md b/CHANGES.md
index c3c4019c7da..045b95c4e76 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -67,6 +67,7 @@
* Support for X source added (Java/Python)
([#X](https://github.com/apache/beam/issues/X)).
* Added `schema_update_options` to `WriteToBigQuery` file loads, allowing
BigQuery load jobs to add nullable fields or relax required fields when
appending data (Python) ([#21141](https://github.com/apache/beam/issues/21141)).
* 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 6798e2f2bd7..b18c573cf51 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.metrics.Metrics;
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 @@ import org.slf4j.LoggerFactory;
* <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 ≤ 3), {@link SqlTypes#TIMESTAMP}
(4–6), or {@link NanosInstant} (≥ 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>
@@ -147,6 +149,11 @@ import org.slf4j.LoggerFactory;
* <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
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 4d9f072e598..bf2f53944a0 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.ClickHouseOutputStream;
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 @@ public class ClickHouseWriter {
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());
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 1b9fdffd4c8..65b4910773d 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 @@ package org.apache.beam.sdk.io.clickhouse;
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.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 abstract class TableSchema implements Serializable {
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 abstract class TableSchema implements Serializable {
DATE,
DATETIME,
DATETIME64,
+ DECIMAL,
ENUM8,
ENUM16,
FIXEDSTRING,
@@ -260,9 +272,18 @@ public abstract class TableSchema 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 abstract class TableSchema implements Serializable {
.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 abstract class TableSchema implements Serializable {
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 @@ public abstract class TableSchema implements Serializable {
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 53ad991b67c..c3e08f8acba 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;
+}
+{
+ (
+ <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;
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 da435f20684..7a5a39ae83d 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 static org.junit.Assert.assertTrue;
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 class ClickHouseIOIT extends BaseClickHouseTest {
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 89f5c8b7c85..20b11c3a462 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 class ClickHouseWriterTest {
"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 2ce9c27d02b..48f9f95377b 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 @@ package org.apache.beam.sdk.io.clickhouse;
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 class TableSchemaTest {
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 class TableSchemaTest {
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 class TableSchemaTest {
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<String, ColumnType> m1 = new HashMap<>();
