diff --git a/rust/frontend/src/binder/select.rs b/rust/frontend/src/binder/select.rs
index ddfa92c11026..e7d1d843a10f 100644
--- a/rust/frontend/src/binder/select.rs
+++ b/rust/frontend/src/binder/select.rs
@@ -22,6 +22,7 @@ use risingwave_sqlparser::ast::{Expr, Select, SelectItem};
 use super::bind_context::{Clause, ColumnBinding};
 use super::UNNAMED_COLUMN;
 use crate::binder::{Binder, Relation};
+use crate::catalog::{is_row_id_column_name, ROWID_PREFIX};
 use crate::expr::{Expr as _, ExprImpl, InputRef};
 
 #[derive(Debug)]
@@ -118,6 +119,14 @@ impl Binder {
                     aliases.push(alias);
                 }
                 SelectItem::ExprWithAlias { expr, alias } => {
+                    if is_row_id_column_name(&alias.value) {
+                        return Err(ErrorCode::InternalError(format!(
+                            "column name prefixed with {:?} are reserved word.",
+                            ROWID_PREFIX
+                        ))
+                        .into());
+                    }
+
                     let expr = self.bind_expr(expr)?;
                     select_list.push(expr);
                     aliases.push(Some(alias.value));
diff --git a/rust/frontend/src/catalog/mod.rs b/rust/frontend/src/catalog/mod.rs
index ec802bbac433..90b884b63e10 100644
--- a/rust/frontend/src/catalog/mod.rs
+++ b/rust/frontend/src/catalog/mod.rs
@@ -30,6 +30,16 @@ pub(crate) type SchemaId = u32;
 pub(crate) type TableId = risingwave_common::catalog::TableId;
 pub(crate) type ColumnId = risingwave_common::catalog::ColumnId;
 
+pub const ROWID_PREFIX: &str = "_row_id";
+
+pub fn gen_row_id_column_name(idx: usize) -> String {
+    ROWID_PREFIX.to_string() + "#" + &idx.to_string()
+}
+
+pub fn is_row_id_column_name(name: &str) -> bool {
+    name.starts_with(ROWID_PREFIX)
+}
+
 #[derive(Error, Debug)]
 pub enum CatalogError {
     #[error("{0} not found: {1}")]
diff --git a/rust/frontend/src/catalog/table_catalog.rs b/rust/frontend/src/catalog/table_catalog.rs
index 1749038e933b..26e4305efced 100644
--- a/rust/frontend/src/catalog/table_catalog.rs
+++ b/rust/frontend/src/catalog/table_catalog.rs
@@ -159,8 +159,8 @@ mod tests {
     use risingwave_pb::plan::{ColumnCatalog as ProstColumnCatalog, ColumnDesc as ProstColumnDesc};
 
     use crate::catalog::column_catalog::ColumnCatalog;
+    use crate::catalog::gen_row_id_column_name;
     use crate::catalog::table_catalog::TableCatalog;
-    use crate::handler::create_table::ROWID_NAME;
 
     #[test]
     fn test_into_table_catalog() {
@@ -173,7 +173,7 @@ mod tests {
                 ProstColumnCatalog {
                     column_desc: Some(ProstColumnDesc {
                         column_id: 0,
-                        name: ROWID_NAME.to_string(),
+                        name: gen_row_id_column_name(0),
                         field_descs: vec![],
                         column_type: Some(DataType::Int32.to_protobuf()),
                         type_name: String::new(),
@@ -220,7 +220,7 @@ mod tests {
                         column_desc: ColumnDesc {
                             data_type: DataType::Int32,
                             column_id: ColumnId::new(0),
-                            name: ROWID_NAME.to_string(),
+                            name: gen_row_id_column_name(0),
                             field_descs: vec![],
                             type_name: String::new()
                         },
@@ -258,7 +258,7 @@ mod tests {
                     column_desc: ColumnDesc {
                         data_type: DataType::Int32,
                         column_id: ColumnId::new(0),
-                        name: ROWID_NAME.to_string(),
+                        name: gen_row_id_column_name(0),
                         field_descs: vec![],
                         type_name: String::new()
                     },
diff --git a/rust/frontend/src/handler/create_mv.rs b/rust/frontend/src/handler/create_mv.rs
index f6741d3f7334..a8a5bbd83452 100644
--- a/rust/frontend/src/handler/create_mv.rs
+++ b/rust/frontend/src/handler/create_mv.rs
@@ -47,7 +47,7 @@ pub fn gen_create_mv_plan(
 
     let mut plan_root = Planner::new(context).plan_query(bound)?;
     plan_root.set_required_dist(Distribution::any().clone());
-    let materialize = plan_root.gen_create_mv_plan(table_name);
+    let materialize = plan_root.gen_create_mv_plan(table_name)?;
     let table = materialize.table().to_prost(schema_id, database_id);
     let plan: PlanRef = materialize.into();
 
diff --git a/rust/frontend/src/handler/create_source.rs b/rust/frontend/src/handler/create_source.rs
index fd48a2d4179d..43ef7a07aa93 100644
--- a/rust/frontend/src/handler/create_source.rs
+++ b/rust/frontend/src/handler/create_source.rs
@@ -26,7 +26,7 @@ use risingwave_source::ProtobufParser;
 use risingwave_sqlparser::ast::{CreateSourceStatement, ProtobufSchema, SourceSchema};
 
 use crate::binder::expr::bind_data_type;
-use crate::handler::create_table::ROWID_NAME;
+use crate::catalog::gen_row_id_column_name;
 use crate::session::OptimizerContext;
 
 fn extract_protobuf_table_schema(schema: &ProtobufSchema) -> Result<Vec<ColumnCatalog>> {
@@ -55,7 +55,7 @@ pub(super) async fn handle_create_source(
     let mut column_catalogs = vec![ColumnCatalog {
         column_desc: Some(ColumnDesc {
             column_id: 0,
-            name: ROWID_NAME.to_string(),
+            name: gen_row_id_column_name(0),
             column_type: Some(DataType::Int32.to_protobuf()),
             field_descs: vec![],
             type_name: "".to_string(),
@@ -142,6 +142,7 @@ pub mod tests {
 
     use super::*;
     use crate::catalog::column_catalog::ColumnCatalog;
+    use crate::catalog::gen_row_id_column_name;
     use crate::test_utils::LocalFrontend;
 
     /// Returns the file.
@@ -220,8 +221,9 @@ pub mod tests {
             let city_type = DataType::Struct {
                 fields: vec![DataType::Varchar, DataType::Varchar].into(),
             };
+            let row_id_col_name = gen_row_id_column_name(0);
             let expected_columns = maplit::hashmap! {
-                ROWID_NAME => DataType::Int32,
+                row_id_col_name.as_str() => DataType::Int32,
                 "id" => DataType::Int32,
                 "country.zipcode" => DataType::Varchar,
                 "zipcode" => DataType::Int64,
diff --git a/rust/frontend/src/handler/create_table.rs b/rust/frontend/src/handler/create_table.rs
index 2a56c98078d0..4274c94615e2 100644
--- a/rust/frontend/src/handler/create_table.rs
+++ b/rust/frontend/src/handler/create_table.rs
@@ -16,7 +16,7 @@ use fixedbitset::FixedBitSet;
 use itertools::Itertools;
 use pgwire::pg_response::{PgResponse, StatementType};
 use risingwave_common::catalog::{ColumnDesc, ColumnId, TableId};
-use risingwave_common::error::Result;
+use risingwave_common::error::{ErrorCode, Result};
 use risingwave_common::types::DataType;
 use risingwave_pb::catalog::source::Info;
 use risingwave_pb::catalog::{Source as ProstSource, Table as ProstTable, TableSourceInfo};
@@ -25,13 +25,12 @@ use risingwave_sqlparser::ast::{ColumnDef, ObjectName};
 
 use crate::binder::expr::bind_data_type;
 use crate::binder::Binder;
+use crate::catalog::{gen_row_id_column_name, is_row_id_column_name, ROWID_PREFIX};
 use crate::optimizer::plan_node::StreamSource;
 use crate::optimizer::property::{Distribution, Order};
 use crate::optimizer::{PlanRef, PlanRoot};
 use crate::session::{OptimizerContext, OptimizerContextRef, SessionImpl};
 
-pub const ROWID_NAME: &str = "_row_id";
-
 pub fn gen_create_table_plan(
     session: &SessionImpl,
     context: OptimizerContextRef,
@@ -51,12 +50,20 @@ pub fn gen_create_table_plan(
         column_descs.push(ColumnDesc {
             data_type: DataType::Int64,
             column_id: ColumnId::new(0),
-            name: ROWID_NAME.to_string(),
+            name: gen_row_id_column_name(0),
             field_descs: vec![],
             type_name: "".to_string(),
         });
         // Then user columns.
         for (i, column) in columns.into_iter().enumerate() {
+            if is_row_id_column_name(&column.name.value) {
+                return Err(ErrorCode::InternalError(format!(
+                    "column name prefixed with {:?} are reserved word.",
+                    ROWID_PREFIX
+                ))
+                .into());
+            }
+
             column_descs.push(ColumnDesc {
                 data_type: bind_data_type(&column.data_type)?,
                 column_id: ColumnId::new((i + 1) as i32),
@@ -106,7 +113,7 @@ pub fn gen_create_table_plan(
             Order::any().clone(),
             required_cols,
         )
-        .gen_create_mv_plan(table_name)
+        .gen_create_mv_plan(table_name)?
     };
     let table = materialize.table().to_prost(schema_id, database_id);
 
@@ -154,7 +161,7 @@ mod tests {
     use risingwave_common::catalog::{DEFAULT_DATABASE_NAME, DEFAULT_SCHEMA_NAME};
     use risingwave_common::types::DataType;
 
-    use super::*;
+    use crate::catalog::gen_row_id_column_name;
     use crate::test_utils::LocalFrontend;
 
     #[tokio::test]
@@ -188,8 +195,9 @@ mod tests {
             .map(|col| (col.name(), col.data_type().clone()))
             .collect::<HashMap<&str, DataType>>();
 
+        let row_id_col_name = gen_row_id_column_name(0);
         let expected_columns = maplit::hashmap! {
-            ROWID_NAME => DataType::Int64,
+            row_id_col_name.as_str() => DataType::Int64,
             "v1" => DataType::Int16,
             "v2" => DataType::Int32,
             "v3" => DataType::Int64,
diff --git a/rust/frontend/src/optimizer/mod.rs b/rust/frontend/src/optimizer/mod.rs
index ac995cbcd7c6..ac278f697ea4 100644
--- a/rust/frontend/src/optimizer/mod.rs
+++ b/rust/frontend/src/optimizer/mod.rs
@@ -25,6 +25,7 @@ use fixedbitset::FixedBitSet;
 use itertools::Itertools as _;
 use property::{Distribution, Order};
 use risingwave_common::catalog::Schema;
+use risingwave_common::error::Result;
 
 use self::heuristic::{ApplyOrder, HeuristicOptimizer};
 use self::plan_node::{LogicalProject, StreamMaterialize};
@@ -170,7 +171,7 @@ impl PlanRoot {
     ///
     /// The `MaterializeExecutor` won't be generated at this stage, and will be attached in
     /// `gen_create_mv_plan`.
-    pub fn gen_create_mv_plan(&mut self, mv_name: String) -> StreamMaterialize {
+    pub fn gen_create_mv_plan(&mut self, mv_name: String) -> Result<StreamMaterialize> {
         let stream_plan = match self.plan.convention() {
             Convention::Logical => {
                 let plan = self.gen_optimized_logical_plan();
diff --git a/rust/frontend/src/optimizer/plan_node/logical_project.rs b/rust/frontend/src/optimizer/plan_node/logical_project.rs
index 226747df442a..f7147e881168 100644
--- a/rust/frontend/src/optimizer/plan_node/logical_project.rs
+++ b/rust/frontend/src/optimizer/plan_node/logical_project.rs
@@ -41,7 +41,7 @@ pub struct LogicalProject {
 impl LogicalProject {
     pub fn new(input: PlanRef, exprs: Vec<ExprImpl>, expr_alias: Vec<Option<String>>) -> Self {
         let ctx = input.ctx();
-        let schema = Self::derive_schema(&exprs, &expr_alias);
+        let schema = Self::derive_schema(&exprs, &expr_alias, input.schema());
         let pk_indices = Self::derive_pk(input.schema(), input.pk_indices(), &exprs);
         for expr in &exprs {
             assert_input_ref!(expr, input.schema().fields().len());
@@ -126,13 +126,21 @@ impl LogicalProject {
         LogicalProject::new(input, exprs, alias).into()
     }
 
-    fn derive_schema(exprs: &[ExprImpl], expr_alias: &[Option<String>]) -> Schema {
+    fn derive_schema(
+        exprs: &[ExprImpl],
+        expr_alias: &[Option<String>],
+        input_schema: &Schema,
+    ) -> Schema {
+        let o2i = Self::o2i_col_mapping_inner(input_schema.len(), exprs);
         let fields = exprs
             .iter()
             .zip_eq(expr_alias.iter())
             .enumerate()
             .map(|(id, (expr, alias))| {
-                let name = alias.clone().unwrap_or(format!("expr#{}", id));
+                let name = alias.clone().unwrap_or(match o2i.try_map(id) {
+                    Some(input_idx) => input_schema.fields()[input_idx].name.clone(),
+                    None => format!("expr#{}", id),
+                });
                 Field {
                     name,
                     data_type: expr.return_type(),
diff --git a/rust/frontend/src/optimizer/plan_node/stream_materialize.rs b/rust/frontend/src/optimizer/plan_node/stream_materialize.rs
index f0987a93a1c1..4739f657874c 100644
--- a/rust/frontend/src/optimizer/plan_node/stream_materialize.rs
+++ b/rust/frontend/src/optimizer/plan_node/stream_materialize.rs
@@ -12,11 +12,14 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::HashSet;
 use std::fmt;
 
 use fixedbitset::FixedBitSet;
 use itertools::Itertools;
-use risingwave_common::catalog::{ColumnDesc, OrderedColumnDesc, Schema, TableId};
+use risingwave_common::catalog::{ColumnDesc, Field, OrderedColumnDesc, Schema, TableId};
+use risingwave_common::error::ErrorCode::InternalError;
+use risingwave_common::error::Result;
 use risingwave_common::util::sort_util::OrderType;
 use risingwave_pb::expr::InputRefExpr;
 use risingwave_pb::plan::ColumnOrder;
@@ -25,7 +28,7 @@ use risingwave_pb::stream_plan::stream_node::Node as ProstStreamNode;
 use super::{PlanRef, PlanTreeNodeUnary, ToStreamProst};
 use crate::catalog::column_catalog::ColumnCatalog;
 use crate::catalog::table_catalog::TableCatalog;
-use crate::catalog::ColumnId;
+use crate::catalog::{gen_row_id_column_name, is_row_id_column_name, ColumnId};
 use crate::optimizer::plan_node::{PlanBase, PlanNode};
 use crate::optimizer::property::{Order, WithSchema};
 
@@ -39,22 +42,56 @@ pub struct StreamMaterialize {
 }
 
 impl StreamMaterialize {
-    fn derive_plan_base(input: &PlanRef) -> PlanBase {
+    fn derive_plan_base(input: &PlanRef) -> Result<PlanBase> {
         let ctx = input.ctx();
-        let schema = input.schema();
+
+        let schema = Self::derive_schema(input.schema())?;
         let pk_indices = input.pk_indices();
 
-        PlanBase::new_stream(
+        Ok(PlanBase::new_stream(
             ctx,
-            schema.clone(),
+            schema,
             pk_indices.to_vec(),
             input.distribution().clone(),
             input.append_only(),
-        )
+        ))
+    }
+
+    fn derive_schema(schema: &Schema) -> Result<Schema> {
+        let mut col_names = HashSet::new();
+        for field in schema.fields() {
+            if is_row_id_column_name(&field.name) {
+                continue;
+            }
+            if !col_names.insert(field.name.clone()) {
+                return Err(InternalError(format!(
+                    "column {} specified more than once",
+                    field.name
+                ))
+                .into());
+            }
+        }
+        let mut row_id_count = 0;
+        let fields = schema
+            .fields()
+            .iter()
+            .map(|field| match is_row_id_column_name(&field.name) {
+                true => {
+                    let field = Field {
+                        data_type: field.data_type.clone(),
+                        name: gen_row_id_column_name(row_id_count),
+                    };
+                    row_id_count += 1;
+                    field
+                }
+                false => field.clone(),
+            })
+            .collect();
+        Ok(Schema { fields })
     }
     #[must_use]
     pub fn new(input: PlanRef, table: TableCatalog) -> Self {
-        let base = Self::derive_plan_base(&input);
+        let base = Self::derive_plan_base(&input).unwrap();
         Self { base, input, table }
     }
 
@@ -64,8 +101,8 @@ impl StreamMaterialize {
         mv_name: String,
         user_order_by: Order,
         user_cols: FixedBitSet,
-    ) -> Self {
-        let base = Self::derive_plan_base(&input);
+    ) -> Result<Self> {
+        let base = Self::derive_plan_base(&input)?;
         let schema = &base.schema;
         let pk_indices = &base.pk_indices;
         // Materialize executor won't change the append-only behavior of the stream, so it depends
@@ -115,7 +152,7 @@ impl StreamMaterialize {
             pk_desc,
         };
 
-        Self { base, input, table }
+        Ok(Self { base, input, table })
     }
 
     /// Get a reference to the stream materialize's table.
diff --git a/rust/frontend/test_runner/tests/testdata/basic_query_1.yaml b/rust/frontend/test_runner/tests/testdata/basic_query_1.yaml
index c34f067532c5..2fa02129451c 100644
--- a/rust/frontend/test_runner/tests/testdata/basic_query_1.yaml
+++ b/rust/frontend/test_runner/tests/testdata/basic_query_1.yaml
@@ -10,8 +10,8 @@
     BatchExchange { order: [], dist: Single }
       BatchScan { table: t, columns: [v1, v2] }
   stream_plan: |
-    StreamMaterialize { columns: [v1, v2, _row_id(hidden)], pk_columns: [_row_id] }
-      StreamTableScan { table: t, columns: [v1, v2, _row_id], pk_indices: [2] }
+    StreamMaterialize { columns: [v1, v2, _row_id#0(hidden)], pk_columns: [_row_id#0] }
+      StreamTableScan { table: t, columns: [v1, v2, _row_id#0], pk_indices: [2] }
 - sql: |
     create table t (v1 bigint, v2 double precision);
     select t2.* from t;
@@ -24,9 +24,9 @@
       BatchFilter { predicate: (((((1:Int32 > 2:Int32) AND (1:Int32 = 1:Int32)) AND (3:Int32 < 1:Int32)) AND (4:Int32 <> 1:Int32)) OR (((1:Int32 = 1:Int32) AND (2:Int32 >= 1:Int32)) AND (1:Int32 <= 2:Int32))) }
         BatchScan { table: t, columns: [] }
   stream_plan: |
-    StreamMaterialize { columns: [_row_id(hidden)], pk_columns: [_row_id] }
+    StreamMaterialize { columns: [_row_id#0(hidden)], pk_columns: [_row_id#0] }
       StreamFilter { predicate: (((((1:Int32 > 2:Int32) AND (1:Int32 = 1:Int32)) AND (3:Int32 < 1:Int32)) AND (4:Int32 <> 1:Int32)) OR (((1:Int32 = 1:Int32) AND (2:Int32 >= 1:Int32)) AND (1:Int32 <= 2:Int32))) }
-        StreamTableScan { table: t, columns: [_row_id], pk_indices: [0] }
+        StreamTableScan { table: t, columns: [_row_id#0], pk_indices: [0] }
 - sql: |
     create table t (v1 int);
     select * from t where v1<1;
@@ -35,9 +35,9 @@
       BatchFilter { predicate: ($0 < 1:Int32) }
         BatchScan { table: t, columns: [v1] }
   stream_plan: |
-    StreamMaterialize { columns: [v1, _row_id(hidden)], pk_columns: [_row_id] }
+    StreamMaterialize { columns: [v1, _row_id#0(hidden)], pk_columns: [_row_id#0] }
       StreamFilter { predicate: ($0 < 1:Int32) }
-        StreamTableScan { table: t, columns: [v1, _row_id], pk_indices: [1] }
+        StreamTableScan { table: t, columns: [v1, _row_id#0], pk_indices: [1] }
 - sql: |
     create table t ();
     select (((((false is not true) is true) is not false) is false) is not null) is null from t;
@@ -53,9 +53,9 @@
       BatchProject { exprs: [Case(($0 = 1:Int32), 1:Int32::Decimal, ($0 = 2:Int32), 2:Int32::Decimal, Normalized(0.0):Decimal)], expr_alias: [ ] }
         BatchScan { table: t, columns: [v1] }
   stream_plan: |
-    StreamMaterialize { columns: [expr#0, expr#1(hidden)], pk_columns: [expr#1] }
+    StreamMaterialize { columns: [expr#0, _row_id#0(hidden)], pk_columns: [_row_id#0] }
       StreamProject { exprs: [Case(($0 = 1:Int32), 1:Int32::Decimal, ($0 = 2:Int32), 2:Int32::Decimal, Normalized(0.0):Decimal), $1], expr_alias: [ ,  ] }
-        StreamTableScan { table: t, columns: [v1, _row_id], pk_indices: [1] }
+        StreamTableScan { table: t, columns: [v1, _row_id#0], pk_indices: [1] }
 - sql: |
     select length(trim(trailing '1' from '12'))+length(trim(leading '2' from '23'))+length(trim(both '3' from '34'));
   batch_plan: |
@@ -89,4 +89,4 @@
   batch_plan: |
     BatchDelete { table: t }
       BatchFilter { predicate: ($1 = 1:Int32) }
-        BatchScan { table: t, columns: [_row_id, v1, v2] }
+        BatchScan { table: t, columns: [_row_id#0, v1, v2] }
diff --git a/rust/frontend/test_runner/tests/testdata/basic_query_2.yaml b/rust/frontend/test_runner/tests/testdata/basic_query_2.yaml
index 657511026bc7..7e4a909a59a2 100644
--- a/rust/frontend/test_runner/tests/testdata/basic_query_2.yaml
+++ b/rust/frontend/test_runner/tests/testdata/basic_query_2.yaml
@@ -3,7 +3,7 @@
     delete from t;
   batch_plan: |
     BatchDelete { table: t }
-      BatchScan { table: t, columns: [_row_id, v1, v2] }
+      BatchScan { table: t, columns: [_row_id#0, v1, v2] }
 - sql: |
     create table t (v1 int, v2 int);
     select v1 from t;
@@ -11,8 +11,8 @@
     BatchExchange { order: [], dist: Single }
       BatchScan { table: t, columns: [v1] }
   stream_plan: |
-    StreamMaterialize { columns: [v1, _row_id(hidden)], pk_columns: [_row_id] }
-      StreamTableScan { table: t, columns: [v1, _row_id], pk_indices: [1] }
+    StreamMaterialize { columns: [v1, _row_id#0(hidden)], pk_columns: [_row_id#0] }
+      StreamTableScan { table: t, columns: [v1, _row_id#0], pk_indices: [1] }
 - sql: |
     values(cast(1 as bigint));
   batch_plan: |
@@ -44,49 +44,51 @@
     create table t1 (v1 int, v2 int);
     create table t2 (v1 int, v2 int);
     create table t3 (v1 int, v2 int);
-    select * from t1 join t2 on (t1.v1 = t2.v1) join t3 on (t2.v2 = t3.v2);
+    select t1.v1 as t1_v1, t1.v2 as t1_v2, t2.v1 as t2_v1, t2.v2 as t2_v2, t3.v1 as t3_v1, t3.v2 as t3_v2 from t1 join t2 on (t1.v1 = t2.v1) join t3 on (t2.v2 = t3.v2);
   batch_plan: |
     BatchExchange { order: [], dist: Single }
-      BatchHashJoin { type: Inner, predicate: $3 = $5 }
-        BatchExchange { order: [], dist: HashShard([3]) }
-          BatchHashJoin { type: Inner, predicate: $0 = $2 }
-            BatchExchange { order: [], dist: HashShard([0]) }
-              BatchScan { table: t1, columns: [v1, v2] }
-            BatchExchange { order: [], dist: HashShard([0]) }
-              BatchScan { table: t2, columns: [v1, v2] }
-        BatchExchange { order: [], dist: HashShard([1]) }
-          BatchScan { table: t3, columns: [v1, v2] }
+      BatchProject { exprs: [$0, $1, $2, $3, $4, $5], expr_alias: [t1_v1, t1_v2, t2_v1, t2_v2, t3_v1, t3_v2] }
+        BatchHashJoin { type: Inner, predicate: $3 = $5 }
+          BatchExchange { order: [], dist: HashShard([3]) }
+            BatchHashJoin { type: Inner, predicate: $0 = $2 }
+              BatchExchange { order: [], dist: HashShard([0]) }
+                BatchScan { table: t1, columns: [v1, v2] }
+              BatchExchange { order: [], dist: HashShard([0]) }
+                BatchScan { table: t2, columns: [v1, v2] }
+          BatchExchange { order: [], dist: HashShard([1]) }
+            BatchScan { table: t3, columns: [v1, v2] }
   stream_plan: |
-    StreamMaterialize { columns: [v1, v2, _row_id, v1, v2(hidden), _row_id(hidden), v1, v2, _row_id(hidden)], pk_columns: [_row_id, _row_id, _row_id] }
-      StreamHashJoin { type: Inner, predicate: $4 = $7 }
-        StreamExchange { dist: HashShard([4]) }
-          StreamHashJoin { type: Inner, predicate: $0 = $3 }
-            StreamExchange { dist: HashShard([0]) }
-              StreamTableScan { table: t1, columns: [v1, v2, _row_id], pk_indices: [2] }
-            StreamExchange { dist: HashShard([0]) }
-              StreamTableScan { table: t2, columns: [v1, v2, _row_id], pk_indices: [2] }
-        StreamExchange { dist: HashShard([1]) }
-          StreamTableScan { table: t3, columns: [v1, v2, _row_id], pk_indices: [2] }
+    StreamMaterialize { columns: [t1_v1, t1_v2, t2_v1, t2_v2, t3_v1, t3_v2, _row_id#0(hidden), _row_id#1(hidden)], pk_columns: [t2_v1, _row_id#0, _row_id#1] }
+      StreamProject { exprs: [$0, $1, $2, $3, $6, $7, $5, $8], expr_alias: [t1_v1, t1_v2, t2_v1, t2_v2, t3_v1, t3_v2,  ,  ] }
+        StreamHashJoin { type: Inner, predicate: $4 = $7 }
+          StreamExchange { dist: HashShard([4]) }
+            StreamHashJoin { type: Inner, predicate: $0 = $3 }
+              StreamExchange { dist: HashShard([0]) }
+                StreamTableScan { table: t1, columns: [v1, v2, _row_id#0], pk_indices: [2] }
+              StreamExchange { dist: HashShard([0]) }
+                StreamTableScan { table: t2, columns: [v1, v2, _row_id#0], pk_indices: [2] }
+          StreamExchange { dist: HashShard([1]) }
+            StreamTableScan { table: t3, columns: [v1, v2, _row_id#0], pk_indices: [2] }
 - sql: |
     create table t1 (v1 int not null, v2 int not null);
     create table t2 (v1 int not null, v2 int not null);
-    select t1.v2, t2.v2 from t1 join t2 on t1.v1 = t2.v1;
+    select t1.v2 as t1_v2, t2.v2 as t2_v2 from t1 join t2 on t1.v1 = t2.v1;
   batch_plan: |
     BatchExchange { order: [], dist: Single }
-      BatchProject { exprs: [$1, $3], expr_alias: [v2, v2] }
+      BatchProject { exprs: [$1, $3], expr_alias: [t1_v2, t2_v2] }
         BatchHashJoin { type: Inner, predicate: $0 = $2 }
           BatchExchange { order: [], dist: HashShard([0]) }
             BatchScan { table: t1, columns: [v1, v2] }
           BatchExchange { order: [], dist: HashShard([0]) }
             BatchScan { table: t2, columns: [v1, v2] }
   stream_plan: |
-    StreamMaterialize { columns: [v2, v2, expr#2(hidden), expr#3(hidden)], pk_columns: [expr#2, expr#3] }
-      StreamProject { exprs: [$1, $4, $2, $5], expr_alias: [v2, v2,  ,  ] }
+    StreamMaterialize { columns: [t1_v2, t2_v2, _row_id#0(hidden), _row_id#1(hidden)], pk_columns: [_row_id#0, _row_id#1] }
+      StreamProject { exprs: [$1, $4, $2, $5], expr_alias: [t1_v2, t2_v2,  ,  ] }
         StreamHashJoin { type: Inner, predicate: $0 = $3 }
           StreamExchange { dist: HashShard([0]) }
-            StreamTableScan { table: t1, columns: [v1, v2, _row_id], pk_indices: [2] }
+            StreamTableScan { table: t1, columns: [v1, v2, _row_id#0], pk_indices: [2] }
           StreamExchange { dist: HashShard([0]) }
-            StreamTableScan { table: t2, columns: [v1, v2, _row_id], pk_indices: [2] }
+            StreamTableScan { table: t2, columns: [v1, v2, _row_id#0], pk_indices: [2] }
 - sql: select 1
   batch_plan: |
     BatchProject { exprs: [1:Int32], expr_alias: [ ] }
@@ -109,7 +111,7 @@
           StreamExchange { dist: HashShard([0]) }
             StreamProject { exprs: [$0, $1, $2, $0, $3], expr_alias: [ ,  ,  ,  ,  ] }
               StreamExchange { dist: HashShard([0]) }
-                StreamTableScan { table: t, columns: [v1, v2, v3, _row_id], pk_indices: [3] }
+                StreamTableScan { table: t, columns: [v1, v2, v3, _row_id#0], pk_indices: [3] }
 - sql: |
     create table t(v1 int, v2 int, v3 int);
     select min(v1) + max(v2) * count(v3) from t;
@@ -119,10 +121,10 @@
         BatchExchange { order: [], dist: Single }
           BatchScan { table: t, columns: [v1, v2, v3] }
   stream_plan: |
-    StreamMaterialize { columns: [expr#0, expr#1(hidden), expr#2(hidden), expr#3(hidden), expr#4(hidden)], pk_columns: [expr#1, expr#2, expr#3, expr#4] }
+    StreamMaterialize { columns: [expr#0, agg#0(hidden), agg#1(hidden), agg#2(hidden), agg#3(hidden)], pk_columns: [agg#0, agg#1, agg#2, agg#3] }
       StreamProject { exprs: [($1 + ($2 * $3)), $0, $1, $2, $3], expr_alias: [ ,  ,  ,  ,  ] }
         StreamSimpleAgg { aggs: [count, min($0), max($1), count($2)] }
-          StreamTableScan { table: t, columns: [v1, v2, v3, _row_id], pk_indices: [3] }
+          StreamTableScan { table: t, columns: [v1, v2, v3, _row_id#0], pk_indices: [3] }
 - sql: |
     create table t(v1 int, v2 int);
     select v1 from t group by v2;
@@ -143,4 +145,4 @@
         StreamHashAgg { group_keys: [$0], aggs: [count, min($1), sum($2), count($2)] }
           StreamProject { exprs: [$2, $0, ($0 + $1), $3], expr_alias: [ ,  ,  ,  ] }
             StreamExchange { dist: HashShard([2]) }
-              StreamTableScan { table: t, columns: [v1, v2, v3, _row_id], pk_indices: [3] }
+              StreamTableScan { table: t, columns: [v1, v2, v3, _row_id#0], pk_indices: [3] }
diff --git a/rust/frontend/test_runner/tests/testdata/column_pruning.yaml b/rust/frontend/test_runner/tests/testdata/column_pruning.yaml
index 6118228746e5..825644b83788 100644
--- a/rust/frontend/test_runner/tests/testdata/column_pruning.yaml
+++ b/rust/frontend/test_runner/tests/testdata/column_pruning.yaml
@@ -3,7 +3,7 @@
     select v1 from t
   logical_plan: |
     LogicalProject { exprs: [$1], expr_alias: [v1] }
-      LogicalScan { table: t, columns: [_row_id, v1, v2] }
+      LogicalScan { table: t, columns: [_row_id#0, v1, v2] }
   optimized_logical_plan: |
     LogicalScan { table: t, columns: [v1] }
 - sql: |
@@ -13,7 +13,7 @@
   logical_plan: |
     LogicalProject { exprs: [$1], expr_alias: [v1] }
       LogicalFilter { predicate: ($2 > 2:Int32) }
-        LogicalScan { table: t, columns: [_row_id, v1, v2, v3] }
+        LogicalScan { table: t, columns: [_row_id#0, v1, v2, v3] }
   optimized_logical_plan: |
     LogicalProject { exprs: [$0], expr_alias: [v1] }
       LogicalFilter { predicate: ($1 > 2:Int32) }
@@ -26,8 +26,8 @@
   logical_plan: |
     LogicalProject { exprs: [$1, $5], expr_alias: [v1, v1] }
       LogicalJoin { type: Inner, on: ($2 = $6) }
-        LogicalScan { table: t1, columns: [_row_id, v1, v2, v3] }
-        LogicalScan { table: t2, columns: [_row_id, v1, v2, v3] }
+        LogicalScan { table: t1, columns: [_row_id#0, v1, v2, v3] }
+        LogicalScan { table: t2, columns: [_row_id#0, v1, v2, v3] }
   optimized_logical_plan: |
     LogicalProject { exprs: [$0, $2], expr_alias: [v1, v1] }
       LogicalJoin { type: Inner, on: ($1 = $3) }
@@ -42,7 +42,7 @@
       LogicalAgg { group_keys: [], agg_calls: [count($0)] }
         LogicalProject { exprs: [$1], expr_alias: [ ] }
           LogicalFilter { predicate: ($2 > 2:Int32) }
-            LogicalScan { table: t, columns: [_row_id, v1, v2, v3] }
+            LogicalScan { table: t, columns: [_row_id#0, v1, v2, v3] }
   optimized_logical_plan: |
     LogicalAgg { group_keys: [], agg_calls: [count($0)] }
       LogicalProject { exprs: [$0], expr_alias: [ ] }
@@ -54,7 +54,7 @@
     select 1 from t
   logical_plan: |
     LogicalProject { exprs: [1:Int32], expr_alias: [ ] }
-      LogicalScan { table: t, columns: [_row_id, v1, v2, v3] }
+      LogicalScan { table: t, columns: [_row_id#0, v1, v2, v3] }
   optimized_logical_plan: |
     LogicalProject { exprs: [1:Int32], expr_alias: [ ] }
       LogicalScan { table: t, columns: [] }
@@ -65,7 +65,7 @@
   logical_plan: |
     LogicalProject { exprs: [1:Int32], expr_alias: [ ] }
       LogicalFilter { predicate: ($2 > 1:Int32) }
-        LogicalScan { table: t, columns: [_row_id, v1, v2, v3] }
+        LogicalScan { table: t, columns: [_row_id#0, v1, v2, v3] }
   optimized_logical_plan: |
     LogicalProject { exprs: [1:Int32], expr_alias: [ ] }
       LogicalFilter { predicate: ($0 > 1:Int32) }
@@ -78,7 +78,7 @@
     LogicalProject { exprs: [$0], expr_alias: [ ] }
       LogicalAgg { group_keys: [], agg_calls: [count($0)] }
         LogicalProject { exprs: [1:Int32], expr_alias: [ ] }
-          LogicalScan { table: t, columns: [_row_id, v1, v2, v3] }
+          LogicalScan { table: t, columns: [_row_id#0, v1, v2, v3] }
   optimized_logical_plan: |
     LogicalAgg { group_keys: [], agg_calls: [count($0)] }
       LogicalProject { exprs: [1:Int32], expr_alias: [ ] }
@@ -92,7 +92,7 @@
       LogicalAgg { group_keys: [], agg_calls: [count($0)] }
         LogicalProject { exprs: [1:Int32], expr_alias: [ ] }
           LogicalFilter { predicate: ($2 > 1:Int32) }
-            LogicalScan { table: t, columns: [_row_id, v1, v2, v3] }
+            LogicalScan { table: t, columns: [_row_id#0, v1, v2, v3] }
   optimized_logical_plan: |
     LogicalAgg { group_keys: [], agg_calls: [count($0)] }
       LogicalProject { exprs: [1:Int32], expr_alias: [ ] }
@@ -107,8 +107,8 @@
     LogicalProject { exprs: [$1, $5], expr_alias: [v1, v1] }
       LogicalFilter { predicate: ($3 < 1:Int32) }
         LogicalJoin { type: Inner, on: ($2 = $6) }
-          LogicalScan { table: t1, columns: [_row_id, v1, v2, v3] }
-          LogicalScan { table: t2, columns: [_row_id, v1, v2, v3] }
+          LogicalScan { table: t1, columns: [_row_id#0, v1, v2, v3] }
+          LogicalScan { table: t2, columns: [_row_id#0, v1, v2, v3] }
   optimized_logical_plan: |
     LogicalProject { exprs: [$0, $2], expr_alias: [v1, v1] }
       LogicalJoin { type: Inner, on: ($1 = $3) }
@@ -126,7 +126,7 @@
       LogicalAgg { group_keys: [], agg_calls: [count($0), count($1)] }
         LogicalProject { exprs: [1:Int32, $1], expr_alias: [ ,  ] }
           LogicalFilter { predicate: ($2 > 1:Int32) }
-            LogicalScan { table: t, columns: [_row_id, v1, v2, v3] }
+            LogicalScan { table: t, columns: [_row_id#0, v1, v2, v3] }
   optimized_logical_plan: |
     LogicalAgg { group_keys: [], agg_calls: [count($0), count($1)] }
       LogicalProject { exprs: [1:Int32, $0], expr_alias: [ ,  ] }
diff --git a/rust/frontend/test_runner/tests/testdata/limit.yaml b/rust/frontend/test_runner/tests/testdata/limit.yaml
index 7a1bb0c8b942..49ad5924d7f9 100644
--- a/rust/frontend/test_runner/tests/testdata/limit.yaml
+++ b/rust/frontend/test_runner/tests/testdata/limit.yaml
@@ -4,14 +4,14 @@
   logical_plan: |
     LogicalLimit { limit: 4, offset: 0 }
       LogicalProject { exprs: [$1], expr_alias: [v] }
-        LogicalScan { table: t, columns: [_row_id, v] }
+        LogicalScan { table: t, columns: [_row_id#0, v] }
 - sql: |
     create table t (v int not null);
     select * from t offset 4;
   logical_plan: |
     LogicalLimit { limit: 9223372036854775807, offset: 4 }
       LogicalProject { exprs: [$1], expr_alias: [v] }
-        LogicalScan { table: t, columns: [_row_id, v] }
+        LogicalScan { table: t, columns: [_row_id#0, v] }
 - sql: |
     create table t (v int not null);
     select * from ( select * from t limit 5 ) limit 4;
@@ -20,4 +20,4 @@
       LogicalProject { exprs: [$0], expr_alias: [v] }
         LogicalLimit { limit: 5, offset: 0 }
           LogicalProject { exprs: [$1], expr_alias: [v] }
-            LogicalScan { table: t, columns: [_row_id, v] }
+            LogicalScan { table: t, columns: [_row_id#0, v] }
diff --git a/rust/frontend/test_runner/tests/testdata/mv_on_mv.yaml b/rust/frontend/test_runner/tests/testdata/mv_on_mv.yaml
index fb7866610c29..e1aaab260d4b 100644
--- a/rust/frontend/test_runner/tests/testdata/mv_on_mv.yaml
+++ b/rust/frontend/test_runner/tests/testdata/mv_on_mv.yaml
@@ -10,10 +10,10 @@
   sql: |
     select m1.v1 as m1v1, m1.v2 as m1v2, m2.v1 as m2v1, m2.v2 as m2v2 from m1 join m2 on m1.v1 = m2.v1;
   stream_plan: |
-    StreamMaterialize { columns: [m1v1, m1v2, m2v1, m2v2, expr#4(hidden), expr#5(hidden)], pk_columns: [expr#4, expr#5] }
+    StreamMaterialize { columns: [m1v1, m1v2, m2v1, m2v2, _row_id#0(hidden), _row_id#1(hidden)], pk_columns: [_row_id#0, _row_id#1] }
       StreamProject { exprs: [$0, $1, $3, $4, $2, $5], expr_alias: [m1v1, m1v2, m2v1, m2v2,  ,  ] }
         StreamHashJoin { type: Inner, predicate: $0 = $3 }
           StreamExchange { dist: HashShard([0]) }
-            StreamTableScan { table: m1, columns: [v1, v2, _row_id], pk_indices: [2] }
+            StreamTableScan { table: m1, columns: [v1, v2, _row_id#0], pk_indices: [2] }
           StreamExchange { dist: HashShard([0]) }
-            StreamTableScan { table: m2, columns: [v1, v2, _row_id], pk_indices: [2] }
+            StreamTableScan { table: m2, columns: [v1, v2, _row_id#0], pk_indices: [2] }
diff --git a/rust/frontend/test_runner/tests/testdata/order_by.yaml b/rust/frontend/test_runner/tests/testdata/order_by.yaml
index efd3ba4f5a90..b63ad3da5212 100644
--- a/rust/frontend/test_runner/tests/testdata/order_by.yaml
+++ b/rust/frontend/test_runner/tests/testdata/order_by.yaml
@@ -11,7 +11,7 @@
   batch_plan: |
     BatchExchange { order: [$1 ASC], dist: Single }
       BatchSort { order: [$1 ASC] }
-        BatchScan { table: t, columns: [_row_id, v1, v2] }
+        BatchScan { table: t, columns: [_row_id#0, v1, v2] }
 - sql: |
     create table t (v1 bigint, v2 double precision);
     select v1, v1+1 from t order by v1;
diff --git a/rust/frontend/test_runner/tests/testdata/predicate_pushdown.yaml b/rust/frontend/test_runner/tests/testdata/predicate_pushdown.yaml
index e4b3e2d0ab2b..1a4d6e2cbb00 100644
--- a/rust/frontend/test_runner/tests/testdata/predicate_pushdown.yaml
+++ b/rust/frontend/test_runner/tests/testdata/predicate_pushdown.yaml
@@ -6,8 +6,8 @@
     LogicalProject { exprs: [$1, $2, $3, $5, $6, $7], expr_alias: [v1, v2, v3, v1, v2, v3] }
       LogicalFilter { predicate: ($6 > 2:Int32) }
         LogicalJoin { type: Inner, on: ($1 = $6) AND ($1 > 1:Int32) }
-          LogicalScan { table: t1, columns: [_row_id, v1, v2, v3] }
-          LogicalScan { table: t2, columns: [_row_id, v1, v2, v3] }
+          LogicalScan { table: t1, columns: [_row_id#0, v1, v2, v3] }
+          LogicalScan { table: t2, columns: [_row_id#0, v1, v2, v3] }
   optimized_logical_plan: |
     LogicalJoin { type: Inner, on: ($0 = $4) }
       LogicalFilter { predicate: ($0 > 1:Int32) }
@@ -21,7 +21,7 @@
     LogicalProject { exprs: [$0, $1], expr_alias: [v1, v2] }
       LogicalFilter { predicate: ($1 > 1:Int32) }
         LogicalProject { exprs: [$1, $2], expr_alias: [v1, v2] }
-          LogicalScan { table: t, columns: [_row_id, v1, v2] }
+          LogicalScan { table: t, columns: [_row_id#0, v1, v2] }
   optimized_logical_plan: |
     LogicalFilter { predicate: ($1 > 1:Int32) }
       LogicalScan { table: t, columns: [v1, v2] }
@@ -32,7 +32,7 @@
     LogicalProject { exprs: [$1], expr_alias: [v1] }
       LogicalFilter { predicate: ($0 > 1:Int32) }
         LogicalProject { exprs: [$2, $1], expr_alias: [v2, v1] }
-          LogicalScan { table: t, columns: [_row_id, v1, v2] }
+          LogicalScan { table: t, columns: [_row_id#0, v1, v2] }
   optimized_logical_plan: |
     LogicalProject { exprs: [$0], expr_alias: [v1] }
       LogicalFilter { predicate: ($1 > 1:Int32) }
@@ -45,7 +45,7 @@
       LogicalFilter { predicate: ($0 > 1:Int32) }
         LogicalProject { exprs: [$2, $1], expr_alias: [a2, v1] }
           LogicalFilter { predicate: ($1 > 2:Int32) }
-            LogicalScan { table: t, columns: [_row_id, v1, v2] }
+            LogicalScan { table: t, columns: [_row_id#0, v1, v2] }
   optimized_logical_plan: |
     LogicalProject { exprs: [$0], expr_alias: [v1] }
       LogicalFilter { predicate: ($1 > 1:Int32) }
@@ -60,7 +60,7 @@
         LogicalProject { exprs: [$0, $1], expr_alias: [v1, min] }
           LogicalAgg { group_keys: [0], agg_calls: [min($1)] }
             LogicalProject { exprs: [$1, $2], expr_alias: [ ,  ] }
-              LogicalScan { table: t, columns: [_row_id, v1, v2, v3, v4] }
+              LogicalScan { table: t, columns: [_row_id#0, v1, v2, v3, v4] }
   optimized_logical_plan: |
     LogicalProject { exprs: [$0, $1], expr_alias: [v1, min] }
       LogicalFilter { predicate: ($1 > 1:Int32) AND ($0 > $1) }
diff --git a/rust/frontend/test_runner/tests/testdata/stream_proto.yaml b/rust/frontend/test_runner/tests/testdata/stream_proto.yaml
index 29790cc74783..da56abb1ee04 100644
--- a/rust/frontend/test_runner/tests/testdata/stream_proto.yaml
+++ b/rust/frontend/test_runner/tests/testdata/stream_proto.yaml
@@ -22,7 +22,7 @@
                     - columnType:
                         typeName: INT64
                         isNullable: true
-                      name: _row_id
+                      name: "_row_id#0"
             pkIndices:
               - 1
             chainNode:
@@ -32,7 +32,7 @@
                 - dataType:
                     typeName: INT64
                     isNullable: true
-                  name: _row_id
+                  name: "_row_id#0"
                 - dataType:
                     typeName: INT32
                     isNullable: true
@@ -88,7 +88,7 @@
             typeName: INT64
             isNullable: true
           columnId: 1
-          name: _row_id
+          name: "_row_id#0"
         isHidden: true
     pkColumnIds:
       - 1
@@ -117,7 +117,7 @@
                 - columnType:
                     typeName: INT64
                     isNullable: true
-                  name: _row_id
+                  name: "_row_id#0"
         pkIndices:
           - 1
         chainNode:
@@ -127,7 +127,7 @@
             - dataType:
                 typeName: INT64
                 isNullable: true
-              name: _row_id
+              name: "_row_id#0"
             - dataType:
                 typeName: INT32
                 isNullable: true
@@ -162,7 +162,7 @@
             typeName: INT64
             isNullable: true
           columnId: 1
-          name: _row_id
+          name: "_row_id#0"
         isHidden: true
     pkColumnIds:
       - 1
@@ -191,7 +191,7 @@
                 - columnType:
                     typeName: INT64
                     isNullable: true
-                  name: _row_id
+                  name: "_row_id#0"
         pkIndices:
           - 1
         chainNode:
@@ -201,7 +201,7 @@
             - dataType:
                 typeName: INT64
                 isNullable: true
-              name: _row_id
+              name: "_row_id#0"
             - dataType:
                 typeName: INT32
                 isNullable: true
@@ -240,7 +240,7 @@
             typeName: INT64
             isNullable: true
           columnId: 1
-          name: _row_id
+          name: "_row_id#0"
         isHidden: true
     pkColumnIds:
       - 1
@@ -270,7 +270,7 @@
                     - columnType:
                         typeName: INT64
                         isNullable: true
-                      name: _row_id
+                      name: "_row_id#0"
             pkIndices:
               - 1
             chainNode:
@@ -280,7 +280,7 @@
                 - dataType:
                     typeName: INT64
                     isNullable: true
-                  name: _row_id
+                  name: "_row_id#0"
                 - dataType:
                     typeName: INT32
                     isNullable: true
@@ -379,7 +379,7 @@
                                 - columnType:
                                     typeName: INT64
                                     isNullable: true
-                                  name: _row_id
+                                  name: "_row_id#0"
                         pkIndices:
                           - 2
                         chainNode:
@@ -389,7 +389,7 @@
                             - dataType:
                                 typeName: INT64
                                 isNullable: true
-                              name: _row_id
+                              name: "_row_id#0"
                             - dataType:
                                 typeName: INT32
                                 isNullable: true
@@ -417,7 +417,7 @@
                         - dataType:
                             typeName: INT64
                             isNullable: true
-                          name: _row_id
+                          name: "_row_id#0"
                       strategy:
                         type: HASH
                         columnIndices:
@@ -505,7 +505,7 @@
             typeName: INT32
             isNullable: true
           columnId: 1
-          name: "expr#1"
+          name: v2
         isHidden: true
     pkColumnIds:
       - 1
diff --git a/rust/frontend/test_runner/tests/testdata/subquery.yaml b/rust/frontend/test_runner/tests/testdata/subquery.yaml
index a64811ac57be..351ff319f7d8 100644
--- a/rust/frontend/test_runner/tests/testdata/subquery.yaml
+++ b/rust/frontend/test_runner/tests/testdata/subquery.yaml
@@ -5,7 +5,7 @@
     LogicalProject { exprs: [$0], expr_alias: [v1] }
       LogicalFilter { predicate: ($1 > 1:Int32) }
         LogicalProject { exprs: [$1, $2], expr_alias: [v1, v2] }
-          LogicalScan { table: t, columns: [_row_id, v1, v2] }
+          LogicalScan { table: t, columns: [_row_id#0, v1, v2] }
 - sql: |
     /* merge and then eliminate */
     create table t (v1 bigint, v2 double precision);
@@ -13,7 +13,7 @@
   logical_plan: |
     LogicalProject { exprs: [$0, $1], expr_alias: [v1, v2] }
       LogicalProject { exprs: [$1, $2], expr_alias: [a1, a2] }
-        LogicalScan { table: t, columns: [_row_id, v1, v2] }
+        LogicalScan { table: t, columns: [_row_id#0, v1, v2] }
   optimized_logical_plan: |
     LogicalScan { table: t, columns: [v1, v2] }
 - sql: |
@@ -27,7 +27,7 @@
     LogicalProject { exprs: [$1], expr_alias: [v3] }
       LogicalFilter { predicate: ($0 > 1:Int32) }
         LogicalProject { exprs: [$2, $1], expr_alias: [v2, v3] }
-          LogicalScan { table: t, columns: [_row_id, v1, v2] }
+          LogicalScan { table: t, columns: [_row_id#0, v1, v2] }
 - sql: |
     /* consecutive projects are merged */
     create table t (v1 bigint, v2 double precision);
@@ -35,7 +35,7 @@
   logical_plan: |
     LogicalProject { exprs: [$0, 2:Int32], expr_alias: [v1,  ] }
       LogicalProject { exprs: [$1, $2, 1:Int32], expr_alias: [v1, v2,  ] }
-        LogicalScan { table: t, columns: [_row_id, v1, v2] }
+        LogicalScan { table: t, columns: [_row_id#0, v1, v2] }
   optimized_logical_plan: |
     LogicalProject { exprs: [$0, 2:Int32], expr_alias: [v1,  ] }
       LogicalScan { table: t, columns: [v1] }
@@ -45,7 +45,7 @@
   logical_plan: |
     LogicalProject { exprs: [$0, $1], expr_alias: [v1, v2] }
       LogicalProject { exprs: [$1, $2], expr_alias: [v1, v2] }
-        LogicalScan { table: t, columns: [_row_id, v1, v2] }
+        LogicalScan { table: t, columns: [_row_id#0, v1, v2] }
   optimized_logical_plan: |
     LogicalScan { table: t, columns: [v1, v2] }
 - sql: |
@@ -56,8 +56,8 @@
     LogicalProject { exprs: [$0, $1, $3, $4], expr_alias: [v1, v2, v1, v2] }
       LogicalJoin { type: Inner, on: true:Boolean }
         LogicalProject { exprs: [$1, $2], expr_alias: [v1, v2] }
-          LogicalScan { table: t, columns: [_row_id, v1, v2] }
-        LogicalScan { table: t, columns: [_row_id, v1, v2] }
+          LogicalScan { table: t, columns: [_row_id#0, v1, v2] }
+        LogicalScan { table: t, columns: [_row_id#0, v1, v2] }
 - sql: |
     /* table alias not supported yet */
     create table t (v1 bigint, v2 double precision);
diff --git a/rust/frontend/test_runner/tests/testdata/subquery_expr.yaml b/rust/frontend/test_runner/tests/testdata/subquery_expr.yaml
index b6872ec3515b..31794f1df228 100644
--- a/rust/frontend/test_runner/tests/testdata/subquery_expr.yaml
+++ b/rust/frontend/test_runner/tests/testdata/subquery_expr.yaml
@@ -12,18 +12,18 @@
   logical_plan: |
     LogicalProject { exprs: [$2, 1:Int32], expr_alias: [ ,  ] }
       LogicalJoin { type: LeftOuter, on: true:Boolean }
-        LogicalScan { table: t, columns: [_row_id, x] }
+        LogicalScan { table: t, columns: [_row_id#0, x] }
         LogicalProject { exprs: [$1], expr_alias: [x] }
-          LogicalScan { table: t, columns: [_row_id, x] }
+          LogicalScan { table: t, columns: [_row_id#0, x] }
 - sql: |
     create table t(x int);
     select (select x from t) + 1 from t;
   logical_plan: |
     LogicalProject { exprs: [($2 + 1:Int32)], expr_alias: [ ] }
       LogicalJoin { type: LeftOuter, on: true:Boolean }
-        LogicalScan { table: t, columns: [_row_id, x] }
+        LogicalScan { table: t, columns: [_row_id#0, x] }
         LogicalProject { exprs: [$1], expr_alias: [x] }
-          LogicalScan { table: t, columns: [_row_id, x] }
+          LogicalScan { table: t, columns: [_row_id#0, x] }
 - sql: |
     create table t(x int);
     select (select x from t), (select 1);
@@ -33,7 +33,7 @@
         LogicalJoin { type: LeftOuter, on: true:Boolean }
           LogicalValues { rows: [[]], schema: Schema { fields: [] } }
           LogicalProject { exprs: [$1], expr_alias: [x] }
-            LogicalScan { table: t, columns: [_row_id, x] }
+            LogicalScan { table: t, columns: [_row_id#0, x] }
         LogicalProject { exprs: [1:Int32], expr_alias: [ ] }
           LogicalValues { rows: [[]], schema: Schema { fields: [] } }
 - sql: |
@@ -42,12 +42,12 @@
   logical_plan: |
     LogicalProject { exprs: [($1 + $2)], expr_alias: [v3] }
       LogicalJoin { type: LeftOuter, on: true:Boolean }
-        LogicalScan { table: t, columns: [_row_id, x] }
+        LogicalScan { table: t, columns: [_row_id#0, x] }
         LogicalProject { exprs: [($1 + $2)], expr_alias: [v2] }
           LogicalJoin { type: LeftOuter, on: true:Boolean }
-            LogicalScan { table: t, columns: [_row_id, x] }
+            LogicalScan { table: t, columns: [_row_id#0, x] }
             LogicalProject { exprs: [$1], expr_alias: [v1] }
-              LogicalScan { table: t, columns: [_row_id, x] }
+              LogicalScan { table: t, columns: [_row_id#0, x] }
 - sql: |
     select (select 1, 2);
   binder_error: 'Bind error: subquery must return only one column'
@@ -62,7 +62,7 @@
           LogicalProject { exprs: [($0 >= 1:Int32)], expr_alias: [ ] }
             LogicalAgg { group_keys: [], agg_calls: [count] }
               LogicalProject { exprs: [$1], expr_alias: [x] }
-                LogicalScan { table: t, columns: [_row_id, x] }
+                LogicalScan { table: t, columns: [_row_id#0, x] }
 - sql: |
     create table t1(x int);
     create table t2(x int);
@@ -71,11 +71,11 @@
     LogicalProject { exprs: [$1], expr_alias: [x] }
       LogicalFilter { predicate: $2 }
         LogicalJoin { type: LeftOuter, on: true:Boolean }
-          LogicalScan { table: t1, columns: [_row_id, x] }
+          LogicalScan { table: t1, columns: [_row_id#0, x] }
           LogicalProject { exprs: [($0 >= 1:Int32)], expr_alias: [ ] }
             LogicalAgg { group_keys: [], agg_calls: [count] }
               LogicalProject { exprs: [$1], expr_alias: [x] }
-                LogicalScan { table: t2, columns: [_row_id, x] }
+                LogicalScan { table: t2, columns: [_row_id#0, x] }
 - sql: |
     create table t(x int);
     select x from t where exists (select * from t);
@@ -83,11 +83,11 @@
     LogicalProject { exprs: [$1], expr_alias: [x] }
       LogicalFilter { predicate: $2 }
         LogicalJoin { type: LeftOuter, on: true:Boolean }
-          LogicalScan { table: t, columns: [_row_id, x] }
+          LogicalScan { table: t, columns: [_row_id#0, x] }
           LogicalProject { exprs: [($0 >= 1:Int32)], expr_alias: [ ] }
             LogicalAgg { group_keys: [], agg_calls: [count] }
               LogicalProject { exprs: [$1], expr_alias: [x] }
-                LogicalScan { table: t, columns: [_row_id, x] }
+                LogicalScan { table: t, columns: [_row_id#0, x] }
 - sql: |
     create table t1(x int);
     create table t2(x int);
@@ -96,6 +96,6 @@
     LogicalProject { exprs: [$1], expr_alias: [x] }
       LogicalFilter { predicate: ($1 > $2) }
         LogicalJoin { type: LeftOuter, on: true:Boolean }
-          LogicalScan { table: t1, columns: [_row_id, x] }
+          LogicalScan { table: t1, columns: [_row_id#0, x] }
           LogicalProject { exprs: [$1], expr_alias: [x] }
-            LogicalScan { table: t2, columns: [_row_id, x] }
+            LogicalScan { table: t2, columns: [_row_id#0, x] }
diff --git a/rust/frontend/test_runner/tests/testdata/tpch.yaml b/rust/frontend/test_runner/tests/testdata/tpch.yaml
index 6959cf300efc..a271c0aba72a 100644
--- a/rust/frontend/test_runner/tests/testdata/tpch.yaml
+++ b/rust/frontend/test_runner/tests/testdata/tpch.yaml
@@ -80,7 +80,7 @@
         StreamHashAgg { group_keys: [$0, $1], aggs: [count, sum($2), sum($3), sum($4), sum($5), sum($6), count($6), sum($7), count($7), sum($8), count($8), count] }
           StreamProject { exprs: [$4, $5, $0, $1, ($1 * (1:Int32 - $2)), (($1 * (1:Int32 - $2)) * (1:Int32 + $3)), $0, $1, $2, $6], expr_alias: [ ,  ,  ,  ,  ,  ,  ,  ,  ,  ] }
             StreamExchange { dist: HashShard([4, 5]) }
-              StreamTableScan { table: lineitem, columns: [l_quantity, l_extendedprice, l_discount, l_tax, l_returnflag, l_linestatus, _row_id], pk_indices: [6] }
+              StreamTableScan { table: lineitem, columns: [l_quantity, l_extendedprice, l_discount, l_tax, l_returnflag, l_linestatus, _row_id#0], pk_indices: [6] }
 - id: tpch_q3
   before:
     - create_tables
@@ -143,13 +143,13 @@
                       StreamProject { exprs: [$0, $2], expr_alias: [ ,  ] }
                         StreamExchange { dist: HashShard([0]) }
                           StreamFilter { predicate: ($1 = "FURNITURE":Varchar) AND true:Boolean AND true:Boolean }
-                            StreamTableScan { table: customer, columns: [c_custkey, c_mktsegment, _row_id], pk_indices: [2] }
+                            StreamTableScan { table: customer, columns: [c_custkey, c_mktsegment, _row_id#0], pk_indices: [2] }
                       StreamExchange { dist: HashShard([1]) }
                         StreamFilter { predicate:  }
-                          StreamTableScan { table: orders, columns: [o_orderkey, o_custkey, o_orderdate, o_shippriority, _row_id], pk_indices: [4] }
+                          StreamTableScan { table: orders, columns: [o_orderkey, o_custkey, o_orderdate, o_shippriority, _row_id#0], pk_indices: [4] }
                 StreamExchange { dist: HashShard([0]) }
                   StreamFilter { predicate:  }
-                    StreamTableScan { table: lineitem, columns: [l_orderkey, l_extendedprice, l_discount, _row_id], pk_indices: [3] }
+                    StreamTableScan { table: lineitem, columns: [l_orderkey, l_extendedprice, l_discount, _row_id#0], pk_indices: [3] }
 - id: tpch_q6
   before:
     - create_tables
@@ -171,9 +171,9 @@
             BatchFilter { predicate: ($0 < 24:Int32) }
               BatchScan { table: lineitem, columns: [l_quantity, l_extendedprice, l_discount] }
   stream_plan: |
-    StreamMaterialize { columns: [revenue, expr#1(hidden)], pk_columns: [expr#1, revenue] }
+    StreamMaterialize { columns: [revenue, agg#0(hidden)], pk_columns: [agg#0, revenue] }
       StreamProject { exprs: [$1, $0], expr_alias: [revenue,  ] }
         StreamSimpleAgg { aggs: [count, sum($0)] }
           StreamProject { exprs: [($1 * $2), $3], expr_alias: [ ,  ] }
             StreamFilter { predicate: ($0 < 24:Int32) }
-              StreamTableScan { table: lineitem, columns: [l_quantity, l_extendedprice, l_discount, _row_id], pk_indices: [3] }
+              StreamTableScan { table: lineitem, columns: [l_quantity, l_extendedprice, l_discount, _row_id#0], pk_indices: [3] }