From 5d2858d1a97820442126767135b348875c8d3f0a Mon Sep 17 00:00:00 2001
From: Adam Recsko <adam.recsko@icloud.com>
Date: Wed, 9 Sep 2026 17:39:14 +0200
Subject: [PATCH] [rust] Return TableNotExist for missing tables

Check table existence before refreshing its metadata so get_table
returns
  TableNotExist instead of InvalidTableException for a missing table.
  Add an integration test covering the expected error classification.
Fixes #4270
---
 .../crates/fluss/src/client/connection.rs     | 20 ++++++-----------
 .../crates/fluss/tests/integration/admin.rs   | 22 +++++++++++++++++++
 2 files changed, 29 insertions(+), 13 deletions(-)

diff --git a/fluss-rust/crates/fluss/src/client/connection.rs b/fluss-rust/crates/fluss/src/client/connection.rs
index 71985cb1a8f..2f763eeacc9 100644
--- a/fluss-rust/crates/fluss/src/client/connection.rs
+++ b/fluss-rust/crates/fluss/src/client/connection.rs
@@ -21,7 +21,7 @@ use crate::client::lookup::LookupClient;
 use crate::client::metadata::Metadata;
 use crate::client::table::FlussTable;
 use crate::config::Config;
-use crate::error::{Error, FlussError, Result};
+use crate::error::{Error, Result};
 use crate::metadata::TablePath;
 
 #[cfg(feature = "integration_tests")]
@@ -183,19 +183,13 @@ impl FlussConnection {
     }
 
     pub async fn get_table(&self, table_path: &TablePath) -> Result<FlussTable<'_>> {
+        if self.metadata.fetch_table_id(table_path).await?.is_none() {
+            return Err(Error::table_not_exist(format!(
+                "Table not found: {table_path}"
+            )));
+        }
         self.metadata.update_table_metadata(table_path).await?;
-        let table_info = self
-            .metadata
-            .get_cluster()
-            .get_table(table_path)
-            .map_err(|e| {
-                if e.api_error() == Some(FlussError::InvalidTableException) {
-                    Error::table_not_exist(format!("Table not found: {table_path}"))
-                } else {
-                    e
-                }
-            })?
-            .clone();
+        let table_info = self.metadata.get_cluster().get_table(table_path)?.clone();
         Ok(FlussTable::new(self, self.metadata.clone(), table_info))
     }
 }
diff --git a/fluss-rust/crates/fluss/tests/integration/admin.rs b/fluss-rust/crates/fluss/tests/integration/admin.rs
index 64218abdf3d..211d0ba4cb6 100644
--- a/fluss-rust/crates/fluss/tests/integration/admin.rs
+++ b/fluss-rust/crates/fluss/tests/integration/admin.rs
@@ -426,6 +426,28 @@ mod admin_test {
         );
     }
 
+    #[tokio::test]
+    async fn test_get_nonexistent_table_returns_table_not_exist() {
+        let cluster = get_shared_cluster();
+        let connection = cluster.get_fluss_connection().await;
+        let table_path = TablePath::new(
+            "fluss",
+            format!("nonexistent_table_{}", uuid::Uuid::new_v4().simple()),
+        );
+
+        let error = match connection.get_table(&table_path).await {
+            Ok(_) => panic!("getting a nonexistent table should fail"),
+            Err(error) => error,
+        };
+
+        assert_eq!(
+            error.api_error(),
+            Some(FlussError::TableNotExist),
+            "Expected TableNotExist error, got {:?}",
+            error
+        );
+    }
+
     /// Helper to assert that an error is a FlussAPIError with the expected code.
     fn assert_api_error(error: fluss::error::Error, expected: FlussError) {
         assert_eq!(