test: add "numbers" table in distributed mode (GreptimeTeam#1374)

src/catalog/src/remote/manager.rs

@@ -33,7 +33,6 @@ use table::engine::manager::TableEngineManagerRef;
 use table::engine::EngineContext;
 use table::metadata::TableId;
 use table::requests::{CreateTableRequest, OpenTableRequest};
-use table::table::numbers::NumbersTable;
 use table::TableRef;
 use tokio::sync::Mutex;
 
@@ -423,15 +422,6 @@ impl CatalogManager for RemoteCatalogManager {
             })?;
         handle_system_table_request(self, engine, &mut system_table_requests).await?;
         info!("All system table opened");
-
-        self.catalog(DEFAULT_CATALOG_NAME)
-            .unwrap()
-            .unwrap()
-            .schema(DEFAULT_SCHEMA_NAME)
-            .unwrap()
-            .unwrap()
-            .register_table("numbers".to_string(), Arc::new(NumbersTable::default()))
-            .unwrap();
         Ok(())
     }
 

src/catalog/tests/remote_catalog_tests.rs

@@ -173,7 +173,6 @@ mod tests {
             .schema(DEFAULT_SCHEMA_NAME)
             .unwrap()
             .unwrap();
-        assert_eq!(vec!["numbers"], default_schema.table_names().unwrap());
 
         // register a new table with an nonexistent catalog
         let catalog_name = DEFAULT_CATALOG_NAME.to_string();
@@ -209,14 +208,7 @@ mod tests {
             table,
         };
         assert!(catalog_manager.register_table(reg_req).await.unwrap());
-        assert_eq!(
-            HashSet::from([table_name, "numbers".to_string()]),
-            default_schema
-                .table_names()
-                .unwrap()
-                .into_iter()
-                .collect::<HashSet<_>>()
-        );
+        assert_eq!(vec![table_name], default_schema.table_names().unwrap());
     }
 
     #[tokio::test]

src/frontend/src/catalog.rs

@@ -32,12 +32,14 @@ use catalog::{
     RegisterSchemaRequest, RegisterSystemTableRequest, RegisterTableRequest, RenameTableRequest,
     SchemaProvider, SchemaProviderRef,
 };
+use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
 use common_error::prelude::BoxedError;
 use common_telemetry::error;
 use futures::StreamExt;
 use meta_client::rpc::TableName;
 use partition::manager::PartitionRuleManagerRef;
 use snafu::prelude::*;
+use table::table::numbers::NumbersTable;
 use table::TableRef;
 
 use crate::datanode::DatanodeClients;
