Caideyipi commented on code in PR #18058:
URL: https://github.com/apache/iotdb/pull/18058#discussion_r3503114116
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/writeback/WriteBackSink.java:
##########
@@ -103,35 +115,126 @@
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.WRITE_BACK_CONNECTOR_SKIP_IF_DEFAULT_VALUE;
import static
org.apache.iotdb.commons.utils.ErrorHandlingCommonUtils.getRootCause;
import static
org.apache.iotdb.db.exception.metadata.DatabaseNotSetException.DATABASE_NOT_SET;
+import static org.apache.tsfile.common.constant.TsFileConstant.PATH_SEPARATOR;
@TreeModel
@TableModel
public class WriteBackSink implements PipeConnector {
private static final Logger LOGGER =
LoggerFactory.getLogger(WriteBackSink.class);
+ private static final String CONNECTOR_IOTDB_DATABASE_KEY =
"connector.database";
+ private static final String SINK_IOTDB_DATABASE_KEY = "sink.database";
// Simulate the behavior of the client-to-server communication
// for correctly handling data insertion in IoTDBReceiverAgent#receive method
- private static final Coordinator COORDINATOR = Coordinator.getInstance();
- private static final SessionManager SESSION_MANAGER =
SessionManager.getInstance();
public static final AtomicLong id = new AtomicLong();
private InternalClientSession session;
private boolean skipIfNoPrivileges;
private boolean useEventUserName;
private UserEntity userEntity;
-
- private static final SqlParser RELATIONAL_SQL_PARSER = new SqlParser();
+ private String targetTableModelDatabaseName;
+ private String invalidTargetTableModelDatabaseName;
+ private String targetTreeModelDatabaseName;
private static final Set<String> ALREADY_CREATED_DATABASES =
ConcurrentHashMap.newKeySet();
+ private static SessionManager getSessionManager() {
+ return SessionManagerHolder.INSTANCE;
+ }
+
+ private static SqlParser getRelationalSqlParser() {
+ return SqlParserHolder.INSTANCE;
+ }
+
+ private static class SessionManagerHolder {
+
+ private static final SessionManager INSTANCE =
SessionManager.getInstance();
+
+ private SessionManagerHolder() {
+ // empty constructor
+ }
+ }
+
+ private static class SqlParserHolder {
+
+ private static final SqlParser INSTANCE = new SqlParser();
+
+ private SqlParserHolder() {
+ // empty constructor
+ }
+ }
+
@Override
public void validate(final PipeParameterValidator validator) throws
Exception {
validator.validateSynonymAttributes(
Arrays.asList(CONNECTOR_IOTDB_USER_KEY, SINK_IOTDB_USER_KEY),
Arrays.asList(CONNECTOR_IOTDB_USERNAME_KEY, SINK_IOTDB_USERNAME_KEY),
false);
+ validator.validateSynonymAttributes(
+ Collections.singletonList(CONNECTOR_IOTDB_DATABASE_KEY),
+ Collections.singletonList(SINK_IOTDB_DATABASE_KEY),
+ false);
+
+ final String targetDatabase =
+ validator
+ .getParameters()
+ .getStringByKeys(CONNECTOR_IOTDB_DATABASE_KEY,
SINK_IOTDB_DATABASE_KEY);
+ if (Objects.nonNull(targetDatabase)) {
+ validateTargetDatabase(targetDatabase);
+ }
+ }
+
+ private static void validateTargetDatabase(final String targetDatabase) {
+ if (PathUtils.isTableModelDatabase(targetDatabase)) {
+ validateTableModelDatabaseName(targetDatabase);
+
validateAndNormalizeTreeModelDatabaseName(PathUtils.qualifyDatabaseName(targetDatabase));
+ return;
+ }
+
+ validateAndNormalizeTreeModelDatabaseName(targetDatabase);
+ }
+
+ private static void validateTableModelDatabaseName(final String
databaseName) {
+ try {
+ TableConfigTaskVisitor.validateDatabaseName(databaseName);
+ } catch (final Exception e) {
+ throw new PipeException(
+ String.format(
+ "The table-model database %s is invalid. It should not contain
'%s', should match "
+ + "the pattern %s, and the length should not exceed %d",
+ databaseName, PATH_SEPARATOR, IoTDBConfig.DATABASE_PATTERN,
MAX_DATABASE_NAME_LENGTH),
Review Comment:
Addressed. I moved the newly added write-back sink database
validation/rewrite exception messages into DataNodePipeMessages for both en and
zh locales.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/writeback/WriteBackSink.java:
##########
@@ -215,6 +325,30 @@ public void customize(
}
}
+ private void customizeTargetDatabase(final String targetDatabase) {
+ targetTableModelDatabaseName = null;
+ invalidTargetTableModelDatabaseName = null;
+ targetTreeModelDatabaseName = null;
+
+ if (PathUtils.isTableModelDatabase(targetDatabase)) {
+ targetTableModelDatabaseName =
targetDatabase.toLowerCase(Locale.ENGLISH);
+ targetTreeModelDatabaseName =
+ validateAndNormalizeTreeModelDatabaseName(
+ PathUtils.qualifyDatabaseName(targetTableModelDatabaseName));
+ return;
+ }
+
+ targetTreeModelDatabaseName =
validateAndNormalizeTreeModelDatabaseName(targetDatabase);
+ final String tableModelDatabaseName =
+
PathUtils.unQualifyDatabaseName(targetTreeModelDatabaseName).toLowerCase(Locale.ENGLISH);
+ try {
+ TableConfigTaskVisitor.validateDatabaseName(tableModelDatabaseName);
+ targetTableModelDatabaseName = tableModelDatabaseName;
+ } catch (final Exception e) {
+ invalidTargetTableModelDatabaseName = tableModelDatabaseName;
+ }
+ }
Review Comment:
The write-back sink can receive table-model and tree-model insertion events
in the same pipe, while the configured target database is a single parameter. I
kept both normalized forms so each model can be rewritten with the same target
semantics. I also added comments explaining this, and kept the invalid
table-model conversion marker for cases such as root.target.db that are valid
tree-model targets but cannot be used by table-model events.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/writeback/WriteBackSink.java:
##########
@@ -103,35 +115,126 @@
import static
org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.WRITE_BACK_CONNECTOR_SKIP_IF_DEFAULT_VALUE;
import static
org.apache.iotdb.commons.utils.ErrorHandlingCommonUtils.getRootCause;
import static
org.apache.iotdb.db.exception.metadata.DatabaseNotSetException.DATABASE_NOT_SET;
+import static org.apache.tsfile.common.constant.TsFileConstant.PATH_SEPARATOR;
@TreeModel
@TableModel
public class WriteBackSink implements PipeConnector {
private static final Logger LOGGER =
LoggerFactory.getLogger(WriteBackSink.class);
+ private static final String CONNECTOR_IOTDB_DATABASE_KEY =
"connector.database";
+ private static final String SINK_IOTDB_DATABASE_KEY = "sink.database";
// Simulate the behavior of the client-to-server communication
// for correctly handling data insertion in IoTDBReceiverAgent#receive method
- private static final Coordinator COORDINATOR = Coordinator.getInstance();
- private static final SessionManager SESSION_MANAGER =
SessionManager.getInstance();
public static final AtomicLong id = new AtomicLong();
private InternalClientSession session;
private boolean skipIfNoPrivileges;
private boolean useEventUserName;
private UserEntity userEntity;
-
- private static final SqlParser RELATIONAL_SQL_PARSER = new SqlParser();
+ private String targetTableModelDatabaseName;
+ private String invalidTargetTableModelDatabaseName;
+ private String targetTreeModelDatabaseName;
private static final Set<String> ALREADY_CREATED_DATABASES =
ConcurrentHashMap.newKeySet();
+ private static SessionManager getSessionManager() {
+ return SessionManagerHolder.INSTANCE;
+ }
+
+ private static SqlParser getRelationalSqlParser() {
+ return SqlParserHolder.INSTANCE;
+ }
+
+ private static class SessionManagerHolder {
+
+ private static final SessionManager INSTANCE =
SessionManager.getInstance();
+
+ private SessionManagerHolder() {
+ // empty constructor
+ }
+ }
+
+ private static class SqlParserHolder {
+
+ private static final SqlParser INSTANCE = new SqlParser();
+
+ private SqlParserHolder() {
+ // empty constructor
+ }
+ }
+
@Override
public void validate(final PipeParameterValidator validator) throws
Exception {
validator.validateSynonymAttributes(
Arrays.asList(CONNECTOR_IOTDB_USER_KEY, SINK_IOTDB_USER_KEY),
Arrays.asList(CONNECTOR_IOTDB_USERNAME_KEY, SINK_IOTDB_USERNAME_KEY),
false);
+ validator.validateSynonymAttributes(
+ Collections.singletonList(CONNECTOR_IOTDB_DATABASE_KEY),
+ Collections.singletonList(SINK_IOTDB_DATABASE_KEY),
+ false);
+
+ final String targetDatabase =
+ validator
+ .getParameters()
+ .getStringByKeys(CONNECTOR_IOTDB_DATABASE_KEY,
SINK_IOTDB_DATABASE_KEY);
+ if (Objects.nonNull(targetDatabase)) {
+ validateTargetDatabase(targetDatabase);
+ }
+ }
+
+ private static void validateTargetDatabase(final String targetDatabase) {
+ if (PathUtils.isTableModelDatabase(targetDatabase)) {
+ validateTableModelDatabaseName(targetDatabase);
+
validateAndNormalizeTreeModelDatabaseName(PathUtils.qualifyDatabaseName(targetDatabase));
+ return;
+ }
+
+ validateAndNormalizeTreeModelDatabaseName(targetDatabase);
+ }
+
+ private static void validateTableModelDatabaseName(final String
databaseName) {
+ try {
+ TableConfigTaskVisitor.validateDatabaseName(databaseName);
+ } catch (final Exception e) {
+ throw new PipeException(
+ String.format(
+ "The table-model database %s is invalid. It should not contain
'%s', should match "
+ + "the pattern %s, and the length should not exceed %d",
+ databaseName, PATH_SEPARATOR, IoTDBConfig.DATABASE_PATTERN,
MAX_DATABASE_NAME_LENGTH),
Review Comment:
Addressed. I moved the newly added write-back sink database
validation/rewrite exception messages into DataNodePipeMessages for both en and
zh locales.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/writeback/WriteBackSink.java:
##########
@@ -215,6 +325,30 @@ public void customize(
}
}
+ private void customizeTargetDatabase(final String targetDatabase) {
+ targetTableModelDatabaseName = null;
+ invalidTargetTableModelDatabaseName = null;
+ targetTreeModelDatabaseName = null;
+
+ if (PathUtils.isTableModelDatabase(targetDatabase)) {
+ targetTableModelDatabaseName =
targetDatabase.toLowerCase(Locale.ENGLISH);
+ targetTreeModelDatabaseName =
+ validateAndNormalizeTreeModelDatabaseName(
+ PathUtils.qualifyDatabaseName(targetTableModelDatabaseName));
+ return;
+ }
+
+ targetTreeModelDatabaseName =
validateAndNormalizeTreeModelDatabaseName(targetDatabase);
+ final String tableModelDatabaseName =
+
PathUtils.unQualifyDatabaseName(targetTreeModelDatabaseName).toLowerCase(Locale.ENGLISH);
+ try {
+ TableConfigTaskVisitor.validateDatabaseName(tableModelDatabaseName);
+ targetTableModelDatabaseName = tableModelDatabaseName;
+ } catch (final Exception e) {
+ invalidTargetTableModelDatabaseName = tableModelDatabaseName;
+ }
+ }
Review Comment:
The write-back sink can receive table-model and tree-model insertion events
in the same pipe, while the configured target database is a single parameter. I
kept both normalized forms so each model can be rewritten with the same target
semantics. I also added comments explaining this, and kept the invalid
table-model conversion marker for cases such as root.target.db that are valid
tree-model targets but cannot be used by table-model events.
--
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]
