MartijnVisser commented on code in PR #171: URL: https://github.com/apache/flink-connector-jdbc/pull/171#discussion_r4081216300
########## flink-connector-jdbc-clickhouse/src/main/java/org/apache/flink/connector/jdbc/clickhouse/database/dialect/ClickHouseDialectConverter.java: ########## @@ -0,0 +1,238 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.connector.jdbc.clickhouse.database.dialect; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.connector.jdbc.core.database.dialect.AbstractDialectConverter; +import org.apache.flink.table.data.ArrayData; +import org.apache.flink.table.data.DecimalData; +import org.apache.flink.table.data.GenericArrayData; +import org.apache.flink.table.data.GenericMapData; +import org.apache.flink.table.data.MapData; +import org.apache.flink.table.data.TimestampData; +import org.apache.flink.table.types.logical.ArrayType; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.LogicalTypeRoot; +import org.apache.flink.table.types.logical.MapType; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.table.types.logical.utils.LogicalTypeUtils; + +import java.lang.reflect.Array; +import java.sql.Date; +import java.sql.Time; +import java.sql.Timestamp; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.LocalTime; +import java.util.HashMap; +import java.util.Map; + +/** + * Runtime converter that responsible to convert between JDBC object and Flink internal object for + * ClickHouse. + */ +@Internal +public class ClickHouseDialectConverter extends AbstractDialectConverter { + + private static final long serialVersionUID = 1L; + + public ClickHouseDialectConverter(RowType rowType) { + super(rowType); + } + + @Override + public JdbcDeserializationConverter createInternalConverter(LogicalType type) { + LogicalTypeRoot root = type.getTypeRoot(); + + if (root == LogicalTypeRoot.ARRAY) { + ArrayType arrayType = (ArrayType) type; + return createClickHouseArrayConverter(arrayType); + } else if (root == LogicalTypeRoot.MAP) { + MapType mapType = (MapType) type; + return createClickHouseMapConverter(mapType); + } else { + return createPrimitiveConverter(type); + } + } + + @Override + public JdbcSerializationConverter createExternalConverter(LogicalType type) { + LogicalTypeRoot root = type.getTypeRoot(); + + if (root == LogicalTypeRoot.ARRAY) { + return (val, index, statement) -> + statement.setObject(index, toExternalSerializer(val.getArray(index), type)); + } else if (root == LogicalTypeRoot.MAP) { + return (val, index, statement) -> + statement.setObject(index, toExternalSerializer(val.getMap(index), type)); + } else { + return super.createExternalConverter(type); Review Comment: TIMESTAMP shifts by the JVM timezone on write: `testAppend` and `testUpsert` store values an hour early under Europe/Amsterdam, and pass under UTC as in CI. The driver shifts `setTimestamp` and `setObject(LocalDateTime)`; `setString` round-trips. ########## flink-connector-jdbc-clickhouse/src/main/java/org/apache/flink/connector/jdbc/clickhouse/database/dialect/ClickHouseDialectConverter.java: ########## @@ -0,0 +1,238 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.connector.jdbc.clickhouse.database.dialect; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.connector.jdbc.core.database.dialect.AbstractDialectConverter; +import org.apache.flink.table.data.ArrayData; +import org.apache.flink.table.data.DecimalData; +import org.apache.flink.table.data.GenericArrayData; +import org.apache.flink.table.data.GenericMapData; +import org.apache.flink.table.data.MapData; +import org.apache.flink.table.data.TimestampData; +import org.apache.flink.table.types.logical.ArrayType; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.LogicalTypeRoot; +import org.apache.flink.table.types.logical.MapType; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.table.types.logical.utils.LogicalTypeUtils; + +import java.lang.reflect.Array; +import java.sql.Date; +import java.sql.Time; +import java.sql.Timestamp; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.LocalTime; +import java.util.HashMap; +import java.util.Map; + +/** + * Runtime converter that responsible to convert between JDBC object and Flink internal object for + * ClickHouse. + */ +@Internal +public class ClickHouseDialectConverter extends AbstractDialectConverter { + + private static final long serialVersionUID = 1L; + + public ClickHouseDialectConverter(RowType rowType) { + super(rowType); + } + + @Override + public JdbcDeserializationConverter createInternalConverter(LogicalType type) { + LogicalTypeRoot root = type.getTypeRoot(); + + if (root == LogicalTypeRoot.ARRAY) { + ArrayType arrayType = (ArrayType) type; + return createClickHouseArrayConverter(arrayType); + } else if (root == LogicalTypeRoot.MAP) { + MapType mapType = (MapType) type; + return createClickHouseMapConverter(mapType); + } else { + return createPrimitiveConverter(type); + } + } + + @Override + public JdbcSerializationConverter createExternalConverter(LogicalType type) { + LogicalTypeRoot root = type.getTypeRoot(); + + if (root == LogicalTypeRoot.ARRAY) { + return (val, index, statement) -> + statement.setObject(index, toExternalSerializer(val.getArray(index), type)); + } else if (root == LogicalTypeRoot.MAP) { + return (val, index, statement) -> + statement.setObject(index, toExternalSerializer(val.getMap(index), type)); + } else { + return super.createExternalConverter(type); + } + } + + // adding support to MAP and ARRAY types + private static Object toExternalSerializer(Object value, LogicalType type) { + switch (type.getTypeRoot()) { + case BOOLEAN: + return value; + case TINYINT: + return value instanceof Number ? ((Number) value).byteValue() : value; + case SMALLINT: + return value instanceof Number ? ((Number) value).shortValue() : value; + case INTEGER: + case INTERVAL_YEAR_MONTH: + return value instanceof Number ? ((Number) value).intValue() : value; + case BIGINT: + case INTERVAL_DAY_TIME: + return value instanceof Number ? ((Number) value).longValue() : value; + case FLOAT: + return value instanceof Number ? ((Number) value).floatValue() : value; + case DOUBLE: + return value instanceof Number ? ((Number) value).doubleValue() : value; + case CHAR: + case VARCHAR: + return value.toString(); Review Comment: A null ARRAY<STRING> element or MAP value throws an NPE here, as `getElementOrNull` returns null; the DATE, TIME, TIMESTAMP and DECIMAL casts too. One null check at the top of `toExternalSerializer` covers all. ########## flink-connector-jdbc-clickhouse/src/main/java/org/apache/flink/connector/jdbc/clickhouse/database/dialect/ClickHouseDialect.java: ########## @@ -0,0 +1,139 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.connector.jdbc.clickhouse.database.dialect; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.connector.jdbc.core.database.dialect.AbstractDialect; +import org.apache.flink.connector.jdbc.core.database.dialect.JdbcDialectConverter; +import org.apache.flink.table.types.logical.LogicalTypeRoot; +import org.apache.flink.table.types.logical.RowType; + +import java.util.Arrays; +import java.util.EnumSet; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; + +import static java.lang.String.format; + +/** JDBC dialect for ClickHouse. */ +@Internal +public class ClickHouseDialect extends AbstractDialect { + + private static final long serialVersionUID = 1L; + + // Define MAX/MIN precision of TIMESTAMP type according to ClickHouse docs: + // https://clickhouse.com/docs/sql-reference/data-types/datetime64 + private static final int MAX_TIMESTAMP_PRECISION = 9; + private static final int MIN_TIMESTAMP_PRECISION = 0; + + // Define MAX/MIN precision of DECIMAL type according to ClickHouse docs: + // https://clickhouse.com/docs/sql-reference/data-types/decimal + private static final int MAX_DECIMAL_PRECISION = 76; + private static final int MIN_DECIMAL_PRECISION = 1; + + @Override + public JdbcDialectConverter getRowConverter(RowType rowType) { + return new ClickHouseDialectConverter(rowType); + } + + @Override + public String getLimitClause(long limit) { + return "LIMIT " + limit; + } + + @Override + public Optional<String> defaultDriverName() { + return Optional.of("com.clickhouse.jdbc.Driver"); + } + + @Override + public String dialectName() { + return "ClickHouse"; + } + + @Override + public String quoteIdentifier(String identifier) { + return identifier; + } + + // ClickHouse does not support Upsert statements + // Instead you can create a table with ReplacingMergeTree engine; + // https://clickhouse.com/docs/engines/table-engines/mergetree-family/replacingmergetree + @Override + public Optional<String> getUpsertStatement( + String tableName, String[] fieldNames, String[] uniqueKeyFields) { + return Optional.empty(); Review Comment: Here the sink falls back to a SELECT per key and an `ALTER TABLE ... UPDATE` mutation per existing key, which ClickHouse calls heavy and not designed for frequent use. Why not INSERTs into a ReplacingMergeTree? ########## flink-connector-jdbc-clickhouse/src/main/java/org/apache/flink/connector/jdbc/clickhouse/database/dialect/ClickHouseDialect.java: ########## @@ -0,0 +1,139 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.connector.jdbc.clickhouse.database.dialect; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.connector.jdbc.core.database.dialect.AbstractDialect; +import org.apache.flink.connector.jdbc.core.database.dialect.JdbcDialectConverter; +import org.apache.flink.table.types.logical.LogicalTypeRoot; +import org.apache.flink.table.types.logical.RowType; + +import java.util.Arrays; +import java.util.EnumSet; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; + +import static java.lang.String.format; + +/** JDBC dialect for ClickHouse. */ +@Internal +public class ClickHouseDialect extends AbstractDialect { + + private static final long serialVersionUID = 1L; + + // Define MAX/MIN precision of TIMESTAMP type according to ClickHouse docs: + // https://clickhouse.com/docs/sql-reference/data-types/datetime64 + private static final int MAX_TIMESTAMP_PRECISION = 9; + private static final int MIN_TIMESTAMP_PRECISION = 0; + + // Define MAX/MIN precision of DECIMAL type according to ClickHouse docs: + // https://clickhouse.com/docs/sql-reference/data-types/decimal + private static final int MAX_DECIMAL_PRECISION = 76; + private static final int MIN_DECIMAL_PRECISION = 1; + + @Override + public JdbcDialectConverter getRowConverter(RowType rowType) { + return new ClickHouseDialectConverter(rowType); + } + + @Override + public String getLimitClause(long limit) { + return "LIMIT " + limit; + } + + @Override + public Optional<String> defaultDriverName() { + return Optional.of("com.clickhouse.jdbc.Driver"); + } + + @Override + public String dialectName() { + return "ClickHouse"; + } + + @Override + public String quoteIdentifier(String identifier) { Review Comment: Why not quote with backticks? Unquoted, a column like `order-id` breaks the generated SQL. ########## pom.xml: ########## @@ -230,6 +238,24 @@ under the License. <version>4.5.13</version> </dependency> + <dependency> + <groupId>org.apache.httpcomponents.core5</groupId> + <artifactId>httpcore5</artifactId> + <version>5.3.4</version> + </dependency> + + <dependency> + <groupId>org.apache.httpcomponents.core5</groupId> + <artifactId>httpcore5-h2</artifactId> + <version>5.3.4</version> + </dependency> + + <dependency> + <groupId>org.apache.httpcomponents.client5</groupId> + <artifactId>httpclient5</artifactId> Review Comment: This pins httpclient5 for every module, not just the ClickHouse tests: `flink-connector-jdbc-core` gets 5.4.4 at runtime through openlineage-java, which declares 5.6.1 (5.6.4 on main). These and commons-io belong in the ClickHouse module's pom. ########## flink-connector-jdbc-clickhouse/src/test/java/org/apache/flink/connector/jdbc/clickhouse/table/ClickHouseDynamicTableSinkITCase.java: ########## @@ -0,0 +1,231 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.connector.jdbc.clickhouse.table; + +import org.apache.flink.connector.jdbc.clickhouse.ClickHouseTestBase; +import org.apache.flink.connector.jdbc.clickhouse.database.dialect.ClickHouseDialect; +import org.apache.flink.connector.jdbc.core.table.sink.JdbcDynamicTableSinkITCase; +import org.apache.flink.connector.jdbc.testutils.tables.TableRow; +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.EnvironmentSettings; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.table.planner.factories.TestValuesTableFactory; +import org.apache.flink.types.Row; +import org.apache.flink.types.RowKind; + +import java.math.BigDecimal; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; + +import static org.apache.flink.connector.jdbc.clickhouse.ClickHouseTestBase.tableRow; +import static org.apache.flink.connector.jdbc.testutils.tables.TableBuilder.dbType; +import static org.apache.flink.connector.jdbc.testutils.tables.TableBuilder.field; +import static org.apache.flink.connector.jdbc.testutils.tables.TableBuilder.pkField; +import static org.assertj.core.api.Assertions.assertThat; + +/** The Table Sink ITCase for {@link ClickHouseDialect}. */ +class ClickHouseDynamicTableSinkITCase extends JdbcDynamicTableSinkITCase + implements ClickHouseTestBase { + + @Override + protected TableRow createUpsertOutputTable() { + return tableRow( + "dynamicSinkForUpsert", + pkField("cnt", dbType("Int64"), DataTypes.BIGINT().notNull()), + pkField("lencnt", dbType("Int64"), DataTypes.BIGINT().notNull()), + field("cTag", dbType("Int32"), DataTypes.INT().notNull()), + field("ts", dbType("DateTime64(6)"), DataTypes.TIMESTAMP())); + } + + @Override + protected TableRow createAppendOutputTable() { + return tableRow( + "dynamicSinkForAppend", + pkField("id", dbType("Int32"), DataTypes.INT().notNull()), + field("num", dbType("Int64"), DataTypes.BIGINT().notNull()), + field("ts", dbType("DateTime64(6)"), DataTypes.TIMESTAMP())); + } + + @Override + protected TableRow createBatchOutputTable() { + return tableRow( + "dynamicSinkForBatch", + field("NAME", dbType("String"), DataTypes.VARCHAR(20).notNull()), + field("SCORE", dbType("Int64"), DataTypes.BIGINT().notNull())); + } + + @Override + protected TableRow createRealOutputTable() { + return tableRow("REAL_TABLE", field("real_data", dbType("Float32"), DataTypes.FLOAT())); + } + + @Override + protected TableRow createUserOutputTable() { + return tableRow( + "USER_TABLE", + pkField("user_id", dbType("String"), DataTypes.VARCHAR(20).notNull()), + pkField("user_name", dbType("String"), DataTypes.VARCHAR(20).notNull()), + field("email", dbType("String"), DataTypes.VARCHAR(255)), + field("balance", dbType("Decimal(18, 2)"), DataTypes.DECIMAL(18, 2)), + field( + "map_col", + dbType("Map(Int, String)"), + DataTypes.MAP(DataTypes.INT(), DataTypes.STRING())), + field("array_col", dbType("Array(String)"), DataTypes.ARRAY(DataTypes.STRING())), + field("balance2", dbType("Decimal(18, 2)"), DataTypes.DECIMAL(18, 2))); + } + + @Override + protected List<Row> testUserData() { + return Arrays.asList( + Row.of( + "user1", + "Tom", + "[email protected]", + new BigDecimal("8.10"), + new HashMap<Integer, String>() { + { + put(1, "tom123"); + put(2, "tom234"); + } + }, + new String[] {"tommy", "thomas"}, + new BigDecimal("16.20")), + Row.of( + "user3", + "Bailey", + "[email protected]", + new BigDecimal("9.99"), + new HashMap<Integer, String>() { + { + put(3, "bly123"); + put(4, "bly234"); + } + }, + new String[] {"bailey", "bll"}, + new BigDecimal("19.98")), + Row.of( + "user4", + "Tina", + "[email protected]", + new BigDecimal("11.30"), + new HashMap<Integer, String>() { + { + put(3, "tina123"); + put(4, "tina3333"); + } + }, + new String[] {"tnn", "tina"}, + new BigDecimal("22.60"))); + } + + @Override + protected void testReadingFromChangelogSource() throws Exception { Review Comment: This override has only inserts, so the base test's updates and deletes never reach ClickHouse. The base test passes locally with a five-column user table; I think MAP and ARRAY fit a separate test. ########## flink-connector-jdbc-clickhouse/src/test/java/org/apache/flink/connector/jdbc/clickhouse/table/ClickHouseDynamicTableSinkITCase.java: ########## @@ -0,0 +1,231 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.connector.jdbc.clickhouse.table; + +import org.apache.flink.connector.jdbc.clickhouse.ClickHouseTestBase; +import org.apache.flink.connector.jdbc.clickhouse.database.dialect.ClickHouseDialect; +import org.apache.flink.connector.jdbc.core.table.sink.JdbcDynamicTableSinkITCase; +import org.apache.flink.connector.jdbc.testutils.tables.TableRow; +import org.apache.flink.table.api.DataTypes; +import org.apache.flink.table.api.EnvironmentSettings; +import org.apache.flink.table.api.TableEnvironment; +import org.apache.flink.table.planner.factories.TestValuesTableFactory; +import org.apache.flink.types.Row; +import org.apache.flink.types.RowKind; + +import java.math.BigDecimal; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; + +import static org.apache.flink.connector.jdbc.clickhouse.ClickHouseTestBase.tableRow; +import static org.apache.flink.connector.jdbc.testutils.tables.TableBuilder.dbType; +import static org.apache.flink.connector.jdbc.testutils.tables.TableBuilder.field; +import static org.apache.flink.connector.jdbc.testutils.tables.TableBuilder.pkField; +import static org.assertj.core.api.Assertions.assertThat; + +/** The Table Sink ITCase for {@link ClickHouseDialect}. */ +class ClickHouseDynamicTableSinkITCase extends JdbcDynamicTableSinkITCase + implements ClickHouseTestBase { + + @Override + protected TableRow createUpsertOutputTable() { + return tableRow( + "dynamicSinkForUpsert", + pkField("cnt", dbType("Int64"), DataTypes.BIGINT().notNull()), + pkField("lencnt", dbType("Int64"), DataTypes.BIGINT().notNull()), Review Comment: The base test keys this on (cnt, cTag), so a lencnt change tests `getUpdateStatement`. With lencnt in the key it is a new row instead. I've verified the base layout passes, so I'd keep it. ########## flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/utils/JdbcTypeUtilTest.java: ########## @@ -35,7 +35,6 @@ class JdbcTypeUtilTest { void testTypeConversions() { assertThat(logicalTypeToSqlType(LogicalTypeRoot.INTEGER)).isEqualTo(Types.INTEGER); testUnsupportedType(LogicalTypeRoot.RAW); - testUnsupportedType(LogicalTypeRoot.MAP); Review Comment: This drops the MAP assertion without a replacement. Please assert that MAP maps to `Types.JAVA_OBJECT`, and put the core change in its own commit. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
