(flink) branch release-1.17 updated: [FLINK-34379][table] Fix adding catalogtable logic

[jingge]

This is an automated email from the ASF dual-hosted git repository.

jingge pushed a commit to branch release-1.17
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/release-1.17 by this push:
     new 7fe25989a65 [FLINK-34379][table] Fix adding catalogtable logic
7fe25989a65 is described below

commit 7fe25989a65474fd6184d7936d0c2167f9080528
Author: Jeyhun Karimov <je.kari...@gmail.com>
AuthorDate: Wed May 15 09:29:15 2024 +0200

    [FLINK-34379][table] Fix adding catalogtable logic
---
 .../utils/DynamicPartitionPruningUtils.java        | 23 ++++++----------------
 1 file changed, 6 insertions(+), 17 deletions(-)

diff --git 
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/DynamicPartitionPruningUtils.java
 
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/DynamicPartitionPruningUtils.java
index 089e13333fd..c61cae238f2 100644
--- 
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/DynamicPartitionPruningUtils.java
+++ 
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/utils/DynamicPartitionPruningUtils.java
@@ -23,6 +23,7 @@ import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.streaming.api.transformations.SourceTransformation;
 import org.apache.flink.table.catalog.CatalogTable;
 import org.apache.flink.table.catalog.ContextResolvedTable;
+import org.apache.flink.table.catalog.ObjectIdentifier;
 import org.apache.flink.table.connector.source.DataStreamScanProvider;
 import org.apache.flink.table.connector.source.DynamicTableSource;
 import org.apache.flink.table.connector.source.ScanTableSource;
@@ -61,10 +62,10 @@ import org.apache.calcite.util.ImmutableIntList;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
-import java.util.HashSet;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
-import java.util.Set;
 import java.util.stream.Collectors;
 
 /** Planner utils for Dynamic partition Pruning. */
@@ -117,7 +118,7 @@ public class DynamicPartitionPruningUtils {
         private final RelNode relNode;
         private boolean hasFilter;
         private boolean hasPartitionedScan;
-        private final Set<ContextResolvedTable> tables = new HashSet<>();
+        private final Map<ObjectIdentifier, ContextResolvedTable> tables = new 
HashMap<>();
 
         public DppDimSideChecker(RelNode relNode) {
             this.relNode = relNode;
@@ -234,20 +235,8 @@ public class DynamicPartitionPruningUtils {
         }
 
         private void setTables(ContextResolvedTable catalogTable) {
-            if (tables.size() == 0) {
-                tables.add(catalogTable);
-            } else {
-                boolean hasAdded = false;
-                for (ContextResolvedTable thisTable : new ArrayList<>(tables)) 
{
-                    if (hasAdded) {
-                        break;
-                    }
-                    if 
(!thisTable.getIdentifier().equals(catalogTable.getIdentifier())) {
-                        tables.add(catalogTable);
-                        hasAdded = true;
-                    }
-                }
-            }
+            ObjectIdentifier identifier = catalogTable.getIdentifier();
+            tables.putIfAbsent(identifier, catalogTable);
         }
     }