chl-wxp commented on PR #11176:
URL: https://github.com/apache/seatunnel/pull/11176#issuecomment-4778077969
> > > Thank you for your contribution. I have a small question here: How
should I handle adding any prefix or suffix to all target tables? I understand
that this should not be marked as abandoned.
> >
> >
> > Use `table` with table placeholders instead. For example, use `table =
"prefix_${table_name}_suffix"` instead of configuring `tablePrefix` and
`tableSuffix`.
>
> ```java
> ReadonlyConfig catalogOptions = getCatalogOptions(context);
> Optional<String> optionalTable =
config.getOptional(JdbcSinkOptions.TABLE);
> Optional<String> optionalDatabase =
config.getOptional(JdbcSinkOptions.DATABASE);
> // source table info
> TableIdentifier tableId = catalogTable.getTableId();
> // sink table info
> String sinkDatabaseName =
>
optionalDatabase.orElse(catalogTable.getTablePath().getDatabaseName());
> String sinkTableNameBefore =
>
optionalTable.orElse(catalogTable.getTablePath().getTableName());
> String[] sinkTableSplitArray = sinkTableNameBefore.split("\\.");
> String sinkTableName =
sinkTableSplitArray[sinkTableSplitArray.length - 1];
> String sinkSchemaName;
> if (sinkTableSplitArray.length > 1) {
> sinkSchemaName =
sinkTableSplitArray[sinkTableSplitArray.length - 2];
> } else {
> sinkSchemaName = null;
> }
> if
(StringUtils.isNotBlank(catalogOptions.get(JdbcSinkOptions.SCHEMA))) {
> sinkSchemaName = catalogOptions.get(JdbcSinkOptions.SCHEMA);
> }
> // prefix / suffix
> String tempTableName;
> String prefix = catalogOptions.get(JdbcSinkOptions.TABLE_PREFIX);
> String suffix = catalogOptions.get(JdbcSinkOptions.TABLE_SUFFIX);
> if (StringUtils.isNotEmpty(prefix) ||
StringUtils.isNotEmpty(suffix)) {
> tempTableName = StringUtils.isNotEmpty(prefix) ? prefix +
sinkTableName : sinkTableName;
> tempTableName = StringUtils.isNotEmpty(suffix) ? tempTableName
+ suffix : tempTableName;
> } else {
> tempTableName = sinkTableName;
> }
> // without replace, keep original directly
> String finalSchemaName = sinkSchemaName;
> String finalTableName = tempTableName;
> // rebuild identifier
> TableIdentifier newTableId =
> TableIdentifier.of(
> tableId.getCatalogName(),
> sinkDatabaseName,
> finalSchemaName,
> finalTableName);
> ```
>
> Thank you for your reply. After reviewing the source code, I found that
the configuration table = `"prefix_${table_name}_suffix"` does not work. This
kind of configuration causes catalogTable to point to a non-existent source
`schemaName`, which is exactly why `tablePrefix` and `tableSuffix` were
introduced.
Replacement occurs at
`seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/TablePlaceholderProcessor.java`
--
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]
