This is an automated email from the ASF dual-hosted git repository.
JNSimba pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new ab043a433ce [fix](fe) Normalize MySQL JDBC URLs for streaming jobs
(#65647)
ab043a433ce is described below
commit ab043a433cef0246bd81fbe76011a83b9a13d17b
Author: wudi <[email protected]>
AuthorDate: Wed Jul 22 16:41:31 2026 +0800
[fix](fe) Normalize MySQL JDBC URLs for streaming jobs (#65647)
### What problem does this PR solve?
MySQL Connector/J interprets TINYINT(1) as BOOLEAN and
YEAR as a date unless JDBC URL options override those defaults.
Streaming jobs did not normalize the URL consistently across the FROM
MYSQL and cdc_stream TVF paths, which could change automatically
discovered Doris column types and YEAR zero values. This change adds a
shared streaming JDBC URL normalizer and applies it before metadata
discovery and CDC reads. MySQL streaming URLs now use
yearIsDateType=false, tinyInt1isBit=false, useUnicode=true, and
characterEncoding=utf-8. PostgreSQL URLs remain unchanged, and
rewriteBatchedStatements is intentionally not added because streaming
ingestion writes through Stream Load.
---
.../insert/streaming/StreamingInsertJob.java | 17 ++-
.../streaming/StreamingJdbcUrlNormalizer.java | 83 ++++++++++++++
.../offset/jdbc/JdbcTvfSourceOffsetProvider.java | 3 +
.../CdcStreamTableValuedFunction.java | 10 +-
.../streaming/StreamingJdbcUrlNormalizerTest.java | 78 +++++++++++++
.../jdbc/JdbcTvfSourceOffsetProviderTest.java | 46 ++++++++
.../CdcStreamTableValuedFunctionTest.java | 15 +++
.../source/reader/mysql/MySqlSourceReader.java | 8 ++
.../source/reader/mysql/MySqlYearConverter.java | 70 ++++++++++++
.../source/reader/mysql/MySqlSourceReaderTest.java | 29 +++++
.../reader/mysql/MySqlYearConverterTest.java | 75 ++++++++++++
.../cdc/test_streaming_mysql_job_all_type.out | 9 +-
.../test_streaming_mysql_job_integer_boundary.out | 33 +++---
.../cdc/tvf/test_cdc_stream_tvf_mysql.out | 13 +--
...ing_job_cdc_stream_mysql_jdbc_type_defaults.out | 14 +++
.../cdc/test_streaming_mysql_job_all_type.groovy | 4 +-
..._streaming_mysql_job_charset_and_strings.groovy | 6 +-
.../test_streaming_mysql_job_create_alter.groovy | 37 +-----
...est_streaming_mysql_job_integer_boundary.groovy | 41 +++----
.../test_streaming_mysql_job_special_offset.groovy | 11 +-
.../cdc/tvf/test_cdc_stream_tvf_mysql.groovy | 11 +-
..._job_cdc_stream_mysql_jdbc_type_defaults.groovy | 126 +++++++++++++++++++++
22 files changed, 633 insertions(+), 106 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
index b22ccea0b19..28b684a7e03 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
@@ -297,11 +297,12 @@ public class StreamingInsertJob extends
AbstractJob<StreamingJobSchedulerTask, M
private List<String> createTableIfNotExists() throws Exception {
List<String> syncTbls = new ArrayList<>();
+ Map<String, String> effectiveSourceProperties =
buildConvertedSourceProperties(sourceProperties);
// Key: source table name (PG/MySQL); Value: CreateTableCommand for
the Doris target table.
// The two names differ when "table.<src>.target_table" is configured.
LinkedHashMap<String, CreateTableCommand> createTblCmds =
StreamingJobUtils.generateCreateTableCmds(targetDb,
- dataSourceType, sourceProperties, targetProperties);
+ dataSourceType, effectiveSourceProperties,
targetProperties);
Database db =
Env.getCurrentEnv().getInternalCatalog().getDbNullable(targetDb);
Preconditions.checkNotNull(db, "target database %s does not exist",
targetDb);
for (Map.Entry<String, CreateTableCommand> entry :
createTblCmds.entrySet()) {
@@ -536,7 +537,7 @@ public class StreamingInsertJob extends
AbstractJob<StreamingJobSchedulerTask, M
Map<String, String> mergedSourceProperties = new
HashMap<>(this.sourceProperties);
mergedSourceProperties.putAll(alterJobCommand.getSourceProperties());
Map<String, String> newConvertedSourceProperties =
- StreamingJobUtils.convertCertFile(getDbId(),
mergedSourceProperties);
+ buildConvertedSourceProperties(mergedSourceProperties);
this.sourceProperties = mergedSourceProperties;
this.convertedSourceProperties = newConvertedSourceProperties;
logParts.add("source properties: " +
alterJobCommand.getSourceProperties());
@@ -677,11 +678,21 @@ public class StreamingInsertJob extends
AbstractJob<StreamingJobSchedulerTask, M
private Map<String, String> getConvertedSourceProperties() throws
JobException {
if (convertedSourceProperties == null) {
- this.convertedSourceProperties =
StreamingJobUtils.convertCertFile(getDbId(), sourceProperties);
+ this.convertedSourceProperties =
buildConvertedSourceProperties(sourceProperties);
}
return convertedSourceProperties;
}
+ private Map<String, String> buildConvertedSourceProperties(Map<String,
String> inputProperties)
+ throws JobException {
+ Map<String, String> convertedProperties =
+ StreamingJobUtils.convertCertFile(getDbId(), inputProperties);
+ convertedProperties.put(DataSourceConfigKeys.JDBC_URL,
+ StreamingJdbcUrlNormalizer.normalize(dataSourceType,
+
convertedProperties.get(DataSourceConfigKeys.JDBC_URL)));
+ return convertedProperties;
+ }
+
private Map<String, String> getOriginTvfProps() {
if (originTvfProps == null) {
this.originTvfProps = getCurrentTvf().getProperties().getMap();
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJdbcUrlNormalizer.java
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJdbcUrlNormalizer.java
new file mode 100644
index 00000000000..f13d41687df
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJdbcUrlNormalizer.java
@@ -0,0 +1,83 @@
+// 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.job.extensions.insert.streaming;
+
+import org.apache.doris.job.common.DataSourceType;
+
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * Normalizes JDBC URLs before streaming ingestion uses them for metadata
discovery and CDC reads.
+ * Database-specific rules are kept here so every streaming entry point
applies the same read-side
+ * semantics while leaving unrelated JDBC Catalog write optimizations out of
scope.
+ */
+public final class StreamingJdbcUrlNormalizer {
+
+ private StreamingJdbcUrlNormalizer() {
+ }
+
+ public static String normalize(DataSourceType sourceType, String jdbcUrl) {
+ switch (sourceType) {
+ case MYSQL:
+ return normalizeMysql(jdbcUrl);
+ case POSTGRES:
+ return jdbcUrl;
+ default:
+ throw new IllegalArgumentException("Unsupported data source
type: " + sourceType);
+ }
+ }
+
+ private static String normalizeMysql(String jdbcUrl) {
+ String normalizedUrl = jdbcUrl.replace(" ", "");
+ Set<String> params = getParams(normalizedUrl);
+ StringBuilder result = new StringBuilder(normalizedUrl);
+ setDefaultParam(result, params, "yearIsDateType", "false");
+ setDefaultParam(result, params, "tinyInt1isBit", "false");
+ setDefaultParam(result, params, "useUnicode", "true");
+ setDefaultParam(result, params, "characterEncoding", "utf-8");
+ return result.toString();
+ }
+
+ private static void setDefaultParam(StringBuilder jdbcUrl, Set<String>
params, String param, String value) {
+ if (params.contains(param)) {
+ return;
+ }
+ char lastChar = jdbcUrl.charAt(jdbcUrl.length() - 1);
+ if (lastChar != '?' && lastChar != '&') {
+ jdbcUrl.append(jdbcUrl.indexOf("?") < 0 ? '?' : '&');
+ }
+ jdbcUrl.append(param).append('=').append(value);
+ }
+
+ private static Set<String> getParams(String jdbcUrl) {
+ Set<String> params = new HashSet<>();
+ int queryIndex = jdbcUrl.indexOf('?');
+ if (queryIndex < 0) {
+ return params;
+ }
+ for (String pair : jdbcUrl.substring(queryIndex + 1).split("&")) {
+ int equalsIndex = pair.indexOf('=');
+ String name = equalsIndex < 0 ? pair : pair.substring(0,
equalsIndex);
+ if (!name.isEmpty()) {
+ params.add(name);
+ }
+ }
+ return params;
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcTvfSourceOffsetProvider.java
b/fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcTvfSourceOffsetProvider.java
index 2376d3bd65b..0e5bb8fb753 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcTvfSourceOffsetProvider.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcTvfSourceOffsetProvider.java
@@ -27,6 +27,7 @@ import org.apache.doris.job.cdc.split.SnapshotSplit;
import org.apache.doris.job.common.DataSourceType;
import org.apache.doris.job.exception.JobException;
import org.apache.doris.job.extensions.insert.streaming.StreamingInsertJob;
+import
org.apache.doris.job.extensions.insert.streaming.StreamingJdbcUrlNormalizer;
import org.apache.doris.job.offset.Offset;
import org.apache.doris.job.util.StreamingJobUtils;
import org.apache.doris.nereids.analyzer.UnboundTVFRelation;
@@ -105,6 +106,8 @@ public class JdbcTvfSourceOffsetProvider extends
JdbcSourceOffsetProvider {
// Populate default slot/pub into sourceProperties so cleanMeta ->
/api/close
// carries the resolved names for cdcclient ownership-based cleanup.
Map<String, String> effective = new HashMap<>(originTvfProps);
+ effective.put(DataSourceConfigKeys.JDBC_URL,
StreamingJdbcUrlNormalizer.normalize(
+ resolvedType, effective.get(DataSourceConfigKeys.JDBC_URL)));
StreamingJobUtils.populateDefaultSourceProperties(resolvedType,
effective, String.valueOf(jobId));
// Always refresh fields that may be updated via ALTER JOB (e.g.
credentials, parallelism).
this.sourceProperties = effective;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/CdcStreamTableValuedFunction.java
b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/CdcStreamTableValuedFunction.java
index 6dec7cd6bef..e4323cef82a 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/CdcStreamTableValuedFunction.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/CdcStreamTableValuedFunction.java
@@ -27,6 +27,7 @@ import org.apache.doris.job.cdc.DataSourceConfigKeys;
import org.apache.doris.job.cdc.request.FetchRecordRequest;
import org.apache.doris.job.common.DataSourceType;
import
org.apache.doris.job.extensions.insert.streaming.DataSourceConfigValidator;
+import
org.apache.doris.job.extensions.insert.streaming.StreamingJdbcUrlNormalizer;
import org.apache.doris.job.util.StreamingJobUtils;
import org.apache.doris.thrift.TBrokerFileStatus;
import org.apache.doris.thrift.TFileType;
@@ -68,14 +69,17 @@ public class CdcStreamTableValuedFunction extends
ExternalFileTableValuedFunctio
copyProps.remove(INCLUDE_DELETE_SIGN);
copyProps.put("format", "json");
+ DataSourceType sourceType = DataSourceType.valueOf(
+ copyProps.get(DataSourceConfigKeys.TYPE).toUpperCase());
+ copyProps.put(DataSourceConfigKeys.JDBC_URL,
StreamingJdbcUrlNormalizer.normalize(
+ sourceType, copyProps.get(DataSourceConfigKeys.JDBC_URL)));
+
// Standalone TVF: random jobId. TVF-in-job: job.id injected by
rewriteTvfParams.
String jobId = copyProps.computeIfAbsent(JOB_ID_KEY,
k -> UUID.randomUUID().toString().replace("-", ""));
// Default PG slot/pub so cdcclient auto-creates per-job resources
- StreamingJobUtils.populateDefaultSourceProperties(
-
DataSourceType.valueOf(copyProps.get(DataSourceConfigKeys.TYPE).toUpperCase()),
- copyProps, jobId);
+ StreamingJobUtils.populateDefaultSourceProperties(sourceType,
copyProps, jobId);
super.parseCommonProperties(copyProps);
this.processedParams.put(ENABLE_CDC_CLIENT_KEY, "true");
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingJdbcUrlNormalizerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingJdbcUrlNormalizerTest.java
new file mode 100644
index 00000000000..bfdc7ec695d
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingJdbcUrlNormalizerTest.java
@@ -0,0 +1,78 @@
+// 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.job.extensions.insert.streaming;
+
+import org.apache.doris.job.common.DataSourceType;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class StreamingJdbcUrlNormalizerTest {
+
+ @Test
+ public void testNormalizeMysqlJdbcUrl() {
+ String jdbcUrl = StreamingJdbcUrlNormalizer.normalize(
+ DataSourceType.MYSQL, "jdbc:mysql://127.0.0.1:3306/test");
+
+
Assert.assertEquals("jdbc:mysql://127.0.0.1:3306/test?yearIsDateType=false"
+ +
"&tinyInt1isBit=false&useUnicode=true&characterEncoding=utf-8",
+ jdbcUrl);
+ Assert.assertFalse(jdbcUrl.contains("rewriteBatchedStatements"));
+ }
+
+ @Test
+ public void testNormalizeMysqlJdbcUrlPreservesExplicitValues() {
+ String jdbcUrl = "jdbc:mysql://127.0.0.1:3306/test?tinyInt1isBit=true"
+ +
"&yearIsDateType=true&useUnicode=false&characterEncoding=GBK";
+
+ Assert.assertEquals(jdbcUrl,
+ StreamingJdbcUrlNormalizer.normalize(DataSourceType.MYSQL,
jdbcUrl));
+
+ String partialJdbcUrl =
"jdbc:mysql://127.0.0.1:3306/test?yearIsDateType=true";
+ Assert.assertEquals(partialJdbcUrl + "&tinyInt1isBit=false"
+ + "&useUnicode=true&characterEncoding=utf-8",
+ StreamingJdbcUrlNormalizer.normalize(DataSourceType.MYSQL,
partialJdbcUrl));
+ }
+
+ @Test
+ public void testNormalizeMysqlJdbcUrlMatchesExactParameterNames() {
+ String jdbcUrl = "jdbc:mysql://127.0.0.1:3306/test?YEARISDATETYPE=true"
+ + "&custom=characterEncoding=utf-8";
+
+ Assert.assertEquals(jdbcUrl +
"&yearIsDateType=false&tinyInt1isBit=false"
+ + "&useUnicode=true&characterEncoding=utf-8",
+ StreamingJdbcUrlNormalizer.normalize(DataSourceType.MYSQL,
jdbcUrl));
+ }
+
+ @Test
+ public void testNormalizeMysqlJdbcUrlIsIdempotent() {
+ String jdbcUrl =
"jdbc:mysql://127.0.0.1:3306/test?yearIsDateType=false"
+ +
"&tinyInt1isBit=false&useUnicode=true&characterEncoding=utf-8";
+
+ Assert.assertEquals(jdbcUrl,
+ StreamingJdbcUrlNormalizer.normalize(DataSourceType.MYSQL,
jdbcUrl));
+ }
+
+ @Test
+ public void testNormalizePostgresJdbcUrlDoesNotChange() {
+ String jdbcUrl = "jdbc:postgresql://127.0.0.1:5432/test";
+
+ Assert.assertEquals(jdbcUrl,
+ StreamingJdbcUrlNormalizer.normalize(DataSourceType.POSTGRES,
jdbcUrl));
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/job/offset/jdbc/JdbcTvfSourceOffsetProviderTest.java
b/fe/fe-core/src/test/java/org/apache/doris/job/offset/jdbc/JdbcTvfSourceOffsetProviderTest.java
new file mode 100644
index 00000000000..09f336fc442
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/job/offset/jdbc/JdbcTvfSourceOffsetProviderTest.java
@@ -0,0 +1,46 @@
+// 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.job.offset.jdbc;
+
+import org.apache.doris.job.cdc.DataSourceConfigKeys;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class JdbcTvfSourceOffsetProviderTest {
+
+ @Test
+ public void testEnsureInitializedNormalizesMysqlJdbcUrl() throws Exception
{
+ JdbcTvfSourceOffsetProvider provider = new
JdbcTvfSourceOffsetProvider();
+ Map<String, String> properties = new HashMap<>();
+ properties.put(DataSourceConfigKeys.TYPE, "mysql");
+ properties.put(DataSourceConfigKeys.JDBC_URL,
"jdbc:mysql://127.0.0.1:3306/test");
+ properties.put(DataSourceConfigKeys.TABLE, "source_table");
+
+ provider.ensureInitialized(1L, properties);
+
+
Assert.assertEquals("jdbc:mysql://127.0.0.1:3306/test?yearIsDateType=false"
+ +
"&tinyInt1isBit=false&useUnicode=true&characterEncoding=utf-8",
+
provider.getSourceProperties().get(DataSourceConfigKeys.JDBC_URL));
+ Assert.assertEquals("jdbc:mysql://127.0.0.1:3306/test",
+ properties.get(DataSourceConfigKeys.JDBC_URL));
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/tablefunction/CdcStreamTableValuedFunctionTest.java
b/fe/fe-core/src/test/java/org/apache/doris/tablefunction/CdcStreamTableValuedFunctionTest.java
index 1be26fa69da..09eb838f759 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/tablefunction/CdcStreamTableValuedFunctionTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/tablefunction/CdcStreamTableValuedFunctionTest.java
@@ -22,9 +22,11 @@ import org.apache.doris.catalog.PrimitiveType;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.datasource.jdbc.client.JdbcClient;
import org.apache.doris.job.cdc.DataSourceConfigKeys;
+import org.apache.doris.job.cdc.request.FetchRecordRequest;
import org.apache.doris.job.common.DataSourceType;
import org.apache.doris.job.util.StreamingJobUtils;
+import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.Assert;
import org.junit.Test;
import org.mockito.MockedStatic;
@@ -37,6 +39,8 @@ import java.util.Map;
public class CdcStreamTableValuedFunctionTest {
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
@Test
public void testDeleteSignIsExcludedByDefault() throws Exception {
List<Column> columns = getTableColumns(baseProperties());
@@ -70,6 +74,17 @@ public class CdcStreamTableValuedFunctionTest {
Assert.assertTrue(exception.getMessage().contains("include_delete_sign"));
}
+ @Test
+ public void testMysqlJdbcUrlIsNormalizedInPayload() throws Exception {
+ CdcStreamTableValuedFunction function = new
CdcStreamTableValuedFunction(baseProperties());
+
+ FetchRecordRequest request = OBJECT_MAPPER.readValue(
+ function.getBackendConnectProperties().get("http.payload"),
FetchRecordRequest.class);
+
Assert.assertEquals("jdbc:mysql://localhost:3306/test_db?yearIsDateType=false"
+ +
"&tinyInt1isBit=false&useUnicode=true&characterEncoding=utf-8",
+ request.getConfig().get(DataSourceConfigKeys.JDBC_URL));
+ }
+
private List<Column> getTableColumns(Map<String, String> properties)
throws Exception {
JdbcClient jdbcClient = Mockito.mock(JdbcClient.class);
List<Column> sourceColumns = new ArrayList<>();
diff --git
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReader.java
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReader.java
index 53c74af0cb6..b9277075513 100644
---
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReader.java
+++
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReader.java
@@ -102,6 +102,7 @@ import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.mysql.cj.conf.ConnectionUrl;
+import com.mysql.cj.conf.PropertyKey;
import io.debezium.connector.mysql.MySqlConnection;
import io.debezium.connector.mysql.MySqlConnectorConfig;
import io.debezium.connector.mysql.MySqlPartition;
@@ -1018,6 +1019,13 @@ public class MySqlSourceReader extends
AbstractCdcSourceReader {
// Keep genuinely ancient (<100) DATE/DATETIME years; MySQL already
completes 2-digit years.
dbzProps.setProperty("enable.time.adjuster", "false");
+ // The converter is valid only when snapshot JDBC exposes YEAR values
as numbers.
+ if ("false"
+ .equalsIgnoreCase(
+
jdbcProperteis.getProperty(PropertyKey.yearIsDateType.getKeyName()))) {
+ dbzProps.setProperty("converters", "dorisYear");
+ dbzProps.setProperty("dorisYear.type",
MySqlYearConverter.class.getName());
+ }
configFactory.debeziumProperties(dbzProps);
configFactory.heartbeatInterval(Duration.ofMillis(DEBEZIUM_HEARTBEAT_INTERVAL_MS));
diff --git
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlYearConverter.java
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlYearConverter.java
new file mode 100644
index 00000000000..f5aa1646639
--- /dev/null
+++
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlYearConverter.java
@@ -0,0 +1,70 @@
+// 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.source.reader.mysql;
+
+import org.apache.kafka.connect.data.SchemaBuilder;
+
+import java.sql.Date;
+import java.util.Properties;
+
+import io.debezium.spi.converter.CustomConverter;
+import io.debezium.spi.converter.RelationalColumn;
+
+/** Preserves MySQL's special zero YEAR value in both snapshot and binlog
records. */
+public class MySqlYearConverter implements CustomConverter<SchemaBuilder,
RelationalColumn> {
+ private static final int BINLOG_ZERO_YEAR = 1900;
+
+ @Override
+ public void configure(Properties props) {}
+
+ @Override
+ public void converterFor(
+ RelationalColumn field, ConverterRegistration<SchemaBuilder>
registration) {
+ if (!"YEAR".equalsIgnoreCase(field.typeName())) {
+ return;
+ }
+ String qualifiedColumnName = field.dataCollection() + "." +
field.name();
+ registration.register(
+ io.debezium.time.Year.builder(),
+ value -> MySqlYearConverter.convertYear(value,
qualifiedColumnName));
+ }
+
+ static Integer convertYear(Object value, String qualifiedColumnName) {
+ if (value == null) {
+ return null;
+ }
+ if (value instanceof java.time.Year) {
+ int year = ((java.time.Year) value).getValue();
+ return year == BINLOG_ZERO_YEAR ? 0 : year;
+ }
+ if (value instanceof Date) {
+ return ((Date) value).toLocalDate().getYear();
+ }
+ if (value instanceof Number) {
+ return ((Number) value).intValue();
+ }
+ if (value instanceof String) {
+ return Integer.valueOf((String) value);
+ }
+ throw new IllegalArgumentException(
+ "Unsupported value type for MySQL YEAR column "
+ + qualifiedColumnName
+ + ": "
+ + value.getClass().getName());
+ }
+}
diff --git
a/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReaderTest.java
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReaderTest.java
index 2422d6c7c98..4136e4e617a 100644
---
a/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReaderTest.java
+++
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReaderTest.java
@@ -20,6 +20,7 @@ package org.apache.doris.cdcclient.source.reader.mysql;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -165,6 +166,34 @@ public class MySqlSourceReaderTest {
assertTrue(reader.getTableSchemas().containsKey(otherTableId));
}
+ @Test
+ void mysqlConfigRegistersYearConverterOnlyWhenYearIsDateTypeIsFalse()
throws Exception {
+ MySqlSourceConfig falseConfig =
+ sourceConfig(
+ "initial",
+ Map.of(
+ DataSourceConfigKeys.JDBC_URL,
+
"jdbc:mysql://localhost:3306/testdb?yearIsDateType=false"));
+
+ assertEquals("dorisYear",
falseConfig.getDbzProperties().getProperty("converters"));
+ assertEquals(
+
"org.apache.doris.cdcclient.source.reader.mysql.MySqlYearConverter",
+ falseConfig.getDbzProperties().getProperty("dorisYear.type"));
+
+ MySqlSourceConfig trueConfig =
+ sourceConfig(
+ "initial",
+ Map.of(
+ DataSourceConfigKeys.JDBC_URL,
+
"jdbc:mysql://localhost:3306/testdb?yearIsDateType=true"));
+ assertNull(trueConfig.getDbzProperties().getProperty("converters"));
+
assertNull(trueConfig.getDbzProperties().getProperty("dorisYear.type"));
+
+ MySqlSourceConfig missingConfig = sourceConfig("initial");
+ assertNull(missingConfig.getDbzProperties().getProperty("converters"));
+
assertNull(missingConfig.getDbzProperties().getProperty("dorisYear.type"));
+ }
+
// Drive the real generateMySqlConfig JSON-offset path and return the
rebuilt startup offset.
private BinlogOffset startupBinlogOffset(String offsetJson) throws
Exception {
return sourceConfig(offsetJson).getStartupOptions().binlogOffset;
diff --git
a/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlYearConverterTest.java
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlYearConverterTest.java
new file mode 100644
index 00000000000..4264567130b
--- /dev/null
+++
b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlYearConverterTest.java
@@ -0,0 +1,75 @@
+// 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.source.reader.mysql;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import org.junit.jupiter.api.Test;
+
+import java.sql.Date;
+import java.time.LocalDate;
+import java.time.Year;
+
+class MySqlYearConverterTest {
+ private static final String YEAR_COLUMN = "inventory.orders.order_year";
+
+ @Test
+ void preservesSnapshotYearZero() {
+ assertEquals(0, MySqlYearConverter.convertYear((short) 0,
YEAR_COLUMN));
+ }
+
+ @Test
+ void restoresBinlogYearZero() {
+ assertEquals(0, MySqlYearConverter.convertYear(Year.of(1900),
YEAR_COLUMN));
+ }
+
+ @Test
+ void preservesRegularYears() {
+ assertEquals(2000, MySqlYearConverter.convertYear((short) 2000,
YEAR_COLUMN));
+ assertEquals(2024, MySqlYearConverter.convertYear(Year.of(2024),
YEAR_COLUMN));
+ }
+
+ @Test
+ void acceptsDateFromYearIsDateType() {
+ assertEquals(
+ 2024,
+ MySqlYearConverter.convertYear(Date.valueOf("2024-01-01"),
YEAR_COLUMN));
+ }
+
+ @Test
+ void preservesNull() {
+ assertNull(MySqlYearConverter.convertYear(null, YEAR_COLUMN));
+ }
+
+ @Test
+ void unsupportedTypeIncludesColumnAndType() {
+ IllegalArgumentException exception =
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ MySqlYearConverter.convertYear(
+ LocalDate.of(2024, 1, 1),
YEAR_COLUMN));
+
+ assertEquals(
+ "Unsupported value type for MySQL YEAR column "
+ + "inventory.orders.order_year: java.time.LocalDate",
+ exception.getMessage());
+ }
+}
diff --git
a/regression-test/data/job_p0/streaming_job/cdc/test_streaming_mysql_job_all_type.out
b/regression-test/data/job_p0/streaming_job/cdc/test_streaming_mysql_job_all_type.out
index e4878ae8359..0d7e893c805 100644
---
a/regression-test/data/job_p0/streaming_job/cdc/test_streaming_mysql_job_all_type.out
+++
b/regression-test/data/job_p0/streaming_job/cdc/test_streaming_mysql_job_all_type.out
@@ -12,7 +12,7 @@ decimal4_u text Yes false \N NONE
decimal5_u text Yes false \N NONE
double_u double Yes false \N NONE
float_u float Yes false \N NONE
-boolean boolean Yes false \N NONE
+boolean tinyint Yes false \N NONE
tinyint tinyint Yes false \N NONE
smallint smallint Yes false \N NONE
mediumint int Yes false \N NONE
@@ -46,9 +46,8 @@ varbinary text Yes false \N NONE
enum text Yes false \N NONE
-- !select_all_types_null --
-1 120 50000 1000000 9000000000 1000 12345.67
987654.12345 123456789.1234567890 99999999.123456789012345678901234567890
123.45 12.34 true -5 -300 -20000 -500000 -8000000000 -123.45
-12.34 -1000 -1234.56 -98765.43210 -123456789.1234567890
-99999999.123456789012345678901234567890 2023 08:30 08:30:00.123
08:30:00.123456 2023-06-15 2023-06-15T08:30 2023-06-15T08:30
2023-06-15T08:30:00.123 2023-06-15T08:30:00.123456 abc user_001
normal text content c2ltcGxlIGJsb2IgZGF0YQ== {"id": 1, "name": "u
[...]
+1 120 50000 1000000 9000000000 1000 12345.67
987654.12345 123456789.1234567890 99999999.123456789012345678901234567890
123.45 12.34 1 -5 -300 -20000 -500000 -8000000000 -123.45
-12.34 -1000 -1234.56 -98765.43210 -123456789.1234567890
-99999999.123456789012345678901234567890 0 08:30 08:30:00.123
08:30:00.123456 2023-06-15 2023-06-15T08:30 2023-06-15T08:30
2023-06-15T08:30:00.123 2023-06-15T08:30:00.123456 abc user_001
normal text content c2ltcGxlIGJsb2IgZGF0YQ== {"id": 1, "name":
"userA"} [...]
-- !select_all_types_null2 --
-1 120 50000 1000000 9000000000 1000 12345.67
987654.12345 123456789.1234567890 99999999.123456789012345678901234567890
123.45 12.34 true -5 -300 -20000 -500000 -8000000000 -123.45
-12.34 -1000 -1234.56 -98765.43210 -123456789.1234567890
-99999999.123456789012345678901234567890 2023 08:30 08:30:00.123
08:30:00.123456 2023-06-15 2023-06-15T08:30 2023-06-15T08:30
2023-06-15T08:30:00.123 2023-06-15T08:30:00.123456 abc user_001
normal text content c2ltcGxlIGJsb2IgZGF0YQ== {"id": 1, "name": "u
[...]
-2 100 100000 1000000000 100000000000 12345 12345.67
123456789.12345 12345678901234567890.1234567890
123456789012345678901234567890.123456789012345678901234567890 12345.6789
123.456 true -10 -1000 -100000 -100000000 -1000000000000
-12345.6789 -123.456 -12345 -12345.67 -123456789.12345
-12345678901234567890.1234567890
-123456789012345678901234567890.123456789012345678901234567890 2024
12:34:56 12:34:56.789 12:34:56.789123 2024-01-01
2024-01-01T12:34:56 2024-01-01T12:34:56 2024-01-01T12:34: [...]
-
+1 120 50000 1000000 9000000000 1000 12345.67
987654.12345 123456789.1234567890 99999999.123456789012345678901234567890
123.45 12.34 1 -5 -300 -20000 -500000 -8000000000 -123.45
-12.34 -1000 -1234.56 -98765.43210 -123456789.1234567890
-99999999.123456789012345678901234567890 0 08:30 08:30:00.123
08:30:00.123456 2023-06-15 2023-06-15T08:30 2023-06-15T08:30
2023-06-15T08:30:00.123 2023-06-15T08:30:00.123456 abc user_001
normal text content c2ltcGxlIGJsb2IgZGF0YQ== {"id": 1, "name":
"userA"} [...]
+2 100 100000 1000000000 100000000000 12345 12345.67
123456789.12345 12345678901234567890.1234567890
123456789012345678901234567890.123456789012345678901234567890 12345.6789
123.456 1 -10 -1000 -100000 -100000000 -1000000000000
-12345.6789 -123.456 -12345 -12345.67 -123456789.12345
-12345678901234567890.1234567890
-123456789012345678901234567890.123456789012345678901234567890 2024
12:34:56 12:34:56.789 12:34:56.789123 2024-01-01
2024-01-01T12:34:56 2024-01-01T12:34:56 2024-01-01T12:34:56. [...]
diff --git
a/regression-test/data/job_p0/streaming_job/cdc/test_streaming_mysql_job_integer_boundary.out
b/regression-test/data/job_p0/streaming_job/cdc/test_streaming_mysql_job_integer_boundary.out
index c372d368e72..9b0040492b4 100644
---
a/regression-test/data/job_p0/streaming_job/cdc/test_streaming_mysql_job_integer_boundary.out
+++
b/regression-test/data/job_p0/streaming_job/cdc/test_streaming_mysql_job_integer_boundary.out
@@ -7,25 +7,24 @@ smallint_u int Yes false \N NONE
mediumint_u int Yes false \N NONE
int_u bigint Yes false \N NONE
bigint_u largeint Yes false \N NONE
-bool_col boolean Yes false \N NONE
+tinyint_1 tinyint Yes false \N NONE
tinyint_s tinyint Yes false \N NONE
-- !select_snapshot --
-1 all_zero 0 0 0 0 0 false 0
-2 signed_max 127 32767 8388607 2147483647
9223372036854775807 true 127
-3 signed_max_plus1 128 32768 8388608 2147483648
9223372036854775808 false -128
-4 unsigned_max 255 65535 16777215 4294967295
18446744073709551615 true -1
-5 mid_value 100 30000 8000000 2000000000
9000000000000000000 false 50
+1 all_zero 0 0 0 0 0 0 0
+2 signed_max 127 32767 8388607 2147483647
9223372036854775807 1 127
+3 signed_max_plus1 128 32768 8388608 2147483648
9223372036854775808 -1 -128
+4 unsigned_max 255 65535 16777215 4294967295
18446744073709551615 127 -1
+5 mid_value 100 30000 8000000 2000000000
9000000000000000000 4 50
-- !select_after_incr --
-1 all_zero 0 0 0 0 18446744073709551615
false 0
-2 signed_max 127 32767 8388607 2147483647
9223372036854775807 false 127
-3 signed_max_plus1 128 32768 8388608 2147483648
9223372036854775808 false -128
-4 unsigned_max 255 65535 16777215 4294967295
18446744073709551615 true -1
-5 mid_value 100 30000 8000000 4294967295
9000000000000000000 false 50
-101 all_zero 0 0 0 0 0 false 0
-102 signed_max 127 32767 8388607 2147483647
9223372036854775807 true 127
-103 signed_max_plus1 128 32768 8388608 2147483648
9223372036854775808 false -128
-104 unsigned_max 255 65535 16777215 4294967295
18446744073709551615 true -1
-105 mid_value 100 30000 8000000 2000000000
9000000000000000000 false 50
-
+1 all_zero 0 0 0 0 18446744073709551615
0 0
+2 signed_max 127 32767 8388607 2147483647
9223372036854775807 4 127
+3 signed_max_plus1 128 32768 8388608 2147483648
9223372036854775808 -1 -128
+4 unsigned_max 255 65535 16777215 4294967295
18446744073709551615 127 -1
+5 mid_value 100 30000 8000000 4294967295
9000000000000000000 4 50
+101 all_zero 0 0 0 0 0 0 0
+102 signed_max 127 32767 8388607 2147483647
9223372036854775807 1 127
+103 signed_max_plus1 128 32768 8388608 2147483648
9223372036854775808 -1 -128
+104 unsigned_max 255 65535 16777215 4294967295
18446744073709551615 127 -1
+105 mid_value 100 30000 8000000 2000000000
9000000000000000000 4 50
diff --git
a/regression-test/data/job_p0/streaming_job/cdc/tvf/test_cdc_stream_tvf_mysql.out
b/regression-test/data/job_p0/streaming_job/cdc/tvf/test_cdc_stream_tvf_mysql.out
index 02b13ffe0b2..17277ad47e7 100644
---
a/regression-test/data/job_p0/streaming_job/cdc/tvf/test_cdc_stream_tvf_mysql.out
+++
b/regression-test/data/job_p0/streaming_job/cdc/tvf/test_cdc_stream_tvf_mysql.out
@@ -1,11 +1,10 @@
-- This file is automatically generated. You should know what you did if you
want to edit this
-- !select_tvf --
-C1 3
-C1 99
-D1 4
-D1 4
+C1 3 4 0
+C1 99 4 0
+D1 4 -1 2024
+D1 4 -1 2024
-- !select_tvf_dml --
-C1 99
-D1 4
-
+C1 99 4 0
+D1 4 -1 2024
diff --git
a/regression-test/data/job_p0/streaming_job/cdc/tvf/test_streaming_job_cdc_stream_mysql_jdbc_type_defaults.out
b/regression-test/data/job_p0/streaming_job/cdc/tvf/test_streaming_job_cdc_stream_mysql_jdbc_type_defaults.out
new file mode 100644
index 00000000000..c3c7d34ee6e
--- /dev/null
+++
b/regression-test/data/job_p0/streaming_job/cdc/tvf/test_streaming_job_cdc_stream_mysql_jdbc_type_defaults.out
@@ -0,0 +1,14 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !snapshot_data --
+A1 -1 0
+B1 0 2024
+C1 2 0
+D1 4 2025
+
+-- !final_data --
+A1 -1 0
+B1 0 2024
+C1 2 0
+D1 4 2025
+E1 127 0
+F1 -128 2026
diff --git
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_all_type.groovy
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_all_type.groovy
index 1955476bee1..0b5753ac930 100644
---
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_all_type.groovy
+++
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_all_type.groovy
@@ -90,9 +90,9 @@ suite("test_streaming_mysql_job_all_type",
"p0,external,mysql,external_docker,ex
`enum` enum('Value1', 'Value2', 'Value3')
) engine=innodb charset=utf8;
"""
- // mock snapshot data
+ // Use numeric YEAR 0 to verify yearIsDateType=false preserves
0000 instead of 2000.
sql """
- insert into ${mysqlDb}.${table1} values
(1,120,50000,1000000,9000000000,1000,12345.67,987654.12345,123456789.1234567890,99999999.123456789012345678901234567890,123.45,12.34,true,-5,-300,-20000,-500000,-8000000000,-123.45,-12.34,-1000,-1234.56,-98765.43210,-123456789.1234567890,-99999999.123456789012345678901234567890,2023,'08:30:00','08:30:00.123','08:30:00.123456','2023-06-15','2023-06-15
08:30:00','2023-06-15 08:30:00','2023-06-15 08:30:00.123','2023-06-15
08:30:00.123456', [...]
+ insert into ${mysqlDb}.${table1} values
(1,120,50000,1000000,9000000000,1000,12345.67,987654.12345,123456789.1234567890,99999999.123456789012345678901234567890,123.45,12.34,true,-5,-300,-20000,-500000,-8000000000,-123.45,-12.34,-1000,-1234.56,-98765.43210,-123456789.1234567890,-99999999.123456789012345678901234567890,0,'08:30:00','08:30:00.123','08:30:00.123456','2023-06-15','2023-06-15
08:30:00','2023-06-15 08:30:00','2023-06-15 08:30:00.123','2023-06-15
08:30:00.123456','ab [...]
"""
}
diff --git
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_charset_and_strings.groovy
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_charset_and_strings.groovy
index dcec811ed27..f89e937f1a6 100644
---
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_charset_and_strings.groovy
+++
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_charset_and_strings.groovy
@@ -31,8 +31,8 @@ import static java.util.concurrent.TimeUnit.SECONDS
// Same themes run twice: ids 1-6 via JDBC/snapshot, ids 101-106 via binlog.
// Plus UPDATEs that switch a row's payload through binlog.
//
-// JDBC URL uses characterEncoding=utf8 to keep the driver round-trip stable;
-// gbk and latin1 columns are converted by MySQL on insert.
+// The streaming job URL omits Unicode and character encoding options to verify
+// that the MySQL defaults are added before metadata discovery and snapshot
reads.
suite("test_streaming_mysql_job_charset_and_strings",
"p0,external,mysql,external_docker,external_docker_mysql,nondatalake") {
def jobName = "test_streaming_mysql_job_charset_and_strings_name"
def currentDb = (sql "select database()")[0][0]
@@ -98,7 +98,7 @@ suite("test_streaming_mysql_job_charset_and_strings",
"p0,external,mysql,externa
sql """CREATE JOB ${jobName}
ON STREAMING
FROM MYSQL (
- "jdbc_url" =
"jdbc:mysql://${externalEnvIp}:${mysql_port}?serverTimezone=UTC&useUnicode=true&characterEncoding=utf8",
+ "jdbc_url" =
"jdbc:mysql://${externalEnvIp}:${mysql_port}?serverTimezone=UTC",
"driver_url" = "${driver_url}",
"driver_class" = "com.mysql.cj.jdbc.Driver",
"user" = "root",
diff --git
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_create_alter.groovy
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_create_alter.groovy
index a0dbd5e36ce..2e6443db633 100644
---
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_create_alter.groovy
+++
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_create_alter.groovy
@@ -334,18 +334,9 @@ suite("test_streaming_mysql_job_create_alter",
"p0,external,mysql,external_docke
test {
sql """ALTER JOB ${jobName}
FROM MYSQL (
- "jdbc_url" = "jdbc:mysql://${externalEnvIp}:${mysql_port}",
- "driver_url" = "${driver_url}",
- "driver_class" = "com.mysql.cj.jdbc.Driver",
- "user" = "root",
- "password" = "123456",
- "database" = "${mysqlDb}",
- "include_tables" = "changeTable",
- "offset" = "initial"
- )
- TO DATABASE ${currentDb} (
- "table.create.properties.replication_num" = "1"
+ "include_tables" = "changeTable"
)
+ TO DATABASE ${currentDb}
"""
exception "The include_tables property cannot be modified in ALTER
JOB"
}
@@ -354,19 +345,9 @@ suite("test_streaming_mysql_job_create_alter",
"p0,external,mysql,external_docke
test {
sql """ALTER JOB ${jobName}
FROM MYSQL (
- "jdbc_url" = "jdbc:mysql://${externalEnvIp}:${mysql_port}",
- "driver_url" = "${driver_url}",
- "driver_class" = "com.mysql.cj.jdbc.Driver",
- "user" = "root",
- "password" = "123456",
- "database" = "${mysqlDb}",
- "include_tables" = "${table1}",
- "exclude_tables" = "xxxx",
- "offset" = "initial"
- )
- TO DATABASE ${currentDb} (
- "table.create.properties.replication_num" = "1"
+ "exclude_tables" = "xxxx"
)
+ TO DATABASE ${currentDb}
"""
exception "The exclude_tables property cannot be modified in ALTER
JOB"
}
@@ -433,15 +414,7 @@ suite("test_streaming_mysql_job_create_alter",
"p0,external,mysql,external_docke
test {
sql """ALTER JOB ${jobName}
FROM MYSQL (
- "jdbc_url" = "jdbc:mysql://${externalEnvIp}:${mysql_port}",
- "driver_url" = "${driver_url}",
- "driver_class" = "com.mysql.cj.jdbc.Driver",
- "user" = "root",
- "password" = "123456",
- "database" = "${mysqlDb}",
- "include_tables" = "${table1}",
- "offset" = "initial",
- "xxx"="xxx"
+ "xxx" = "xxx"
)
TO DATABASE ${currentDb}
"""
diff --git
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_integer_boundary.groovy
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_integer_boundary.groovy
index be00ffa7389..80e29d18da0 100644
---
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_integer_boundary.groovy
+++
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_integer_boundary.groovy
@@ -22,7 +22,7 @@ import static java.util.concurrent.TimeUnit.SECONDS
// Guard MySQL integer-type boundaries that all_type does NOT cover:
// 1) UNSIGNED boundaries (esp. BIGINT UNSIGNED >= 2^63, which overflows
Java long)
-// 2) TINYINT(1) <-> BOOLEAN mapping (MySQL ambiguity not present in PG)
+// 2) TINYINT(1) remains TINYINT and preserves non-boolean values
//
// Coverage is run twice: ids 1-5 cover the snapshot path, ids 101-105 repeat
// the same boundary themes through the binlog path. Plus UPDATEs that switch
@@ -60,20 +60,20 @@ suite("test_streaming_mysql_job_integer_boundary",
"p0,external,mysql,external_d
`mediumint_u` mediumint unsigned,
`int_u` int unsigned,
`bigint_u` bigint unsigned,
- `bool_col` tinyint(1),
+ `tinyint_1` tinyint(1),
`tinyint_s` tinyint
) engine=innodb default charset=utf8;
"""
// ----- Snapshot rows: 5 boundary themes via JDBC path -----
- // tinyint_u smallint_u
mediumint_u int_u bigint_u bool tinyint_s
- sql """insert into ${mysqlDb}.${table1} values (1, 'all_zero',
0, 0, 0, 0, 0,
0, 0)"""
- sql """insert into ${mysqlDb}.${table1} values (2, 'signed_max',
127, 32767, 8388607, 2147483647, 9223372036854775807,
1, 127)"""
+ // tinyint_u smallint_u
mediumint_u int_u bigint_u tinyint(1) tinyint_s
+ sql """insert into ${mysqlDb}.${table1} values (1, 'all_zero',
0, 0, 0, 0, 0,
0, 0)"""
+ sql """insert into ${mysqlDb}.${table1} values (2, 'signed_max',
127, 32767, 8388607, 2147483647, 9223372036854775807,
1, 127)"""
// signed_max+1: every column passes its signed boundary; bigint_u
steps into Java long overflow range.
- sql """insert into ${mysqlDb}.${table1} values (3,
'signed_max_plus1', 128, 32768, 8388608, 2147483648,
9223372036854775808, 0, -128)"""
+ sql """insert into ${mysqlDb}.${table1} values (3,
'signed_max_plus1', 128, 32768, 8388608, 2147483648,
9223372036854775808, -1, -128)"""
// unsigned_max: each column at its unsigned upper bound. bigint_u
is 2^64-1.
- sql """insert into ${mysqlDb}.${table1} values (4, 'unsigned_max',
255, 65535, 16777215, 4294967295, 18446744073709551615,
1, -1)"""
- sql """insert into ${mysqlDb}.${table1} values (5, 'mid_value',
100, 30000, 8000000, 2000000000, 9000000000000000000,
0, 50)"""
+ sql """insert into ${mysqlDb}.${table1} values (4, 'unsigned_max',
255, 65535, 16777215, 4294967295, 18446744073709551615,
127, -1)"""
+ sql """insert into ${mysqlDb}.${table1} values (5, 'mid_value',
100, 30000, 8000000, 2000000000, 9000000000000000000,
4, 50)"""
}
sql """CREATE JOB ${jobName}
@@ -113,18 +113,18 @@ suite("test_streaming_mysql_job_integer_boundary",
"p0,external,mysql,external_d
// Verify type mapping in Doris (tinyint_u->smallint,
bigint_u->largeint, etc.)
qt_desc_integer_boundary """desc ${currentDb}.${table1};"""
- qt_select_snapshot """select id, tag, tinyint_u, smallint_u,
mediumint_u, int_u, bigint_u, bool_col, tinyint_s from ${currentDb}.${table1}
order by id;"""
+ qt_select_snapshot """select id, tag, tinyint_u, smallint_u,
mediumint_u, int_u, bigint_u, tinyint_1, tinyint_s from ${currentDb}.${table1}
order by id;"""
// ===== Binlog phase: repeat the SAME 5 boundary themes through
binlog path =====
connect("root", "123456",
"jdbc:mysql://${externalEnvIp}:${mysql_port}") {
- sql """insert into ${mysqlDb}.${table1} values (101, 'all_zero',
0, 0, 0, 0, 0,
0, 0)"""
- sql """insert into ${mysqlDb}.${table1} values (102, 'signed_max',
127, 32767, 8388607, 2147483647, 9223372036854775807,
1, 127)"""
- sql """insert into ${mysqlDb}.${table1} values (103,
'signed_max_plus1', 128, 32768, 8388608, 2147483648,
9223372036854775808, 0, -128)"""
- sql """insert into ${mysqlDb}.${table1} values (104,
'unsigned_max', 255, 65535, 16777215, 4294967295,
18446744073709551615, 1, -1)"""
- sql """insert into ${mysqlDb}.${table1} values (105, 'mid_value',
100, 30000, 8000000, 2000000000, 9000000000000000000,
0, 50)"""
+ sql """insert into ${mysqlDb}.${table1} values (101, 'all_zero',
0, 0, 0, 0, 0,
0, 0)"""
+ sql """insert into ${mysqlDb}.${table1} values (102, 'signed_max',
127, 32767, 8388607, 2147483647, 9223372036854775807,
1, 127)"""
+ sql """insert into ${mysqlDb}.${table1} values (103,
'signed_max_plus1', 128, 32768, 8388608, 2147483648,
9223372036854775808, -1, -128)"""
+ sql """insert into ${mysqlDb}.${table1} values (104,
'unsigned_max', 255, 65535, 16777215, 4294967295,
18446744073709551615, 127, -1)"""
+ sql """insert into ${mysqlDb}.${table1} values (105, 'mid_value',
100, 30000, 8000000, 2000000000, 9000000000000000000,
4, 50)"""
sql """update ${mysqlDb}.${table1} set
bigint_u=18446744073709551615 where id=1"""
- sql """update ${mysqlDb}.${table1} set bool_col=0 where id=2"""
+ sql """update ${mysqlDb}.${table1} set tinyint_1=4 where id=2"""
sql """update ${mysqlDb}.${table1} set int_u=4294967295 where
id=5"""
}
@@ -135,15 +135,16 @@ suite("test_streaming_mysql_job_integer_boundary",
"p0,external,mysql,external_d
{
def cnt = sql """select count(1) from
${currentDb}.${table1}"""
def upd1 = sql """select cast(bigint_u as string) from
${currentDb}.${table1} where id=1"""
- def upd2 = sql """select bool_col from
${currentDb}.${table1} where id=2"""
+ def upd2 = sql """select tinyint_1 from
${currentDb}.${table1} where id=2"""
def upd5 = sql """select int_u from
${currentDb}.${table1} where id=5"""
def b1 = upd1.get(0).get(0) == null ? '' :
upd1.get(0).get(0).toString()
- def b2 = upd2.get(0).get(0)
+ def t2 = upd2.get(0).get(0)
def b5 = upd5.get(0).get(0)
- log.info("incr count=" + cnt + " id1.bigint_u=" + b1 +
" id2.bool_col=" + b2 + " id5.int_u=" + b5)
+ log.info("incr count=" + cnt + " id1.bigint_u=" + b1
+ + " id2.tinyint_1=" + t2 + " id5.int_u=" + b5)
cnt.get(0).get(0) == 10 &&
b1 == '18446744073709551615' &&
- b2 != null && b2.toString() == 'false' &&
+ t2 != null && t2.toString() == '4' &&
b5 != null && b5.toString() == '4294967295'
}
)
@@ -155,7 +156,7 @@ suite("test_streaming_mysql_job_integer_boundary",
"p0,external,mysql,external_d
throw ex
}
- qt_select_after_incr """select id, tag, tinyint_u, smallint_u,
mediumint_u, int_u, bigint_u, bool_col, tinyint_s from ${currentDb}.${table1}
order by id;"""
+ qt_select_after_incr """select id, tag, tinyint_u, smallint_u,
mediumint_u, int_u, bigint_u, tinyint_1, tinyint_s from ${currentDb}.${table1}
order by id;"""
sql """DROP JOB IF EXISTS where jobname = '${jobName}'"""
diff --git
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_special_offset.groovy
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_special_offset.groovy
index 3c988239225..c4234913661 100644
---
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_special_offset.groovy
+++
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_special_offset.groovy
@@ -182,18 +182,9 @@ suite("test_streaming_mysql_job_special_offset",
"p0,external,mysql,external_doc
test {
sql """ALTER JOB ${jobName}
FROM MYSQL (
- "jdbc_url" =
"jdbc:mysql://${externalEnvIp}:${mysql_port}",
- "driver_url" = "${driver_url}",
- "driver_class" = "com.mysql.cj.jdbc.Driver",
- "user" = "root",
- "password" = "123456",
- "database" = "${mysqlDb}",
- "include_tables" = "${table1}",
"offset" = "latest"
)
- TO DATABASE ${currentDb} (
- "table.create.properties.replication_num" = "1"
- )
+ TO DATABASE ${currentDb}
"""
exception "The offset in source properties cannot be modified in
ALTER JOB"
}
diff --git
a/regression-test/suites/job_p0/streaming_job/cdc/tvf/test_cdc_stream_tvf_mysql.groovy
b/regression-test/suites/job_p0/streaming_job/cdc/tvf/test_cdc_stream_tvf_mysql.groovy
index ba6f0eacac3..744ab7539ac 100644
---
a/regression-test/suites/job_p0/streaming_job/cdc/tvf/test_cdc_stream_tvf_mysql.groovy
+++
b/regression-test/suites/job_p0/streaming_job/cdc/tvf/test_cdc_stream_tvf_mysql.groovy
@@ -146,17 +146,20 @@ suite("test_cdc_stream_tvf_mysql",
"p0,external,mysql,external_docker,external_d
sql """CREATE TABLE ${mysqlDb}.${table1} (
`name` varchar(200) NOT NULL,
`age` int DEFAULT NULL,
+ `tinyint_1` tinyint(1) DEFAULT NULL,
+ `year_zero` year DEFAULT NULL,
PRIMARY KEY (`name`)
) ENGINE=InnoDB"""
- sql """INSERT INTO ${mysqlDb}.${table1} (name, age) VALUES ('A1',
1);"""
- sql """INSERT INTO ${mysqlDb}.${table1} (name, age) VALUES ('B1',
2);"""
+ sql """INSERT INTO ${mysqlDb}.${table1} VALUES ('A1', 1, 0, 0);"""
+ sql """INSERT INTO ${mysqlDb}.${table1} VALUES ('B1', 2, 1,
2023);"""
def result = sql_return_maparray "show master status"
def file = result[0]["File"]
def position = result[0]["Position"]
offset = """{"file":"${file}","pos":"${position}"}"""
- sql """INSERT INTO ${mysqlDb}.${table1} (name, age) VALUES ('C1',
3);"""
- sql """INSERT INTO ${mysqlDb}.${table1} (name, age) VALUES ('D1',
4);"""
+ // Values 4 and YEAR 0 distinguish numeric JDBC semantics from
boolean/date semantics.
+ sql """INSERT INTO ${mysqlDb}.${table1} VALUES ('C1', 3, 4, 0);"""
+ sql """INSERT INTO ${mysqlDb}.${table1} VALUES ('D1', 4, -1,
2024);"""
// capture offset before UPDATE/DELETE events
def result2 = sql_return_maparray "show master status"
diff --git
a/regression-test/suites/job_p0/streaming_job/cdc/tvf/test_streaming_job_cdc_stream_mysql_jdbc_type_defaults.groovy
b/regression-test/suites/job_p0/streaming_job/cdc/tvf/test_streaming_job_cdc_stream_mysql_jdbc_type_defaults.groovy
new file mode 100644
index 00000000000..b246d653627
--- /dev/null
+++
b/regression-test/suites/job_p0/streaming_job/cdc/tvf/test_streaming_job_cdc_stream_mysql_jdbc_type_defaults.groovy
@@ -0,0 +1,126 @@
+// 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.
+
+
+import org.awaitility.Awaitility
+
+import static java.util.concurrent.TimeUnit.SECONDS
+
+/**
+ * Verify that a MySQL cdc_stream job uses numeric TINYINT(1) and YEAR
semantics
+ * consistently while fetching snapshot splits and scanning snapshot/binlog
rows.
+ */
+suite("test_streaming_job_cdc_stream_mysql_jdbc_type_defaults",
+ "p0,external,mysql,external_docker,external_docker_mysql,nondatalake")
{
+ def jobName = "test_cdc_stream_mysql_jdbc_defaults_job"
+ def currentDb = (sql "select database()")[0][0]
+
+ sql """DROP JOB IF EXISTS where jobname = '${jobName}'"""
+ sql """DROP TABLE IF EXISTS
${currentDb}.test_cdc_stream_mysql_jdbc_defaults_tbl FORCE"""
+
+ sql """
+ CREATE TABLE ${currentDb}.test_cdc_stream_mysql_jdbc_defaults_tbl (
+ `name` varchar(32) NULL,
+ `tinyint_value` tinyint NULL,
+ `year_value` smallint NULL
+ ) ENGINE=OLAP
+ DUPLICATE KEY(`name`)
+ DISTRIBUTED BY HASH(`name`) BUCKETS AUTO
+ PROPERTIES ("replication_allocation" = "tag.location.default: 1")
+ """
+
+ String enabled = context.config.otherConfigs.get("enableJdbcTest")
+ if (enabled != null && enabled.equalsIgnoreCase("true")) {
+ String mysqlPort = context.config.otherConfigs.get("mysql_57_port")
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String s3Endpoint = getS3Endpoint()
+ String bucket = getS3BucketName()
+ String driverUrl =
"https://${bucket}.${s3Endpoint}/regression/jdbc_driver/mysql-connector-j-8.4.0.jar";
+
+ connect("root", "123456",
"jdbc:mysql://${externalEnvIp}:${mysqlPort}") {
+ sql """CREATE DATABASE IF NOT EXISTS test_cdc_db"""
+ sql """DROP TABLE IF EXISTS
test_cdc_db.test_cdc_stream_mysql_jdbc_defaults_src"""
+ sql """
+ CREATE TABLE
test_cdc_db.test_cdc_stream_mysql_jdbc_defaults_src (
+ `name` varchar(32) NOT NULL,
+ `tinyint_value` tinyint(1) NOT NULL,
+ `year_value` year NOT NULL
+ ) ENGINE=InnoDB
+ """
+ sql """INSERT INTO
test_cdc_db.test_cdc_stream_mysql_jdbc_defaults_src VALUES ('A1', -1, 0)"""
+ sql """INSERT INTO
test_cdc_db.test_cdc_stream_mysql_jdbc_defaults_src VALUES ('B1', 0, 2024)"""
+ sql """INSERT INTO
test_cdc_db.test_cdc_stream_mysql_jdbc_defaults_src VALUES ('C1', 2, 0)"""
+ sql """INSERT INTO
test_cdc_db.test_cdc_stream_mysql_jdbc_defaults_src VALUES ('D1', 4, 2025)"""
+ }
+
+ sql """
+ CREATE JOB ${jobName}
+ ON STREAMING DO INSERT INTO
${currentDb}.test_cdc_stream_mysql_jdbc_defaults_tbl
+ (name, tinyint_value, year_value)
+ SELECT name, tinyint_value, year_value FROM cdc_stream(
+ "type" = "mysql",
+ "jdbc_url" =
"jdbc:mysql://${externalEnvIp}:${mysqlPort}",
+ "driver_url" = "${driverUrl}",
+ "driver_class" = "com.mysql.cj.jdbc.Driver",
+ "user" = "root",
+ "password" = "123456",
+ "database" = "test_cdc_db",
+ "table" =
"test_cdc_stream_mysql_jdbc_defaults_src",
+ "offset" = "initial",
+ "snapshot_split_key" = "tinyint_value",
+ "snapshot_split_size" = "1"
+ )
+ """
+
+ try {
+ Awaitility.await().atMost(300, SECONDS).pollInterval(2,
SECONDS).until({
+ def rows = sql """SELECT count(1) FROM
${currentDb}.test_cdc_stream_mysql_jdbc_defaults_tbl"""
+ log.info("snapshot rows: " + rows)
+ (rows.get(0).get(0) as int) == 4
+ })
+ } catch (Exception ex) {
+ log.info("job: " + (sql """select * from jobs("type"="insert")
where Name='${jobName}'"""))
+ log.info("tasks: " + (sql """select * from tasks("type"="insert")
where JobName='${jobName}'"""))
+ throw ex
+ }
+
+ order_qt_snapshot_data """SELECT * FROM
${currentDb}.test_cdc_stream_mysql_jdbc_defaults_tbl"""
+
+ connect("root", "123456",
"jdbc:mysql://${externalEnvIp}:${mysqlPort}") {
+ sql """INSERT INTO
test_cdc_db.test_cdc_stream_mysql_jdbc_defaults_src VALUES ('E1', 127, 0)"""
+ sql """INSERT INTO
test_cdc_db.test_cdc_stream_mysql_jdbc_defaults_src VALUES ('F1', -128, 2026)"""
+ }
+
+ try {
+ Awaitility.await().atMost(120, SECONDS).pollInterval(2,
SECONDS).until({
+ def rows = sql """
+ SELECT count(1) FROM
${currentDb}.test_cdc_stream_mysql_jdbc_defaults_tbl
+ WHERE name IN ('E1', 'F1')
+ """
+ log.info("incremental rows: " + rows)
+ (rows.get(0).get(0) as int) == 2
+ })
+ } catch (Exception ex) {
+ log.info("job: " + (sql """select * from jobs("type"="insert")
where Name='${jobName}'"""))
+ log.info("tasks: " + (sql """select * from tasks("type"="insert")
where JobName='${jobName}'"""))
+ throw ex
+ }
+
+ order_qt_final_data """SELECT * FROM
${currentDb}.test_cdc_stream_mysql_jdbc_defaults_tbl"""
+ sql """DROP JOB IF EXISTS where jobname = '${jobName}'"""
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