@@ -373,24 +375,35 @@ impl SchemaProvider for FrontendSchemaProvider {
 
         std::thread::spawn(|| {
             common_runtime::block_on_read(async move {
+                let mut tables = vec![];
+                if catalog_name == DEFAULT_CATALOG_NAME && schema_name == DEFAULT_SCHEMA_NAME {
+                    tables.push("numbers".to_string());
+                }
+
                 let key = build_table_global_prefix(catalog_name, schema_name);
                 let mut iter = backend.range(key.as_bytes());
-                let mut res = HashSet::new();
 
                 while let Some(r) = iter.next().await {
                     let Kv(k, _) = r?;
                     let key = TableGlobalKey::parse(String::from_utf8_lossy(&k))
                         .context(InvalidCatalogValueSnafu)?;
-                    res.insert(key.table_name);
+                    tables.push(key.table_name);
                 }
-                Ok(res.into_iter().collect())
+                Ok(tables)
             })
         })
         .join()
         .unwrap()
     }
 
     async fn table(&self, name: &str) -> catalog::error::Result<Option<TableRef>> {
+        if self.catalog_name == DEFAULT_CATALOG_NAME
+            && self.schema_name == DEFAULT_SCHEMA_NAME
+            && name == "numbers"
+        {
+            return Ok(Some(Arc::new(NumbersTable::default())));
+        }
+
         let table_global_key = TableGlobalKey {
             catalog_name: self.catalog_name.clone(),
             schema_name: self.schema_name.clone(),

src/frontend/src/tests/instance_test.rs

@@ -324,8 +324,6 @@ async fn test_execute_query(instance: Arc<dyn MockInstance>) {
 
 #[apply(both_instances_cases)]
 async fn test_execute_show_databases_tables(instance: Arc<dyn MockInstance>) {
-    let is_distributed_mode = instance.is_distributed_mode();
-
     let instance = instance.frontend();
 
     let output = execute_sql(&instance, "show databases").await;
@@ -358,24 +356,14 @@ async fn test_execute_show_databases_tables(instance: Arc<dyn MockInstance>) {
         _ => unreachable!(),
     }
 
-    let expected = if is_distributed_mode {
-        "\
-+---------+
-| Tables  |
-+---------+
-| scripts |
-+---------+\
-"
-    } else {
-        "\
+    let expected = "\
 +---------+
 | Tables  |
 +---------+
 | numbers |
 | scripts |
 +---------+\
-"
-    };
+";
     let output = execute_sql(&instance, "show tables").await;
     check_unordered_output_stream(output, expected).await;
 
@@ -385,26 +373,15 @@ async fn test_execute_show_databases_tables(instance: Arc<dyn MockInstance>) {
     ).await;
 
     let output = execute_sql(&instance, "show tables").await;
-    let expected = if is_distributed_mode {
-        "\
-+---------+
-| Tables  |
-+---------+
-| demo    |
-| scripts |
-+---------+\
-"
-    } else {
-        "\
+    let expected = "\
 +---------+
 | Tables  |
 +---------+
 | demo    |
 | numbers |
 | scripts |
 +---------+\
-"
-    };
+";
     check_unordered_output_stream(output, expected).await;
 
     // show tables like [string]
@@ -686,8 +663,7 @@ async fn test_insert_with_default_value(instance: Arc<dyn MockInstance>) {
     test_insert_with_default_value_for_type(instance.frontend(), "bigint").await;
 }
 
-// should apply to both instance. tracked in #1294
-#[apply(standalone_instance_case)]
+#[apply(both_instances_cases)]
 async fn test_use_database(instance: Arc<dyn MockInstance>) {
     let instance = instance.frontend();
 
@@ -928,8 +904,6 @@ async fn test_execute_copy_from_s3(instance: Arc<dyn MockInstance>) {
 
 #[apply(both_instances_cases)]
 async fn test_information_schema(instance: Arc<dyn MockInstance>) {
-    let is_distributed_mode = instance.is_distributed_mode();
-
     let instance = instance.frontend();
 
     let sql = "create table another_table(i bigint time index)";
@@ -942,24 +916,14 @@ async fn test_information_schema(instance: Arc<dyn MockInstance>) {
     let sql = "select table_catalog, table_schema, table_name, table_type from information_schema.tables where table_type != 'SYSTEM VIEW' order by table_name";
 
     let output = execute_sql(&instance, sql).await;
-    let expected = if is_distributed_mode {
-        "\
-+---------------+--------------------+------------+------------+
-| table_catalog | table_schema       | table_name | table_type |
-+---------------+--------------------+------------+------------+
-| greptime      | public             | scripts    | BASE TABLE |
-| greptime      | information_schema | tables     | VIEW       |
-+---------------+--------------------+------------+------------+"
-    } else {
-        "\
+    let expected = "\
 +---------------+--------------------+------------+------------+
 | table_catalog | table_schema       | table_name | table_type |
 +---------------+--------------------+------------+------------+
 | greptime      | public             | numbers    | BASE TABLE |
 | greptime      | public             | scripts    | BASE TABLE |
 | greptime      | information_schema | tables     | VIEW       |
-+---------------+--------------------+------------+------------+"
-    };
++---------------+--------------------+------------+------------+";
     check_output_stream(output, expected).await;
 
     let output = execute_sql_with(&instance, sql, query_ctx).await;

src/frontend/src/tests/test_util.rs

@@ -26,28 +26,18 @@ use crate::tests::{
 
 pub(crate) trait MockInstance {
     fn frontend(&self) -> Arc<Instance>;
-
-    fn is_distributed_mode(&self) -> bool;
 }
 
 impl MockInstance for MockStandaloneInstance {
     fn frontend(&self) -> Arc<Instance> {
         self.instance.clone()
     }
-
-    fn is_distributed_mode(&self) -> bool {
-        false
-    }
 }
 
 impl MockInstance for MockDistributedInstance {
     fn frontend(&self) -> Arc<Instance> {
         self.frontend.clone()
     }
-
-    fn is_distributed_mode(&self) -> bool {
-        true
-    }
 }
 
 pub(crate) async fn standalone() -> Arc<dyn MockInstance> {