refactor: add test feature gate to numbers table (#7148)

* refactor: add test feature gate to numbers table

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* chore: add debug_assertions

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* refactor: extract numbers table provider

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

* chore: address CR issues

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

---------

Signed-off-by: shuiyisong <xixing.sys@gmail.com>

src/catalog/src/kvbackend/builder.rs

@@ -29,6 +29,7 @@ use crate::information_schema::{InformationExtensionRef, InformationSchemaProvid
 use crate::kvbackend::KvBackendCatalogManager;
 use crate::kvbackend::manager::{CATALOG_CACHE_MAX_CAPACITY, SystemCatalog};
 use crate::process_manager::ProcessManagerRef;
+use crate::system_schema::numbers_table_provider::NumbersTableProvider;
 use crate::system_schema::pg_catalog::PGCatalogProvider;
 
 pub struct KvBackendCatalogManagerBuilder {
@@ -119,6 +120,7 @@ impl KvBackendCatalogManagerBuilder {
                     DEFAULT_CATALOG_NAME.to_string(),
                     me.clone(),
                 )),
+                numbers_table_provider: NumbersTableProvider,
                 backend,
                 process_manager,
                 #[cfg(feature = "enterprise")]

src/catalog/src/kvbackend/manager.rs

@@ -18,8 +18,7 @@ use std::sync::{Arc, Weak};
 
 use async_stream::try_stream;
 use common_catalog::consts::{
-    DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, INFORMATION_SCHEMA_NAME, NUMBERS_TABLE_ID,
-    PG_CATALOG_NAME,
+    DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, INFORMATION_SCHEMA_NAME, PG_CATALOG_NAME,
 };
 use common_error::ext::BoxedError;
 use common_meta::cache::{
@@ -45,7 +44,6 @@ use table::TableRef;
 use table::dist_table::DistTable;
 use table::metadata::{TableId, TableInfoRef};
 use table::table::PartitionRules;
-use table::table::numbers::{NUMBERS_TABLE_NAME, NumbersTable};
 use table::table_name::TableName;
 use tokio::sync::Semaphore;
 use tokio_stream::wrappers::ReceiverStream;
@@ -61,6 +59,7 @@ use crate::information_schema::{InformationExtensionRef, InformationSchemaProvid
 use crate::kvbackend::TableCacheRef;
 use crate::process_manager::ProcessManagerRef;
 use crate::system_schema::SystemSchemaProvider;
+use crate::system_schema::numbers_table_provider::NumbersTableProvider;
 use crate::system_schema::pg_catalog::PGCatalogProvider;
 
 /// Access all existing catalog, schema and tables.
@@ -555,6 +554,7 @@ pub(super) struct SystemCatalog {
     // system_schema_provider for default catalog
     pub(super) information_schema_provider: Arc<InformationSchemaProvider>,
     pub(super) pg_catalog_provider: Arc<PGCatalogProvider>,
+    pub(super) numbers_table_provider: NumbersTableProvider,
     pub(super) backend: KvBackendRef,
     pub(super) process_manager: Option<ProcessManagerRef>,
     #[cfg(feature = "enterprise")]
@@ -584,9 +584,7 @@ impl SystemCatalog {
             PG_CATALOG_NAME if channel == Channel::Postgres => {
                 self.pg_catalog_provider.table_names()
             }
-            DEFAULT_SCHEMA_NAME => {
-                vec![NUMBERS_TABLE_NAME.to_string()]
-            }
+            DEFAULT_SCHEMA_NAME => self.numbers_table_provider.table_names(),
             _ => vec![],
         }
     }
@@ -604,7 +602,7 @@ impl SystemCatalog {
         if schema == INFORMATION_SCHEMA_NAME {
             self.information_schema_provider.table(table).is_some()
         } else if schema == DEFAULT_SCHEMA_NAME {
-            table == NUMBERS_TABLE_NAME
+            self.numbers_table_provider.table_exists(table)
         } else if schema == PG_CATALOG_NAME && channel == Channel::Postgres {
             self.pg_catalog_provider.table(table).is_some()
         } else {
@@ -649,8 +647,8 @@ impl SystemCatalog {
                     });
                 pg_catalog_provider.table(table_name)
             }
-        } else if schema == DEFAULT_SCHEMA_NAME && table_name == NUMBERS_TABLE_NAME {
-            Some(NumbersTable::table(NUMBERS_TABLE_ID))
+        } else if schema == DEFAULT_SCHEMA_NAME {
+            self.numbers_table_provider.table(table_name)
         } else {
             None
         }

src/catalog/src/system_schema.rs

@@ -14,6 +14,7 @@
 
 pub mod information_schema;
 mod memory_table;
+pub mod numbers_table_provider;
 pub mod pg_catalog;
 pub mod predicate;
 mod utils;

src/catalog/src/system_schema/numbers_table_provider.rs

@@ -0,0 +1,59 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#[cfg(any(test, feature = "testing", debug_assertions))]
+use common_catalog::consts::NUMBERS_TABLE_ID;
+use table::TableRef;
+#[cfg(any(test, feature = "testing", debug_assertions))]
+use table::table::numbers::NUMBERS_TABLE_NAME;
+#[cfg(any(test, feature = "testing", debug_assertions))]
+use table::table::numbers::NumbersTable;
+
+// NumbersTableProvider is a dedicated provider for feature-gating the numbers table.
+#[derive(Clone)]
+pub struct NumbersTableProvider;
+
+#[cfg(any(test, feature = "testing", debug_assertions))]
+impl NumbersTableProvider {
+    pub(crate) fn table_exists(&self, name: &str) -> bool {
+        name == NUMBERS_TABLE_NAME
+    }
+
+    pub(crate) fn table_names(&self) -> Vec<String> {
+        vec![NUMBERS_TABLE_NAME.to_string()]
+    }
+
+    pub(crate) fn table(&self, name: &str) -> Option<TableRef> {
+        if name == NUMBERS_TABLE_NAME {
+            Some(NumbersTable::table(NUMBERS_TABLE_ID))
+        } else {
+            None
+        }
+    }
+}
+
+#[cfg(not(any(test, feature = "testing", debug_assertions)))]
+impl NumbersTableProvider {
+    pub(crate) fn table_exists(&self, _name: &str) -> bool {
+        false
+    }
+
+    pub(crate) fn table_names(&self) -> Vec<String> {
+        vec![]
+    }
+
+    pub(crate) fn table(&self, _name: &str) -> Option<TableRef> {
+        None
+    }
+}