diff --git a/src/catalog/src/kvbackend/manager.rs b/src/catalog/src/kvbackend/manager.rs
index 2c5e83b29e..0cfab086d1 100644
--- a/src/catalog/src/kvbackend/manager.rs
+++ b/src/catalog/src/kvbackend/manager.rs
@@ -13,7 +13,7 @@
 // limitations under the License.
 
 use std::any::Any;
-use std::collections::{BTreeSet, HashSet};
+use std::collections::BTreeSet;
 use std::sync::{Arc, Weak};
 
 use async_stream::try_stream;
@@ -166,32 +166,29 @@ impl KvBackendCatalogManager {
                 .context(TableMetadataManagerSnafu)?
         {
             let mut new_table_info = (*table.table_info()).clone();
-            // Gather all column names from the logical table
-            let logical_column_names: HashSet<_> = new_table_info
-                .meta
-                .schema
-                .column_schemas()
-                .iter()
-                .map(|col| &col.name)
-                .collect();
 
-            // Only preserve partition key indices where the corresponding columns exist in logical table
+            // Remap partition key indices from physical table to logical table
             new_table_info.meta.partition_key_indices = physical_table_info_value
                 .table_info
                 .meta
                 .partition_key_indices
                 .iter()
-                .filter(|&&index| {
+                .filter_map(|&physical_index| {
+                    // Get the column name from the physical table using the physical index
                     physical_table_info_value
                         .table_info
                         .meta
                         .schema
                         .column_schemas
-                        .get(index)
-                        .map(|physical_column| logical_column_names.contains(&physical_column.name))
-                        .unwrap_or(false)
+                        .get(physical_index)
+                        .and_then(|physical_column| {
+                            // Find the corresponding index in the logical table schema
+                            new_table_info
+                                .meta
+                                .schema
+                                .column_index_by_name(physical_column.name.as_str())
+                        })
                 })
-                .cloned()
                 .collect();
 
             let new_table = DistTable::table(Arc::new(new_table_info));
diff --git a/tests/cases/standalone/common/create/metric_engine_partition.result b/tests/cases/standalone/common/create/metric_engine_partition.result
index a9740e6e24..db3249e382 100644
--- a/tests/cases/standalone/common/create/metric_engine_partition.result
+++ b/tests/cases/standalone/common/create/metric_engine_partition.result
@@ -1,9 +1,12 @@
 create table metric_engine_partition (
     ts timestamp time index,
-    host string primary key,
+    host string,
     cpu double,
+    `one_partition_key` string,
+    `another_partition_key` string,
+    primary key(host, `one_partition_key`, `another_partition_key`)
 )
-partition on columns (host) (
+partition on columns (host, `one_partition_key`, `another_partition_key`) (
     host <= 'host1',
     host > 'host1' and host <= 'host2',
     host > 'host2'
@@ -64,26 +67,26 @@ Affected Rows: 0
 
 show create table logical_table_2;
 
-+-----------------+-------------------------------------------------+
-| Table           | Create Table                                    |
-+-----------------+-------------------------------------------------+
-| logical_table_2 | CREATE TABLE IF NOT EXISTS "logical_table_2" (  |
-|                 |   "cpu" DOUBLE NULL,                            |
-|                 |   "host" STRING NULL,                           |
-|                 |   "ts" TIMESTAMP(3) NOT NULL,                   |
-|                 |   TIME INDEX ("ts"),                            |
-|                 |   PRIMARY KEY ("host")                          |
-|                 | )                                               |
-|                 | PARTITION ON COLUMNS ("host") (                 |
-|                 |   host <= 'host1',                              |
-|                 |   host > 'host2',                               |
-|                 |   host > 'host1' AND host <= 'host2'            |
-|                 | )                                               |
-|                 | ENGINE=metric                                   |
-|                 | WITH(                                           |
-|                 |   on_physical_table = 'metric_engine_partition' |
-|                 | )                                               |
-+-----------------+-------------------------------------------------+
++-----------------+-------------------------------------------------------------------------------+
+| Table           | Create Table                                                                  |
++-----------------+-------------------------------------------------------------------------------+
+| logical_table_2 | CREATE TABLE IF NOT EXISTS "logical_table_2" (                                |
+|                 |   "cpu" DOUBLE NULL,                                                          |
+|                 |   "host" STRING NULL,                                                         |
+|                 |   "ts" TIMESTAMP(3) NOT NULL,                                                 |
+|                 |   TIME INDEX ("ts"),                                                          |
+|                 |   PRIMARY KEY ("host")                                                        |
+|                 | )                                                                             |
+|                 | PARTITION ON COLUMNS ("host", "one_partition_key", "another_partition_key") ( |
+|                 |   host <= 'host1',                                                            |
+|                 |   host > 'host2',                                                             |
+|                 |   host > 'host1' AND host <= 'host2'                                          |
+|                 | )                                                                             |
+|                 | ENGINE=metric                                                                 |
+|                 | WITH(                                                                         |
+|                 |   on_physical_table = 'metric_engine_partition'                               |
+|                 | )                                                                             |
++-----------------+-------------------------------------------------------------------------------+
 
 select count(*) from logical_table_2;
 
diff --git a/tests/cases/standalone/common/create/metric_engine_partition.sql b/tests/cases/standalone/common/create/metric_engine_partition.sql
index 28d5e45157..d068c88465 100644
--- a/tests/cases/standalone/common/create/metric_engine_partition.sql
+++ b/tests/cases/standalone/common/create/metric_engine_partition.sql
@@ -1,9 +1,12 @@
 create table metric_engine_partition (
     ts timestamp time index,
-    host string primary key,
+    host string,
     cpu double,
+    `one_partition_key` string,
+    `another_partition_key` string,
+    primary key(host, `one_partition_key`, `another_partition_key`)
 )
-partition on columns (host) (
+partition on columns (host, `one_partition_key`, `another_partition_key`) (
     host <= 'host1',
     host > 'host1' and host <= 'host2',
     host > 'host2'