[query] refactor: impl ClustersTable table with embedded TableInfo
@@ -20,6 +20,7 @@ use common_catalog::TableIOContext;
use common_datablocks::DataBlock;
use common_datavalues::prelude::*;
use common_exception::Result;
+use common_meta_types::TableInfo;
use common_planners::Extras;
use common_planners::Part;
use common_planners::ReadDataSourcePlan;
@@ -28,34 +29,40 @@ use common_streams::DataBlockStream;
use common_streams::SendableDataBlockStream;
use crate::catalogs::Table;
-use crate::catalogs::ToTableInfo;
pub struct ClustersTable {
- table_id: u64,
- schema: DataSchemaRef,
+ table_info: TableInfo,
}
impl ClustersTable {
pub fn create(table_id: u64) -> Self {
- ClustersTable {
- table_id,
- schema: DataSchemaRefExt::create(vec![
+ let schema = DataSchemaRefExt::create(vec![
DataField::new("name", DataType::String, false),
DataField::new("host", DataType::String, false),
DataField::new("port", DataType::UInt16, false),
- ]),
- }
+ ]);
+
+ let table_info = TableInfo {
+ name: "clusters".to_string(),
+ table_id,
+ schema,
+ engine: "SystemClusters".to_string(),
+ is_local: true,
+ ..Default::default()
+ };
+
+ ClustersTable { table_info }
}
}
#[async_trait::async_trait]
impl Table for ClustersTable {
fn name(&self) -> &str {
- "clusters"
+ &self.table_info.name
}
fn engine(&self) -> &str {
- "SystemClusters"
+ &self.table_info.engine
}
fn as_any(&self) -> &dyn Any {
@@ -63,15 +70,15 @@ impl Table for ClustersTable {
}
fn schema(&self) -> Result<DataSchemaRef> {
- Ok(self.schema.clone())
+ Ok(self.table_info.schema.clone())
}
fn get_id(&self) -> u64 {
- self.table_id
+ self.table_info.table_id
}
fn is_local(&self) -> bool {
- true
+ self.table_info.is_local
}
fn read_plan(
@@ -81,7 +88,7 @@ impl Table for ClustersTable {
_partition_num_hint: Option<usize>,
) -> Result<ReadDataSourcePlan> {
Ok(ReadDataSourcePlan {
- table_info: self.to_table_info("system")?,
+ table_info: self.table_info.clone(),
parts: vec![Part {
name: "".to_string(),
version: 0,
@@ -114,13 +121,16 @@ impl Table for ClustersTable {
}
Ok(Box::pin(DataBlockStream::create(
- self.schema.clone(),
+ self.table_info.schema.clone(),
None,
- vec![DataBlock::create_by_array(self.schema.clone(), vec![
+ vec![DataBlock::create_by_array(
+ self.table_info.schema.clone(),
+ vec![
names.finish().into_series(),
addresses.finish().into_series(),
addresses_port.finish().into_series(),
- ])],
+ ],
+ )],
)))
}
}
