chore(rust): report the TableNotFound error while dropping non-exist table (#1022)

this will work after upgrading lance with
https://github.com/lancedb/lance/pull/1995 merged
see #884 for details

Signed-off-by: BubbleCal <bubble-cal@outlook.com>

rust/lancedb/src/connection.rs

@@ -623,7 +623,17 @@ impl ConnectionInternal for Database {
     async fn drop_table(&self, name: &str) -> Result<()> {
         let dir_name = format!("{}.{}", name, LANCE_EXTENSION);
         let full_path = self.base_path.child(dir_name.clone());
-        self.object_store.remove_dir_all(full_path).await?;
+        self.object_store
+            .remove_dir_all(full_path)
+            .await
+            .map_err(|err| match err {
+                // this error is not lance::Error::DatasetNotFound,
+                // as the method `remove_dir_all` may be used to remove something not be a dataset
+                lance::Error::NotFound { .. } => Error::TableNotFound {
+                    name: name.to_owned(),
+                },
+                _ => Error::from(err),
+            })?;
         Ok(())
     }
 
@@ -634,8 +644,6 @@ impl ConnectionInternal for Database {
 
 #[cfg(test)]
 mod tests {
-    use std::fs::create_dir_all;
-
     use arrow_schema::{DataType, Field, Schema};
     use tempfile::tempdir;
 
@@ -691,13 +699,49 @@ mod tests {
         // let db = Database::connect("s3://bucket/path/to/database").await.unwrap();
     }
 
+    #[tokio::test]
+    #[ignore = "this can't pass due to https://github.com/lancedb/lancedb/issues/1019, enable it after the bug fixed"]
+    async fn test_open_table() {
+        let tmp_dir = tempdir().unwrap();
+        let uri = tmp_dir.path().to_str().unwrap();
+        let db = connect(uri).execute().await.unwrap();
+
+        assert_eq!(db.table_names().await.unwrap().len(), 0);
+
+        db.open_table("invalid_table").execute().await.unwrap();
+        // open non-exist table
+        assert!(matches!(
+            db.open_table("invalid_table").execute().await,
+            Err(crate::Error::TableNotFound { .. })
+        ));
+
+        assert_eq!(db.table_names().await.unwrap().len(), 0);
+
+        let schema = Arc::new(Schema::new(vec![Field::new("x", DataType::Int32, false)]));
+        db.create_empty_table("table1", schema)
+            .execute()
+            .await
+            .unwrap();
+        db.open_table("table1").execute().await.unwrap();
+        let tables = db.table_names().await.unwrap();
+        assert_eq!(tables, vec!["table1".to_owned()]);
+    }
+
     #[tokio::test]
     async fn drop_table() {
         let tmp_dir = tempdir().unwrap();
-        create_dir_all(tmp_dir.path().join("table1.lance")).unwrap();
 
         let uri = tmp_dir.path().to_str().unwrap();
         let db = connect(uri).execute().await.unwrap();
+
+        // drop non-exist table
+        // TODO(BubbleCal): enable this after upgrading lance with https://github.com/lancedb/lance/pull/1995 merged
+        // assert!(matches!(
+        //     db.drop_table("invalid_table").await,
+        //     Err(crate::Error::TableNotFound { .. }),
+        // ));
+
+        create_dir_all(tmp_dir.path().join("table1.lance")).unwrap();
         db.drop_table("table1").await.unwrap();
 
         let tables = db.table_names().await.unwrap();