chore: remove space_and_table from the table_impl (#1114)

## Rationale
Some fields are not necessary in the `TableImpl`, so let's remove them.

## Detailed Changes
Remove verbose fields in the `TableImpl`.

## Test Plan
Existing tests.

Author: ShiKaiWi

analytic_engine/src/instance/read.rs

@@ -37,7 +37,6 @@ use crate::{
         merge::{MergeBuilder, MergeConfig, MergeIterator},
         IterOptions, RecordBatchWithKeyIterator,
     },
-    space::SpaceAndTable,
     sst::factory::{ReadFrequency, SstReadOptions},
     table::{
         data::TableData,
@@ -85,20 +84,15 @@ impl Instance {
     /// `read_parallelism` output streams.
     pub async fn partitioned_read_from_table(
         &self,
-        space_table: &SpaceAndTable,
+        table_data: &TableData,
         request: ReadRequest,
     ) -> Result<PartitionedStreams> {
         debug!(
             "Instance read from table, space_id:{}, table:{}, table_id:{:?}, request:{:?}",
-            space_table.space().id,
-            space_table.table_data().name,
-            space_table.table_data().id,
-            request
+            table_data.space_id, table_data.name, table_data.id, request
         );
 
-        let table_data = space_table.table_data();
         let table_options = table_data.table_options();
-
         // Collect metrics.
         table_data.metrics.on_read_request_begin();
         let need_merge_sort = need_merge_sort_streams(&table_options, &request);

analytic_engine/src/instance/write.rs

@@ -26,7 +26,7 @@ use crate::{
     },
     memtable::{key::KeySequence, PutContext},
     payload::WritePayload,
-    space::{SpaceAndTable, SpaceRef},
+    space::SpaceRef,
     table::{data::TableDataRef, version::MemTableForWrite},
 };
 
@@ -249,15 +249,17 @@ pub struct Writer<'a> {
 impl<'a> Writer<'a> {
     pub fn new(
         instance: InstanceRef,
-        space_table: SpaceAndTable,
+        space: SpaceRef,
+        table_data: TableDataRef,
         serial_exec: &'a mut TableOpSerialExecutor,
     ) -> Writer<'a> {
-        assert_eq!(space_table.table_data().id, serial_exec.table_id());
+        // Ensure the writer has permission to handle the write of the table.
+        assert_eq!(table_data.id, serial_exec.table_id());
 
         Self {
             instance,
-            space: space_table.space().clone(),
-            table_data: space_table.table_data().clone(),
+            space,
+            table_data,
             serial_exec,
         }
     }

analytic_engine/src/memtable/factory.rs

@@ -24,7 +24,7 @@ pub struct Options {
     pub arena_block_size: u32,
     /// Log sequence at the memtable creation.
     pub creation_sequence: SequenceNumber,
-    /// Memory usage colllector
+    /// Memory usage collector
     pub collector: CollectorRef,
 }
 

analytic_engine/src/space.rs

@@ -19,6 +19,8 @@ use crate::{
     table::data::{TableDataRef, TableDataSet},
 };
 
+pub type SpaceId = u32;
+
 /// Holds references to the table data and its space
 ///
 /// REQUIRE: The table must belongs to the space
@@ -69,10 +71,6 @@ impl fmt::Debug for SpaceAndTable {
     }
 }
 
-/// Space id
-// TODO(yingwen): Or just use something like uuid as space id?
-pub type SpaceId = u32;
-
 #[derive(Debug)]
 pub struct SpaceContext {
     /// Catalog name

analytic_engine/src/table/mod.rs

@@ -38,7 +38,7 @@ use trace_metric::MetricsCollector;
 use self::data::TableDataRef;
 use crate::{
     instance::{alter::Alterer, write::Writer, InstanceRef},
-    space::{SpaceAndTable, SpaceId},
+    space::{SpaceAndTable, SpaceRef},
 };
 
 pub mod data;
@@ -53,21 +53,21 @@ const GET_METRICS_COLLECTOR_NAME: &str = "get";
 const ADDITIONAL_PENDING_WRITE_CAP_RATIO: usize = 10;
 
 struct WriteRequests {
-    pub space_table: SpaceAndTable,
-    pub instance: InstanceRef,
+    pub space: SpaceRef,
     pub table_data: TableDataRef,
+    pub instance: InstanceRef,
     pub pending_writes: Arc<Mutex<PendingWriteQueue>>,
 }
 
 impl WriteRequests {
     pub fn new(
-        space_table: SpaceAndTable,
         instance: InstanceRef,
+        space: SpaceRef,
         table_data: TableDataRef,
         pending_writes: Arc<Mutex<PendingWriteQueue>>,
     ) -> Self {
         Self {
-            space_table,
+            space,
             instance,
             table_data,
             pending_writes,
@@ -77,15 +77,12 @@ impl WriteRequests {
 
 /// Table trait implementation
 pub struct TableImpl {
-    space_table: SpaceAndTable,
+    space: SpaceRef,
     /// Instance
     instance: InstanceRef,
     /// Engine type
     engine_type: String,
 
-    space_id: SpaceId,
-    table_id: TableId,
-
     /// Holds a strong reference to prevent the underlying table from being
     /// dropped when this handle exist.
     table_data: TableDataRef,
@@ -98,13 +95,11 @@ impl TableImpl {
     pub fn new(instance: InstanceRef, space_table: SpaceAndTable) -> Self {
         let pending_writes = Mutex::new(PendingWriteQueue::new(instance.max_rows_in_write_queue));
         let table_data = space_table.table_data().clone();
-        let space_id = space_table.space().space_id();
+        let space = space_table.space().clone();
         Self {
-            space_table,
+            space,
             instance,
             engine_type: ANALYTIC_ENGINE_TYPE.to_string(),
-            space_id,
-            table_id: table_data.id,
             table_data,
             pending_writes: Arc::new(pending_writes),
         }
@@ -114,8 +109,8 @@ impl TableImpl {
 impl fmt::Debug for TableImpl {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("TableImpl")
-            .field("space_id", &self.space_id)
-            .field("table_id", &self.table_id)
+            .field("space_id", &self.space.id)
+            .field("table_id", &self.table_data.id)
             .finish()
     }
 }
@@ -285,8 +280,8 @@ impl TableImpl {
                 // take responsibilities for merging and writing the
                 // requests in the queue.
                 let write_requests = WriteRequests::new(
-                    self.space_table.clone(),
                     self.instance.clone(),
+                    self.space.clone(),
                     self.table_data.clone(),
                     self.pending_writes.clone(),
                 );
@@ -341,7 +336,8 @@ impl TableImpl {
 
         let mut writer = Writer::new(
             write_requests.instance,
-            write_requests.space_table,
+            write_requests.space,
+            write_requests.table_data.clone(),
             &mut serial_exec,
         );
         let write_res = writer
@@ -424,11 +420,7 @@ impl Table for TableImpl {
     }
 
     async fn write(&self, request: WriteRequest) -> Result<usize> {
-        let _timer = self
-            .space_table
-            .table_data()
-            .metrics
-            .start_table_total_timer();
+        let _timer = self.table_data.metrics.start_table_total_timer();
 
         if self.should_queue_write_request(&request) {
             return self.write_with_pending_queue(request).await;
@@ -437,7 +429,8 @@ impl Table for TableImpl {
         let mut serial_exec = self.table_data.serial_exec.lock().await;
         let mut writer = Writer::new(
             self.instance.clone(),
-            self.space_table.clone(),
+            self.space.clone(),
+            self.table_data.clone(),
             &mut serial_exec,
         );
         writer
@@ -451,7 +444,7 @@ impl Table for TableImpl {
         request.opts.read_parallelism = 1;
         let mut streams = self
             .instance
-            .partitioned_read_from_table(&self.space_table, request)
+            .partitioned_read_from_table(&self.table_data, request)
             .await
             .box_err()
             .context(Scan { table: self.name() })?;
@@ -542,7 +535,7 @@ impl Table for TableImpl {
     async fn partitioned_read(&self, request: ReadRequest) -> Result<PartitionedStreams> {
         let streams = self
             .instance
-            .partitioned_read_from_table(&self.space_table, request)
+            .partitioned_read_from_table(&self.table_data, request)
             .await
             .box_err()
             .context(Scan { table: self.name() })?;