nzw921rx commented on PR #11176:
URL: https://github.com/apache/seatunnel/pull/11176#issuecomment-4778039102
> > 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.
--
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]
