diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/btree/BTreeIndexTopoBuilder.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/btree/BTreeIndexTopoBuilder.java
index 0a8cda3baf36..c98f7452f7bd 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/btree/BTreeIndexTopoBuilder.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/btree/BTreeIndexTopoBuilder.java
@@ -72,7 +72,7 @@
 /** The {@link BTreeIndexTopoBuilder} for BTree index in Flink. */
 public class BTreeIndexTopoBuilder {
 
-    public static void buildIndex(
+    public static boolean buildIndex(
             StreamExecutionEnvironment env,
             Supplier<BTreeGlobalIndexBuilder> indexBuilderSupplier,
             FileStoreTable table,
@@ -90,12 +90,12 @@ public static void buildIndex(
 
             List<DataSplit> splits = splitByContiguousRowRange(indexBuilder.scan());
             if (splits.isEmpty()) {
-                return;
+                return false;
             }
             Map<BinaryRow, Map<Range, List<DataSplit>>> partitionRangeSplits =
                     groupSplitsByRange(splits);
             if (partitionRangeSplits.isEmpty()) {
-                return;
+                return false;
             }
 
             // 2. Select necessary columns (index field + ROW_ID)
@@ -160,23 +160,25 @@ public static void buildIndex(
             commit(table, allCommitMessages);
         }
 
-        env.execute("Create btree global index for table: " + table.name());
+        return true;
     }
 
-    public static void buildIndex(
+    public static void buildIndexAndExecute(
             StreamExecutionEnvironment env,
             FileStoreTable table,
             String indexColumn,
             PartitionPredicate partitionPredicate,
             Options userOptions)
             throws Exception {
-        buildIndex(
+        if (buildIndex(
                 env,
                 () -> new BTreeGlobalIndexBuilder(table),
                 table,
                 Collections.singletonList(indexColumn),
                 partitionPredicate,
-                userOptions);
+                userOptions)) {
+            env.execute("Create btree global index for table: " + table.name());
+        }
     }
 
     protected static DataStream<Committable> executeForPartitionRange(
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CreateGlobalIndexProcedure.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CreateGlobalIndexProcedure.java
index 56675c18e4ce..664b69be4f9f 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CreateGlobalIndexProcedure.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/CreateGlobalIndexProcedure.java
@@ -119,7 +119,7 @@ public String[] call(
         // Build global index based on index type
         indexType = indexType.toLowerCase().trim();
         if ("btree".equals(indexType)) {
-            BTreeIndexTopoBuilder.buildIndex(
+            BTreeIndexTopoBuilder.buildIndexAndExecute(
                     procedureContext.getExecutionEnvironment(),
                     table,
                     indexColumn,