nzw921rx commented on code in PR #12561:
URL: https://github.com/apache/seatunnel/pull/12561#discussion_r4374259021
##########
seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/PostgresIncrementalSourceFactory.java:
##########
@@ -121,6 +125,15 @@ TableSource<T, SplitT, StateT> restoreSource(
CatalogTableUtils.mergeCatalogTableConfig(
catalogTables, tableConfigs.get(), s ->
TablePath.of(s, true));
}
+ // pgoutput streams only published columns; snapshot-only jobs
never stream
+ if (PGOUTPUT.equalsIgnoreCase(
+
config.get(PostgresIncrementalSourceOptions.DECODING_PLUGIN_NAME)
+ .trim())
Review Comment:
### [P2] Gate publication filtering on the effective decoding plugin
This guard reads the top-level `decoding.plugin.name` (default `pgoutput`),
but that is not necessarily the plugin used by Debezium.
[PostgresSourceConfigFactory](https://github.com/apache/seatunnel/blob/17aa4e8e6440117aaae8862d5a017563f31af6a5/seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/config/PostgresSourceConfigFactory.java#L70-L76)
first sets `plugin.name`, then
[PgBaseSourceConfigFactory.create()](https://github.com/apache/seatunnel/blob/17aa4e8e6440117aaae8862d5a017563f31af6a5/seatunnel-connectors-v2/connector-cdc/connector-cdc-pg-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/pgbase/config/PgBaseSourceConfigFactory.java#L84-L85)
overwrites it with `props.putAll(dbzProperties)`.
For example, an existing job can omit the top-level plugin option and use:
```hocon
schema-changes.enabled = false
debezium {
"plugin.name" = "decoderbufs"
"publication.name" = "unused_pub"
}
```
The actual plugin is `decoderbufs`, yet this condition still executes the
pgoutput publication filter. On PG15+, if `unused_pub` exists for the captured
table with an explicit column list such as `(id, price)`, an ordinary column
such as `note` is incorrectly removed from the output schema. If that
publication is INSERT-only and omits the required primary key, source creation
can instead fail the new primary-key validation. That publication does not
govern decoderbufs output, and the base version did not filter these jobs.
Please decide whether to filter from the final effective `plugin.name` after
merging the Debezium passthrough properties, and add a regression test for this
configuration precedence.
--
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]
