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-12696-262d623d9eddc49be3dc0184b39e540736bcce19 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 31100cb5301274fdb1f27ba539d2f46ade44dd9e Author: Jast <[email protected]> AuthorDate: Mon Oct 12 05:25:22 2026 +0000 [Fix][Connector-V2][TiDB-CDC] Emit null instead of NPE for columns missing from live table info (#12696) --- .../source/converter/DefaultDataConverter.java | 4 + .../source/converter/DefaultDataConverterTest.java | 102 +++++++++++++++++++++ 2 files changed, 106 insertions(+) diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-tidb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/tidb/source/converter/DefaultDataConverter.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-tidb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/tidb/source/converter/DefaultDataConverter.java index f71437a903..416e216fe6 100644 --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-tidb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/tidb/source/converter/DefaultDataConverter.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-tidb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/tidb/source/converter/DefaultDataConverter.java @@ -52,7 +52,11 @@ public class DefaultDataConverter implements DataConverter<Object[]> { String fieldName = rowType.getFieldName(fieldIndex); TiColumnInfo columnInfo = tableInfo.getColumn(fieldName); if (columnInfo == null) { + // The planned schema can contain a column that the live TiKV table info no + // longer has (e.g. dropped/renamed after job planning); emit null for it + // instead of dereferencing the missing column info below. fields[fieldIndex] = null; + continue; } DataType dataType = columnInfo.getType(); Object value = values[columnInfo.getOffset()]; diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-tidb/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/tidb/source/converter/DefaultDataConverterTest.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-tidb/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/tidb/source/converter/DefaultDataConverterTest.java new file mode 100644 index 0000000000..d062363981 --- /dev/null +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-tidb/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/tidb/source/converter/DefaultDataConverterTest.java @@ -0,0 +1,102 @@ +/* + * 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.cdc.tidb.source.converter; + +import org.apache.seatunnel.api.table.type.BasicType; +import org.apache.seatunnel.api.table.type.SeaTunnelDataType; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.api.table.type.SeaTunnelRowType; + +import org.junit.jupiter.api.Test; +import org.tikv.common.meta.CIStr; +import org.tikv.common.meta.TiColumnInfo; +import org.tikv.common.meta.TiTableInfo; +import org.tikv.common.types.IntegerType; +import org.tikv.common.types.StringType; + +import java.util.Arrays; +import java.util.Collections; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; + +class DefaultDataConverterTest { + + private static final long TABLE_ID = 42L; + + @Test + void convertShouldEmitNullForColumnMissingFromLiveTableInfo() throws Exception { + // The planned row type still contains "dropped_col" while the live TiKV table info + // no longer has it (e.g. the column was dropped after job planning). + TiTableInfo tableInfo = tableInfo(); + SeaTunnelRowType rowType = + new SeaTunnelRowType( + new String[] {"id", "name", "dropped_col"}, + new SeaTunnelDataType<?>[] { + BasicType.LONG_TYPE, BasicType.STRING_TYPE, BasicType.STRING_TYPE + }); + Object[] values = {7L, "Alice"}; + + SeaTunnelRow row = new DefaultDataConverter().convert(values, tableInfo, rowType); + + assertEquals(3, row.getArity()); + assertEquals(7L, row.getField(0)); + assertEquals("Alice", row.getField(1)); + assertNull(row.getField(2)); + } + + @Test + void convertShouldMapAllColumnsWhenSchemasMatch() throws Exception { + SeaTunnelRowType rowType = + new SeaTunnelRowType( + new String[] {"id", "name"}, + new SeaTunnelDataType<?>[] {BasicType.LONG_TYPE, BasicType.STRING_TYPE}); + Object[] values = {7L, "Alice"}; + + SeaTunnelRow row = new DefaultDataConverter().convert(values, tableInfo(), rowType); + + assertEquals(2, row.getArity()); + assertEquals(7L, row.getField(0)); + assertEquals("Alice", row.getField(1)); + } + + private static TiTableInfo tableInfo() { + return new TiTableInfo( + TABLE_ID, + CIStr.newCIStr("test_table"), + "utf8mb4", + "utf8mb4_bin", + true, + Arrays.asList( + new TiColumnInfo(1L, "id", 0, IntegerType.BIGINT, true), + new TiColumnInfo(2L, "name", 1, StringType.VARCHAR, false)), + Collections.emptyList(), + "", + 0L, + 2L, + 0L, + 0L, + null, + null, + null, + 0L, + 0L, + 0L, + null); + } +}
