diff --git a/src/operator/src/insert.rs b/src/operator/src/insert.rs
index 0c38f9a1dc..b6dcf0a231 100644
--- a/src/operator/src/insert.rs
+++ b/src/operator/src/insert.rs
@@ -81,7 +81,7 @@ pub type InserterRef = Arc<Inserter>;
 
 /// Hint for the table type to create automatically.
 #[derive(Clone)]
-enum AutoCreateTableType {
+pub(crate) enum AutoCreateTableType {
     /// A logical table with the physical table name.
     Logical(String),
     /// A physical table.
@@ -680,38 +680,9 @@ impl Inserter {
         create_type: &AutoCreateTableType,
         ctx: &QueryContextRef,
     ) -> Result<CreateTableExpr> {
-        let mut table_options = Vec::with_capacity(4);
-        for key in VALID_TABLE_OPTION_KEYS {
-            if let Some(value) = ctx.extension(key) {
-                table_options.push((key, value));
-            }
-        }
-
         let mut engine_name = default_engine();
-        match create_type {
-            AutoCreateTableType::Logical(physical_table) => {
-                engine_name = METRIC_ENGINE_NAME;
-                table_options.push((LOGICAL_TABLE_METADATA_KEY, physical_table));
-            }
-            AutoCreateTableType::Physical => {
-                if let Some(append_mode) = ctx.extension(APPEND_MODE_KEY) {
-                    table_options.push((APPEND_MODE_KEY, append_mode));
-                }
-                if let Some(merge_mode) = ctx.extension(MERGE_MODE_KEY) {
-                    table_options.push((MERGE_MODE_KEY, merge_mode));
-                }
-            }
-            // Set append_mode to true for log table.
-            // because log tables should keep rows with the same ts and tags.
-            AutoCreateTableType::Log => {
-                table_options.push((APPEND_MODE_KEY, "true"));
-            }
-            AutoCreateTableType::LastNonNull => {
-                table_options.push((MERGE_MODE_KEY, "last_non_null"));
-            }
-            AutoCreateTableType::Trace => {
-                table_options.push((APPEND_MODE_KEY, "true"));
-            }
+        if matches!(create_type, AutoCreateTableType::Logical(_)) {
+            engine_name = METRIC_ENGINE_NAME;
         }
 
         let schema = ctx.current_schema();
@@ -722,11 +693,9 @@ impl Inserter {
             build_create_table_expr(&table_ref, request_schema, engine_name)?;
 
         info!("Table `{table_ref}` does not exist, try creating table");
-        for (k, v) in table_options {
-            create_table_expr
-                .table_options
-                .insert(k.to_string(), v.to_string());
-        }
+
+        // Use the common fill_table_options_for_create function to populate table options
+        fill_table_options_for_create(&mut create_table_expr.table_options, create_type, ctx);
 
         Ok(create_table_expr)
     }
@@ -919,6 +888,47 @@ fn validate_column_count_match(requests: &RowInsertRequests) -> Result<()> {
     Ok(())
 }
 
+/// Fill table options for a new table by create type.
+pub(crate) fn fill_table_options_for_create(
+    table_options: &mut std::collections::HashMap<String, String>,
+    create_type: &AutoCreateTableType,
+    ctx: &QueryContextRef,
+) {
+    for key in VALID_TABLE_OPTION_KEYS {
+        if let Some(value) = ctx.extension(key) {
+            table_options.insert(key.to_string(), value.to_string());
+        }
+    }
+
+    match create_type {
+        AutoCreateTableType::Logical(physical_table) => {
+            table_options.insert(
+                LOGICAL_TABLE_METADATA_KEY.to_string(),
+                physical_table.to_string(),
+            );
+        }
+        AutoCreateTableType::Physical => {
+            if let Some(append_mode) = ctx.extension(APPEND_MODE_KEY) {
+                table_options.insert(APPEND_MODE_KEY.to_string(), append_mode.to_string());
+            }
+            if let Some(merge_mode) = ctx.extension(MERGE_MODE_KEY) {
+                table_options.insert(MERGE_MODE_KEY.to_string(), merge_mode.to_string());
+            }
+        }
+        // Set append_mode to true for log table.
+        // because log tables should keep rows with the same ts and tags.
+        AutoCreateTableType::Log => {
+            table_options.insert(APPEND_MODE_KEY.to_string(), "true".to_string());
+        }
+        AutoCreateTableType::LastNonNull => {
+            table_options.insert(MERGE_MODE_KEY.to_string(), "last_non_null".to_string());
+        }
+        AutoCreateTableType::Trace => {
+            table_options.insert(APPEND_MODE_KEY.to_string(), "true".to_string());
+        }
+    }
+}
+
 /// Builds a [CreateTableExpr] for the given table and schema.
 pub(crate) fn build_create_table_expr(
     table: &TableReference,
diff --git a/src/operator/src/schema_helper.rs b/src/operator/src/schema_helper.rs
index 41d22045b2..bd1d586eca 100644
--- a/src/operator/src/schema_helper.rs
+++ b/src/operator/src/schema_helper.rs
@@ -18,10 +18,7 @@ use std::collections::HashMap;
 use std::sync::Arc;
 
 use api::v1::alter_table_expr::Kind;
-use api::v1::{
-    AddColumn, AddColumns, AlterTableExpr, ColumnDataType, ColumnDef, ColumnSchema,
-    CreateTableExpr, SemanticType,
-};
+use api::v1::{AlterTableExpr, ColumnDataType, ColumnSchema, CreateTableExpr, SemanticType};
 use catalog::CatalogManagerRef;
 use common_catalog::consts::{
     default_engine, is_readonly_schema, DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME,
@@ -57,7 +54,7 @@ use crate::error::{
     TableMetadataManagerSnafu, TableNotFoundSnafu, UnexpectedSnafu,
 };
 use crate::expr_helper;
-use crate::insert::build_create_table_expr;
+use crate::insert::{build_create_table_expr, fill_table_options_for_create, AutoCreateTableType};
 use crate::statement::ddl::{create_table_info, parse_partitions, verify_alter, NAME_PATTERN_REG};
 
 /// Helper to query and manipulate (CREATE/ALTER) table schemas.
@@ -688,10 +685,12 @@ async fn ensure_logical_tables_for_metrics(
                     METRIC_ENGINE_NAME,
                 )?;
                 create_expr.create_if_not_exists = true;
-                // Add the logical table metadata key to link with physical table
-                create_expr.table_options.insert(
-                    LOGICAL_TABLE_METADATA_KEY.to_string(),
-                    physical_table_name.clone(),
+                let create_type = AutoCreateTableType::Logical(physical_table_name.clone());
+                // Fill table options.
+                fill_table_options_for_create(
+                    &mut create_expr.table_options,
+                    &create_type,
+                    query_ctx,
                 );
 
                 tables_to_create.push(create_expr);