refactor: drop Table trait

src/catalog/src/information_schema.rs

@@ -41,7 +41,7 @@ use table::error::{SchemaConversionSnafu, TablesRecordBatchSnafu};
 use table::metadata::{
     FilterPushDownType, TableInfoBuilder, TableInfoRef, TableMetaBuilder, TableType,
 };
-use table::thin_table::{ThinTable, ThinTableAdapter};
+use table::table::Table;
 use table::TableRef;
 pub use table_names::*;
 
@@ -187,10 +187,9 @@ impl InformationSchemaProvider {
         self.information_table(name).map(|table| {
             let table_info = Self::table_info(self.catalog_name.clone(), &table);
             let filter_pushdown = FilterPushDownType::Inexact;
-            let thin_table = ThinTable::new(table_info, filter_pushdown);
-
             let data_source = Arc::new(InformationTableDataSource::new(table));
-            Arc::new(ThinTableAdapter::new(thin_table, data_source)) as _
+            let thin_table = Table::new(table_info, filter_pushdown, data_source);
+            Arc::new(thin_table)
         })
     }
 

src/query/src/tests/time_range_filter_test.rs

@@ -30,8 +30,8 @@ use store_api::data_source::{DataSource, DataSourceRef};
 use store_api::storage::ScanRequest;
 use table::metadata::FilterPushDownType;
 use table::predicate::TimeRangePredicateBuilder;
+use table::table::Table;
 use table::test_util::MemTable;
-use table::thin_table::{ThinTable, ThinTableAdapter};
 use table::TableRef;
 
 use crate::tests::exec_selection;
@@ -42,16 +42,13 @@ struct MemTableWrapper;
 impl MemTableWrapper {
     pub fn table(table: TableRef, filter: Arc<RwLock<Vec<Expr>>>) -> TableRef {
         let table_info = table.table_info();
-        let thin_table_adapter = table.as_any().downcast_ref::<ThinTableAdapter>().unwrap();
-        let data_source = thin_table_adapter.data_source();
-
-        let thin_table = ThinTable::new(table_info, FilterPushDownType::Exact);
+        let data_source = table.data_source();
         let data_source = Arc::new(DataSourceWrapper {
             inner: data_source,
             filter,
         });
-
-        Arc::new(ThinTableAdapter::new(thin_table, data_source))
+        let thin_table = Table::new(table_info, FilterPushDownType::Exact, data_source);
+        Arc::new(thin_table)
     }
 }
 

src/table/src/dist_table.rs

@@ -21,17 +21,17 @@ use store_api::storage::ScanRequest;
 
 use crate::error::UnsupportedSnafu;
 use crate::metadata::{FilterPushDownType, TableInfoRef};
-use crate::thin_table::{ThinTable, ThinTableAdapter};
+use crate::table::Table;
 use crate::TableRef;
 
 #[derive(Clone)]
 pub struct DistTable;
 
 impl DistTable {
     pub fn table(table_info: TableInfoRef) -> TableRef {
-        let thin_table = ThinTable::new(table_info, FilterPushDownType::Inexact);
         let data_source = Arc::new(DummyDataSource);
-        Arc::new(ThinTableAdapter::new(thin_table, data_source))
+        let thin_table = Table::new(table_info, FilterPushDownType::Inexact, data_source);
+        Arc::new(thin_table)
     }
 }
 

src/table/src/lib.rs

@@ -23,7 +23,6 @@ pub mod stats;
 pub mod table;
 pub mod table_reference;
 pub mod test_util;
-pub mod thin_table;
 
 pub use crate::error::{Error, Result};
 pub use crate::stats::{ColumnStatistics, TableStatistics};

src/table/src/table.rs

@@ -12,53 +12,80 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-pub mod adapter;
-mod metrics;
-pub mod numbers;
-pub mod scan;
-
-use std::any::Any;
 use std::sync::Arc;
 
-use async_trait::async_trait;
 use common_query::logical_plan::Expr;
 use common_recordbatch::SendableRecordBatchStream;
 use datatypes::schema::SchemaRef;
+use snafu::ResultExt;
+use store_api::data_source::DataSourceRef;
 use store_api::storage::ScanRequest;
 
-use crate::error::Result;
+use crate::error::{Result, TablesRecordBatchSnafu};
 use crate::metadata::{FilterPushDownType, TableId, TableInfoRef, TableType};
 
