[rust] ignore missing partitions during scanner metadata updates

fluss-rust/crates/fluss/src/client/metadata.rs

@@ -359,6 +359,32 @@ impl Metadata {
         Ok(())
     }
 
+    /// Checks that the partition ids exist, refreshing metadata for uncached ids.
+    pub async fn check_and_update_partition_metadata_by_ids(
+        &self,
+        table_path: &TablePath,
+        partition_ids: &[PartitionId],
+    ) -> Result<()> {
+        let cluster_binding = self.cluster.read().clone();
+        let need_update_partition_ids: Vec<PartitionId> = partition_ids
+            .iter()
+            .filter(|partition_id| cluster_binding.get_partition_name(**partition_id).is_none())
+            .copied()
+            .collect::<HashSet<_>>()
+            .into_iter()
+            .collect();
+
+        if !need_update_partition_ids.is_empty() {
+            self.update_tables_metadata(
+                &HashSet::from([table_path]),
+                &HashSet::new(),
+                need_update_partition_ids,
+            )
+            .await?;
+        }
+        Ok(())
+    }
+
     /// Resolves the partition id, refreshing metadata once if not cached.
     /// Returns `None` when the partition does not exist — `PartitionNotExists`
     /// server errors are swallowed so callers can short-circuit to an empty result.

fluss-rust/crates/fluss/src/client/table/scanner.rs

@@ -829,7 +829,7 @@ impl LogScannerInner {
         let table_bucket =
             TableBucket::new_with_partition(self.table_id, Some(partition_id), bucket);
         self.metadata
-            .check_and_update_table_metadata(from_ref(&self.table_path))
+            .check_and_update_partition_metadata_by_ids(&self.table_path, &[partition_id])
             .await?;
         self.log_scanner_status
             .assign_scan_bucket(table_bucket, offset);
@@ -866,9 +866,19 @@ impl LogScannerInner {
             });
         }
 
-        self.metadata
-            .check_and_update_table_metadata(from_ref(&self.table_path))
-            .await?;
+        if self.is_partitioned_table {
+            let partition_ids: Vec<PartitionId> = bucket_offsets
+                .keys()
+                .filter_map(TableBucket::partition_id)
+                .collect();
+            self.metadata
+                .check_and_update_partition_metadata_by_ids(&self.table_path, &partition_ids)
+                .await?;
+        } else {
+            self.metadata
+                .check_and_update_table_metadata(from_ref(&self.table_path))
+                .await?;
+        }
 
         self.log_scanner_status.assign_scan_buckets(bucket_offsets);
         Ok(())
@@ -1415,15 +1425,21 @@ impl LogFetcher {
             Ok(())
         };
 
