This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new e9c446638c [Fix][API] Fix backward compatibility issue in
CatalogFactory optionRule validation (#11165)
e9c446638c is described below
commit e9c446638c55fa16fcc0bc03de8eb0eff0ae26c8
Author: zhiwei.niu <[email protected]>
AuthorDate: Thu Jun 25 20:44:26 2026 +0800
[Fix][API] Fix backward compatibility issue in CatalogFactory optionRule
validation (#11165)
Co-authored-by: Daniel <[email protected]>
---
.../introduction/concepts/incompatible-changes.md | 7 ++
.../introduction/concepts/incompatible-changes.md | 7 ++
.../SqlServerIncrementalSourceFactoryTest.java | 33 ++++++
.../jdbc/catalog/dm/DamengCatalogFactory.java | 2 +-
.../jdbc/catalog/duckdb/DuckDBCatalogFactory.java | 16 ++-
.../jdbc/catalog/highgo/HighGoCatalogFactory.java | 2 +-
.../jdbc/catalog/iris/IrisCatalogFactory.java | 2 +-
.../catalog/kingbase/KingbaseCatalogFactory.java | 2 +-
.../jdbc/catalog/mysql/MySqlCatalogFactory.java | 2 +-
.../catalog/opengauss/OpenGaussCatalogFactory.java | 2 +-
.../jdbc/catalog/oracle/OracleCatalogFactory.java | 31 +++++-
.../jdbc/catalog/psql/PostgresCatalogFactory.java | 2 +-
.../catalog/redshift/RedshiftCatalogFactory.java | 2 +-
.../catalog/saphana/SapHanaCatalogFactory.java | 30 ++++-
.../jdbc/catalog/saphana/SapHanaURLParser.java | 3 +-
.../catalog/sqlserver/SqlServerCatalogFactory.java | 31 +++++-
.../jdbc/catalog/tidb/TiDBCatalogFactory.java | 2 +-
.../jdbc/catalog/xugu/XuguCatalogFactory.java | 5 +-
.../seatunnel/jdbc/config/JdbcCommonOptions.java | 50 +++++----
.../jdbc/catalog/JdbcCatalogFactoryTest.java | 122 +++++++++++++++++++--
.../lance/catalog/LanceCatalogFactory.java | 4 +
21 files changed, 301 insertions(+), 56 deletions(-)
diff --git a/docs/en/introduction/concepts/incompatible-changes.md
b/docs/en/introduction/concepts/incompatible-changes.md
index e5db071e74..6ea2dc2ac1 100644
--- a/docs/en/introduction/concepts/incompatible-changes.md
+++ b/docs/en/introduction/concepts/incompatible-changes.md
@@ -61,6 +61,13 @@ You need to check this document before you upgrade to
related version.
### Configuration Changes
+- **Breaking Change: CatalogFactory creation path now validates
`optionRule()`**
+ - **Affected component**: `seatunnel-api` —
`FactoryUtil.createOptionalCatalog()`
+ - **Description**: The `FactoryUtil.createOptionalCatalog()` method now
calls `ConfigValidator.validate(catalogFactory.optionRule())` before creating a
catalog instance. Previously, no validation was performed on the catalog
factory's option rules during catalog creation.
+ - **Impact**: Catalog factories whose `optionRule()` declares options as
`required` that are not always present in the config passed to
`createOptionalCatalog()` will now throw `OptionValidationException`. This
primarily affects the JDBC connector path via `JdbcCatalogUtils.findCatalog()`.
+ - **Migration Guide**: If you have a custom `CatalogFactory` implementation,
ensure that its `optionRule()` accurately reflects which options are truly
mandatory vs optional in the config that reaches it at runtime.
+
+
### Connector Changes
- **Breaking Change: Iceberg Connector — source table primary key is no longer
silently inherited**
diff --git a/docs/zh/introduction/concepts/incompatible-changes.md
b/docs/zh/introduction/concepts/incompatible-changes.md
index 97907ef653..8a6d7b0a51 100644
--- a/docs/zh/introduction/concepts/incompatible-changes.md
+++ b/docs/zh/introduction/concepts/incompatible-changes.md
@@ -60,6 +60,13 @@
### 配置变更
+- **破坏性变更：CatalogFactory 创建路径现在会校验 `optionRule()`**
+ - **影响范围**：`seatunnel-api` — `FactoryUtil.createOptionalCatalog()`
+ - **变更说明**：`FactoryUtil.createOptionalCatalog()` 方法现在在创建 catalog 实例之前会调用
`ConfigValidator.validate(catalogFactory.optionRule())` 进行校验。此前，catalog 创建路径不会对
catalog factory 的 option rules 执行任何校验。
+ - **影响**：如果 catalog factory 的 `optionRule()` 将某些选项声明为 `required`，而传入
`createOptionalCatalog()` 的配置中这些选项并不总是存在，则会抛出
`OptionValidationException`。这主要影响通过 `JdbcCatalogUtils.findCatalog()` 触发的 JDBC
连接器路径。
+ - **迁移指南**：如果您有自定义的 `CatalogFactory` 实现，请确保其 `optionRule()`
准确反映在运行时到达它的配置中，哪些选项是真正必填的，哪些是可选的。
+
+
### 连接器变更
- **破坏性变更：Iceberg 连接器 — 不再自动继承源表主键**
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/SqlServerIncrementalSourceFactoryTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/SqlServerIncrementalSourceFactoryTest.java
index af86c4fc11..e814f76ee5 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/SqlServerIncrementalSourceFactoryTest.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/SqlServerIncrementalSourceFactoryTest.java
@@ -17,12 +17,45 @@
package org.apache.seatunnel.connectors.seatunnel.cdc.sqlserver.source;
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.table.catalog.Catalog;
+import org.apache.seatunnel.api.table.factory.FactoryUtil;
+
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
class SqlServerIncrementalSourceFactoryTest {
@Test
public void testOptionRule() {
Assertions.assertNotNull((new
SqlServerIncrementalSourceFactory()).optionRule());
}
+
+ /**
+ * SQLServer CDC source creation must accept the driver-specific
databaseName URL syntax during
+ * the submission-time catalog validation step.
+ */
+ @Test
+ public void testCreateOptionalCatalogWithSqlServerStyleUrl() {
+ Map<String, Object> config = new HashMap<>();
+ config.put("url",
"jdbc:sqlserver://localhost:1433;databaseName=seatunnel");
+ config.put("username", "sa");
+ config.put("password", "Password!");
+ config.put("database-names", Arrays.asList("seatunnel"));
+ config.put("table-names", Arrays.asList("seatunnel.dbo.orders"));
+
+ Optional<Catalog> catalog =
+ FactoryUtil.createOptionalCatalog(
+ "SqlServer",
+ ReadonlyConfig.fromMap(config),
+ Thread.currentThread().getContextClassLoader(),
+ "SqlServer");
+
+ Assertions.assertTrue(catalog.isPresent());
+ catalog.ifPresent(Catalog::close);
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/dm/DamengCatalogFactory.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/dm/DamengCatalogFactory.java
index ead206a230..5494d1b147 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/dm/DamengCatalogFactory.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/dm/DamengCatalogFactory.java
@@ -51,6 +51,6 @@ public class DamengCatalogFactory implements CatalogFactory {
@Override
public OptionRule optionRule() {
- return JdbcCommonOptions.BASE_CATALOG_RULE.build();
+ return JdbcCommonOptions.baseCatalogRule().build();
}
}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/duckdb/DuckDBCatalogFactory.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/duckdb/DuckDBCatalogFactory.java
index 53999ace8f..b6232fe2c5 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/duckdb/DuckDBCatalogFactory.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/duckdb/DuckDBCatalogFactory.java
@@ -28,13 +28,6 @@ import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.DatabaseI
import com.google.auto.service.AutoService;
-import static
org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcCommonOptions.DECIMAL_TYPE_NARROWING;
-import static
org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcCommonOptions.HANDLE_BLOB_AS_STRING;
-import static
org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcCommonOptions.PASSWORD;
-import static
org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcCommonOptions.SCHEMA;
-import static
org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcCommonOptions.URL;
-import static
org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcCommonOptions.USERNAME;
-
/** Factory for {@link DuckDBCatalog} */
@AutoService(Factory.class)
public class DuckDBCatalogFactory implements CatalogFactory {
@@ -58,8 +51,13 @@ public class DuckDBCatalogFactory implements CatalogFactory {
@Override
public OptionRule optionRule() {
return OptionRule.builder()
- .required(URL)
- .optional(USERNAME, PASSWORD, SCHEMA, DECIMAL_TYPE_NARROWING,
HANDLE_BLOB_AS_STRING)
+ .required(JdbcCommonOptions.URL)
+ .optional(
+ JdbcCommonOptions.USERNAME,
+ JdbcCommonOptions.PASSWORD,
+ JdbcCommonOptions.SCHEMA,
+ JdbcCommonOptions.DECIMAL_TYPE_NARROWING,
+ JdbcCommonOptions.HANDLE_BLOB_AS_STRING)
.build();
}
}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/highgo/HighGoCatalogFactory.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/highgo/HighGoCatalogFactory.java
index e612541075..caa8e59b10 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/highgo/HighGoCatalogFactory.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/highgo/HighGoCatalogFactory.java
@@ -45,6 +45,6 @@ public class HighGoCatalogFactory implements CatalogFactory {
@Override
public OptionRule optionRule() {
- return JdbcCommonOptions.BASE_CATALOG_RULE.build();
+ return JdbcCommonOptions.baseCatalogRule().build();
}
}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/iris/IrisCatalogFactory.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/iris/IrisCatalogFactory.java
index 33ed340f27..e951b07695 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/iris/IrisCatalogFactory.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/iris/IrisCatalogFactory.java
@@ -50,6 +50,6 @@ public class IrisCatalogFactory implements CatalogFactory {
@Override
public OptionRule optionRule() {
- return JdbcCommonOptions.BASE_CATALOG_RULE.build();
+ return JdbcCommonOptions.baseCatalogRule().build();
}
}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/kingbase/KingbaseCatalogFactory.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/kingbase/KingbaseCatalogFactory.java
index b535ce5f50..9453115d13 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/kingbase/KingbaseCatalogFactory.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/kingbase/KingbaseCatalogFactory.java
@@ -51,6 +51,6 @@ public class KingbaseCatalogFactory implements CatalogFactory
{
@Override
public OptionRule optionRule() {
- return JdbcCommonOptions.BASE_CATALOG_RULE.build();
+ return JdbcCommonOptions.baseCatalogRule().build();
}
}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/mysql/MySqlCatalogFactory.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/mysql/MySqlCatalogFactory.java
index 052f8763ab..537b8c768a 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/mysql/MySqlCatalogFactory.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/mysql/MySqlCatalogFactory.java
@@ -51,6 +51,6 @@ public class MySqlCatalogFactory implements CatalogFactory {
@Override
public OptionRule optionRule() {
- return JdbcCommonOptions.BASE_CATALOG_RULE.build();
+ return JdbcCommonOptions.baseCatalogRule().build();
}
}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/opengauss/OpenGaussCatalogFactory.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/opengauss/OpenGaussCatalogFactory.java
index c80b163e74..bffb1800bf 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/opengauss/OpenGaussCatalogFactory.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/opengauss/OpenGaussCatalogFactory.java
@@ -51,6 +51,6 @@ public class OpenGaussCatalogFactory implements
CatalogFactory {
@Override
public OptionRule optionRule() {
- return JdbcCommonOptions.BASE_CATALOG_RULE.build();
+ return JdbcCommonOptions.baseCatalogRule().build();
}
}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/oracle/OracleCatalogFactory.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/oracle/OracleCatalogFactory.java
index 782130bd96..bd2794a751 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/oracle/OracleCatalogFactory.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/oracle/OracleCatalogFactory.java
@@ -17,8 +17,12 @@
package org.apache.seatunnel.connectors.seatunnel.jdbc.catalog.oracle;
+import org.apache.seatunnel.shade.org.apache.commons.lang3.StringUtils;
+
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.ConditionExtension;
import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.configuration.util.OptionValidationException;
import org.apache.seatunnel.api.table.catalog.Catalog;
import org.apache.seatunnel.api.table.factory.CatalogFactory;
import org.apache.seatunnel.api.table.factory.Factory;
@@ -53,6 +57,31 @@ public class OracleCatalogFactory implements CatalogFactory {
@Override
public OptionRule optionRule() {
- return JdbcCommonOptions.BASE_CATALOG_RULE.build();
+ return JdbcCommonOptions.baseCatalogRule(new
OracleUrlValidator()).build();
+ }
+
+ static class OracleUrlValidator implements ConditionExtension<String> {
+ @Override
+ public String description() {
+ return "Oracle JDBC URL must contain a service name (e.g.
jdbc:oracle:thin:@host:port/service)";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, String url) {
+ if (url == null || url.trim().isEmpty()) {
+ return false;
+ }
+ try {
+ JdbcUrlUtil.UrlInfo info = OracleURLParser.parse(url);
+ return StringUtils.isNotBlank(info.getHost())
+ && info.getDefaultDatabase().isPresent();
+ } catch (IllegalArgumentException e) {
+ throw new OptionValidationException(
+ String.format(
+ "Invalid Oracle JDBC URL format: [%s], "
+ + "expected pattern:
jdbc:oracle:thin:@host:port/service",
+ url));
+ }
+ }
}
}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/psql/PostgresCatalogFactory.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/psql/PostgresCatalogFactory.java
index 7217432eb6..3215eace02 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/psql/PostgresCatalogFactory.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/psql/PostgresCatalogFactory.java
@@ -51,6 +51,6 @@ public class PostgresCatalogFactory implements CatalogFactory
{
@Override
public OptionRule optionRule() {
- return JdbcCommonOptions.BASE_CATALOG_RULE.build();
+ return JdbcCommonOptions.baseCatalogRule().build();
}
}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/redshift/RedshiftCatalogFactory.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/redshift/RedshiftCatalogFactory.java
index 7931ef3826..728bc48977 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/redshift/RedshiftCatalogFactory.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/redshift/RedshiftCatalogFactory.java
@@ -51,6 +51,6 @@ public class RedshiftCatalogFactory implements CatalogFactory
{
@Override
public OptionRule optionRule() {
- return JdbcCommonOptions.BASE_CATALOG_RULE.build();
+ return JdbcCommonOptions.baseCatalogRule().build();
}
}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/saphana/SapHanaCatalogFactory.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/saphana/SapHanaCatalogFactory.java
index ba7dd361b6..0e1f73abae 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/saphana/SapHanaCatalogFactory.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/saphana/SapHanaCatalogFactory.java
@@ -17,8 +17,12 @@
package org.apache.seatunnel.connectors.seatunnel.jdbc.catalog.saphana;
+import org.apache.seatunnel.shade.org.apache.commons.lang3.StringUtils;
+
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.ConditionExtension;
import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.configuration.util.OptionValidationException;
import org.apache.seatunnel.api.table.catalog.Catalog;
import org.apache.seatunnel.api.table.factory.CatalogFactory;
import org.apache.seatunnel.api.table.factory.Factory;
@@ -51,6 +55,30 @@ public class SapHanaCatalogFactory implements CatalogFactory
{
@Override
public OptionRule optionRule() {
- return JdbcCommonOptions.BASE_CATALOG_RULE.build();
+ return JdbcCommonOptions.baseCatalogRule(new
SapHanaUrlValidator()).build();
+ }
+
+ static class SapHanaUrlValidator implements ConditionExtension<String> {
+ @Override
+ public String description() {
+ return "SAP HANA JDBC URL must be a valid format (e.g.
jdbc:sap://host:port)";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, String url) {
+ if (url == null || url.trim().isEmpty()) {
+ return false;
+ }
+ try {
+ JdbcUrlUtil.UrlInfo info = SapHanaURLParser.parse(url);
+ return StringUtils.isNotBlank(info.getHost());
+ } catch (IllegalArgumentException e) {
+ throw new OptionValidationException(
+ String.format(
+ "Invalid SAP HANA JDBC URL format: [%s], "
+ + "expected pattern:
jdbc:sap://host:port",
+ url));
+ }
+ }
}
}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/saphana/SapHanaURLParser.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/saphana/SapHanaURLParser.java
index 6c6e66fbfe..704947d481 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/saphana/SapHanaURLParser.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/saphana/SapHanaURLParser.java
@@ -25,7 +25,8 @@ import java.util.regex.Pattern;
public class SapHanaURLParser {
private static final Pattern HANA_URL_PATTERN =
-
Pattern.compile("^(?<url>jdbc:sap://(?<host>[^:]+):(?<port>\\d+)/\\?(?<params>.*?))$");
+ Pattern.compile(
+
"^(?<url>jdbc:sap://(?<host>[^:]+):(?<port>\\d+)(/\\?(?<params>.*?))?)$");
public static JdbcUrlUtil.UrlInfo parse(String url) {
Matcher matcher = HANA_URL_PATTERN.matcher(url);
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/sqlserver/SqlServerCatalogFactory.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/sqlserver/SqlServerCatalogFactory.java
index aa345997d2..0624347c3e 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/sqlserver/SqlServerCatalogFactory.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/sqlserver/SqlServerCatalogFactory.java
@@ -17,8 +17,12 @@
package org.apache.seatunnel.connectors.seatunnel.jdbc.catalog.sqlserver;
+import org.apache.seatunnel.shade.org.apache.commons.lang3.StringUtils;
+
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.ConditionExtension;
import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.configuration.util.OptionValidationException;
import org.apache.seatunnel.api.table.catalog.Catalog;
import org.apache.seatunnel.api.table.factory.CatalogFactory;
import org.apache.seatunnel.api.table.factory.Factory;
@@ -51,6 +55,31 @@ public class SqlServerCatalogFactory implements
CatalogFactory {
@Override
public OptionRule optionRule() {
- return JdbcCommonOptions.BASE_CATALOG_RULE.build();
+ return JdbcCommonOptions.baseCatalogRule(new
SqlServerUrlValidator()).build();
+ }
+
+ /** Validates URL format only; database may be provided via separate
config option. */
+ static class SqlServerUrlValidator implements ConditionExtension<String> {
+ @Override
+ public String description() {
+ return "SqlServer JDBC URL must be a valid format (e.g.
jdbc:sqlserver://host:port;databaseName=db)";
+ }
+
+ @Override
+ public boolean evaluate(ReadonlyConfig config, String url) {
+ if (url == null || url.trim().isEmpty()) {
+ return false;
+ }
+ try {
+ JdbcUrlUtil.UrlInfo info = SqlServerURLParser.parse(url);
+ return info != null && StringUtils.isNotBlank(info.getHost());
+ } catch (IllegalArgumentException e) {
+ throw new OptionValidationException(
+ String.format(
+ "Invalid SqlServer JDBC URL format: [%s], "
+ + "expected pattern:
jdbc:sqlserver://host:port[;databaseName=db]",
+ url));
+ }
+ }
}
}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/tidb/TiDBCatalogFactory.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/tidb/TiDBCatalogFactory.java
index beb2945b87..153062851e 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/tidb/TiDBCatalogFactory.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/tidb/TiDBCatalogFactory.java
@@ -50,6 +50,6 @@ public class TiDBCatalogFactory implements CatalogFactory {
@Override
public OptionRule optionRule() {
- return JdbcCommonOptions.BASE_CATALOG_RULE.build();
+ return JdbcCommonOptions.baseCatalogRule().build();
}
}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/xugu/XuguCatalogFactory.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/xugu/XuguCatalogFactory.java
index 63f35bfd35..cd239e7e9b 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/xugu/XuguCatalogFactory.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/xugu/XuguCatalogFactory.java
@@ -23,7 +23,6 @@ import org.apache.seatunnel.api.table.catalog.Catalog;
import org.apache.seatunnel.api.table.factory.CatalogFactory;
import org.apache.seatunnel.api.table.factory.Factory;
import org.apache.seatunnel.common.utils.JdbcUrlUtil;
-import
org.apache.seatunnel.connectors.seatunnel.jdbc.catalog.oracle.OracleURLParser;
import org.apache.seatunnel.connectors.seatunnel.jdbc.config.JdbcCommonOptions;
import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.DatabaseIdentifier;
@@ -40,7 +39,7 @@ public class XuguCatalogFactory implements CatalogFactory {
@Override
public Catalog createCatalog(String catalogName, ReadonlyConfig options) {
String urlWithDatabase = options.get(JdbcCommonOptions.URL);
- JdbcUrlUtil.UrlInfo urlInfo = OracleURLParser.parse(urlWithDatabase);
+ JdbcUrlUtil.UrlInfo urlInfo = JdbcUrlUtil.getUrlInfo(urlWithDatabase);
return new XuguCatalog(
catalogName,
options.get(JdbcCommonOptions.USERNAME),
@@ -52,6 +51,6 @@ public class XuguCatalogFactory implements CatalogFactory {
@Override
public OptionRule optionRule() {
- return JdbcCommonOptions.BASE_CATALOG_RULE.build();
+ return JdbcCommonOptions.baseCatalogRule().build();
}
}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcCommonOptions.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcCommonOptions.java
index 2c636be5c4..bff276da22 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcCommonOptions.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/config/JdbcCommonOptions.java
@@ -17,12 +17,15 @@
package org.apache.seatunnel.connectors.seatunnel.jdbc.config;
+import org.apache.seatunnel.shade.org.apache.commons.lang3.StringUtils;
+
import org.apache.seatunnel.api.configuration.Option;
import org.apache.seatunnel.api.configuration.Options;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import org.apache.seatunnel.api.configuration.util.ConditionExtension;
import org.apache.seatunnel.api.configuration.util.Conditions;
import org.apache.seatunnel.api.configuration.util.OptionRule;
+import org.apache.seatunnel.api.configuration.util.OptionValidationException;
import org.apache.seatunnel.common.utils.JdbcUrlUtil;
import java.util.Map;
@@ -162,38 +165,42 @@ public class JdbcCommonOptions {
public static final Option<String> REGION =
Options.key("region").stringType().noDefaultValue().withDescription("region");
- /** @deprecated Use {@link #baseCatalogRule()} instead to avoid shared
mutable state. */
- @Deprecated public static final OptionRule.Builder BASE_CATALOG_RULE =
baseCatalogRule();
-
/**
- * Returns a fresh {@link OptionRule.Builder} with the base validation
rules shared by all JDBC
- * catalog factories (MySQL, PostgreSQL, Oracle, etc.).
+ * Returns a fresh {@link OptionRule.Builder} with the base validation
rules shared by standard
+ * JDBC catalog factories (MySQL, PostgreSQL, etc.) that use generic
{@code host:port/database}
+ * URL format and require username/password authentication.
*
- * <p>These rules are evaluated at <b>submission time</b> via {@code
- * ConfigValidator.validate(factory.optionRule())} in the {@code
FactoryUtil} entry path. They
- * enforce that the JDBC URL contains a database name, and that
username/password are provided.
+ * <p>Catalog factories with non-standard URL formats (SqlServer, Oracle,
SapHana) should use
+ * {@link #baseCatalogRule(ConditionExtension)} with their own URL
validator.
*
- * <p>Individual catalog factories may append additional rules (e.g.
OceanBase requires {@code
- * compatible_mode}) before calling {@code .build()}.
+ * <p>Catalog factories that do not require authentication (e.g. DuckDB)
should define their own
+ * {@code optionRule()} directly.
*/
public static OptionRule.Builder baseCatalogRule() {
+ return baseCatalogRule(new UrlContainsDatabaseValidator());
+ }
+
+ /**
+ * Returns a fresh {@link OptionRule.Builder} with a custom URL validator.
Use this for
+ * databases whose JDBC URL does not follow the standard {@code
host:port/database} format (e.g.
+ * SqlServer, Oracle, SapHana).
+ */
+ public static OptionRule.Builder
baseCatalogRule(ConditionExtension<String> urlValidator) {
return OptionRule.builder()
- .required(URL, Conditions.extension(URL, new
UrlContainsDatabaseValidator()))
+ .required(URL, Conditions.extension(URL, urlValidator))
.required(USERNAME, PASSWORD)
.optional(SCHEMA, DECIMAL_TYPE_NARROWING,
HANDLE_BLOB_AS_STRING);
}
/**
- * Submission-time validator that ensures the JDBC URL contains a database
name.
- *
- * <p>This validator is attached to the {@code url} option via {@link
- * Conditions#extension(Option, ConditionExtension)} and is evaluated by
{@code ConfigValidator}
- * before the catalog/source/sink factory creates its connector instance.
+ * Validates that the JDBC URL has a valid format with at least a host
component. Database name
+ * is optional to maintain backward compatibility with connectors (e.g.
StarRocks, Doris) that
+ * specify the database in the query or table_path instead of the URL.
*/
public static class UrlContainsDatabaseValidator implements
ConditionExtension<String> {
@Override
public String description() {
- return "JDBC URL must contain a database name";
+ return "JDBC URL must be a valid format:
jdbc:<scheme>://host:port[/database]";
}
@Override
@@ -202,9 +209,14 @@ public class JdbcCommonOptions {
return false;
}
try {
- return
JdbcUrlUtil.getUrlInfo(url).getDefaultDatabase().isPresent();
+ JdbcUrlUtil.UrlInfo urlInfo = JdbcUrlUtil.getUrlInfo(url);
+ return StringUtils.isNotBlank(urlInfo.getHost());
} catch (IllegalArgumentException e) {
- return false;
+ throw new OptionValidationException(
+ String.format(
+ "Invalid JDBC URL format: [%s], "
+ + "expected pattern:
jdbc:<scheme>://host:port[/database]",
+ url));
}
}
}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/JdbcCatalogFactoryTest.java
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/JdbcCatalogFactoryTest.java
index 5aadf30058..2b827e3248 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/JdbcCatalogFactoryTest.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/JdbcCatalogFactoryTest.java
@@ -21,9 +21,14 @@ import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import org.apache.seatunnel.api.configuration.util.ConfigValidator;
import org.apache.seatunnel.api.configuration.util.OptionRule;
import org.apache.seatunnel.api.configuration.util.OptionValidationException;
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.catalog.dm.DamengCatalogFactory;
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.catalog.duckdb.DuckDBCatalogFactory;
import
org.apache.seatunnel.connectors.seatunnel.jdbc.catalog.mysql.MySqlCatalogFactory;
import
org.apache.seatunnel.connectors.seatunnel.jdbc.catalog.oceanbase.OceanBaseCatalogFactory;
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.catalog.oracle.OracleCatalogFactory;
import
org.apache.seatunnel.connectors.seatunnel.jdbc.catalog.psql.PostgresCatalogFactory;
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.catalog.saphana.SapHanaCatalogFactory;
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.catalog.sqlserver.SqlServerCatalogFactory;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -40,6 +45,8 @@ class JdbcCatalogFactoryTest {
ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule);
}
+ // ==================== Standard URL validators (MySQL / Postgres)
====================
+
@Test
void testValidCatalogConfig() {
Map<String, Object> cfg = new HashMap<>();
@@ -68,39 +75,42 @@ class JdbcCatalogFactoryTest {
}
@Test
- void testUrlWithoutDatabaseFails() {
+ void testBlankUrlFails() {
Map<String, Object> cfg = new HashMap<>();
- cfg.put("url", "jdbc:mysql://host:3306");
+ cfg.put("url", "");
cfg.put("username", "root");
cfg.put("password", "pass");
Assertions.assertThrows(OptionValidationException.class, () ->
validate(mysqlRule, cfg));
}
@Test
- void testBlankUrlFails() {
+ void testMissingCredentialsFails() {
Map<String, Object> cfg = new HashMap<>();
- cfg.put("url", "");
- cfg.put("username", "root");
- cfg.put("password", "pass");
+ cfg.put("url", "jdbc:mysql://localhost:3306/mydb");
Assertions.assertThrows(OptionValidationException.class, () ->
validate(mysqlRule, cfg));
}
@Test
- void testMissingUsernameFails() {
+ void testMissingPasswordFails() {
Map<String, Object> cfg = new HashMap<>();
- cfg.put("url", "jdbc:mysql://host:3306/mydb");
- cfg.put("password", "pass");
+ cfg.put("url", "jdbc:mysql://localhost:3306/mydb");
+ cfg.put("username", "root");
Assertions.assertThrows(OptionValidationException.class, () ->
validate(mysqlRule, cfg));
}
@Test
- void testMissingPasswordFails() {
+ void testCatalogConfigMimicsExtractCatalogConfig() {
Map<String, Object> cfg = new HashMap<>();
- cfg.put("url", "jdbc:mysql://host:3306/mydb");
+ cfg.put("url", "jdbc:mysql://localhost:3306/mydb");
cfg.put("username", "root");
- Assertions.assertThrows(OptionValidationException.class, () ->
validate(mysqlRule, cfg));
+ cfg.put("password", "pass");
+ cfg.put("decimal_type_narrowing", true);
+ cfg.put("handle_blob_as_string", false);
+ Assertions.assertDoesNotThrow(() -> validate(mysqlRule, cfg));
}
+ // ==================== OceanBase ====================
+
@Test
void testOceanBaseWithoutCompatibleModeFails() {
OptionRule obRule = new OceanBaseCatalogFactory().optionRule();
@@ -110,4 +120,92 @@ class JdbcCatalogFactoryTest {
cfg.put("password", "pass");
Assertions.assertThrows(OptionValidationException.class, () ->
validate(obRule, cfg));
}
+
+ @Test
+ void testOceanBaseValidConfig() {
+ OptionRule obRule = new OceanBaseCatalogFactory().optionRule();
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put("url", "jdbc:oceanbase://localhost:2881/mydb");
+ cfg.put("username", "root");
+ cfg.put("password", "pass");
+ cfg.put("compatible_mode", "mysql");
+ Assertions.assertDoesNotThrow(() -> validate(obRule, cfg));
+ }
+
+ // ==================== Dameng (no database in URL) ====================
+
+ @Test
+ void testDamengCatalogUrlWithoutDatabase() {
+ OptionRule dmRule = new DamengCatalogFactory().optionRule();
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put("url", "jdbc:dm://e2e_dmdb:5236");
+ cfg.put("username", "SYSDBA");
+ cfg.put("password", "SYSDBA");
+ Assertions.assertDoesNotThrow(() -> validate(dmRule, cfg));
+ }
+
+ // ==================== DuckDB (no credentials required)
====================
+
+ @Test
+ void testDuckDBCatalogConfigNoCredentials() {
+ OptionRule rule = new DuckDBCatalogFactory().optionRule();
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put("url", "jdbc:duckdb:/tmp/test.db");
+ Assertions.assertDoesNotThrow(() -> validate(rule, cfg));
+ }
+
+ // ==================== SqlServer dialect validator ====================
+
+ @Test
+ void testSqlServerCatalogUrlWithDatabaseNameProperty() {
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put("url",
"jdbc:sqlserver://localhost:1433;databaseName=seatunnel");
+ cfg.put("username", "sa");
+ cfg.put("password", "Password!");
+ Assertions.assertDoesNotThrow(
+ () -> validate(new SqlServerCatalogFactory().optionRule(),
cfg));
+ }
+
+ @Test
+ void testSqlServerCatalogUrlWithoutDatabasePasses() {
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put("url", "jdbc:sqlserver://localhost:1433;encrypt=false");
+ cfg.put("username", "sa");
+ cfg.put("password", "Password!");
+ Assertions.assertDoesNotThrow(
+ () -> validate(new SqlServerCatalogFactory().optionRule(),
cfg));
+ }
+
+ // ==================== Oracle dialect validator ====================
+
+ @Test
+ void testOracleCatalogThinUrlWithoutDoubleSlash() {
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put("url", "jdbc:oracle:thin:@localhost:1521/ORCLCDB");
+ cfg.put("username", "system");
+ cfg.put("password", "oracle");
+ Assertions.assertDoesNotThrow(() -> validate(new
OracleCatalogFactory().optionRule(), cfg));
+ }
+
+ // ==================== SapHana dialect validator ====================
+
+ @Test
+ void testSapHanaCatalogHostOnlyUrl() {
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put("url", "jdbc:sap://localhost:39017");
+ cfg.put("username", "SYSTEM");
+ cfg.put("password", "Password1");
+ Assertions.assertDoesNotThrow(
+ () -> validate(new SapHanaCatalogFactory().optionRule(), cfg));
+ }
+
+ @Test
+ void testSapHanaCatalogWithDatabaseParam() {
+ Map<String, Object> cfg = new HashMap<>();
+ cfg.put("url", "jdbc:sap://localhost:39017/?databaseName=HXE");
+ cfg.put("username", "SYSTEM");
+ cfg.put("password", "Password1");
+ Assertions.assertDoesNotThrow(
+ () -> validate(new SapHanaCatalogFactory().optionRule(), cfg));
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-lance/src/main/java/org/apache/seatunnel/connectors/seatunnel/lance/catalog/LanceCatalogFactory.java
b/seatunnel-connectors-v2/connector-lance/src/main/java/org/apache/seatunnel/connectors/seatunnel/lance/catalog/LanceCatalogFactory.java
index 7d26b4e8ef..a96529bcd6 100644
---
a/seatunnel-connectors-v2/connector-lance/src/main/java/org/apache/seatunnel/connectors/seatunnel/lance/catalog/LanceCatalogFactory.java
+++
b/seatunnel-connectors-v2/connector-lance/src/main/java/org/apache/seatunnel/connectors/seatunnel/lance/catalog/LanceCatalogFactory.java
@@ -21,7 +21,11 @@ import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import org.apache.seatunnel.api.configuration.util.OptionRule;
import org.apache.seatunnel.api.table.catalog.Catalog;
import org.apache.seatunnel.api.table.factory.CatalogFactory;
+import org.apache.seatunnel.api.table.factory.Factory;
+import com.google.auto.service.AutoService;
+
+@AutoService(Factory.class)
public class LanceCatalogFactory implements CatalogFactory {
@Override
public Catalog createCatalog(String catalogName, ReadonlyConfig
readonlyConfig) {
