Copilot commented on code in PR #65647: URL: https://github.com/apache/doris/pull/65647#discussion_r3586077744
########## fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJdbcUrlNormalizer.java: ########## @@ -0,0 +1,82 @@ +// 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; + +/** + * 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(" ", ""); + normalizedUrl = forceBooleanParam( + normalizedUrl, "yearIsDateType", "true", "false"); + normalizedUrl = forceBooleanParam( + normalizedUrl, "tinyInt1isBit", "true", "false"); + normalizedUrl = forceBooleanParam( + normalizedUrl, "useUnicode", "false", "true"); + return setParam(normalizedUrl, "characterEncoding", "utf-8"); + } + + private static String forceBooleanParam(String jdbcUrl, String param, + String unexpectedValue, String expectedValue) { + String unexpectedParam = param + "=" + unexpectedValue; + String expectedParam = param + "=" + expectedValue; + if (jdbcUrl.contains(expectedParam)) { + return jdbcUrl; + } + if (jdbcUrl.contains(unexpectedParam)) { + return jdbcUrl.replace(unexpectedParam, expectedParam); + } + return appendParam(jdbcUrl, expectedParam); + } + + private static String setParam(String jdbcUrl, String param, String value) { + String expectedParam = param + "=" + value; + if (jdbcUrl.contains(expectedParam)) { + return jdbcUrl; + } + return appendParam(jdbcUrl, expectedParam); + } + + private static String appendParam(String jdbcUrl, String param) { + String delimiter = jdbcUrl.contains("?") ? "&" : "?"; + if (!jdbcUrl.endsWith(delimiter)) { + jdbcUrl += delimiter; + } + return jdbcUrl + param; + } Review Comment: appendParam() can produce a malformed-looking query prefix when the URL already ends with '?' (e.g. "jdbc:mysql://host/db?" becomes "...?&yearIsDateType=false"). It's safer to treat both '?' and '&' as valid trailing delimiters and avoid injecting an extra '&' after '?'. ########## fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingJdbcUrlNormalizerTest.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.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 testNormalizeMysqlJdbcUrlOverridesOppositeValues() { + String jdbcUrl = StreamingJdbcUrlNormalizer.normalize(DataSourceType.MYSQL, + "jdbc:mysql://127.0.0.1:3306/test?tinyInt1isBit=true" + + "&yearIsDateType=true&useUnicode=false&characterEncoding=GBK"); + + Assert.assertTrue(jdbcUrl.contains("tinyInt1isBit=false")); + Assert.assertTrue(jdbcUrl.contains("yearIsDateType=false")); + Assert.assertTrue(jdbcUrl.contains("useUnicode=true")); + Assert.assertTrue(jdbcUrl.contains("characterEncoding=GBK&characterEncoding=utf-8")); + Assert.assertFalse(jdbcUrl.contains("tinyInt1isBit=true")); + Assert.assertFalse(jdbcUrl.contains("yearIsDateType=true")); + Assert.assertFalse(jdbcUrl.contains("useUnicode=false")); + Assert.assertFalse(jdbcUrl.contains("rewriteBatchedStatements")); Review Comment: The test currently asserts that the normalized URL contains duplicate/conflicting characterEncoding parameters ("characterEncoding=GBK&characterEncoding=utf-8"), which bakes in ambiguous behavior and doesn't actually validate that the normalizer enforces utf-8. Once the normalizer is fixed to replace/remove existing characterEncoding, update this test to assert that GBK is not present and utf-8 is present exactly once. ########## fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJdbcUrlNormalizer.java: ########## @@ -0,0 +1,82 @@ +// 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; + +/** + * 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(" ", ""); + normalizedUrl = forceBooleanParam( + normalizedUrl, "yearIsDateType", "true", "false"); + normalizedUrl = forceBooleanParam( + normalizedUrl, "tinyInt1isBit", "true", "false"); + normalizedUrl = forceBooleanParam( + normalizedUrl, "useUnicode", "false", "true"); + return setParam(normalizedUrl, "characterEncoding", "utf-8"); + } + + private static String forceBooleanParam(String jdbcUrl, String param, + String unexpectedValue, String expectedValue) { + String unexpectedParam = param + "=" + unexpectedValue; + String expectedParam = param + "=" + expectedValue; + if (jdbcUrl.contains(expectedParam)) { + return jdbcUrl; + } + if (jdbcUrl.contains(unexpectedParam)) { + return jdbcUrl.replace(unexpectedParam, expectedParam); + } + return appendParam(jdbcUrl, expectedParam); + } + + private static String setParam(String jdbcUrl, String param, String value) { + String expectedParam = param + "=" + value; + if (jdbcUrl.contains(expectedParam)) { + return jdbcUrl; + } + return appendParam(jdbcUrl, expectedParam); + } Review Comment: setParam() does not override an existing parameter with a different value; it appends a second copy instead (e.g. "characterEncoding=GBK&characterEncoding=utf-8"). This makes the normalization result ambiguous and can fail to enforce the intended encoding if the driver honors the first occurrence. The normalizer should replace/remove any existing occurrences of the key and emit exactly one "characterEncoding=utf-8". ########## fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingJdbcUrlNormalizer.java: ########## @@ -0,0 +1,82 @@ +// 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; + +/** + * 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(" ", ""); + normalizedUrl = forceBooleanParam( + normalizedUrl, "yearIsDateType", "true", "false"); + normalizedUrl = forceBooleanParam( + normalizedUrl, "tinyInt1isBit", "true", "false"); + normalizedUrl = forceBooleanParam( + normalizedUrl, "useUnicode", "false", "true"); + return setParam(normalizedUrl, "characterEncoding", "utf-8"); + } + + private static String forceBooleanParam(String jdbcUrl, String param, + String unexpectedValue, String expectedValue) { + String unexpectedParam = param + "=" + unexpectedValue; + String expectedParam = param + "=" + expectedValue; + if (jdbcUrl.contains(expectedParam)) { + return jdbcUrl; + } + if (jdbcUrl.contains(unexpectedParam)) { + return jdbcUrl.replace(unexpectedParam, expectedParam); + } + return appendParam(jdbcUrl, expectedParam); + } Review Comment: forceBooleanParam() returns early if the URL already contains the expected value, even if a conflicting duplicate with the unexpected value is also present (e.g. "...tinyInt1isBit=false&tinyInt1isBit=true"). In that case the normalizer does not actually enforce a single effective value, and Connector/J may honor the conflicting entry depending on how it parses duplicates. Consider parsing the query string and removing all existing occurrences of the key before adding the desired "key=value" once (boundary-aware on ?/&). -- 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]