-        // TODO: Handle PartitionNotExist error like java side
-        update_result.or_else(|e| {
-            if let Error::RpcError { source, .. } = &e
+        update_result.or_else(|error| {
+            if matches!(error.api_error(), Some(FlussError::PartitionNotExists)) {
+                warn!(
+                    "Received PartitionNotExists while updating scanner metadata; ignoring it: {error}"
+                );
+                Ok(())
+            } else if let Error::RpcError { source, .. } = &error
                 && matches!(source, RpcError::ConnectionError(_) | RpcError::Poisoned(_))
             {
-                warn!("Retrying after encountering error while updating table metadata: {e}");
+                warn!(
+                    "Retrying after encountering error while updating table metadata: {error}"
+                );
                 Ok(())
             } else {
-                Err(e)
+                Err(error)
             }
         })?;
         Ok(())
@@ -1583,14 +1599,33 @@ impl LogFetcher {
 
                     let error = FlussError::for_code(error_code);
                     if Self::should_invalidate_table_meta(error) {
-                        // TODO: Consider triggering table meta invalidation from sender/lookup paths.
                         let table_id = table_bucket.table_id();
                         let cluster = metadata.get_cluster();
                         if let Some(table_path) = cluster.get_table_path_by_id(table_id) {
-                            let physical_tables = HashSet::from([PhysicalTablePath::of(Arc::new(
-                                table_path.clone(),
-                            ))]);
-                            metadata.invalidate_physical_table_meta(&physical_tables);
+                            let physical_table_path = match table_bucket.partition_id() {
+                                Some(partition_id) => {
+                                    match cluster.get_partition_name(partition_id) {
+                                        Some(partition_name) => {
+                                            Some(PhysicalTablePath::of_partitioned(
+                                                Arc::new(table_path.clone()),
+                                                Some(partition_name.clone()),
+                                            ))
+                                        }
+                                        None => {
+                                            warn!(
+                                                "Partition id {partition_id} is missing from partition_name_by_id while invalidating metadata for table {table_path}"
+                                            );
+                                            None
+                                        }
+                                    }
+                                }
+                                None => Some(PhysicalTablePath::of(Arc::new(table_path.clone()))),
+                            };
+                            if let Some(physical_table_path) = physical_table_path {
+                                metadata.invalidate_physical_table_meta(&HashSet::from([
+                                    physical_table_path,
+                                ]));
+                            }
                         } else {
                             warn!(
                                 "Table id {table_id} is missing from table_path_by_id while invalidating table metadata"

fluss-rust/crates/fluss/src/cluster/cluster.rs

@@ -108,12 +108,8 @@ impl Cluster {
         &self,
         physical_tables_to_invalid: &HashSet<PhysicalTablePath>,
     ) -> Self {
-        let table_paths: HashSet<&TablePath> = physical_tables_to_invalid
-            .iter()
-            .map(|path| path.get_table_path())
-            .collect();
         let (available_locations_by_path, available_locations_by_bucket) =
-            self.filter_bucket_locations_by_path(&table_paths);
+            self.filter_bucket_locations_by_physical_path(physical_tables_to_invalid);
 
         Cluster::new(
             self.coordinator_server.clone(),
@@ -177,6 +173,32 @@ impl Cluster {
         (available_locations_by_path, available_locations_by_bucket)
     }
 
+    fn filter_bucket_locations_by_physical_path(
+        &self,
+        physical_table_paths: &HashSet<PhysicalTablePath>,
+    ) -> (
+        HashMap<Arc<PhysicalTablePath>, Vec<BucketLocation>>,
+        HashMap<TableBucket, BucketLocation>,
+    ) {
+        let available_locations_by_path = self
+            .available_locations_by_path
+            .iter()
+            .filter(|&(path, _)| !physical_table_paths.contains(path.as_ref()))
+            .map(|(path, locations)| (path.clone(), locations.clone()))
+            .collect();
+
+        let available_locations_by_bucket = self
+            .available_locations_by_bucket
+            .iter()
+            .filter(|&(_bucket, location)| {
+                !physical_table_paths.contains(location.physical_table_path.as_ref())
+            })
+            .map(|(bucket, location)| (bucket.clone(), location.clone()))
+            .collect();
+
+        (available_locations_by_path, available_locations_by_bucket)
+    }
+
     pub fn from_metadata_response(
         metadata_response: MetadataResponse,
         origin_cluster: Option<&Cluster>,
@@ -540,6 +562,67 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_invalidate_physical_table_meta_only_invalidates_exact_partition() {
+        let table_path = Arc::new(TablePath::new("db", "table"));
+        let partition_1 = Arc::new(PhysicalTablePath::of_partitioned(
+            Arc::clone(&table_path),
+            Some("p1".to_string()),
+        ));
+        let partition_2 = Arc::new(PhysicalTablePath::of_partitioned(
+            Arc::clone(&table_path),
+            Some("p2".to_string()),
+        ));
+        let bucket_1 = TableBucket::new_with_partition(1, Some(10), 0);
+        let bucket_2 = TableBucket::new_with_partition(1, Some(20), 0);
+        let leader = ServerNode::new(1, "ts1-host".to_string(), 9124, ServerType::TabletServer);
+        let location_1 = BucketLocation::new(
+            bucket_1.clone(),
+            Some(leader.clone()),
+            Arc::clone(&partition_1),
+        );
+        let location_2 = BucketLocation::new(
+            bucket_2.clone(),
+            Some(leader.clone()),
+            Arc::clone(&partition_2),
+        );
+        let cluster = Cluster::new(
+            None,
+            HashMap::from([(leader.id(), leader)]),
+            HashMap::from([
+                (Arc::clone(&partition_1), vec![location_1.clone()]),
+                (Arc::clone(&partition_2), vec![location_2.clone()]),
+            ]),
+            HashMap::from([
+                (bucket_1.clone(), location_1),
+                (bucket_2.clone(), location_2),
+            ]),
+            HashMap::new(),
+            HashMap::new(),
+            HashMap::from([
+                (Arc::clone(&partition_1), 10),
+                (Arc::clone(&partition_2), 20),
+            ]),
+        );
+
+        let updated_cluster =
+            cluster.invalidate_physical_table_meta(&HashSet::from([partition_1.as_ref().clone()]));
+
+        assert!(updated_cluster.leader_for(&bucket_1).is_none());
+        assert!(updated_cluster.leader_for(&bucket_2).is_some());
+        assert!(
+            updated_cluster
+                .get_available_buckets_for_table_path(partition_1.as_ref())
+                .is_empty()
+        );
+        assert_eq!(
+            updated_cluster
+                .get_available_buckets_for_table_path(partition_2.as_ref())
+                .len(),
+            1
+        );
+    }
+
     #[test]
     fn test_get_server_nodes_empty_cluster() {
         let cluster = Cluster::default();