diff --git a/src/query/expression/src/schema.rs b/src/query/expression/src/schema.rs
index e14304408aaa6..be369c3c5ab14 100644
--- a/src/query/expression/src/schema.rs
+++ b/src/query/expression/src/schema.rs
@@ -65,7 +65,6 @@ pub const CHANGE_ACTION_COL_NAME: &str = "change$action";
 pub const CHANGE_IS_UPDATE_COL_NAME: &str = "change$is_update";
 pub const CHANGE_ROW_ID_COL_NAME: &str = "change$row_id";
 
-pub const ROW_NUMBER_COL_NAME: &str = "_row_number";
 pub const PREDICATE_COLUMN_NAME: &str = "_predicate";
 
 // stream column id.
@@ -94,7 +93,6 @@ pub static INTERNAL_COLUMNS: LazyLock<HashSet<&'static str>> = LazyLock::new(||
         CHANGE_ACTION_COL_NAME,
         CHANGE_IS_UPDATE_COL_NAME,
         CHANGE_ROW_ID_COL_NAME,
-        ROW_NUMBER_COL_NAME,
         PREDICATE_COLUMN_NAME,
         ORIGIN_VERSION_COL_NAME,
         ORIGIN_BLOCK_ID_COL_NAME,
diff --git a/src/query/sql/src/executor/physical_plans/physical_hash_join.rs b/src/query/sql/src/executor/physical_plans/physical_hash_join.rs
index 641648cce95ee..45a2efe6772fa 100644
--- a/src/query/sql/src/executor/physical_plans/physical_hash_join.rs
+++ b/src/query/sql/src/executor/physical_plans/physical_hash_join.rs
@@ -27,7 +27,6 @@ use databend_common_expression::DataField;
 use databend_common_expression::DataSchemaRef;
 use databend_common_expression::DataSchemaRefExt;
 use databend_common_expression::RemoteExpr;
-use databend_common_expression::ROW_NUMBER_COL_NAME;
 use databend_common_functions::BUILTIN_FUNCTIONS;
 use databend_storages_common_table_meta::table::get_change_type;
 
@@ -375,12 +374,6 @@ impl PhysicalPlanBuilder {
             None
         };
 
-        // for distributed merge into, there is a field called "_row_number", but
-        // it's not an internal row_number, we need to add it here
-        if let Some((index, _)) = build_schema.column_with_name(ROW_NUMBER_COL_NAME) {
-            build_projections.insert(index);
-        }
-
         let mut merged_fields =
             Vec::with_capacity(probe_projections.len() + build_projections.len());
         let mut probe_fields = Vec::with_capacity(probe_projections.len());
@@ -503,12 +496,6 @@ impl PhysicalPlanBuilder {
             }
         }
 
-        // for distributed merge into, there is a field called "_row_number", but
-        // it's not an internal row_number, we need to add it here
-        if let Some((index, _)) = projected_schema.column_with_name(ROW_NUMBER_COL_NAME) {
-            projections.insert(index);
-        }
-
         let mut output_fields = Vec::with_capacity(column_projections.len());
         for (i, field) in merged_fields.iter().enumerate() {
             if projections.contains(&i) {