This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12840-e2957d27afa29d4de4324cf642450757d865b902 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit def3d4bc635aaedc80123552adf99431bb5aac75 Author: Goutam Adwant <[email protected]> AuthorDate: Mon Oct 19 12:17:11 2026 +0000 [Fix][Connector-V2] Preserve ClickHouse sink rows with incomplete keys (#12840) --- .../ReduceBufferedBatchStatementExecutor.java | 45 ++++-- .../ReduceBufferedBatchStatementExecutorTest.java | 155 +++++++++++++++++++++ .../clickhouse/ClickhouseSinkCDCChangelogIT.java | 92 ++++++++++++ .../fake_to_clickhouse_nullable_composite_key.conf | 48 +++++++ .../resources/fake_to_clickhouse_nullable_key.conf | 49 +++++++ .../fake_to_clickhouse_nullable_update_key.conf | 48 +++++++ 6 files changed, 425 insertions(+), 12 deletions(-) diff --git a/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/executor/ReduceBufferedBatchStatementExecutor.java b/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/executor/ReduceBufferedBatchStatementExecutor.java index 3169f4bf8c..cf860559d5 100644 --- a/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/executor/ReduceBufferedBatchStatementExecutor.java +++ b/seatunnel-connectors-v2/connector-clickhouse/src/main/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/executor/ReduceBufferedBatchStatementExecutor.java @@ -17,8 +17,6 @@ package org.apache.seatunnel.connectors.seatunnel.clickhouse.sink.client.executor; -import org.apache.seatunnel.shade.org.apache.commons.lang3.tuple.Pair; - import org.apache.seatunnel.api.table.type.RowKind; import org.apache.seatunnel.api.table.type.SeaTunnelRow; @@ -29,8 +27,6 @@ import lombok.RequiredArgsConstructor; import java.sql.Connection; import java.sql.SQLException; import java.util.LinkedHashMap; -import java.util.Map; -import java.util.Set; import java.util.function.Function; @AllArgsConstructor @@ -42,8 +38,7 @@ public class ReduceBufferedBatchStatementExecutor implements JdbcBatchStatementE @NonNull private final Function<SeaTunnelRow, SeaTunnelRow> valueTransform; private boolean ignoreUpdateBefore; - @NonNull private final LinkedHashMap<SeaTunnelRow, Pair<Boolean, SeaTunnelRow>> buffer = - new LinkedHashMap<>(); + @NonNull private final LinkedHashMap<Object, BufferedRow> buffer = new LinkedHashMap<>(); @Override public void prepareStatements(Connection connection) throws SQLException { @@ -60,25 +55,39 @@ public class ReduceBufferedBatchStatementExecutor implements JdbcBatchStatementE SeaTunnelRow key = keyExtractor.apply(record); boolean changeFlag = changeFlag(record.getRowKind()); SeaTunnelRow value = valueTransform.apply(record); - buffer.put(key, Pair.of(changeFlag, value)); + BufferedRow bufferedRow = new BufferedRow(changeFlag, key, value); + // Incomplete keys cannot identify one row, so they must not participate in reduction. + Object bufferKey = hasNullField(key) ? bufferedRow : key; + buffer.put(bufferKey, bufferedRow); + } + + private static boolean hasNullField(SeaTunnelRow key) { + if (key == null) { + return true; + } + for (Object field : key.getFields()) { + if (field == null) { + return true; + } + } + return false; } @Override public void executeBatch() throws SQLException { Boolean preChangeFlag = null; - Set<Map.Entry<SeaTunnelRow, Pair<Boolean, SeaTunnelRow>>> entrySet = buffer.entrySet(); - for (Map.Entry<SeaTunnelRow, Pair<Boolean, SeaTunnelRow>> entry : entrySet) { - Boolean currentChangeFlag = entry.getValue().getKey(); + for (BufferedRow bufferedRow : buffer.values()) { + Boolean currentChangeFlag = bufferedRow.changeFlag; if (currentChangeFlag) { if (preChangeFlag != null && !preChangeFlag) { deleteExecutor.executeBatch(); } - insertOrUpdateExecutor.addToBatch(entry.getValue().getValue()); + insertOrUpdateExecutor.addToBatch(bufferedRow.value); } else { if (preChangeFlag != null && preChangeFlag) { insertOrUpdateExecutor.executeBatch(); } - deleteExecutor.addToBatch(entry.getKey()); + deleteExecutor.addToBatch(bufferedRow.key); } preChangeFlag = currentChangeFlag; } @@ -114,4 +123,16 @@ public class ReduceBufferedBatchStatementExecutor implements JdbcBatchStatementE throw new UnsupportedOperationException("Unsupported rowKind: " + rowKind); } } + + private static class BufferedRow { + private final boolean changeFlag; + private final SeaTunnelRow key; + private final SeaTunnelRow value; + + private BufferedRow(boolean changeFlag, SeaTunnelRow key, SeaTunnelRow value) { + this.changeFlag = changeFlag; + this.key = key; + this.value = value; + } + } } diff --git a/seatunnel-connectors-v2/connector-clickhouse/src/test/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/executor/ReduceBufferedBatchStatementExecutorTest.java b/seatunnel-connectors-v2/connector-clickhouse/src/test/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/executor/ReduceBufferedBatchStatementExecutorTest.java new file mode 100644 index 0000000000..16d369fbba --- /dev/null +++ b/seatunnel-connectors-v2/connector-clickhouse/src/test/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/sink/client/executor/ReduceBufferedBatchStatementExecutorTest.java @@ -0,0 +1,155 @@ +/* + * 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.seatunnel.connectors.seatunnel.clickhouse.sink.client.executor; + +import org.apache.seatunnel.api.table.type.RowKind; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.sql.Connection; +import java.sql.SQLException; +import java.util.ArrayList; +import java.util.List; +import java.util.function.Function; + +class ReduceBufferedBatchStatementExecutorTest { + + @Test + void preservesDistinctRowsWithNullKeys() throws SQLException { + List<SeaTunnelRow> inserts = new ArrayList<>(); + ReduceBufferedBatchStatementExecutor executor = + executor(inserts, new ArrayList<>(), row -> extractKey(row, 1)); + + executor.addToBatch(row(RowKind.INSERT, 1, null, "first")); + executor.addToBatch(row(RowKind.INSERT, 2, null, "second")); + executor.executeBatch(); + + Assertions.assertEquals(2, inserts.size()); + Assertions.assertEquals("first", inserts.get(0).getField(2)); + Assertions.assertEquals("second", inserts.get(1).getField(2)); + } + + @Test + void preservesRowsWhenTheExtractorReturnsNoKey() throws SQLException { + List<SeaTunnelRow> inserts = new ArrayList<>(); + ReduceBufferedBatchStatementExecutor executor = + executor(inserts, new ArrayList<>(), row -> null); + + executor.addToBatch(row(RowKind.INSERT, 1, "first")); + executor.addToBatch(row(RowKind.INSERT, 2, "second")); + executor.executeBatch(); + + Assertions.assertEquals(2, inserts.size()); + } + + @Test + void preservesDistinctRowsWithAnIncompleteCompositeKey() throws SQLException { + List<SeaTunnelRow> inserts = new ArrayList<>(); + ReduceBufferedBatchStatementExecutor executor = + executor(inserts, new ArrayList<>(), row -> extractKey(row, 1, 2)); + + executor.addToBatch(row(RowKind.INSERT, 1, null, "east", "first")); + executor.addToBatch(row(RowKind.INSERT, 2, null, "east", "second")); + executor.executeBatch(); + + Assertions.assertEquals(2, inserts.size()); + } + + @Test + void stillReducesCompleteKeysToTheLastChange() throws SQLException { + List<SeaTunnelRow> inserts = new ArrayList<>(); + ReduceBufferedBatchStatementExecutor executor = + executor(inserts, new ArrayList<>(), row -> extractKey(row, 1)); + + executor.addToBatch(row(RowKind.INSERT, 1, 7, "first")); + executor.addToBatch(row(RowKind.INSERT, 2, 7, "second")); + executor.executeBatch(); + + Assertions.assertEquals(1, inserts.size()); + Assertions.assertEquals("second", inserts.get(0).getField(2)); + } + + @Test + void preservesDistinctDeletesWithNullKeys() throws SQLException { + List<SeaTunnelRow> inserts = new ArrayList<>(); + List<SeaTunnelRow> deletes = new ArrayList<>(); + ReduceBufferedBatchStatementExecutor executor = + executor(inserts, deletes, row -> extractKey(row, 1)); + + executor.addToBatch(row(RowKind.DELETE, 1, null, "first")); + executor.addToBatch(row(RowKind.DELETE, 2, null, "second")); + executor.executeBatch(); + + Assertions.assertTrue(inserts.isEmpty()); + Assertions.assertEquals(2, deletes.size()); + Assertions.assertNull(deletes.get(0).getField(0)); + } + + private static ReduceBufferedBatchStatementExecutor executor( + List<SeaTunnelRow> inserts, + List<SeaTunnelRow> deletes, + Function<SeaTunnelRow, SeaTunnelRow> keyExtractor) { + return new ReduceBufferedBatchStatementExecutor( + new RecordingExecutor(inserts), + new RecordingExecutor(deletes), + keyExtractor, + Function.identity(), + false); + } + + private static SeaTunnelRow extractKey(SeaTunnelRow row, int... fieldIndexes) { + Object[] fields = new Object[fieldIndexes.length]; + for (int i = 0; i < fieldIndexes.length; i++) { + fields[i] = row.getField(fieldIndexes[i]); + } + SeaTunnelRow key = new SeaTunnelRow(fields); + key.setTableId(row.getTableId()); + key.setRowKind(row.getRowKind()); + return key; + } + + private static SeaTunnelRow row(RowKind kind, Object... fields) { + SeaTunnelRow row = new SeaTunnelRow(fields); + row.setRowKind(kind); + return row; + } + + private static class RecordingExecutor implements JdbcBatchStatementExecutor { + private final List<SeaTunnelRow> rows; + + private RecordingExecutor(List<SeaTunnelRow> rows) { + this.rows = rows; + } + + @Override + public void prepareStatements(Connection connection) {} + + @Override + public void addToBatch(SeaTunnelRow row) { + rows.add(row); + } + + @Override + public void executeBatch() {} + + @Override + public void closeStatements() {} + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-clickhouse-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/ClickhouseSinkCDCChangelogIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-clickhouse-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/ClickhouseSinkCDCChangelogIT.java index 5d55346166..651ac89649 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-clickhouse-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/ClickhouseSinkCDCChangelogIT.java +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-clickhouse-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/clickhouse/ClickhouseSinkCDCChangelogIT.java @@ -155,6 +155,98 @@ public class ClickhouseSinkCDCChangelogIT extends TestSuiteBase implements TestR dropSinkTable(); } + @TestTemplate + public void testNullableKeyRowsAreNotReduced(TestContainer container) throws Exception { + try (Statement statement = connection.createStatement()) { + statement.execute( + "CREATE TABLE default.sink_nullable_key (code Nullable(Int32), marker String)" + + " ENGINE = MergeTree ORDER BY code SETTINGS allow_nullable_key = 1"); + try { + Container.ExecResult result = + container.executeJob("/fake_to_clickhouse_nullable_key.conf"); + Assertions.assertEquals(0, result.getExitCode()); + + Set<String> markers = new HashSet<>(); + int rowCount = 0; + try (ResultSet rows = + statement.executeQuery("SELECT marker FROM default.sink_nullable_key")) { + while (rows.next()) { + markers.add(rows.getString(1)); + rowCount++; + } + } + Assertions.assertEquals(3, rowCount); + Assertions.assertEquals( + new HashSet<>(Arrays.asList("first", "second", "new")), markers); + } finally { + statement.execute("DROP TABLE default.sink_nullable_key"); + } + } + } + + @TestTemplate + public void testNullableCompositeKeyRowsAreNotReduced(TestContainer container) + throws Exception { + try (Statement statement = connection.createStatement()) { + statement.execute( + "CREATE TABLE default.sink_nullable_composite_key" + + " (code Nullable(Int32), region String, marker String)" + + " ENGINE = MergeTree ORDER BY (code, region)" + + " SETTINGS allow_nullable_key = 1"); + try { + Container.ExecResult result = + container.executeJob("/fake_to_clickhouse_nullable_composite_key.conf"); + Assertions.assertEquals(0, result.getExitCode()); + + Set<String> markers = new HashSet<>(); + int rowCount = 0; + try (ResultSet rows = + statement.executeQuery( + "SELECT marker FROM default.sink_nullable_composite_key")) { + while (rows.next()) { + markers.add(rows.getString(1)); + rowCount++; + } + } + Assertions.assertEquals(2, rowCount); + Assertions.assertEquals(new HashSet<>(Arrays.asList("first", "second")), markers); + } finally { + statement.execute("DROP TABLE default.sink_nullable_composite_key"); + } + } + } + + @TestTemplate + public void testNullableKeyUpdateAfterRowsAreNotReduced(TestContainer container) + throws Exception { + try (Statement statement = connection.createStatement()) { + statement.execute( + "CREATE TABLE default.sink_nullable_update_key" + + " (code Nullable(Int32), marker String)" + + " ENGINE = MergeTree ORDER BY code SETTINGS allow_nullable_key = 1"); + try { + Container.ExecResult result = + container.executeJob("/fake_to_clickhouse_nullable_update_key.conf"); + Assertions.assertEquals(0, result.getExitCode()); + + Set<String> markers = new HashSet<>(); + int rowCount = 0; + try (ResultSet rows = + statement.executeQuery( + "SELECT marker FROM default.sink_nullable_update_key")) { + while (rows.next()) { + markers.add(rows.getString(1)); + rowCount++; + } + } + Assertions.assertEquals(2, rowCount); + Assertions.assertEquals(new HashSet<>(Arrays.asList("first", "second")), markers); + } finally { + statement.execute("DROP TABLE default.sink_nullable_update_key"); + } + } + } + @TestTemplate public void testClickhouseLogEngineTable(TestContainer container) throws Exception { initializeClickhouseLogEngineTable(); diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-clickhouse-e2e/src/test/resources/fake_to_clickhouse_nullable_composite_key.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-clickhouse-e2e/src/test/resources/fake_to_clickhouse_nullable_composite_key.conf new file mode 100644 index 0000000000..1bd8a4d0f3 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-clickhouse-e2e/src/test/resources/fake_to_clickhouse_nullable_composite_key.conf @@ -0,0 +1,48 @@ +# +# 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. +# + +env { + parallelism = 1 + job.mode = "BATCH" +} + +source { + FakeSource { + schema { + fields { + code = int + region = string + marker = string + } + } + rows = [ + { kind = INSERT, fields = [null, "east", "first"] }, + { kind = INSERT, fields = [null, "east", "second"] } + ] + } +} + +sink { + Clickhouse { + host = "clickhouse:8123" + database = "default" + table = "sink_nullable_composite_key" + username = "default" + password = "" + primary_key = "code, region" + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-clickhouse-e2e/src/test/resources/fake_to_clickhouse_nullable_key.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-clickhouse-e2e/src/test/resources/fake_to_clickhouse_nullable_key.conf new file mode 100644 index 0000000000..f3bbbe351b --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-clickhouse-e2e/src/test/resources/fake_to_clickhouse_nullable_key.conf @@ -0,0 +1,49 @@ +# +# 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. +# + +env { + parallelism = 1 + job.mode = "BATCH" +} + +source { + FakeSource { + schema { + fields { + code = int + marker = string + } + } + rows = [ + { kind = INSERT, fields = [null, "first"] }, + { kind = INSERT, fields = [null, "second"] }, + { kind = INSERT, fields = [7, "old"] }, + { kind = INSERT, fields = [7, "new"] } + ] + } +} + +sink { + Clickhouse { + host = "clickhouse:8123" + database = "default" + table = "sink_nullable_key" + username = "default" + password = "" + primary_key = "code" + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-clickhouse-e2e/src/test/resources/fake_to_clickhouse_nullable_update_key.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-clickhouse-e2e/src/test/resources/fake_to_clickhouse_nullable_update_key.conf new file mode 100644 index 0000000000..cbac88e9d3 --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-clickhouse-e2e/src/test/resources/fake_to_clickhouse_nullable_update_key.conf @@ -0,0 +1,48 @@ +# +# 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. +# + +env { + parallelism = 1 + job.mode = "BATCH" +} + +source { + FakeSource { + schema { + fields { + code = int + marker = string + } + } + rows = [ + { kind = UPDATE_AFTER, fields = [null, "first"] }, + { kind = UPDATE_AFTER, fields = [null, "second"] } + ] + } +} + +sink { + Clickhouse { + host = "clickhouse:8123" + database = "default" + table = "sink_nullable_update_key" + username = "default" + password = "" + primary_key = "code" + support_upsert = true + } +}
