refactor(index): pass resolved bucket mode to index manifests

src/paimon/core/global_index/global_index_live_row_filter.cpp

@@ -118,8 +118,9 @@ Result<std::unique_ptr<SnapshotReader>> CreateSnapshotReader(
 
     PAIMON_ASSIGN_OR_RAISE(
         std::unique_ptr<IndexManifestFile> index_manifest_file,
-        IndexManifestFile::Create(fs, options.GetManifestFormat(), options.GetManifestCompression(),
-                                  path_factory, options.GetBucket(), pool, options));
+        IndexManifestFile::Create(
+            fs, options.GetManifestFormat(), options.GetManifestCompression(), path_factory,
+            ResolveBucketMode(options.GetBucket(), table_schema), pool, options));
     auto index_file_handler = std::make_unique<IndexFileHandler>(
         fs, std::move(index_manifest_file), std::make_shared<IndexFilePathFactories>(path_factory),
         options.DeletionVectorsBitmap64(), pool);

src/paimon/core/global_index/global_index_scan_impl.cpp

@@ -33,6 +33,7 @@
 #include "paimon/core/global_index/global_index_evaluator_impl.h"
 #include "paimon/core/global_index/global_index_live_row_filter.h"
 #include "paimon/core/index/index_file_handler.h"
+#include "paimon/core/table/bucket_mode.h"
 #include "paimon/global_index/bitmap_global_index_result.h"
 #include "paimon/global_index/global_indexer.h"
 #include "paimon/global_index/global_indexer_factory.h"
