diff --git a/rust/frontend/src/catalog/database_catalog.rs b/rust/frontend/src/catalog/database_catalog.rs
index 4fccbc8131456..c64954eaf61a9 100644
--- a/rust/frontend/src/catalog/database_catalog.rs
+++ b/rust/frontend/src/catalog/database_catalog.rs
@@ -41,7 +41,7 @@ impl DatabaseCatalog {
         self.schema_by_name.get_mut(schema)
     }
 
-    pub fn id(&self) -> u32 {
+    pub fn id(&self) -> u64 {
         self.id
     }
 }
diff --git a/rust/frontend/src/catalog/mod.rs b/rust/frontend/src/catalog/mod.rs
index 8bc6c08f65830..4b01be076bd41 100644
--- a/rust/frontend/src/catalog/mod.rs
+++ b/rust/frontend/src/catalog/mod.rs
@@ -4,15 +4,15 @@ use risingwave_common::error::ErrorCode;
 use thiserror::Error;
 
 pub(crate) mod catalog_service;
-mod column_catalog;
+pub(crate) mod column_catalog;
 mod database_catalog;
 mod schema_catalog;
 mod table_catalog;
 
-pub(crate) type DatabaseId = u32;
-pub(crate) type SchemaId = u32;
-pub(crate) type TableId = u32;
-pub(crate) type ColumnId = u32;
+pub(crate) type DatabaseId = u64;
+pub(crate) type SchemaId = u64;
+pub(crate) type TableId = u64;
+pub(crate) type ColumnId = u64;
 
 #[derive(Error, Debug)]
 pub enum CatalogError {
diff --git a/rust/frontend/src/catalog/table_catalog.rs b/rust/frontend/src/catalog/table_catalog.rs
index a1faae9359419..7cfb833da4ba4 100644
--- a/rust/frontend/src/catalog/table_catalog.rs
+++ b/rust/frontend/src/catalog/table_catalog.rs
@@ -1,4 +1,4 @@
-use std::sync::atomic::{AtomicU32, Ordering};
+use std::sync::atomic::{AtomicU64, Ordering};
 
 use risingwave_common::array::RwError;
 use risingwave_common::error::Result;
@@ -9,7 +9,7 @@ use crate::catalog::{ColumnId, TableId};
 
 pub struct TableCatalog {
     table_id: TableId,
-    next_column_id: AtomicU32,
+    next_column_id: AtomicU64,
     column_by_name: Vec<(String, ColumnCatalog)>,
     primary_keys: Vec<ColumnId>,
 }
@@ -18,7 +18,7 @@ impl TableCatalog {
     pub fn new(table_id: TableId) -> Self {
         Self {
             table_id,
-            next_column_id: AtomicU32::new(0),
+            next_column_id: AtomicU64::new(0),
             column_by_name: vec![],
             primary_keys: vec![],
         }
@@ -52,7 +52,7 @@ impl TableCatalog {
         self.table_id
     }
 
-    pub fn get_pks(&self) -> Vec<u32> {
+    pub fn get_pks(&self) -> Vec<u64> {
         self.primary_keys.clone()
     }
 }
@@ -61,7 +61,7 @@ impl TryFrom<&Table> for TableCatalog {
     type Error = RwError;
 
     fn try_from(tb: &Table) -> Result<Self> {
-        let mut table_catalog = Self::new(tb.get_table_ref_id()?.table_id as u32);
+        let mut table_catalog = Self::new(tb.get_table_ref_id()?.table_id as u64);
         for col in &tb.column_descs {
             table_catalog.add_column(
                 &col.name,
diff --git a/rust/frontend/src/optimizer/plan_node/batch_seq_scan.rs b/rust/frontend/src/optimizer/plan_node/batch_seq_scan.rs
index fad5059f9089e..131892122adb0 100644
--- a/rust/frontend/src/optimizer/plan_node/batch_seq_scan.rs
+++ b/rust/frontend/src/optimizer/plan_node/batch_seq_scan.rs
@@ -3,14 +3,22 @@ use std::fmt;
 use risingwave_common::catalog::Schema;
 
 use super::{IntoPlanRef, PlanRef, ToDistributedBatch};
+use crate::optimizer::plan_node::LogicalScan;
 use crate::optimizer::property::{WithDistribution, WithOrder, WithSchema};
+
 #[derive(Debug, Clone)]
 pub struct BatchSeqScan {
-    // TODO(catalog)
+    logical: LogicalScan,
 }
 impl WithSchema for BatchSeqScan {
     fn schema(&self) -> &Schema {
-        todo!()
+        self.logical.schema()
+    }
+}
+
+impl BatchSeqScan {
+    pub fn new(logical: LogicalScan) -> Self {
+        Self { logical }
     }
 }
 
@@ -24,6 +32,6 @@ impl WithOrder for BatchSeqScan {}
 impl WithDistribution for BatchSeqScan {}
 impl ToDistributedBatch for BatchSeqScan {
     fn to_distributed(&self) -> PlanRef {
-        todo!()
+        self.clone().into_plan_ref()
     }
 }
diff --git a/rust/frontend/src/optimizer/plan_node/logical_scan.rs b/rust/frontend/src/optimizer/plan_node/logical_scan.rs
index 0f4395fec178c..ecbe5378be3f3 100644
--- a/rust/frontend/src/optimizer/plan_node/logical_scan.rs
+++ b/rust/frontend/src/optimizer/plan_node/logical_scan.rs
@@ -3,15 +3,20 @@ use std::fmt;
 use risingwave_common::catalog::Schema;
 
 use super::{ColPrunable, IntoPlanRef, PlanRef, ToBatch, ToStream};
+use crate::catalog::{ColumnId, TableId};
+use crate::optimizer::plan_node::BatchSeqScan;
 use crate::optimizer::property::{WithDistribution, WithOrder, WithSchema};
 
 #[derive(Debug, Clone)]
+#[allow(dead_code)]
 pub struct LogicalScan {
-    // TODO(catalog)
+    table_id: TableId,
+    columns: Vec<ColumnId>,
+    schema: Schema,
 }
 impl WithSchema for LogicalScan {
     fn schema(&self) -> &Schema {
-        todo!()
+        &self.schema
     }
 }
 impl_plan_tree_node_for_leaf! {LogicalScan}
@@ -26,7 +31,7 @@ impl fmt::Display for LogicalScan {
 impl ColPrunable for LogicalScan {}
 impl ToBatch for LogicalScan {
     fn to_batch(&self) -> PlanRef {
-        todo!()
+        BatchSeqScan::new(self.clone()).into_plan_ref()
     }
 }
 impl ToStream for LogicalScan {