github-actions[bot] commented on code in PR #66950:
URL: https://github.com/apache/doris/pull/66950#discussion_r3820317528
##########
fe/fe-core/src/main/java/org/apache/doris/job/util/StreamingJobUtils.java:
##########
@@ -417,8 +418,21 @@ public static LinkedHashMap<String, CreateTableCommand>
generateCreateTableCmds(
// Validate and apply exclude_columns for this table
Set<String> excludeColumns = parseExcludeColumns(properties,
table);
+ if (targetDatabase.isTableExist(targetTableName)) {
Review Comment:
[P1] Validate source types against the CDC path before bypassing mapping
An existing Doris target does not make every source type safe for CDC. For
`initial`/`snapshot`, a PostgreSQL user-enum primary key passes
`getPrimaryKeys()` here, but the later split reader delegates it to Flink CDC
3.6
[`PostgresTypeUtils`](https://github.com/apache/flink-cdc/blob/release-3.6.0/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/utils/PostgresTypeUtils.java#L75-L159),
whose default throws for that type. Independently, an unsupported
custom/composite non-key column is omitted from Debezium 1.9 events because
Doris neither overrides nor exposes its documented default
[`include.unknown.datatypes=false`](https://debezium.io/documentation/reference/1.9/connectors/postgresql.html#postgresql-property-include-unknown-datatypes);
a nullable precreated target then silently receives NULL/default. The new test
uses an INT key and a natively supported enum STRING, so it covers neither
case.
Please classify non-excluded columns against downstream reader support on this
path (including snapshot split-key support), or implement lossless handling,
and reject unsupported schemas during CREATE JOB rather than crashing later or
losing values.
##########
regression-test/data/job_p0/streaming_job/cdc/test_streaming_postgres_job_precreated_target.out:
##########
@@ -0,0 +1,4 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
Review Comment:
[P2] Generate this result with the regression runner
The PR checklist says this suite was not run and its expected output was
derived, while the repository's testing rules require `.out` files to be
generated by the prescribed regression script and never handwritten. The
current External Regression pass supports that the value matches runtime
behavior, but it does not establish the required provenance of the committed
artifact. Please regenerate this file through the suite runner and update the
checklist with the actual run/result.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java:
##########
@@ -301,18 +301,15 @@ private void checkRequiredSourceProperties() {
private List<String> createTableIfNotExists() throws Exception {
List<String> syncTbls = new ArrayList<>();
Map<String, String> effectiveSourceProperties =
buildConvertedSourceProperties(sourceProperties);
- // Key: source table name; Value: CreateTableCommand for the Doris
target table.
- // The two names differ when "table.<src>.target_table" is configured.
- LinkedHashMap<String, CreateTableCommand> createTblCmds =
+ // Key: source table name; Value: CREATE TABLE command, or empty if
the target already exists.
+ // The source and target table names differ when
"table.<src>.target_table" is configured.
+ LinkedHashMap<String, Optional<CreateTableCommand>> createTblCmds =
StreamingJobUtils.generateCreateTableCmds(targetDb,
dataSourceType, effectiveSourceProperties,
targetProperties);
- Database db =
Env.getCurrentEnv().getInternalCatalog().getDbNullable(targetDb);
- Preconditions.checkNotNull(db, "target database %s does not exist",
targetDb);
- for (Map.Entry<String, CreateTableCommand> entry :
createTblCmds.entrySet()) {
+ for (Map.Entry<String, Optional<CreateTableCommand>> entry :
createTblCmds.entrySet()) {
String srcTable = entry.getKey();
- CreateTableCommand createTblCmd = entry.getValue();
- if
(!db.isTableExist(createTblCmd.getCreateTableInfo().getTableName())) {
- createTblCmd.run(ConnectContext.get(), null);
+ if (entry.getValue().isPresent()) {
Review Comment:
[P1] Recheck target state when consuming generated entries
Both Optional states can be stale by this point. If a target recorded as
empty is dropped during the remaining metadata pass, this loop skips DDL, still
adds the source to `syncTbls`, and registers a job whose target is missing.
Conversely, two sources mapped to the same initially absent target produce two
present commands; after the first creates it, the second still runs, and
`InternalCatalog.createTable()` checks capacity/database quota before its
IF-NOT-EXISTS lookup, so a redundant CREATE can fail after the first consumed
the remaining quota. The old code re-resolved the database after generation and
checked current existence while consuming every command. Please carry the
target identity, re-resolve/recheck it here, fail clearly if a skipped target
vanished, and skip commands whose target now exists.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
