Copilot commented on code in PR #64741: URL: https://github.com/apache/doris/pull/64741#discussion_r3465057900
########## fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/MySqlTypeConsistencyITCase.java: ########## @@ -0,0 +1,206 @@ +// 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.doris.cdcclient.itcase; + +import static org.assertj.core.api.Assertions.assertThat; + +import org.apache.doris.cdcclient.common.Env; +import org.apache.doris.cdcclient.itcase.CdcClientReadHarness.SnapshotResult; +import org.apache.doris.job.cdc.split.SnapshotSplit; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.testcontainers.containers.MySQLContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; +import org.testcontainers.utility.DockerImageName; + +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.Statement; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import java.util.concurrent.atomic.AtomicLong; + +/** + * Guards that the snapshot phase (JDBC) and the binlog phase (Debezium decoding) — two different + * converter paths — deserialize the same MySQL value identically. The identical value is inserted + * once before capture (snapshot, id 1) and once after (binlog, id 2); every column must match + * across phases. Any difference is a real bug (the MySQL TIME out-of-range issue was one instance). + * + * <p>JSON is compared by parsed value, not text: MySQL returns snapshot JSON with key reordering and + * spaces ({@code {"a": [2, 3], "k": 1}}) while the binlog path emits compact JSON + * ({@code {"a":[2,3],"k":1}}) — same value, different text, so a semantic comparison is used. + */ +@Testcontainers +class MySqlTypeConsistencyITCase { + + private static final String ROOT_USER = "root"; + private static final String ROOT_PASSWORD = "123456"; + private static final ObjectMapper MAPPER = new ObjectMapper(); + private static final AtomicLong JOB_ID_SEQ = new AtomicLong(821_000); + + // Identical column tuple used for both the snapshot row (id 1) and the binlog row (id 2). + private static final String VALUES = + "200," // c_tinyint_u TINYINT UNSIGNED + + "4000000000," // c_int_u INT UNSIGNED + + "18446744073709551615," // c_bigint_u BIGINT UNSIGNED + + "12345678901234.567890," // c_decimal DECIMAL(20,6) + + "1.5," // c_float FLOAT + + "3.141592653589793," // c_double DOUBLE + + "1," // c_bool TINYINT(1) + + "'b'," // c_enum + + "'x,z'," // c_set + + "b'10100101'," // c_bit BIT(8) + + "'{\"k\": 1, \"a\": [2, 3]}'," // c_json + + "'2023-08-15 12:34:56.123456'," // c_datetime DATETIME(6) + + "'2023-08-15 12:34:56.123456'," // c_timestamp TIMESTAMP(6) + + "'2023-08-15'," // c_date DATE + + "'12:34:56.123456'," // c_time TIME(6) + + "2023," // c_year YEAR + + "0x0102030405," // c_varbinary VARBINARY(8) + + "'abcde'"; // c_char CHAR(5) + + @Container + static final MySQLContainer<?> MYSQL = + new MySQLContainer<>(DockerImageName.parse("mysql:8.0")) + .withConfigurationOverride("docker/server-allow-ancient-date-time") + .withDatabaseName("cdc_test") + .withUsername("cdc") + .withPassword("123456") + .withEnv("MYSQL_ROOT_PASSWORD", ROOT_PASSWORD); + + private String jobId; + private String database; + + @BeforeEach + void setUp() throws Exception { + jobId = String.valueOf(JOB_ID_SEQ.incrementAndGet()); + database = "typescan_db_" + jobId; + try (Connection conn = rootConnection(""); + Statement st = conn.createStatement()) { + st.execute("CREATE DATABASE " + database); + st.execute("USE " + database); + st.execute( + "CREATE TABLE t_scan (" + + "id INT NOT NULL," + + "c_tinyint_u TINYINT UNSIGNED," + + "c_int_u INT UNSIGNED," + + "c_bigint_u BIGINT UNSIGNED," + + "c_decimal DECIMAL(20,6)," + + "c_float FLOAT," + + "c_double DOUBLE," + + "c_bool TINYINT(1)," + + "c_enum ENUM('a','b','c')," + + "c_set SET('x','y','z')," + + "c_bit BIT(8)," + + "c_json JSON," + + "c_datetime DATETIME(6)," + + "c_timestamp TIMESTAMP(6) NULL," + + "c_date DATE," + + "c_time TIME(6)," + + "c_year YEAR," + + "c_varbinary VARBINARY(8)," + + "c_char CHAR(5)," + + "PRIMARY KEY (id))"); + st.execute("INSERT INTO t_scan VALUES (1," + VALUES + ")"); + } + } + + @AfterEach + void tearDown() throws Exception { + Env.getCurrentEnv().close(jobId); + try (Connection conn = rootConnection(""); + Statement st = conn.createStatement()) { + st.execute("DROP DATABASE IF EXISTS " + database); + } + } + + @Test + void snapshotAndBinlogDeserializeIdentically() throws Exception { + try (CdcClientReadHarness harness = + CdcClientReadHarness.mysql( + jobId, + MYSQL.getHost(), + MYSQL.getMappedPort(MySQLContainer.MYSQL_PORT), + ROOT_USER, + ROOT_PASSWORD, + database, + "t_scan", + "initial")) { + + List<SnapshotSplit> splits = harness.fetchAllSnapshotSplits("t_scan"); + SnapshotResult snapshot = harness.readSnapshot(splits); + JsonNode snap = MAPPER.readTree(snapshot.records().get(0)); + + // identical value inserted after capture starts -> binlog path + try (Connection conn = rootConnection(database); + Statement st = conn.createStatement()) { + st.execute("INSERT INTO t_scan VALUES (2," + VALUES + ")"); + } + List<String> binlog = harness.readBinlogUntil(snapshot, splits, 1, Duration.ofSeconds(60)); + JsonNode bin = MAPPER.readTree(binlog.get(0)); + + List<String> mismatches = new ArrayList<>(); + Iterator<String> fields = snap.fieldNames(); + while (fields.hasNext()) { + String col = fields.next(); + if (col.equals("id") || col.startsWith("__DORIS")) { + continue; + } + String snapVal = snap.get(col).asText(); + JsonNode binNode = bin.get(col); + String binVal = binNode == null ? "<missing>" : binNode.asText(); + if (!equalOrJsonEquivalent(snapVal, binVal)) { + mismatches.add(col + ": snapshot=[" + snapVal + "] binlog=[" + binVal + "]"); + } + } + mismatches.forEach(m -> System.out.println("[TYPE SCAN][MISMATCH] " + m)); + assertThat(mismatches).as("snapshot vs binlog per-column mismatches").isEmpty(); + } + } + + // Text equality, falling back to parsed-JSON equality so JSON columns that differ only in + // whitespace/key-order are treated as equal while a genuinely different value still fails. + private boolean equalOrJsonEquivalent(String a, String b) { + if (a.equals(b)) { + return true; + } + try { + return MAPPER.readTree(a).equals(MAPPER.readTree(b)); + } catch (Exception e) { + return false; + } + } Review Comment: `equalOrJsonEquivalent` falls back to parsed-JSON equality for *all* columns. Since numbers/booleans are valid JSON, this can incorrectly treat non-JSON type differences as “equal” (e.g. "1.50" vs "1.5"), weakening the intended type-consistency guard. Restrict the semantic JSON comparison to the actual JSON column(s) and require exact text equality for everything else. ########## fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/deserialize/DebeziumJsonDeserializer.java: ########## @@ -412,19 +412,50 @@ private Object convertToArray(Schema fieldSchema, Object dbzObj) { return dbzObj.toString(); } + // Format a since-midnight time value given as total microseconds (may be negative or exceed + // 24h) as MySQL TIME literal text: ±HH:MM:SS[.ffffff], stripping trailing fractional zeros. + private static String formatTimeText(long microsTotal) { + boolean negative = microsTotal < 0; + long abs = Math.abs(microsTotal); + long hours = abs / 3_600_000_000L; + long rem = abs % 3_600_000_000L; + long minutes = rem / 60_000_000L; + rem %= 60_000_000L; + long seconds = rem / 1_000_000L; + long frac = rem % 1_000_000L; + String sign = negative ? "-" : ""; + if (frac == 0) { + return String.format("%s%02d:%02d:%02d", sign, hours, minutes, seconds); + } + String fracStr = String.format("%06d", frac).replaceAll("0+$", ""); + return String.format("%s%02d:%02d:%02d.%s", sign, hours, minutes, seconds, fracStr); + } Review Comment: `formatTimeText` uses `String.format` and a regex-based `replaceAll("0+$", "")` on every out-of-range TIME conversion. This allocates heavily (Formatter + regex Pattern/Matcher) and can become a hotspot for MySQL TIME-as-duration workloads (values frequently >= 24h). Consider switching to a small `StringBuilder` formatter and trimming trailing zeros without regex. ########## fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/PostgresTypeConsistencyITCase.java: ########## @@ -0,0 +1,196 @@ +// 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.doris.cdcclient.itcase; + +import static org.assertj.core.api.Assertions.assertThat; + +import org.apache.doris.cdcclient.common.Env; +import org.apache.doris.job.cdc.split.SnapshotSplit; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.testcontainers.containers.PostgreSQLContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; +import org.testcontainers.utility.DockerImageName; + +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.Statement; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import java.util.concurrent.atomic.AtomicLong; + +/** + * PostgreSQL counterpart of {@link MySqlTypeConsistencyITCase}: the same value inserted before + * capture (snapshot, JDBC path) and after (binlog, logical-decoding path) must deserialize + * identically per column. JSON/JSONB are compared by parsed value to tolerate whitespace/key-order + * differences while still catching a genuinely different value. + */ +@Testcontainers +class PostgresTypeConsistencyITCase { + + private static final ObjectMapper MAPPER = new ObjectMapper(); + private static final AtomicLong JOB_ID_SEQ = new AtomicLong(595_000); + + // Identical column tuple for both the snapshot row (id 1) and the binlog row (id 2). + private static final String VALUES = + "12345678901234.567890," // c_numeric NUMERIC(20,6) + + "1.5," // c_real REAL + + "3.141592653589793," // c_double DOUBLE PRECISION + + "true," // c_bool BOOLEAN + + "'hello'," // c_text TEXT + + "'world'," // c_varchar VARCHAR(20) + + "'abcde'," // c_char CHAR(5) + + "'{\"k\": 1, \"a\": [2, 3]}'," // c_json JSON + + "'{\"k\": 1, \"a\": [2, 3]}'," // c_jsonb JSONB + + "B'10100101'," // c_bit BIT(8) + + "B'101'," // c_varbit VARBIT(16) + + "'\\x0102030405'," // c_bytea BYTEA + + "'11111111-1111-1111-1111-111111111111'," // c_uuid UUID + + "'2023-08-15'," // c_date DATE + + "'12:34:56.123456'," // c_time TIME(6) + + "'12:34:56.123456+08'," // c_timetz TIMETZ + + "'2023-08-15 12:34:56.123456'," // c_timestamp TIMESTAMP(6) + + "'2023-08-15 12:34:56.123456+08'," // c_timestamptz TIMESTAMPTZ + + "'1 day 02:03:04'," // c_interval INTERVAL + + "'192.168.0.1'," // c_inet INET + + "'{1,2,3}'," // c_int_arr INT[] + + "'{a,b}'," // c_text_arr TEXT[] + + "'12.34'"; // c_money MONEY + + @Container + static final PostgreSQLContainer<?> POSTGRES = + new PostgreSQLContainer<>(DockerImageName.parse("postgres:14")) + .withCommand("postgres", "-c", "wal_level=logical"); + + private String jobId; + private String table; + + @BeforeEach + void setUp() throws Exception { + jobId = String.valueOf(JOB_ID_SEQ.incrementAndGet()); + table = "t_scan_" + jobId; + try (Connection conn = connect(); + Statement st = conn.createStatement()) { + st.execute("DROP TABLE IF EXISTS " + table); + st.execute( + "CREATE TABLE " + table + " (" + + "id INT PRIMARY KEY," + + "c_numeric NUMERIC(20,6)," + + "c_real REAL," + + "c_double DOUBLE PRECISION," + + "c_bool BOOLEAN," + + "c_text TEXT," + + "c_varchar VARCHAR(20)," + + "c_char CHAR(5)," + + "c_json JSON," + + "c_jsonb JSONB," + + "c_bit BIT(8)," + + "c_varbit VARBIT(16)," + + "c_bytea BYTEA," + + "c_uuid UUID," + + "c_date DATE," + + "c_time TIME(6)," + + "c_timetz TIMETZ," + + "c_timestamp TIMESTAMP(6)," + + "c_timestamptz TIMESTAMPTZ," + + "c_interval INTERVAL," + + "c_inet INET," + + "c_int_arr INT[]," + + "c_text_arr TEXT[]," + + "c_money MONEY)"); + st.execute("ALTER TABLE " + table + " REPLICA IDENTITY FULL"); + st.execute("INSERT INTO " + table + " VALUES (1," + VALUES + ")"); + } + } + + @AfterEach + void tearDown() throws Exception { + Env.getCurrentEnv().close(jobId); + try (Connection conn = connect(); + Statement st = conn.createStatement()) { + st.execute("DROP TABLE IF EXISTS " + table); + } + } + + @Test + void snapshotAndBinlogDeserializeIdentically() throws Exception { + try (CdcClientReadHarness harness = + CdcClientReadHarness.postgres( + jobId, + POSTGRES.getHost(), + POSTGRES.getMappedPort(PostgreSQLContainer.POSTGRESQL_PORT), + POSTGRES.getUsername(), + POSTGRES.getPassword(), + POSTGRES.getDatabaseName(), + "public", + table, + "initial")) { + + List<SnapshotSplit> splits = harness.fetchAllSnapshotSplits(table); + CdcClientReadHarness.SnapshotResult snapshot = harness.readSnapshot(splits); + JsonNode snap = MAPPER.readTree(snapshot.records().get(0)); + + try (Connection conn = connect(); + Statement st = conn.createStatement()) { + st.execute("INSERT INTO " + table + " VALUES (2," + VALUES + ")"); + } + List<String> binlog = harness.readBinlogUntil(snapshot, splits, 1, Duration.ofSeconds(60)); + JsonNode bin = MAPPER.readTree(binlog.get(0)); + + List<String> mismatches = new ArrayList<>(); + Iterator<String> fields = snap.fieldNames(); + while (fields.hasNext()) { + String col = fields.next(); + if (col.equals("id") || col.startsWith("__DORIS")) { + continue; + } + String snapVal = snap.get(col).asText(); + JsonNode binNode = bin.get(col); + String binVal = binNode == null ? "<missing>" : binNode.asText(); + if (!equalOrJsonEquivalent(snapVal, binVal)) { + mismatches.add(col + ": snapshot=[" + snapVal + "] binlog=[" + binVal + "]"); + } + } + mismatches.forEach(m -> System.out.println("[PG TYPE SCAN][MISMATCH] " + m)); + assertThat(mismatches).as("snapshot vs binlog per-column mismatches").isEmpty(); + } + } + + private boolean equalOrJsonEquivalent(String a, String b) { + if (a.equals(b)) { + return true; + } + try { + return MAPPER.readTree(a).equals(MAPPER.readTree(b)); + } catch (Exception e) { + return false; + } + } Review Comment: `equalOrJsonEquivalent` applies parsed-JSON equality to every column. Because JSON parsing succeeds for primitives (numbers/booleans/null) too, this can hide real snapshot-vs-binlog representation differences for non-JSON columns. Limit semantic JSON comparison to the JSON/JSONB columns and require exact text equality for the rest. -- 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