@@ -97,7 +98,8 @@ Result<std::unique_ptr<GlobalIndexScanImpl>> GlobalIndexScanImpl::Create(
         std::unique_ptr<IndexManifestFile> index_manifest_file,
         IndexManifestFile::Create(options.GetFileSystem(), options.GetManifestFormat(),
                                   options.GetManifestCompression(), file_store_path_factory,
-                                  options.GetBucket(), pool, options));
+                                  ResolveBucketMode(options.GetBucket(), table_schema), pool,
+                                  options));
     auto index_file_handler = std::make_unique<IndexFileHandler>(
         options.GetFileSystem(), std::move(index_manifest_file),
         std::make_shared<IndexFilePathFactories>(file_store_path_factory),

src/paimon/core/index/index_file_handler_test.cpp

@@ -35,6 +35,7 @@
 #include "paimon/core/schema/schema_manager.h"
 #include "paimon/core/schema/table_schema.h"
 #include "paimon/core/snapshot.h"
+#include "paimon/core/table/bucket_mode.h"
 #include "paimon/core/utils/file_store_path_factory.h"
 #include "paimon/core/utils/snapshot_manager.h"
 #include "paimon/defs.h"
@@ -81,7 +82,8 @@ class IndexFileHandlerTest : public testing::Test {
                                IndexManifestFile::Create(
                                    core_options.GetFileSystem(), core_options.GetManifestFormat(),
                                    core_options.GetManifestCompression(), path_factory,
-                                   core_options.GetBucket(), memory_pool_, core_options));
+                                   ResolveBucketMode(core_options.GetBucket(), table_schema),
+                                   memory_pool_, core_options));
         auto path_factories = std::make_shared<IndexFilePathFactories>(path_factory);
         return std::make_unique<IndexFileHandler>(
             core_options.GetFileSystem(), std::move(index_manifest_file), path_factories,

src/paimon/core/manifest/index_manifest_file.cpp

@@ -46,7 +46,7 @@ class MemoryPool;
 Result<std::unique_ptr<IndexManifestFile>> IndexManifestFile::Create(
     const std::shared_ptr<FileSystem>& file_system, const std::shared_ptr<FileFormat>& file_format,
     const std::string& compression, const std::shared_ptr<FileStorePathFactory>& path_factory,
-    int32_t bucket_mode, const std::shared_ptr<MemoryPool>& pool, const CoreOptions& options) {
+    BucketMode bucket_mode, const std::shared_ptr<MemoryPool>& pool, const CoreOptions& options) {
     std::shared_ptr<arrow::DataType> data_type =
         VersionedObjectSerializer<IndexManifestEntry>::VersionType(IndexManifestEntry::DataType());
 
@@ -72,7 +72,7 @@ IndexManifestFile::IndexManifestFile(const std::shared_ptr<FileSystem>& file_sys
                                      const std::string& file_format_identifier,
                                      const std::string& compression,
                                      const std::shared_ptr<PathFactory>& path_factory,
-                                     int32_t bucket_mode, const std::shared_ptr<Cache>& cache,
+                                     BucketMode bucket_mode, const std::shared_ptr<Cache>& cache,
                                      const std::shared_ptr<MemoryPool>& pool)
     : ObjectsFile<IndexManifestEntry>(file_system, reader_builder, writer_builder,
                                       file_format_identifier,

src/paimon/core/manifest/index_manifest_file.h

@@ -24,6 +24,7 @@
 
 #include "paimon/core/core_options.h"
 #include "paimon/core/manifest/index_manifest_entry.h"
+#include "paimon/core/table/bucket_mode.h"
 #include "paimon/core/utils/file_store_path_factory.h"
 #include "paimon/core/utils/objects_file.h"
 #include "paimon/format/file_format.h"
@@ -49,7 +50,7 @@ class IndexManifestFile : public ObjectsFile<IndexManifestEntry> {
     static Result<std::unique_ptr<IndexManifestFile>> Create(
         const std::shared_ptr<FileSystem>& file_system,
         const std::shared_ptr<FileFormat>& file_format, const std::string& compression,
-        const std::shared_ptr<FileStorePathFactory>& path_factory, int32_t bucket_mode,
+        const std::shared_ptr<FileStorePathFactory>& path_factory, BucketMode bucket_mode,
         const std::shared_ptr<MemoryPool>& pool, const CoreOptions& options);
 
     /// Write new index files to index manifest.
@@ -62,8 +63,9 @@ class IndexManifestFile : public ObjectsFile<IndexManifestEntry> {
                       const std::shared_ptr<ReaderBuilder>& reader_builder,
                       const std::shared_ptr<WriterBuilder>& writer_builder,
                       const std::string& file_format_identifier, const std::string& compression,
-                      const std::shared_ptr<PathFactory>& path_factory, int32_t bucket_mode,
+                      const std::shared_ptr<PathFactory>& path_factory, BucketMode bucket_mode,
                       const std::shared_ptr<Cache>& cache, const std::shared_ptr<MemoryPool>& pool);
-    const int32_t bucket_mode_;
+
+    const BucketMode bucket_mode_;
 };
 }  // namespace paimon

src/paimon/core/manifest/index_manifest_file_handler.cpp

@@ -197,7 +197,7 @@ IndexManifestFileHandler::GlobalDeletionVectorCombiner::Combine(
 
 Result<std::string> IndexManifestFileHandler::Write(
     const std::optional<std::string>& previous_index_manifest,
-    const std::vector<IndexManifestEntry>& new_index_entries, int32_t bucket_mode,
+    const std::vector<IndexManifestEntry>& new_index_entries, BucketMode bucket_mode,
     IndexManifestFile* index_manifest_file) {
     std::vector<IndexManifestEntry> entries;
     if (previous_index_manifest != std::nullopt) {
@@ -255,17 +255,16 @@ IndexManifestFileHandler::SeparateIndexEntries(
 
 Result<std::unique_ptr<IndexManifestFileHandler::IndexManifestFileCombiner>>
 IndexManifestFileHandler::GetIndexManifestFileCombine(const std::string& index_type,
-                                                      int32_t bucket_mode) {
+                                                      BucketMode bucket_mode) {
     if (index_type != DeletionVectorsIndexFile::DELETION_VECTORS_INDEX && index_type != "HASH") {
         return std::make_unique<GlobalFileNameCombiner>();
     }
-    // `bucket_mode` is the configured bucket, not a resolved BucketMode, standing in for Paimon
-    // Java's BucketMode.BUCKET_UNAWARE check. The two agree on every table that can exist:
-    // SchemaValidation rejects the other unaware bucket, 0, and BucketIdCalculator refuses to
-    // write the primary key table on which -1 means HASH_DYNAMIC instead. Lifting either
-    // restriction means passing the resolved BucketMode here, or a dynamic bucket table would
-    // combine its per-bucket deletion vectors by index file name.
-    if (index_type == DeletionVectorsIndexFile::DELETION_VECTORS_INDEX && bucket_mode == -1) {
+    // A dynamic bucket primary key table also has bucket=-1, but keeps one deletion vector index
+    // per (partition, bucket), so it uses the bucketed combiner. Only BUCKET_UNAWARE merges
+    // deletion vector index entries by index file name, using data file names to reject a data
+    // file covered by more than one deletion vector.
+    if (index_type == DeletionVectorsIndexFile::DELETION_VECTORS_INDEX &&
+        bucket_mode == BucketMode::BUCKET_UNAWARE) {
         return std::make_unique<GlobalDeletionVectorCombiner>();
     }
     return std::make_unique<BucketedCombiner>();

src/paimon/core/manifest/index_manifest_file_handler.h

@@ -37,7 +37,8 @@ class IndexManifestFileHandler {
 
     static Result<std::string> Write(const std::optional<std::string>& previous_index_manifest,
                                      const std::vector<IndexManifestEntry>& new_index_entries,
-                                     int32_t bucket_mode, IndexManifestFile* index_manifest_file);
+                                     BucketMode bucket_mode,
+                                     IndexManifestFile* index_manifest_file);
 
  private:
     class IndexManifestFileCombiner {
@@ -81,6 +82,6 @@ class IndexManifestFileHandler {
         const std::vector<IndexManifestEntry>& index_entries);
 
     static Result<std::unique_ptr<IndexManifestFileCombiner>> GetIndexManifestFileCombine(
-        const std::string& index_type, int32_t bucket_mode);
+        const std::string& index_type, BucketMode bucket_mode);
 };
 }  // namespace paimon