[Core] Honor parallelism in generic global index builds

crates/paimon/src/spec/core_options.rs

@@ -31,6 +31,7 @@ const GLOBAL_INDEX_SEARCH_MODE_OPTION: &str = "global-index.search-mode";
 const SCALAR_INDEX_SEARCH_MODE_OPTION: &str = "scalar-index.search-mode";
 const VECTOR_INDEX_SEARCH_MODE_OPTION: &str = "vector-index.search-mode";
 const FULL_TEXT_INDEX_SEARCH_MODE_OPTION: &str = "full-text-index.search-mode";
+const GLOBAL_INDEX_BUILD_PARALLELISM_OPTION: &str = "global-index.build.parallelism";
 const GLOBAL_INDEX_ROW_COUNT_PER_SHARD_OPTION: &str = "global-index.row-count-per-shard";
 const GLOBAL_INDEX_THREAD_NUM_OPTION: &str = "global-index.thread-num";
 const GLOBAL_INDEX_VINDEX_READ_THREAD_NUM_OPTION: &str = "global-index.vindex.read-thread-num";
@@ -1057,6 +1058,19 @@ impl<'a> CoreOptions<'a> {
         Ok(value)
     }
 
+    /// Number of generic global-index shards prepared concurrently by a local
+    /// build, matching PyPaimon's option and default. Each shard may also use
+    /// native worker threads; the default remains one.
+    pub fn global_index_build_parallelism(&self) -> crate::Result<usize> {
+        match self.options.get(GLOBAL_INDEX_BUILD_PARALLELISM_OPTION) {
+            None => Ok(1),
+            Some(raw) => raw.trim().parse::<usize>().ok().filter(|value| *value > 0)
+                .ok_or_else(|| crate::Error::ConfigInvalid {
+                    message: format!("Option '{GLOBAL_INDEX_BUILD_PARALLELISM_OPTION}' must be greater than 0, got: {raw}"),
+                }),
+        }
+    }
+
     /// Maximum number of concurrent global-index search tasks, mirroring Java
     /// `CoreOptions.GLOBAL_INDEX_THREAD_NUM` (key `global-index.thread-num`,
     /// default 32). Used as the per-operation fan-out limit for sorted BTree and
@@ -2534,6 +2548,35 @@ mod tests {
         }
     }
 
+    #[test]
+    fn test_global_index_build_parallelism() {
+        assert_eq!(
+            CoreOptions::new(&HashMap::new())
+                .global_index_build_parallelism()
+                .unwrap(),
+            1
+        );
+        for (raw, expected) in [("1", 1), (" 3 ", 3), ("+2", 2), ("1000000", 1000000)] {
+            let options =
+                HashMap::from([(GLOBAL_INDEX_BUILD_PARALLELISM_OPTION.into(), raw.into())]);
+            assert_eq!(
+                CoreOptions::new(&options)
+                    .global_index_build_parallelism()
+                    .unwrap(),
+                expected
+            );
+        }
+        for raw in ["0", "-1", "invalid", "18446744073709551616"] {
+            let options =
+                HashMap::from([(GLOBAL_INDEX_BUILD_PARALLELISM_OPTION.into(), raw.into())]);
+            assert!(CoreOptions::new(&options)
+                .global_index_build_parallelism()
+                .unwrap_err()
+                .to_string()
+                .contains("parallelism"));
+        }
+    }
+
     #[test]
     fn test_global_index_thread_num_default_and_custom() {
         // Default mirrors Java (32) when unset.

crates/paimon/src/table/full_text_index_build_builder.rs

@@ -34,7 +34,7 @@ use crate::table::global_index_build_common::{
     copy_local_file_to_output, indexed_row_ranges, validate_existing_index_overlap,
 };
 use crate::table::global_index_types::FULL_TEXT_GLOBAL_INDEX_TYPE;
-use crate::table::{CommitMessage, DataSplitBuilder, RowRange, Table, TableCommit};
+use crate::table::{CommitMessage, DataSplitBuilder, RowRange, Table};
 use crate::{Error, Result};
 use arrow_array::cast::AsArray;
 use arrow_array::{Array, Int64Array, LargeStringArray, RecordBatch, StringArray, StringViewArray};
@@ -98,8 +98,9 @@ impl<'a> FullTextIndexBuildBuilder<'a> {
         // procedure options.
         let mut merged_options = self.table.schema().options().clone();
         merged_options.extend(self.options.clone());
-        let rows_per_shard =
-            CoreOptions::new(&merged_options).global_index_row_count_per_shard()?;
+        let core_options = CoreOptions::new(&merged_options);
+        let rows_per_shard = core_options.global_index_row_count_per_shard()?;
+        let parallelism = core_options.global_index_build_parallelism()?;
 
         let index_field = find_index_field(self.table, index_column)?;
         validate_text_field(index_field)?;
@@ -161,43 +162,34 @@ impl<'a> FullTextIndexBuildBuilder<'a> {
         )
         .await?;
 
-        let commit = TableCommit::new(
-            self.table.clone(),
-            format!(
-                "global-index-{}-create-{}",
-                FULL_TEXT_GLOBAL_INDEX_TYPE,
-                uuid::Uuid::new_v4()
-            ),
-        );
-        let mut messages: Vec<CommitMessage> = Vec::with_capacity(shards.len());
-        for shard in shards {
-            let build_result = self
-                .build_shard(
-                    &shard,
-                    index_column,
-                    &native_options,
-                    index_field.id(),
-                    &index_meta,
-                )
-                .await;
-            let index_file = match build_result {
-                Ok(index_file) => index_file,
-                Err(error) => {
-                    let _ = commit.abort(&messages).await;
-                    return Err(error);
+        let prepared = super::global_index_build_common::preparation::prepare_shards(
+            self.table,
+            shards,
+            parallelism,
+            |shard| {
+                let native_options = &native_options;
+                let index_meta = &index_meta;
+                async move {
+                    let file = self
+                        .build_shard(
+                            &shard,
+                            index_column,
+                            native_options,
+                            index_field.id(),
+                            index_meta,
+                        )
+                        .await?;
+                    // Java skips a shard whose writer saw no rows.
+                    Ok(file.map(|file| {
+                        let mut message = CommitMessage::new(shard.partition_bytes, 0, vec![]);
+                        message.new_index_files = vec![file];
+                        (message, ())
+                    }))
                 }
-            };
-            // Java skips a shard whose writer saw no rows.
-            let Some(index_file) = index_file else {
-                continue;
-            };
-            let mut message = CommitMessage::new(shard.partition_bytes.clone(), 0, vec![]);
-            message.new_index_files = vec![index_file];
-            messages.push(message);
-        }
-        if messages.is_empty() {
-            return Ok((Some(snapshot.id()), vec![]));
-        }
+            },
+        )
+        .await?;
+        let messages = prepared.into_iter().map(|(message, ())| message).collect();
 
         Ok((Some(snapshot.id()), messages))
     }

crates/paimon/src/table/global_index_build_builder.rs

@@ -98,6 +98,8 @@ impl<'a> GlobalIndexBuildBuilder<'a> {
         self
     }
 
+    /// Overrides table options for this build. Full-text and vindex builds use
+    /// `global-index.build.parallelism` to bound concurrent shards (default: 1).
     pub fn with_options(&mut self, options: HashMap<String, String>) -> &mut Self {
         self.options = options;
         self
@@ -346,3 +348,7 @@ mod tests {
         );
     }
 }
+
+#[cfg(test)]
+#[path = "global_index_build_builder/parallel_tests.rs"]
+mod parallel_tests;