nzw921rx commented on code in PR #12015:
URL: https://github.com/apache/seatunnel/pull/12015#discussion_r4310580048
##########
seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/sink/JdbcSinkFactory.java:
##########
@@ -192,6 +178,310 @@ public TableSink createSink(TableSinkFactoryContext
context) {
finalCatalogTable);
}
+ /**
+ * Writes the resolved primary key columns into the sink config map and
rebuilds the catalog
+ * table so that auto-created tables and generated upsert/update/delete
statements use the given
+ * key columns.
+ *
+ * @param map the sink config map that is later turned back into a {@link
ReadonlyConfig}
+ * @param catalogTable the table being processed
+ * @param primaryKeys the resolved key columns
+ * @return a new catalog table whose primary key is replaced with the
resolved columns
+ */
+ private CatalogTable applyPrimaryKeys(
+ Map<String, String> map, CatalogTable catalogTable, List<String>
primaryKeys) {
+ validatePrimaryKeyColumns(primaryKeys,
catalogTable.getTablePath().getTableName());
+ map.put(JdbcSinkOptions.PRIMARY_KEYS.key(), String.join(",",
primaryKeys));
+ return buildCatalogTableWithPrimaryKey(catalogTable, primaryKeys);
+ }
+
+ /**
+ * Rebuilds the catalog table so that its primary key is the given
columns. The option map is
+ * left untouched, so a caller that resolved the columns from the existing
configuration keeps
+ * the value the user wrote.
+ *
+ * @param catalogTable the table being processed
+ * @param primaryKeys the resolved key columns
+ * @return a new catalog table whose primary key is replaced with the
resolved columns
+ */
+ private CatalogTable buildCatalogTableWithPrimaryKey(
+ CatalogTable catalogTable, List<String> primaryKeys) {
+ PrimaryKey configPk =
+ PrimaryKey.of(
+ catalogTable.getTablePath().getTableName() +
"_config_pk", primaryKeys);
+ TableSchema tableSchema = catalogTable.getTableSchema();
+ return CatalogTable.of(
+ catalogTable.getTableId(),
+ TableSchema.builder()
+ .primaryKey(configPk)
+ .constraintKey(tableSchema.getConstraintKeys())
+ .columns(tableSchema.getColumns())
+ .build(),
+ catalogTable.getOptions(),
+ catalogTable.getPartitionKeys(),
+ catalogTable.getComment(),
+ catalogTable.getCatalogName());
+ }
+
+ /**
+ * Resolves the primary key columns using the pre-existing fallback logic
when no multi-table
+ * mapping matches: explicit top-level {@code primary_keys}, otherwise the
catalog primary key,
+ * otherwise the first unique key. When no key can be determined, the
config map is left
+ * unchanged and the sink falls back to plain INSERT.
+ *
+ * @param config the sink config
+ * @param map the sink config map that is later turned back into a {@link
ReadonlyConfig}
+ * @param catalogTable the table being processed
+ * @return the (possibly rebuilt) catalog table
+ */
+ private CatalogTable applyFallbackPrimaryKeys(
+ ReadonlyConfig config, Map<String, String> map, CatalogTable
catalogTable) {
+ PrimaryKey primaryKey = catalogTable.getTableSchema().getPrimaryKey();
+ if (CollectionUtils.isEmpty(config.get(JdbcSinkOptions.PRIMARY_KEYS)))
{
+ if (primaryKey != null &&
!CollectionUtils.isEmpty(primaryKey.getColumnNames())) {
+ map.put(
+ JdbcSinkOptions.PRIMARY_KEYS.key(),
+ String.join(",", primaryKey.getColumnNames()));
+ } else {
+ Optional<ConstraintKey> keyOptional =
+
catalogTable.getTableSchema().getConstraintKeys().stream()
+ .filter(
+ key ->
+
ConstraintKey.ConstraintType.UNIQUE_KEY.equals(
+
key.getConstraintType()))
+ .findFirst();
+ keyOptional.ifPresent(
+ constraintKey ->
+ map.put(
+ JdbcSinkOptions.PRIMARY_KEYS.key(),
+ constraintKey.getColumnNames().stream()
+ .map(
+
ConstraintKey.ConstraintKeyColumn
+
::getColumnName)
+
.collect(Collectors.joining(","))));
+ }
+ return catalogTable;
+ }
+ // The legacy explicit primary_keys branch keeps the pre-existing
behaviour: only the
+ // catalog primary key is rebuilt, the option value is neither
rewritten nor re-validated.
+ return buildCatalogTableWithPrimaryKey(
+ catalogTable, config.get(JdbcSinkOptions.PRIMARY_KEYS));
+ }
+
+ /**
+ * Validates that each resolved primary key column is a non-empty plain
identifier that does not
+ * contain a comma, so it can be safely comma-joined into {@code
PRIMARY_KEYS} and used in
+ * generated SQL. Generated statements quote identifiers through the
dialect, so a blank name
+ * and a comma (the delimiter of the comma-joined {@code PRIMARY_KEYS}
value) are the only
+ * inputs that cannot be represented safely.
+ *
+ * @param primaryKeys the resolved key columns
+ * @param tableName the table being processed, used in the error message
+ * @throws JdbcConnectorException when a column name is blank or contains
a comma
+ */
+ private void validatePrimaryKeyColumns(List<String> primaryKeys, String
tableName) {
+ for (String key : primaryKeys) {
+ if (StringUtils.isBlank(key)) {
+ throw new JdbcConnectorException(
+ JdbcConnectorErrorCode.INVALID_MULTI_TABLE_CONFIG,
+ String.format(
+ "Resolved primary key column for table '%s' is
empty.", tableName));
+ }
+ if (key.contains(",")) {
+ throw new JdbcConnectorException(
+ JdbcConnectorErrorCode.INVALID_MULTI_TABLE_CONFIG,
+ String.format(
+ "Resolved primary key column '%s' for table
'%s' must not contain a comma.",
+ key, tableName));
+ }
+ }
+ }
+
+ /**
+ * Resolves the per-table primary key mapping from {@code
multi_table_config.primary_keys}.
+ *
+ * <p>Each key is a Java regular expression matched against the upstream
table name using full
+ * match semantics; the first matching pattern in declaration order wins.
Every pattern is
+ * compiled eagerly so an invalid expression fails fast with {@code
JDBC-12} instead of
+ * surfacing lazily once a matching table is processed. An unmatched table
returns {@link
+ * Optional#empty()}, leaving the fallback logic to run.
+ *
+ * @param config the sink config
+ * @param catalogTable the upstream table; its name is matched against the
patterns and its
+ * schema is used to expand {@code ${primary_key}} / {@code
${unique_key}}
+ * @return the resolved key columns, or {@link Optional#empty()} when no
pattern matches
+ */
+ Optional<List<String>> resolveMultiTablePrimaryKeys(
+ ReadonlyConfig config, CatalogTable catalogTable) {
+ Map<String, Object> multiTableConfig =
config.get(JdbcSinkOptions.MULTI_TABLE_CONFIG);
+ if (multiTableConfig == null || multiTableConfig.isEmpty()) {
+ return Optional.empty();
+ }
+ Object primaryKeysObj = multiTableConfig.get("primary_keys");
Review Comment:
I understand that configuring `JdbcSinkOptions.MULTI_TABLE_CONFIG `indicates
that the user has an explicit customization request. However, when
`primary_keys` is not configured, the current implementation silently falls
back. I think this case should fail fast instead.
--
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]
