Copilot commented on code in PR #4212:
URL: https://github.com/apache/flink-cdc/pull/4212#discussion_r2909861895
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/route/TableIdRouter.java:
##########
@@ -177,13 +192,34 @@ public List<Set<TableId>>
groupSourceTablesByRouteRule(Set<TableId> tableIdSet)
if (routes.isEmpty()) {
return new ArrayList<>();
}
- return routes.stream()
- .map(
- route ->
- tableIdSet.stream()
- .filter(tableId -> matches(route.f0,
tableId))
- .collect(Collectors.toSet()))
- .collect(Collectors.toList());
+
+ if (routeMode == RouteMode.ALL_MATCH) {
+ return routes.stream()
+ .map(
+ route ->
+ tableIdSet.stream()
+ .filter(tableId ->
matches(route.f0, tableId))
+ .collect(Collectors.toSet()))
+ .collect(Collectors.toList());
+ } else if (routeMode == RouteMode.FIRST_MATCH) {
+ List<Set<TableId>> matchedRules = new ArrayList<>();
+ for (TableId tableId : tableIdSet) {
+ boolean matched = false;
+ for (Tuple3<Pattern, String, String> route : routes) {
+ if (matches(route.f0, tableId)) {
+ matchedRules.add(Set.of(tableId));
Review Comment:
`groupSourceTablesByRouteRule()`'s `FIRST_MATCH` branch is not grouping by
route rule: it returns one set per *input table* (and even adds empty sets for
unmatched tables), so the returned list size depends on `tableIdSet` rather
than the number of routing rules. This breaks the method contract ("groups <=
routing rules") and causes incorrect/non-deterministic schema merging in
`SchemaDerivator.deduceMergedCreateTableEvent()` (tables routing to the same
sink won’t be merged; the last processed table can overwrite the merged
schema). Rework the `FIRST_MATCH` branch to build one group per routing rule
(preserving rule order), assign each table to the first matching rule’s group,
and omit unmatched tables from all groups.
--
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]
