Skip to content

Commit 378e10d

Browse files
EliaaazzzEliaaazzz
authored
ClickHouseIO: Add DateTime64 support for sub-second timestamp precision (#38510)
ClickHouseIO previously only recognized second-precision DateTime, so pipelines emitting sub-second timestamps could not write to DateTime64 columns. This adds TypeName.DATETIME64 with a validated precision (0-9) and ColumnType.dateTime64(precision), plus a no-arg factory that defaults to precision 3. The parser accepts DateTime64[(precision[, 'timezone'])], including a bare DateTime64 and the type nested inside Nullable(...) and Array(...); the timezone argument is accepted but not stored. Beam field-type mapping is Joda DATETIME for precision up to 3, SqlTypes.TIMESTAMP for 4-6, and NanosInstant for 7 and above. The writer encodes the value as a little-endian int64 of epoch_seconds * 10^precision + sub_second_units, using floor division so negative timestamps match ClickHouse. Tests cover the parser, schema mapping, the encoder (Joda and java.time inputs, negative and overflow cases, null input), and ClickHouse testcontainer round-trips for precisions 3, 6 and 9 plus nullable cases. Closes #38466
1 parent 6ed0635 commit 378e10d

8 files changed

Lines changed: 485 additions & 1 deletion

File tree

‎CHANGES.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,7 @@
6767

6868
* Support for reading from Delta Lake added (Java) ([#38551](https://github.com/apache/beam/issues/38551)).
6969
* Support for X source added (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
70+
* ClickHouseIO: support writing `DateTime64(precision[, 'timezone'])` columns with sub-second precision (Java) ([#38466](https://github.com/apache/beam/issues/38466)).
7071

7172
## New Features / Improvements
7273

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

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,8 @@
3838
import org.apache.beam.sdk.schemas.FieldAccessDescriptor;
3939
import org.apache.beam.sdk.schemas.Schema;
4040
import org.apache.beam.sdk.schemas.logicaltypes.FixedBytes;
41+
import org.apache.beam.sdk.schemas.logicaltypes.NanosInstant;
42+
import org.apache.beam.sdk.schemas.logicaltypes.SqlTypes;
4143
import org.apache.beam.sdk.schemas.transforms.Select;
4244
import org.apache.beam.sdk.transforms.DoFn;
4345
import org.apache.beam.sdk.transforms.PTransform;
@@ -137,6 +139,7 @@
137139
* <tr><td>{@link TableSchema.TypeName#UINT64}</td> <td>{@link Schema.TypeName#INT64}</td></tr>
138140
* <tr><td>{@link TableSchema.TypeName#DATE}</td> <td>{@link Schema.TypeName#DATETIME}</td></tr>
139141
* <tr><td>{@link TableSchema.TypeName#DATETIME}</td> <td>{@link Schema.TypeName#DATETIME}</td></tr>
142+
* <tr><td>{@link TableSchema.TypeName#DATETIME64}</td> <td>{@link Schema.TypeName#DATETIME} (precision &le; 3), {@link SqlTypes#TIMESTAMP} (4&ndash;6), or {@link NanosInstant} (&ge; 7)</td></tr>
140143
* <tr><td>{@link TableSchema.TypeName#ARRAY}</td> <td>{@link Schema.TypeName#ARRAY}</td></tr>
141144
* <tr><td>{@link TableSchema.TypeName#ENUM8}</td> <td>{@link Schema.TypeName#STRING}</td></tr>
142145
* <tr><td>{@link TableSchema.TypeName#ENUM16}</td> <td>{@link Schema.TypeName#STRING}</td></tr>

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

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,39 @@
3939
public class ClickHouseWriter {
4040
private static final Instant EPOCH_INSTANT = new Instant(0L);
4141

42+
// 10^0 through 10^9 inclusive — precision is validated in [0, 9] by ColumnType.dateTime64.
43+
private static final long[] POW10 = {
44+
1L, 10L, 100L, 1_000L, 10_000L, 100_000L, 1_000_000L, 10_000_000L, 100_000_000L, 1_000_000_000L
45+
};
46+
47+
/**
48+
* Encodes a timestamp into ClickHouse's {@code DateTime64(precision)} representation: a signed
49+
* 64-bit integer counting ticks of size 10<sup>-precision</sup> seconds since the Unix epoch.
50+
*
51+
* <p>Accepts either a Joda {@link ReadableInstant} (millisecond precision) or a {@link
52+
* java.time.Instant} (nanosecond precision). Sub-tick fractions are truncated toward negative
53+
* infinity, matching ClickHouse's own encoding for negative timestamps.
54+
*/
55+
static long encodeDateTime64(Object value, int precision) {
56+
long epochSecond;
57+
int nanoOfSecond;
58+
if (value instanceof java.time.Instant) {
59+
java.time.Instant inst = (java.time.Instant) value;
60+
epochSecond = inst.getEpochSecond();
61+
nanoOfSecond = inst.getNano();
62+
} else if (value instanceof ReadableInstant) {
63+
long millis = ((ReadableInstant) value).getMillis();
64+
epochSecond = Math.floorDiv(millis, 1000L);
65+
nanoOfSecond = (int) Math.floorMod(millis, 1000L) * 1_000_000;
66+
} else {
67+
throw new IllegalArgumentException(
68+
"DateTime64 requires a Joda ReadableInstant or java.time.Instant, got "
69+
+ (value == null ? "null" : value.getClass().getName()));
70+
}
71+
long subSecondTicks = nanoOfSecond / POW10[9 - precision];
72+
return Math.addExact(Math.multiplyExact(epochSecond, POW10[precision]), subSecondTicks);
73+
}
74+
4275
@SuppressWarnings("unchecked")
4376
static void writeNullableValue(ClickHouseOutputStream stream, ColumnType columnType, Object value)
4477
throws IOException {
@@ -138,6 +171,13 @@ static void writeValue(ClickHouseOutputStream stream, ColumnType columnType, Obj
138171
BinaryStreamUtils.writeUnsignedInt32(stream, epochSeconds);
139172
break;
140173

174+
case DATETIME64:
175+
int precision =
176+
Preconditions.checkNotNull(
177+
columnType.precision(), "DateTime64 column is missing precision");
178+
BinaryStreamUtils.writeInt64(stream, encodeDateTime64(value, precision));
179+
break;
180+
141181
case ARRAY:
142182
List<Object> values = (List<Object>) value;
143183
BinaryStreamUtils.writeVarInt(stream, values.size());

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

Lines changed: 52 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,8 @@
2727
import java.util.stream.Collectors;
2828
import org.apache.beam.sdk.schemas.Schema;
2929
import org.apache.beam.sdk.schemas.logicaltypes.FixedBytes;
30+
import org.apache.beam.sdk.schemas.logicaltypes.NanosInstant;
31+
import org.apache.beam.sdk.schemas.logicaltypes.SqlTypes;
3032
import org.checkerframework.checker.nullness.qual.Nullable;
3133

3234
/**
@@ -39,6 +41,9 @@
3941
})
4042
public abstract class TableSchema implements Serializable {
4143

44+
private static final Schema.FieldType NANOS_INSTANT_TYPE =
45+
Schema.FieldType.logicalType(new NanosInstant());
46+
4247
public abstract List<Column> columns();
4348

4449
public static TableSchema of(Column... columns) {
@@ -76,6 +81,22 @@ public static Schema.FieldType getEquivalentFieldType(ColumnType columnType) {
7681
case DATETIME:
7782
return Schema.FieldType.DATETIME;
7883

84+
case DATETIME64:
85+
// Pick the narrowest Beam logical type that still round-trips the requested precision:
86+
// ≤ 3 (milliseconds) → Joda DATETIME, keeping existing pipelines unchanged.
87+
// 4–6 (down to microseconds) → SqlTypes.TIMESTAMP (MicrosInstant) — interoperable
88+
// with BigQueryIO, Avro and Beam SQL.
89+
// ≥ 7 (sub-microsecond) → NanosInstant, the only built-in type that preserves
90+
// full nanosecond precision through Row construction.
91+
int p = columnType.precision();
92+
if (p <= 3) {
93+
return Schema.FieldType.DATETIME;
94+
} else if (p <= 6) {
95+
return Schema.FieldType.logicalType(SqlTypes.TIMESTAMP);
96+
} else {
97+
return NANOS_INSTANT_TYPE;
98+
}
99+
79100
case STRING:
80101
return Schema.FieldType.STRING;
81102

@@ -163,6 +184,7 @@ public enum TypeName {
163184
// Primitive types
164185
DATE,
165186
DATETIME,
187+
DATETIME64,
166188
ENUM8,
167189
ENUM16,
168190
FIXEDSTRING,
@@ -238,6 +260,9 @@ public abstract static class ColumnType implements Serializable {
238260

239261
public abstract @Nullable Map<String, ColumnType> tupleTypes();
240262

263+
/** Sub-second precision (0–9) of {@code DateTime64}. {@code null} for other types. */
264+
public abstract @Nullable Integer precision();
265+
241266
public ColumnType withNullable(boolean nullable) {
242267
return toBuilder().nullable(nullable).build();
243268
}
@@ -258,6 +283,26 @@ public static ColumnType fixedString(int size) {
258283
.build();
259284
}
260285

286+
/** Default {@code DateTime64} precision in ClickHouse. */
287+
public static final int DEFAULT_DATETIME64_PRECISION = 3;
288+
289+
/** Returns a {@code DateTime64} type with ClickHouse's default precision of 3. */
290+
public static ColumnType dateTime64() {
291+
return dateTime64(DEFAULT_DATETIME64_PRECISION);
292+
}
293+
294+
public static ColumnType dateTime64(int precision) {
295+
if (precision < 0 || precision > 9) {
296+
throw new IllegalArgumentException(
297+
"DateTime64 precision must be in [0, 9], got " + precision);
298+
}
299+
return ColumnType.builder()
300+
.typeName(TypeName.DATETIME64)
301+
.nullable(false)
302+
.precision(precision)
303+
.build();
304+
}
305+
261306
public static ColumnType enum8(Map<String, Integer> enumValues) {
262307
return ColumnType.builder()
263308
.typeName(TypeName.ENUM8)
@@ -296,13 +341,17 @@ public static ColumnType tuple(Map<String, ColumnType> elements) {
296341
*
297342
* @param str string representation of ClickHouse type
298343
* @return type of ClickHouse column
344+
* @throws IllegalArgumentException if {@code str} is not a valid ClickHouse column type
299345
*/
300346
public static ColumnType parse(String str) {
301347
try {
302348
return new org.apache.beam.sdk.io.clickhouse.impl.parser.ColumnTypeParser(
303349
new StringReader(str))
304350
.parse();
305-
} catch (org.apache.beam.sdk.io.clickhouse.impl.parser.ParseException e) {
351+
} catch (org.apache.beam.sdk.io.clickhouse.impl.parser.ParseException
352+
| org.apache.beam.sdk.io.clickhouse.impl.parser.TokenMgrError
353+
| IllegalArgumentException e) {
354+
// Funnel lexical, syntactic and validation failures into one error surface.
306355
throw new IllegalArgumentException("failed to parse", e);
307356
}
308357
}
@@ -367,6 +416,8 @@ abstract static class Builder {
367416

368417
public abstract Builder tupleTypes(Map<String, ColumnType> tupleElements);
369418

419+
public abstract Builder precision(@Nullable Integer precision);
420+
370421
public abstract ColumnType build();
371422
}
372423
}

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

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -79,6 +79,7 @@ TOKEN :
7979
{
8080
< ARRAY : "ARRAY" >
8181
| < DATE : "DATE" >
82+
| < DATETIME64 : "DATETIME64" >
8283
| < DATETIME : "DATETIME" >
8384
| < ENUM8 : "ENUM8" >
8485
| < ENUM16 : "ENUM16" >
@@ -206,6 +207,7 @@ private ColumnType primitive() :
206207
{
207208
TypeName type;
208209
String size;
210+
ColumnType ct;
209211
}
210212
{
211213
(
@@ -214,10 +216,35 @@ private ColumnType primitive() :
214216
(<FIXEDSTRING> <LPAREN> ( size = integer() ) <RPAREN>) {
215217
return ColumnType.fixedString(Integer.valueOf(size));
216218
}
219+
|
220+
(ct = dateTime64()) { return ct; }
217221
)
218222

219223
}
220224

225+
private ColumnType dateTime64() :
226+
{
227+
String precision = null;
228+
}
229+
{
230+
(
231+
<DATETIME64>
232+
(
233+
<LPAREN> ( precision = integer() )
234+
// The timezone is display-only metadata; accept the syntax and ignore the value.
235+
( <COMMA> string() )?
236+
<RPAREN>
237+
)?
238+
)
239+
{
240+
// Bare DateTime64 defaults to precision 3, matching ClickHouse.
241+
if (precision == null) {
242+
return ColumnType.dateTime64();
243+
}
244+
return ColumnType.dateTime64(Integer.parseInt(precision));
245+
}
246+
}
247+
221248
private ColumnType nullable() :
222249
{
223250
ColumnType ct;

‎sdks/java/io/clickhouse/src/test/java/org/apache/beam/sdk/io/clickhouse/ClickHouseIOIT.java‎

Lines changed: 119 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,8 @@
3232
import org.apache.beam.sdk.schemas.Schema.FieldType;
3333
import org.apache.beam.sdk.schemas.annotations.DefaultSchema;
3434
import org.apache.beam.sdk.schemas.logicaltypes.FixedBytes;
35+
import org.apache.beam.sdk.schemas.logicaltypes.NanosInstant;
36+
import org.apache.beam.sdk.schemas.logicaltypes.SqlTypes;
3537
import org.apache.beam.sdk.testing.TestPipeline;
3638
import org.apache.beam.sdk.transforms.Create;
3739
import org.apache.beam.sdk.util.ReleaseInfo;
@@ -49,6 +51,18 @@
4951
@RunWith(JUnit4.class)
5052
public class ClickHouseIOIT extends BaseClickHouseTest {
5153

54+
private static final long MICROS_PER_SECOND = 1_000_000L;
55+
private static final long NANOS_PER_SECOND = 1_000_000_000L;
56+
57+
// Shared DateTime64 test instant 2026-05-15T12:34:56Z; its .789012345 sub-second component
58+
// exercises every precision bucket.
59+
private static final long TEST_EPOCH_SECONDS = 1_778_848_496L;
60+
// Nano-of-second; the trailing 345 is not micro-aligned.
61+
private static final long TEST_NANOS_OF_SECOND = 789_012_345L;
62+
// The same sub-second component truncated to whole microseconds.
63+
private static final long TEST_MICROS_OF_SECOND = 789_012L;
64+
private static final long TEST_MICRO_ALIGNED_NANOS_OF_SECOND = TEST_MICROS_OF_SECOND * 1_000L;
65+
5266
@Rule public TestPipeline pipeline = TestPipeline.create();
5367

5468
@Test
@@ -480,6 +494,111 @@ public void testPojo() throws Exception {
480494
assertEquals(12L, sum1);
481495
}
482496

497+
@Test
498+
public void testDateTime64Millis() throws Exception {
499+
Schema schema = Schema.of(Schema.Field.of("ts", FieldType.DATETIME));
500+
DateTime ts = new DateTime(2026, 5, 15, 12, 34, 56, 789, DateTimeZone.UTC);
501+
Row row = Row.withSchema(schema).addValue(ts).build();
502+
503+
executeSql("CREATE TABLE test_datetime64_ms (ts DateTime64(3, 'UTC')) ENGINE=Log");
504+
505+
pipeline.apply(Create.of(row).withRowSchema(schema)).apply(write("test_datetime64_ms"));
506+
pipeline.run().waitUntilFinish();
507+
508+
// toUnixTimestamp64Milli returns the underlying tick count, which is the most stable thing to
509+
// assert across CH versions (string formatting may include trailing zeros depending on
510+
// version).
511+
long ticks = executeQueryAsLong("SELECT toUnixTimestamp64Milli(ts) FROM test_datetime64_ms");
512+
assertEquals(ts.getMillis(), ticks);
513+
}
514+
515+
@Test
516+
public void testDateTime64Micros() throws Exception {
517+
Schema schema = Schema.of(Schema.Field.of("ts", FieldType.logicalType(SqlTypes.TIMESTAMP)));
518+
// Micro-aligned nanos, so MicrosInstant accepts the value.
519+
java.time.Instant ts =
520+
java.time.Instant.ofEpochSecond(TEST_EPOCH_SECONDS, TEST_MICRO_ALIGNED_NANOS_OF_SECOND);
521+
Row row = Row.withSchema(schema).addValue(ts).build();
522+
523+
executeSql("CREATE TABLE test_datetime64_us (ts DateTime64(6)) ENGINE=Log");
524+
525+
pipeline.apply(Create.of(row).withRowSchema(schema)).apply(write("test_datetime64_us"));
526+
pipeline.run().waitUntilFinish();
527+
528+
long ticks = executeQueryAsLong("SELECT toUnixTimestamp64Micro(ts) FROM test_datetime64_us");
529+
assertEquals(TEST_EPOCH_SECONDS * MICROS_PER_SECOND + TEST_MICROS_OF_SECOND, ticks);
530+
}
531+
532+
@Test
533+
public void testDateTime64Nanos() throws Exception {
534+
// DateTime64(9) must preserve full nanosecond precision. Use NanosInstant directly
535+
// because SqlTypes.TIMESTAMP (MicrosInstant) rejects non-micro-aligned nanos like the
536+
// trailing 345.
537+
Schema schema = Schema.of(Schema.Field.of("ts", FieldType.logicalType(new NanosInstant())));
538+
java.time.Instant ts =
539+
java.time.Instant.ofEpochSecond(TEST_EPOCH_SECONDS, TEST_NANOS_OF_SECOND);
540+
Row row = Row.withSchema(schema).addValue(ts).build();
541+
542+
executeSql("CREATE TABLE test_datetime64_ns (ts DateTime64(9)) ENGINE=Log");
543+
544+
pipeline.apply(Create.of(row).withRowSchema(schema)).apply(write("test_datetime64_ns"));
545+
pipeline.run().waitUntilFinish();
546+
547+
long ticks = executeQueryAsLong("SELECT toUnixTimestamp64Nano(ts) FROM test_datetime64_ns");
548+
assertEquals(TEST_EPOCH_SECONDS * NANOS_PER_SECOND + TEST_NANOS_OF_SECOND, ticks);
549+
}
550+
551+
@Test
552+
public void testNullableDateTime64() throws Exception {
553+
Schema schema =
554+
Schema.of(Schema.Field.nullable("ts", FieldType.logicalType(SqlTypes.TIMESTAMP)));
555+
java.time.Instant ts =
556+
java.time.Instant.ofEpochSecond(TEST_EPOCH_SECONDS, TEST_MICRO_ALIGNED_NANOS_OF_SECOND);
557+
Row row1 = Row.withSchema(schema).addValue(ts).build();
558+
Row row2 = Row.withSchema(schema).addValue(null).build();
559+
560+
executeSql("CREATE TABLE test_nullable_datetime64 (ts Nullable(DateTime64(6))) ENGINE=Log");
561+
562+
pipeline
563+
.apply(Create.of(row1, row2).withRowSchema(schema))
564+
.apply(write("test_nullable_datetime64"));
565+
pipeline.run().waitUntilFinish();
566+
567+
long total = executeQueryAsLong("SELECT COUNT(*) FROM test_nullable_datetime64");
568+
long nonNull = executeQueryAsLong("SELECT COUNT(ts) FROM test_nullable_datetime64");
569+
assertEquals(2L, total);
570+
assertEquals(1L, nonNull);
571+
}
572+
573+
@Test
574+
public void testNullableDateTime64Nanos() throws Exception {
575+
// Nullable columns take the writeNullableValue path; verify it preserves full nanosecond
576+
// precision alongside an actual null.
577+
Schema schema =
578+
Schema.of(Schema.Field.nullable("ts", FieldType.logicalType(new NanosInstant())));
579+
java.time.Instant ts =
580+
java.time.Instant.ofEpochSecond(TEST_EPOCH_SECONDS, TEST_NANOS_OF_SECOND);
581+
Row row1 = Row.withSchema(schema).addValue(ts).build();
582+
Row row2 = Row.withSchema(schema).addValue(null).build();
583+
584+
executeSql("CREATE TABLE test_nullable_datetime64_ns (ts Nullable(DateTime64(9))) ENGINE=Log");
585+
586+
pipeline
587+
.apply(Create.of(row1, row2).withRowSchema(schema))
588+
.apply(write("test_nullable_datetime64_ns"));
589+
pipeline.run().waitUntilFinish();
590+
591+
long total = executeQueryAsLong("SELECT COUNT(*) FROM test_nullable_datetime64_ns");
592+
long nonNull = executeQueryAsLong("SELECT COUNT(ts) FROM test_nullable_datetime64_ns");
593+
long ticks =
594+
executeQueryAsLong(
595+
"SELECT toUnixTimestamp64Nano(ts) FROM test_nullable_datetime64_ns"
596+
+ " WHERE ts IS NOT NULL");
597+
assertEquals(2L, total);
598+
assertEquals(1L, nonNull);
599+
assertEquals(TEST_EPOCH_SECONDS * NANOS_PER_SECOND + TEST_NANOS_OF_SECOND, ticks);
600+
}
601+
483602
@Test
484603
public void testUserAgentInQueryLog() throws Exception {
485604
Schema schema = Schema.of(Schema.Field.of("f0", FieldType.INT64));

0 commit comments

Comments
 (0)