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 2e530b82be [Improve][Connector-V2] Migrate IoTDBv2 SQL dialect
validation to OptionRule (#11839)
2e530b82be is described below
commit 2e530b82beac2ed38a5d16f12068b087066b2a88
Author: liziing <[email protected]>
AuthorDate: Tue Aug 18 11:31:09 2026 +0800
[Improve][Connector-V2] Migrate IoTDBv2 SQL dialect validation to
OptionRule (#11839)
---
.../seatunnel/iotdbv2/sink/IoTDBv2SinkFactory.java | 26 ++-----
.../iotdbv2/source/IoTDBv2SourceFactory.java | 26 ++-----
.../seatunnel/iotdbv2/IoTDBFactoryTest.java | 85 +++++++++++++++++++++-
3 files changed, 99 insertions(+), 38 deletions(-)
diff --git
a/seatunnel-connectors-v2/connector-iotdb-v2/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdbv2/sink/IoTDBv2SinkFactory.java
b/seatunnel-connectors-v2/connector-iotdb-v2/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdbv2/sink/IoTDBv2SinkFactory.java
index ed8e964187..e0aad1ef99 100644
---
a/seatunnel-connectors-v2/connector-iotdb-v2/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdbv2/sink/IoTDBv2SinkFactory.java
+++
b/seatunnel-connectors-v2/connector-iotdb-v2/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdbv2/sink/IoTDBv2SinkFactory.java
@@ -18,15 +18,14 @@
package org.apache.seatunnel.connectors.seatunnel.iotdbv2.sink;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.Conditions;
import org.apache.seatunnel.api.configuration.util.OptionRule;
import org.apache.seatunnel.api.table.connector.TableSink;
import org.apache.seatunnel.api.table.factory.Factory;
import org.apache.seatunnel.api.table.factory.TableSinkFactory;
import org.apache.seatunnel.api.table.factory.TableSinkFactoryContext;
-import org.apache.seatunnel.common.exception.CommonErrorCode;
import
org.apache.seatunnel.connectors.seatunnel.iotdbv2.config.IoTDBv2SinkOptions;
import
org.apache.seatunnel.connectors.seatunnel.iotdbv2.constant.SinkConstants;
-import
org.apache.seatunnel.connectors.seatunnel.iotdbv2.exception.IotdbConnectorException;
import com.google.auto.service.AutoService;
import lombok.extern.slf4j.Slf4j;
@@ -50,6 +49,8 @@ public class IoTDBv2SinkFactory implements TableSinkFactory {
IoTDBv2SinkOptions.KEY_DEVICE)
.optional(
IoTDBv2SinkOptions.SQL_DIALECT,
+ Conditions.matches(IoTDBv2SinkOptions.SQL_DIALECT,
"(?i)^(tree|table)$"))
+ .optional(
IoTDBv2SinkOptions.KEY_TIMESTAMP,
IoTDBv2SinkOptions.KEY_TAG_FIELDS,
IoTDBv2SinkOptions.KEY_ATTRIBUTE_FIELDS,
@@ -69,22 +70,11 @@ public class IoTDBv2SinkFactory implements TableSinkFactory
{
@Override
public TableSink createSink(TableSinkFactoryContext context) {
ReadonlyConfig conf = context.getOptions();
- String targetSqlDialect;
- if (conf.get(IoTDBv2SinkOptions.SQL_DIALECT) != null) {
- String sqlDialect = conf.get(IoTDBv2SinkOptions.SQL_DIALECT);
- if (SinkConstants.TABLE.equalsIgnoreCase(sqlDialect)) {
- targetSqlDialect = SinkConstants.TABLE;
- } else {
- if (SinkConstants.TREE.equalsIgnoreCase(sqlDialect)) {
- targetSqlDialect = SinkConstants.TREE;
- } else {
- throw new IotdbConnectorException(
- CommonErrorCode.ILLEGAL_ARGUMENT, "Sql dialect not
supported");
- }
- }
- } else {
- targetSqlDialect = SinkConstants.TREE;
- }
+ String sqlDialect = conf.get(IoTDBv2SinkOptions.SQL_DIALECT);
+ String targetSqlDialect =
+ SinkConstants.TABLE.equalsIgnoreCase(sqlDialect)
+ ? SinkConstants.TABLE
+ : SinkConstants.TREE;
return () ->
new IoTDBv2Sink(context.getOptions(),
context.getCatalogTable(), targetSqlDialect);
}
diff --git
a/seatunnel-connectors-v2/connector-iotdb-v2/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdbv2/source/IoTDBv2SourceFactory.java
b/seatunnel-connectors-v2/connector-iotdb-v2/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdbv2/source/IoTDBv2SourceFactory.java
index 6580201fba..6019f30ed1 100644
---
a/seatunnel-connectors-v2/connector-iotdb-v2/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdbv2/source/IoTDBv2SourceFactory.java
+++
b/seatunnel-connectors-v2/connector-iotdb-v2/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdbv2/source/IoTDBv2SourceFactory.java
@@ -18,6 +18,7 @@
package org.apache.seatunnel.connectors.seatunnel.iotdbv2.source;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+import org.apache.seatunnel.api.configuration.util.Conditions;
import org.apache.seatunnel.api.configuration.util.OptionRule;
import org.apache.seatunnel.api.options.ConnectorCommonOptions;
import org.apache.seatunnel.api.source.SeaTunnelSource;
@@ -28,10 +29,8 @@ import org.apache.seatunnel.api.table.connector.TableSource;
import org.apache.seatunnel.api.table.factory.Factory;
import org.apache.seatunnel.api.table.factory.TableSourceFactory;
import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext;
-import org.apache.seatunnel.common.exception.CommonErrorCode;
import
org.apache.seatunnel.connectors.seatunnel.iotdbv2.config.IoTDBv2SourceOptions;
import
org.apache.seatunnel.connectors.seatunnel.iotdbv2.constant.SourceConstants;
-import
org.apache.seatunnel.connectors.seatunnel.iotdbv2.exception.IotdbConnectorException;
import com.google.auto.service.AutoService;
@@ -55,6 +54,8 @@ public class IoTDBv2SourceFactory implements
TableSourceFactory {
ConnectorCommonOptions.SCHEMA)
.optional(
IoTDBv2SourceOptions.SQL_DIALECT,
+ Conditions.matches(IoTDBv2SourceOptions.SQL_DIALECT,
"(?i)^(tree|table)$"))
+ .optional(
IoTDBv2SourceOptions.DATABASE,
IoTDBv2SourceOptions.FETCH_SIZE,
IoTDBv2SourceOptions.DEFAULT_THRIFT_BUFFER_SIZE,
@@ -71,22 +72,11 @@ public class IoTDBv2SourceFactory implements
TableSourceFactory {
TableSource<T, SplitT, StateT>
createSource(TableSourceFactoryContext context) {
CatalogTable catalogTable =
CatalogTableUtil.buildWithConfig(context.getOptions());
ReadonlyConfig conf = context.getOptions();
- String targetSqlDialect;
- if (conf.get(IoTDBv2SourceOptions.SQL_DIALECT) != null) {
- String sqlDialect = conf.get(IoTDBv2SourceOptions.SQL_DIALECT);
- if (SourceConstants.TABLE.equalsIgnoreCase(sqlDialect)) {
- targetSqlDialect = SourceConstants.TABLE;
- } else {
- if (SourceConstants.TREE.equalsIgnoreCase(sqlDialect)) {
- targetSqlDialect = SourceConstants.TREE;
- } else {
- throw new IotdbConnectorException(
- CommonErrorCode.ILLEGAL_ARGUMENT, "Sql dialect not
supported");
- }
- }
- } else {
- targetSqlDialect = SourceConstants.TREE;
- }
+ String sqlDialect = conf.get(IoTDBv2SourceOptions.SQL_DIALECT);
+ String targetSqlDialect =
+ SourceConstants.TABLE.equalsIgnoreCase(sqlDialect)
+ ? SourceConstants.TABLE
+ : SourceConstants.TREE;
return () ->
(SeaTunnelSource<T, SplitT, StateT>)
new IoTDBv2Source(catalogTable, context.getOptions(),
targetSqlDialect);
diff --git
a/seatunnel-connectors-v2/connector-iotdb-v2/src/test/java/org/apache/seatunnel/connectors/seatunnel/iotdbv2/IoTDBFactoryTest.java
b/seatunnel-connectors-v2/connector-iotdb-v2/src/test/java/org/apache/seatunnel/connectors/seatunnel/iotdbv2/IoTDBFactoryTest.java
index 23f3824080..2004c910b5 100644
---
a/seatunnel-connectors-v2/connector-iotdb-v2/src/test/java/org/apache/seatunnel/connectors/seatunnel/iotdbv2/IoTDBFactoryTest.java
+++
b/seatunnel-connectors-v2/connector-iotdb-v2/src/test/java/org/apache/seatunnel/connectors/seatunnel/iotdbv2/IoTDBFactoryTest.java
@@ -17,17 +17,98 @@
package org.apache.seatunnel.connectors.seatunnel.iotdbv2;
+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.api.options.ConnectorCommonOptions;
+import
org.apache.seatunnel.connectors.seatunnel.iotdbv2.config.IoTDBv2CommonOptions;
+import
org.apache.seatunnel.connectors.seatunnel.iotdbv2.config.IoTDBv2SinkOptions;
+import
org.apache.seatunnel.connectors.seatunnel.iotdbv2.config.IoTDBv2SourceOptions;
import
org.apache.seatunnel.connectors.seatunnel.iotdbv2.sink.IoTDBv2SinkFactory;
import
org.apache.seatunnel.connectors.seatunnel.iotdbv2.source.IoTDBv2SourceFactory;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
class IoTDBFactoryTest {
+ private final OptionRule sourceRule = new
IoTDBv2SourceFactory().optionRule();
+ private final OptionRule sinkRule = new IoTDBv2SinkFactory().optionRule();
+
@Test
void optionRule() {
- Assertions.assertNotNull((new IoTDBv2SourceFactory()).optionRule());
- Assertions.assertNotNull((new IoTDBv2SinkFactory()).optionRule());
+ Assertions.assertNotNull(sourceRule);
+ Assertions.assertNotNull(sinkRule);
+ }
+
+ @Test
+ void omittedSqlDialectIsAccepted() {
+ Assertions.assertDoesNotThrow(() -> validate(validSourceConfig(),
sourceRule));
+ Assertions.assertDoesNotThrow(() -> validate(validSinkConfig(),
sinkRule));
+ }
+
+ @Test
+ void supportedSqlDialectsAreAcceptedCaseInsensitively() {
+ for (String dialect : Arrays.asList("tree", "table", "TrEe", "TaBlE"))
{
+ Map<String, Object> sourceConfig = validSourceConfig();
+ sourceConfig.put(IoTDBv2CommonOptions.SQL_DIALECT.key(), dialect);
+ Assertions.assertDoesNotThrow(() -> validate(sourceConfig,
sourceRule));
+
+ Map<String, Object> sinkConfig = validSinkConfig();
+ sinkConfig.put(IoTDBv2CommonOptions.SQL_DIALECT.key(), dialect);
+ Assertions.assertDoesNotThrow(() -> validate(sinkConfig,
sinkRule));
+ }
+ }
+
+ @Test
+ void unsupportedSqlDialectIsRejected() {
+ Map<String, Object> sourceConfig = validSourceConfig();
+ sourceConfig.put(IoTDBv2CommonOptions.SQL_DIALECT.key(),
"unsupported");
+ OptionValidationException sourceException =
+ Assertions.assertThrows(
+ OptionValidationException.class, () ->
validate(sourceConfig, sourceRule));
+ Assertions.assertTrue(
+
sourceException.getMessage().contains(IoTDBv2CommonOptions.SQL_DIALECT.key()));
+
+ Map<String, Object> sinkConfig = validSinkConfig();
+ sinkConfig.put(IoTDBv2CommonOptions.SQL_DIALECT.key(), "unsupported");
+ OptionValidationException sinkException =
+ Assertions.assertThrows(
+ OptionValidationException.class, () ->
validate(sinkConfig, sinkRule));
+ Assertions.assertTrue(
+
sinkException.getMessage().contains(IoTDBv2CommonOptions.SQL_DIALECT.key()));
+ }
+
+ private void validate(Map<String, Object> config, OptionRule rule) {
+ ConfigValidator.of(ReadonlyConfig.fromMap(config)).validate(rule);
+ }
+
+ private Map<String, Object> validSourceConfig() {
+ Map<String, Object> config = commonConfig();
+ config.put(IoTDBv2SourceOptions.SQL.key(), "select * from root.test");
+ config.put(ConnectorCommonOptions.SCHEMA.key(),
Collections.emptyMap());
+ return config;
+ }
+
+ private Map<String, Object> validSinkConfig() {
+ Map<String, Object> config = commonConfig();
+ config.put(IoTDBv2SinkOptions.STORAGE_GROUP.key(), "root.test");
+ config.put(IoTDBv2SinkOptions.KEY_DEVICE.key(), "device");
+ return config;
+ }
+
+ private Map<String, Object> commonConfig() {
+ Map<String, Object> config = new HashMap<>();
+ config.put(
+ IoTDBv2CommonOptions.NODE_URLS.key(),
Collections.singletonList("127.0.0.1:6667"));
+ config.put(IoTDBv2CommonOptions.USERNAME.key(), "root");
+ config.put(IoTDBv2CommonOptions.PASSWORD.key(), "root");
+ return config;
}
}
