From d14decbdb48a630b19d59e746d3f5b4d212bf053 Mon Sep 17 00:00:00 2001 From: Suvrat1629 Date: Fri, 5 Sep 2025 10:12:46 +0530 Subject: [PATCH 1/5] Add support for DateTime, Array, Tuple, Float32, Float64 in parseDefaultExpression --- .../beam/sdk/io/clickhouse/TableSchema.java | 153 ++++++++++++------ 1 file changed, 107 insertions(+), 46 deletions(-) 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 b89a88b3fae8..51f50fbe92b9 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 @@ -32,7 +32,7 @@ /** A descriptor for ClickHouse table schema. */ @AutoValue @SuppressWarnings({ - "nullness" // TODO(https://github.com/apache/beam/issues/20497) + "nullness" // TODO(https://github.com/apache/beam/issues/20497) }) public abstract class TableSchema implements Serializable { @@ -50,15 +50,15 @@ public static TableSchema of(Column... columns) { */ public static Schema getEquivalentSchema(TableSchema tableSchema) { return tableSchema.columns().stream() - .map( - x -> { - if (x.columnType().nullable()) { - return Schema.Field.nullable(x.name(), getEquivalentFieldType(x.columnType())); - } else { - return Schema.Field.of(x.name(), getEquivalentFieldType(x.columnType())); - } - }) - .collect(Schema.toSchema()); + .map( + x -> { + if (x.columnType().nullable()) { + return Schema.Field.nullable(x.name(), getEquivalentFieldType(x.columnType())); + } else { + return Schema.Field.of(x.name(), getEquivalentFieldType(x.columnType())); + } + }) + .collect(Schema.toSchema()); } /** @@ -114,9 +114,9 @@ public static Schema.FieldType getEquivalentFieldType(ColumnType columnType) { return Schema.FieldType.BOOLEAN; case TUPLE: List fields = - columnType.tupleTypes().entrySet().stream() - .map(x -> Schema.Field.of(x.getKey(), Schema.FieldType.DATETIME)) - .collect(Collectors.toList()); + columnType.tupleTypes().entrySet().stream() + .map(x -> Schema.Field.of(x.getKey(), Schema.FieldType.DATETIME)) + .collect(Collectors.toList()); Schema.Field[] array = fields.toArray(new Schema.Field[fields.size()]); Schema schema = Schema.of(array); return Schema.FieldType.row(schema); @@ -139,7 +139,7 @@ public abstract static class Column implements Serializable { public boolean materializedOrAlias() { return DefaultType.MATERIALIZED.equals(defaultType()) - || DefaultType.ALIAS.equals(defaultType()); + || DefaultType.ALIAS.equals(defaultType()); } public static Column of(String name, ColumnType columnType) { @@ -147,10 +147,10 @@ public static Column of(String name, ColumnType columnType) { } public static Column of( - String name, - ColumnType columnType, - @Nullable DefaultType defaultType, - @Nullable Object defaultValue) { + String name, + ColumnType columnType, + @Nullable DefaultType defaultType, + @Nullable Object defaultValue) { return new AutoValue_TableSchema_Column(name, columnType, defaultType, defaultValue); } } @@ -248,43 +248,43 @@ public static ColumnType nullable(TypeName typeName) { public static ColumnType fixedString(int size) { return ColumnType.builder() - .typeName(TypeName.FIXEDSTRING) - .nullable(false) - .fixedStringSize(size) - .build(); + .typeName(TypeName.FIXEDSTRING) + .nullable(false) + .fixedStringSize(size) + .build(); } public static ColumnType enum8(Map enumValues) { return ColumnType.builder() - .typeName(TypeName.ENUM8) - .nullable(false) - .enumValues(enumValues) - .build(); + .typeName(TypeName.ENUM8) + .nullable(false) + .enumValues(enumValues) + .build(); } public static ColumnType enum16(Map enumValues) { return ColumnType.builder() - .typeName(TypeName.ENUM16) - .nullable(false) - .enumValues(enumValues) - .build(); + .typeName(TypeName.ENUM16) + .nullable(false) + .enumValues(enumValues) + .build(); } public static ColumnType array(ColumnType arrayElementType) { return ColumnType.builder() - .typeName(TypeName.ARRAY) - // ClickHouse doesn't allow nullable arrays - .nullable(false) - .arrayElementType(arrayElementType) - .build(); + .typeName(TypeName.ARRAY) + // ClickHouse doesn't allow nullable arrays + .nullable(false) + .arrayElementType(arrayElementType) + .build(); } public static ColumnType tuple(Map elements) { return ColumnType.builder() - .typeName(TypeName.TUPLE) - .nullable(false) - .tupleTypes(elements) - .build(); + .typeName(TypeName.TUPLE) + .nullable(false) + .tupleTypes(elements) + .build(); } /** @@ -297,7 +297,7 @@ public static ColumnType parse(String str) { try { return new org.apache.beam.sdk.io.clickhouse.impl.parser.ColumnTypeParser( new StringReader(str)) - .parse(); + .parse(); } catch (org.apache.beam.sdk.io.clickhouse.impl.parser.ParseException e) { throw new IllegalArgumentException("failed to parse", e); } @@ -313,6 +313,10 @@ public static ColumnType parse(String str) { * @return value of ClickHouse expression */ public static Object parseDefaultExpression(ColumnType columnType, String value) { + if (value == null || value.isEmpty()) { + return null; + } + switch (columnType.typeName()) { case INT8: return Byte.valueOf(value); @@ -322,11 +326,6 @@ public static Object parseDefaultExpression(ColumnType columnType, String value) return Integer.valueOf(value); case INT64: return Long.valueOf(value); - case ENUM16: - case ENUM8: - case FIXEDSTRING: - case STRING: - return value; case UINT8: return Short.valueOf(value); case UINT16: @@ -335,8 +334,70 @@ public static Object parseDefaultExpression(ColumnType columnType, String value) return Long.valueOf(value); case UINT64: return Long.valueOf(value); + case ENUM8: + case ENUM16: + case FIXEDSTRING: + case STRING: + return value; case BOOL: return Boolean.valueOf(value); + case FLOAT32: + return Float.valueOf(value); + case FLOAT64: + return Double.valueOf(value); + case DATE: + case DATETIME: + // ClickHouse DateTime/Date format: 'YYYY-MM-DD HH:MM:SS' or 'YYYY-MM-DD' + // Convert to Joda DateTime (used by Beam Schema.FieldType.DATETIME) + try { + String formattedValue = value.contains(" ") ? value : value + " 00:00:00"; + return new org.joda.time.DateTime( + java.time.LocalDateTime.parse( + formattedValue, java.time.format.DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")) + .atZone(java.time.ZoneId.of("UTC")) + .toInstant() + .toEpochMilli()); + } catch (java.time.format.DateTimeParseException e) { + throw new IllegalArgumentException("Invalid DateTime/Date format: " + value, e); + } + case ARRAY: + // ClickHouse Array format: '[1,2,3]' or '["a","b"]' + String cleanedArray = value.replaceAll("[\\[\\]']", "").trim(); + if (cleanedArray.isEmpty()) { + return new java.util.ArrayList<>(); + } + // Recursively parse array elements based on arrayElementType + ColumnType elementType = columnType.arrayElementType(); + if (elementType == null) { + throw new IllegalArgumentException("Array element type not specified for: " + value); + } + return Arrays.stream(cleanedArray.split(",\\s*")) + .map(element -> parseDefaultExpression(elementType, element)) + .collect(Collectors.toList()); + case TUPLE: + // ClickHouse Tuple format: '(1,"a")' or '('1','a')' after tuplePreprocessing + String cleanedTuple = value.replaceAll("[\\(\\)]", "").trim(); + if (cleanedTuple.isEmpty()) { + return new java.util.ArrayList<>(); + } + Map tupleTypes = columnType.tupleTypes(); + if (tupleTypes == null || tupleTypes.isEmpty()) { + throw new IllegalArgumentException("Tuple types not specified for: " + value); + } + // Split tuple elements (accounting for quoted strings) + String[] elements = cleanedTuple.split(",\\s*(?=(?:[^']*'[^']*')*[^']*$)"); + List tupleValues = new java.util.ArrayList<>(); + int index = 0; + for (Map.Entry entry : tupleTypes.entrySet()) { + if (index >= elements.length) { + throw new IllegalArgumentException("Tuple has fewer elements than expected: " + value); + } + // Strip quotes from quoted strings + String elementValue = elements[index].replaceAll("^'|'$", "").trim(); + tupleValues.add(parseDefaultExpression(entry.getValue(), elementValue)); + index++; + } + return tupleValues; default: throw new UnsupportedOperationException("Unsupported type: " + columnType); } @@ -366,4 +427,4 @@ abstract static class Builder { public abstract ColumnType build(); } } -} +} \ No newline at end of file From 9535f8b68ba5ee3e56e9456953fe39bad47c18fb Mon Sep 17 00:00:00 2001 From: Suvrat1629 Date: Fri, 5 Sep 2025 10:14:58 +0530 Subject: [PATCH 2/5] spotless --- .../beam/sdk/io/clickhouse/TableSchema.java | 100 +++++++++--------- 1 file changed, 51 insertions(+), 49 deletions(-) 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 51f50fbe92b9..de62d622b487 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 @@ -32,7 +32,7 @@ /** A descriptor for ClickHouse table schema. */ @AutoValue @SuppressWarnings({ - "nullness" // TODO(https://github.com/apache/beam/issues/20497) + "nullness" // TODO(https://github.com/apache/beam/issues/20497) }) public abstract class TableSchema implements Serializable { @@ -50,15 +50,15 @@ public static TableSchema of(Column... columns) { */ public static Schema getEquivalentSchema(TableSchema tableSchema) { return tableSchema.columns().stream() - .map( - x -> { - if (x.columnType().nullable()) { - return Schema.Field.nullable(x.name(), getEquivalentFieldType(x.columnType())); - } else { - return Schema.Field.of(x.name(), getEquivalentFieldType(x.columnType())); - } - }) - .collect(Schema.toSchema()); + .map( + x -> { + if (x.columnType().nullable()) { + return Schema.Field.nullable(x.name(), getEquivalentFieldType(x.columnType())); + } else { + return Schema.Field.of(x.name(), getEquivalentFieldType(x.columnType())); + } + }) + .collect(Schema.toSchema()); } /** @@ -114,9 +114,9 @@ public static Schema.FieldType getEquivalentFieldType(ColumnType columnType) { return Schema.FieldType.BOOLEAN; case TUPLE: List fields = - columnType.tupleTypes().entrySet().stream() - .map(x -> Schema.Field.of(x.getKey(), Schema.FieldType.DATETIME)) - .collect(Collectors.toList()); + columnType.tupleTypes().entrySet().stream() + .map(x -> Schema.Field.of(x.getKey(), Schema.FieldType.DATETIME)) + .collect(Collectors.toList()); Schema.Field[] array = fields.toArray(new Schema.Field[fields.size()]); Schema schema = Schema.of(array); return Schema.FieldType.row(schema); @@ -139,7 +139,7 @@ public abstract static class Column implements Serializable { public boolean materializedOrAlias() { return DefaultType.MATERIALIZED.equals(defaultType()) - || DefaultType.ALIAS.equals(defaultType()); + || DefaultType.ALIAS.equals(defaultType()); } public static Column of(String name, ColumnType columnType) { @@ -147,10 +147,10 @@ public static Column of(String name, ColumnType columnType) { } public static Column of( - String name, - ColumnType columnType, - @Nullable DefaultType defaultType, - @Nullable Object defaultValue) { + String name, + ColumnType columnType, + @Nullable DefaultType defaultType, + @Nullable Object defaultValue) { return new AutoValue_TableSchema_Column(name, columnType, defaultType, defaultValue); } } @@ -248,43 +248,43 @@ public static ColumnType nullable(TypeName typeName) { public static ColumnType fixedString(int size) { return ColumnType.builder() - .typeName(TypeName.FIXEDSTRING) - .nullable(false) - .fixedStringSize(size) - .build(); + .typeName(TypeName.FIXEDSTRING) + .nullable(false) + .fixedStringSize(size) + .build(); } public static ColumnType enum8(Map enumValues) { return ColumnType.builder() - .typeName(TypeName.ENUM8) - .nullable(false) - .enumValues(enumValues) - .build(); + .typeName(TypeName.ENUM8) + .nullable(false) + .enumValues(enumValues) + .build(); } public static ColumnType enum16(Map enumValues) { return ColumnType.builder() - .typeName(TypeName.ENUM16) - .nullable(false) - .enumValues(enumValues) - .build(); + .typeName(TypeName.ENUM16) + .nullable(false) + .enumValues(enumValues) + .build(); } public static ColumnType array(ColumnType arrayElementType) { return ColumnType.builder() - .typeName(TypeName.ARRAY) - // ClickHouse doesn't allow nullable arrays - .nullable(false) - .arrayElementType(arrayElementType) - .build(); + .typeName(TypeName.ARRAY) + // ClickHouse doesn't allow nullable arrays + .nullable(false) + .arrayElementType(arrayElementType) + .build(); } public static ColumnType tuple(Map elements) { return ColumnType.builder() - .typeName(TypeName.TUPLE) - .nullable(false) - .tupleTypes(elements) - .build(); + .typeName(TypeName.TUPLE) + .nullable(false) + .tupleTypes(elements) + .build(); } /** @@ -297,7 +297,7 @@ public static ColumnType parse(String str) { try { return new org.apache.beam.sdk.io.clickhouse.impl.parser.ColumnTypeParser( new StringReader(str)) - .parse(); + .parse(); } catch (org.apache.beam.sdk.io.clickhouse.impl.parser.ParseException e) { throw new IllegalArgumentException("failed to parse", e); } @@ -352,11 +352,12 @@ public static Object parseDefaultExpression(ColumnType columnType, String value) try { String formattedValue = value.contains(" ") ? value : value + " 00:00:00"; return new org.joda.time.DateTime( - java.time.LocalDateTime.parse( - formattedValue, java.time.format.DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")) - .atZone(java.time.ZoneId.of("UTC")) - .toInstant() - .toEpochMilli()); + java.time.LocalDateTime.parse( + formattedValue, + java.time.format.DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")) + .atZone(java.time.ZoneId.of("UTC")) + .toInstant() + .toEpochMilli()); } catch (java.time.format.DateTimeParseException e) { throw new IllegalArgumentException("Invalid DateTime/Date format: " + value, e); } @@ -372,8 +373,8 @@ public static Object parseDefaultExpression(ColumnType columnType, String value) throw new IllegalArgumentException("Array element type not specified for: " + value); } return Arrays.stream(cleanedArray.split(",\\s*")) - .map(element -> parseDefaultExpression(elementType, element)) - .collect(Collectors.toList()); + .map(element -> parseDefaultExpression(elementType, element)) + .collect(Collectors.toList()); case TUPLE: // ClickHouse Tuple format: '(1,"a")' or '('1','a')' after tuplePreprocessing String cleanedTuple = value.replaceAll("[\\(\\)]", "").trim(); @@ -390,7 +391,8 @@ public static Object parseDefaultExpression(ColumnType columnType, String value) int index = 0; for (Map.Entry entry : tupleTypes.entrySet()) { if (index >= elements.length) { - throw new IllegalArgumentException("Tuple has fewer elements than expected: " + value); + throw new IllegalArgumentException( + "Tuple has fewer elements than expected: " + value); } // Strip quotes from quoted strings String elementValue = elements[index].replaceAll("^'|'$", "").trim(); @@ -427,4 +429,4 @@ abstract static class Builder { public abstract ColumnType build(); } } -} \ No newline at end of file +} From db334ddb42ff3202da811bd5c6961e82dd415b0a Mon Sep 17 00:00:00 2001 From: Suvrat1629 Date: Fri, 5 Sep 2025 12:19:09 +0530 Subject: [PATCH 3/5] test fix (hopefully) --- .../org/apache/beam/sdk/io/clickhouse/TableSchema.java | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) 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 de62d622b487..16164db973b5 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 @@ -18,6 +18,7 @@ package org.apache.beam.sdk.io.clickhouse; import com.google.auto.value.AutoValue; +import com.google.common.base.Splitter; import java.io.Serializable; import java.io.StringReader; import java.util.Arrays; @@ -386,16 +387,17 @@ public static Object parseDefaultExpression(ColumnType columnType, String value) throw new IllegalArgumentException("Tuple types not specified for: " + value); } // Split tuple elements (accounting for quoted strings) - String[] elements = cleanedTuple.split(",\\s*(?=(?:[^']*'[^']*')*[^']*$)"); + List elements = + Splitter.onPattern(",\\s*(?=(?:[^']*'[^']*')*[^']*$)").splitToList(cleanedTuple); List tupleValues = new java.util.ArrayList<>(); int index = 0; for (Map.Entry entry : tupleTypes.entrySet()) { - if (index >= elements.length) { + if (index >= elements.size()) { throw new IllegalArgumentException( "Tuple has fewer elements than expected: " + value); } // Strip quotes from quoted strings - String elementValue = elements[index].replaceAll("^'|'$", "").trim(); + String elementValue = elements.get(index).replaceAll("^'|'$", "").trim(); tupleValues.add(parseDefaultExpression(entry.getValue(), elementValue)); index++; } From 580826cadac7ef8a19c970f55046f02b104dd368 Mon Sep 17 00:00:00 2001 From: Suvrat1629 Date: Fri, 5 Sep 2025 17:31:45 +0530 Subject: [PATCH 4/5] guava import changes --- .../java/org/apache/beam/sdk/io/clickhouse/TableSchema.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 16164db973b5..8a42c1ecc5c0 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 @@ -18,7 +18,6 @@ package org.apache.beam.sdk.io.clickhouse; import com.google.auto.value.AutoValue; -import com.google.common.base.Splitter; import java.io.Serializable; import java.io.StringReader; import java.util.Arrays; @@ -28,6 +27,7 @@ import java.util.stream.Collectors; import org.apache.beam.sdk.schemas.Schema; import org.apache.beam.sdk.schemas.logicaltypes.FixedBytes; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Splitter; import org.checkerframework.checker.nullness.qual.Nullable; /** A descriptor for ClickHouse table schema. */ From 6ff7686b6ca91518c74f5bc64a26ec669fea5110 Mon Sep 17 00:00:00 2001 From: Suvrat1629 Date: Thu, 11 Sep 2025 10:09:34 +0530 Subject: [PATCH 5/5] isEmpty and UTC changes --- .../java/org/apache/beam/sdk/io/clickhouse/TableSchema.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) 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 8a42c1ecc5c0..9cc5194416d3 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 @@ -314,7 +314,7 @@ public static ColumnType parse(String str) { * @return value of ClickHouse expression */ public static Object parseDefaultExpression(ColumnType columnType, String value) { - if (value == null || value.isEmpty()) { + if (value == null) { return null; } @@ -349,7 +349,6 @@ public static Object parseDefaultExpression(ColumnType columnType, String value) case DATE: case DATETIME: // ClickHouse DateTime/Date format: 'YYYY-MM-DD HH:MM:SS' or 'YYYY-MM-DD' - // Convert to Joda DateTime (used by Beam Schema.FieldType.DATETIME) try { String formattedValue = value.contains(" ") ? value : value + " 00:00:00"; return new org.joda.time.DateTime( @@ -358,7 +357,8 @@ public static Object parseDefaultExpression(ColumnType columnType, String value) java.time.format.DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")) .atZone(java.time.ZoneId.of("UTC")) .toInstant() - .toEpochMilli()); + .toEpochMilli(), + org.joda.time.DateTimeZone.UTC); } catch (java.time.format.DateTimeParseException e) { throw new IllegalArgumentException("Invalid DateTime/Date format: " + value, e); }