Fluss Flink Lookup function return null rather than PartitionNotExistException if partition not exists.

loserwang1024 · loserwang1024 · commit 2f99ece29ff3 · 2025-07-17T09:57:33.000+08:00
diff --git a/fluss-client/src/main/java/com/alibaba/fluss/client/lookup/PrefixKeyLookuper.java b/fluss-client/src/main/java/com/alibaba/fluss/client/lookup/PrefixKeyLookuper.java
@@ -20,6 +20,7 @@
 import com.alibaba.fluss.bucketing.BucketingFunction;
 import com.alibaba.fluss.client.metadata.MetadataUpdater;
 import com.alibaba.fluss.client.table.getter.PartitionGetter;
+import com.alibaba.fluss.exception.PartitionNotExistException;
 import com.alibaba.fluss.metadata.DataLakeFormat;
 import com.alibaba.fluss.metadata.TableBucket;
 import com.alibaba.fluss.metadata.TableInfo;
@@ -150,9 +151,18 @@ public CompletableFuture<LookupResult> lookup(InternalRow prefixKey) {
 
         Long partitionId = null;
         if (partitionGetter != null) {
-            partitionId =
-                    getPartitionId(
-                            prefixKey, partitionGetter, tableInfo.getTablePath(), metadataUpdater);
+            try {
+                partitionId =
+                        getPartitionId(
+                                prefixKey,
+                                partitionGetter,
+                                tableInfo.getTablePath(),
+                                metadataUpdater);
+            } catch (PartitionNotExistException e) {
+                CompletableFuture<LookupResult> future = new CompletableFuture<>();
+                future.completeExceptionally(e);
+                return future;
+            }
         }
 
         TableBucket tableBucket = new TableBucket(tableInfo.getTableId(), partitionId, bucketId);
diff --git a/fluss-client/src/main/java/com/alibaba/fluss/client/lookup/PrimaryKeyLookuper.java b/fluss-client/src/main/java/com/alibaba/fluss/client/lookup/PrimaryKeyLookuper.java
@@ -20,6 +20,7 @@
 import com.alibaba.fluss.bucketing.BucketingFunction;
 import com.alibaba.fluss.client.metadata.MetadataUpdater;
 import com.alibaba.fluss.client.table.getter.PartitionGetter;
+import com.alibaba.fluss.exception.PartitionNotExistException;
 import com.alibaba.fluss.metadata.DataLakeFormat;
 import com.alibaba.fluss.metadata.TableBucket;
 import com.alibaba.fluss.metadata.TableInfo;
@@ -107,14 +108,22 @@ public CompletableFuture<LookupResult> lookup(InternalRow lookupKey) {
                 bucketKeyEncoder == primaryKeyEncoder
                         ? pkBytes
                         : bucketKeyEncoder.encodeKey(lookupKey);
-        Long partitionId =
-                partitionGetter == null
-                        ? null
-                        : getPartitionId(
+        Long partitionId = null;
+        if (partitionGetter != null) {
+            try {
+                partitionId =
+                        getPartitionId(
                                 lookupKey,
                                 partitionGetter,
                                 tableInfo.getTablePath(),
                                 metadataUpdater);
+            } catch (PartitionNotExistException e) {
+                CompletableFuture<LookupResult> future = new CompletableFuture<>();
+                future.completeExceptionally(e);
+                return future;
+            }
+        }
+
         int bucketId = bucketingFunction.bucketing(bkBytes, numBuckets);
         TableBucket tableBucket = new TableBucket(tableInfo.getTableId(), partitionId, bucketId);
         return lookupClient
diff --git a/fluss-client/src/main/java/com/alibaba/fluss/client/metadata/MetadataUpdater.java b/fluss-client/src/main/java/com/alibaba/fluss/client/metadata/MetadataUpdater.java
@@ -26,6 +26,7 @@
 import com.alibaba.fluss.config.ConfigOptions;
 import com.alibaba.fluss.config.Configuration;
 import com.alibaba.fluss.exception.FlussRuntimeException;
+import com.alibaba.fluss.exception.PartitionNotExistException;
 import com.alibaba.fluss.exception.RetriableException;
 import com.alibaba.fluss.metadata.PhysicalTablePath;
 import com.alibaba.fluss.metadata.TableBucket;
@@ -187,7 +188,8 @@ public void checkAndUpdateTableMetadata(Set<TablePath> tablePaths) {
      *
      * <p>and update partition metadata .
      */
-    public boolean checkAndUpdatePartitionMetadata(PhysicalTablePath physicalTablePath) {
+    public boolean checkAndUpdatePartitionMetadata(PhysicalTablePath physicalTablePath)
+            throws PartitionNotExistException {
         if (!cluster.getPartitionId(physicalTablePath).isPresent()) {
             updateMetadata(null, Collections.singleton(physicalTablePath), null);
         }
@@ -251,7 +253,8 @@ public void updatePhysicalTableMetadata(Set<PhysicalTablePath> physicalTablePath
     protected void updateMetadata(
             @Nullable Set<TablePath> tablePaths,
             @Nullable Collection<PhysicalTablePath> tablePartitionNames,
-            @Nullable Collection<Long> tablePartitionIds) {
+            @Nullable Collection<Long> tablePartitionIds)
+            throws PartitionNotExistException {
         try {
             synchronized (this) {
                 cluster =
@@ -266,6 +269,9 @@ protected void updateMetadata(
             Throwable t = ExceptionUtils.stripExecutionException(e);
             if (t instanceof RetriableException || t instanceof TimeoutException) {
                 LOG.warn("Failed to update metadata, but the exception is re-triable.", t);
+            } else if (t instanceof PartitionNotExistException) {
+                LOG.warn("Failed to update metadata because partition is not exist", t);
+                throw (PartitionNotExistException) t;
             } else {
                 throw new FlussRuntimeException("Failed to update metadata", t);
             }
diff --git a/fluss-client/src/main/java/com/alibaba/fluss/client/utils/ClientUtils.java b/fluss-client/src/main/java/com/alibaba/fluss/client/utils/ClientUtils.java
@@ -124,7 +124,8 @@ public static Long getPartitionId(
             InternalRow row,
             PartitionGetter partitionGetter,
             TablePath tablePath,
-            MetadataUpdater metadataUpdater) {
+            MetadataUpdater metadataUpdater)
+            throws PartitionNotExistException {
         checkNotNull(partitionGetter, "partitionGetter shouldn't be null.");
         String partitionName = partitionGetter.getPartition(row);
         PhysicalTablePath physicalTablePath = PhysicalTablePath.of(tablePath, partitionName);
diff --git a/fluss-client/src/main/java/com/alibaba/fluss/client/write/DynamicPartitionCreator.java b/fluss-client/src/main/java/com/alibaba/fluss/client/write/DynamicPartitionCreator.java
@@ -108,7 +108,7 @@ private boolean forceCheckPartitionExist(PhysicalTablePath physicalTablePath) {
             idExist = metadataUpdater.checkAndUpdatePartitionMetadata(physicalTablePath);
         } catch (Exception e) {
             Throwable t = ExceptionUtils.stripExecutionException(e);
-            if (t.getCause() instanceof PartitionNotExistException) {
+            if (t instanceof PartitionNotExistException) {
                 if (!dynamicPartitionEnabled) {
                     throw new PartitionNotExistException(
                             String.format(
diff --git a/fluss-client/src/test/java/com/alibaba/fluss/client/table/AutoPartitionedTableITCase.java b/fluss-client/src/test/java/com/alibaba/fluss/client/table/AutoPartitionedTableITCase.java
@@ -331,7 +331,6 @@ void testOperateNotExistPartitionShouldThrowException() throws Exception {
         // test scan a not exist partition's log
         LogScanner logScanner = table.newScan().createLogScanner();
         assertThatThrownBy(() -> logScanner.subscribe(100L, 0, 0))
-                .cause()
                 .isInstanceOf(PartitionNotExistException.class)
                 .hasMessageContaining("Partition not exist for partition ids: [100]");
         logScanner.close();
diff --git a/fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/source/lookup/FlinkAsyncLookupFunction.java b/fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/source/lookup/FlinkAsyncLookupFunction.java
@@ -24,6 +24,7 @@
 import com.alibaba.fluss.client.lookup.Lookuper;
 import com.alibaba.fluss.client.table.Table;
 import com.alibaba.fluss.config.Configuration;
+import com.alibaba.fluss.exception.PartitionNotExistException;
 import com.alibaba.fluss.exception.TableNotExistException;
 import com.alibaba.fluss.flink.row.FlinkAsFlussRow;
 import com.alibaba.fluss.flink.source.lookup.LookupNormalizer.RemainingFilter;
@@ -163,7 +164,10 @@ private void handleLookupFailed(
             int currentRetry,
             InternalRow keyRow,
             @Nullable RemainingFilter remainingFilter) {
-        if (throwable instanceof TableNotExistException) {
+        if (throwable instanceof PartitionNotExistException) {
+            LOG.debug("Partition is not exist. Ignore and return null. ", throwable);
+            resultFuture.complete(Collections.emptyList());
+        } else if (throwable instanceof TableNotExistException) {
             LOG.error("Table '{}' not found ", tablePath, throwable);
             resultFuture.completeExceptionally(
                     new RuntimeException("Fluss table '" + tablePath + "' not found.", throwable));
diff --git a/fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/source/lookup/FlinkLookupFunction.java b/fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/source/lookup/FlinkLookupFunction.java
@@ -24,6 +24,7 @@
 import com.alibaba.fluss.client.lookup.Lookuper;
 import com.alibaba.fluss.client.table.Table;
 import com.alibaba.fluss.config.Configuration;
+import com.alibaba.fluss.exception.PartitionNotExistException;
 import com.alibaba.fluss.flink.row.FlinkAsFlussRow;
 import com.alibaba.fluss.flink.utils.FlinkConversions;
 import com.alibaba.fluss.flink.utils.FlinkUtils;
@@ -141,6 +142,11 @@ public Collection<RowData> lookup(RowData keyRow) {
                     }
                 }
                 return projectedRows;
+            } catch (PartitionNotExistException partitionNotExistException) {
+                LOG.debug(
+                        "Partition is not exist. Ignore and return null. ",
+                        partitionNotExistException);
+                return Collections.emptyList();
             } catch (Exception e) {
                 LOG.error(String.format("Fluss lookup error, retry times = %d", retry), e);
                 if (retry >= maxRetryTimes) {
diff --git a/fluss-flink/fluss-flink-common/src/test/java/com/alibaba/fluss/flink/source/FlinkTableSourceITCase.java b/fluss-flink/fluss-flink-common/src/test/java/com/alibaba/fluss/flink/source/FlinkTableSourceITCase.java
@@ -1255,7 +1255,8 @@ private String prepareDimTableAndSourceTable(
                                 Row.of(1, "name1", 11, partition1),
                                 Row.of(2, "name2", 2, partition1),
                                 Row.of(3, "name33", 33, partition2),
-                                Row.of(10, "name0", 44, partition2));
+                                Row.of(10, "name0", 44, partition2),
+                                Row.of(1, "name1", 11, "empty-partition"));
         Schema.Builder builder =
                 Schema.newBuilder()
                         .column("a", DataTypes.INT())
