cshuo commented on code in PR #4476:
URL: https://github.com/apache/flink-cdc/pull/4476#discussion_r3613430627
##########
flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-hudi/src/main/java/org/apache/flink/cdc/connectors/hudi/sink/coordinator/MultiTableStreamWriteOperatorCoordinator.java:
##########
@@ -956,7 +1028,22 @@ private Configuration createTableSpecificConfig(TableId
tableId) {
Schema cdcSchema =
Preconditions.checkNotNull(
tableSchemas.get(tableId), "Schema for " + tableId +
"should not be null.");
- return ConfigUtils.createTableConfig(baseConfig, cdcSchema, tableId);
+ Configuration tableConfig = ConfigUtils.createTableConfig(baseConfig,
cdcSchema, tableId);
+ applyHiveSyncTableDefaults(tableConfig, tableId);
+ return tableConfig;
+ }
+
+ static void applyHiveSyncTableDefaults(Configuration tableConfig, TableId
tableId) {
+ if (FlinkOptions.HIVE_SYNC_DB
Review Comment:
`hive_sync.db: default` is a valid explicit setting but is treated as unset
here and replaced by the source schema. Use `tableConfig.contains(...)` to
decide whether to derive the default.
--
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]