-/// Table abstraction.
-#[async_trait]
-pub trait Table: Send + Sync {
-    /// Returns the table as [`Any`](std::any::Any) so that it can be
-    /// downcast to a specific implementation.
-    fn as_any(&self) -> &dyn Any;
+pub mod adapter;
+mod metrics;
+pub mod numbers;
+pub mod scan;
+
+pub type TableRef = Arc<Table>;
 
-    /// Get a reference to the schema for this table
-    fn schema(&self) -> SchemaRef;
+#[async_trait::async_trait]
+pub trait TableIdProvider {
+    async fn next_table_id(&self) -> Result<TableId>;
+}
+
+pub type TableIdProviderRef = Arc<dyn TableIdProvider + Send + Sync>;
+
+/// Table handle.
+pub struct Table {
+    table_info: TableInfoRef,
+    filter_pushdown: FilterPushDownType,
+    data_source: DataSourceRef,
+}
+
+impl Table {
+    pub fn new(
+        table_info: TableInfoRef,
+        filter_pushdown: FilterPushDownType,
+        data_source: DataSourceRef,
+    ) -> Self {
+        Self {
+            table_info,
+            filter_pushdown,
+            data_source,
+        }
+    }
+
+    pub fn data_source(&self) -> DataSourceRef {
+        self.data_source.clone()
+    }
+
+    /// Get a reference to the schema for this table.
+    pub fn schema(&self) -> SchemaRef {
+        self.table_info.meta.schema.clone()
+    }
 
     /// Get a reference to the table info.
-    fn table_info(&self) -> TableInfoRef;
+    pub fn table_info(&self) -> TableInfoRef {
+        self.table_info.clone()
+    }
 
     /// Get the type of this table for metadata/catalog purposes.
-    fn table_type(&self) -> TableType;
+    pub fn table_type(&self) -> TableType {
+        self.table_info.table_type
+    }
 
-    async fn scan_to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream>;
+    pub async fn scan_to_stream(&self, request: ScanRequest) -> Result<SendableRecordBatchStream> {
+        self.data_source
+            .get_stream(request)
+            .context(TablesRecordBatchSnafu)
+    }
 
     /// Tests whether the table provider can make use of any or all filter expressions
     /// to optimise data retrieval.
-    fn supports_filters_pushdown(&self, filters: &[&Expr]) -> Result<Vec<FilterPushDownType>> {
-        Ok(vec![FilterPushDownType::Unsupported; filters.len()])
+    pub fn supports_filters_pushdown(&self, filters: &[&Expr]) -> Result<Vec<FilterPushDownType>> {
+        Ok(vec![self.filter_pushdown; filters.len()])
     }
 }
-
-pub type TableRef = Arc<dyn Table>;
-
-#[async_trait::async_trait]
-pub trait TableIdProvider {
-    async fn next_table_id(&self) -> Result<TableId>;
-}
-
-pub type TableIdProviderRef = Arc<dyn TableIdProvider + Send + Sync>;

src/table/src/table/numbers.rs

@@ -32,7 +32,7 @@ use store_api::storage::ScanRequest;
 use crate::metadata::{
     FilterPushDownType, TableId, TableInfoBuilder, TableInfoRef, TableMetaBuilder, TableType,
 };
-use crate::thin_table::{ThinTable, ThinTableAdapter};
+use crate::table::Table;
 use crate::TableRef;
 
 const NUMBER_COLUMN: &str = "number";
@@ -49,12 +49,13 @@ impl NumbersTable {
     }
 
     pub fn table_with_name(table_id: TableId, name: String) -> TableRef {
-        let thin_table = ThinTable::new(
+        let data_source = Arc::new(NumbersDataSource::new(Self::schema()));
+        let thin_table = Table::new(
             Self::table_info(table_id, name, "test_engine".to_string()),
             FilterPushDownType::Unsupported,
+            data_source,
         );
-        let data_source = Arc::new(NumbersDataSource::new(Self::schema()));
-        Arc::new(ThinTableAdapter::new(thin_table, data_source))
+        Arc::new(thin_table)
     }
 
     pub fn schema() -> SchemaRef {

src/table/src/test_util/empty_table.rs

@@ -21,18 +21,22 @@ use store_api::data_source::DataSource;
 use store_api::storage::ScanRequest;
 
 use crate::metadata::{FilterPushDownType, TableInfo};
-use crate::thin_table::{ThinTable, ThinTableAdapter};
+use crate::table::Table;
 use crate::TableRef;
 
 pub struct EmptyTable;
 
 impl EmptyTable {
     pub fn from_table_info(info: &TableInfo) -> TableRef {
-        let thin_table = ThinTable::new(Arc::new(info.clone()), FilterPushDownType::Unsupported);
         let data_source = Arc::new(EmptyDataSource {
             schema: info.meta.schema.clone(),
         });
-        Arc::new(ThinTableAdapter::new(thin_table, data_source))
+        let thin_table = Table::new(
+            Arc::new(info.clone()),
+            FilterPushDownType::Unsupported,
+            data_source,
+        );
+        Arc::new(thin_table)
     }
 }
 

src/table/src/test_util/memtable.rs

@@ -33,7 +33,7 @@ use crate::error::{SchemaConversionSnafu, TableProjectionSnafu, TablesRecordBatc
 use crate::metadata::{
     FilterPushDownType, TableId, TableInfoBuilder, TableMetaBuilder, TableType, TableVersion,
 };
-use crate::thin_table::{ThinTable, ThinTableAdapter};
+use crate::table::Table;
 use crate::TableRef;
 
 pub struct MemTable;
@@ -94,9 +94,9 @@ impl MemTable {
                 .unwrap(),
         );
 
-        let thin_table = ThinTable::new(info, FilterPushDownType::Unsupported);
         let data_source = Arc::new(MemtableDataSource { recordbatch });
-        Arc::new(ThinTableAdapter::new(thin_table, data_source))
+        let thin_table = Table::new(info, FilterPushDownType::Unsupported, data_source);
+        Arc::new(thin_table)
     }
 
     /// Creates a 1 column 100 rows table, with table name "numbers", column name "uint32s" and