diff --git a/src/paimon/core/global_index/global_index_live_row_filter.cpp b/src/paimon/core/global_index/global_index_live_row_filter.cpp index faf9e0408..2c72e1b52 100644 --- a/src/paimon/core/global_index/global_index_live_row_filter.cpp +++ b/src/paimon/core/global_index/global_index_live_row_filter.cpp @@ -118,8 +118,9 @@ Result> CreateSnapshotReader( PAIMON_ASSIGN_OR_RAISE( std::unique_ptr 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( fs, std::move(index_manifest_file), std::make_shared(path_factory), options.DeletionVectorsBitmap64(), pool); diff --git a/src/paimon/core/global_index/global_index_scan_impl.cpp b/src/paimon/core/global_index/global_index_scan_impl.cpp index 8cc1c77a9..e7ae8cfea 100644 --- a/src/paimon/core/global_index/global_index_scan_impl.cpp +++ b/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> GlobalIndexScanImpl::Create( std::unique_ptr 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( options.GetFileSystem(), std::move(index_manifest_file), std::make_shared(file_store_path_factory), diff --git a/src/paimon/core/index/index_file_handler_test.cpp b/src/paimon/core/index/index_file_handler_test.cpp index 64fcf39fb..a7eac7fb7 100644 --- a/src/paimon/core/index/index_file_handler_test.cpp +++ b/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(path_factory); return std::make_unique( core_options.GetFileSystem(), std::move(index_manifest_file), path_factories, diff --git a/src/paimon/core/manifest/index_manifest_file.cpp b/src/paimon/core/manifest/index_manifest_file.cpp index ab8ed8c61..f84babdb1 100644 --- a/src/paimon/core/manifest/index_manifest_file.cpp +++ b/src/paimon/core/manifest/index_manifest_file.cpp @@ -46,7 +46,7 @@ class MemoryPool; Result> IndexManifestFile::Create( const std::shared_ptr& file_system, const std::shared_ptr& file_format, const std::string& compression, const std::shared_ptr& path_factory, - int32_t bucket_mode, const std::shared_ptr& pool, const CoreOptions& options) { + BucketMode bucket_mode, const std::shared_ptr& pool, const CoreOptions& options) { std::shared_ptr data_type = VersionedObjectSerializer::VersionType(IndexManifestEntry::DataType()); @@ -72,7 +72,7 @@ IndexManifestFile::IndexManifestFile(const std::shared_ptr& file_sys const std::string& file_format_identifier, const std::string& compression, const std::shared_ptr& path_factory, - int32_t bucket_mode, const std::shared_ptr& cache, + BucketMode bucket_mode, const std::shared_ptr& cache, const std::shared_ptr& pool) : ObjectsFile(file_system, reader_builder, writer_builder, file_format_identifier, diff --git a/src/paimon/core/manifest/index_manifest_file.h b/src/paimon/core/manifest/index_manifest_file.h index 99122327d..94e223b99 100644 --- a/src/paimon/core/manifest/index_manifest_file.h +++ b/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 { static Result> Create( const std::shared_ptr& file_system, const std::shared_ptr& file_format, const std::string& compression, - const std::shared_ptr& path_factory, int32_t bucket_mode, + const std::shared_ptr& path_factory, BucketMode bucket_mode, const std::shared_ptr& pool, const CoreOptions& options); /// Write new index files to index manifest. @@ -62,8 +63,9 @@ class IndexManifestFile : public ObjectsFile { const std::shared_ptr& reader_builder, const std::shared_ptr& writer_builder, const std::string& file_format_identifier, const std::string& compression, - const std::shared_ptr& path_factory, int32_t bucket_mode, + const std::shared_ptr& path_factory, BucketMode bucket_mode, const std::shared_ptr& cache, const std::shared_ptr& pool); - const int32_t bucket_mode_; + + const BucketMode bucket_mode_; }; } // namespace paimon diff --git a/src/paimon/core/manifest/index_manifest_file_handler.cpp b/src/paimon/core/manifest/index_manifest_file_handler.cpp index 55261de95..5f7422826 100644 --- a/src/paimon/core/manifest/index_manifest_file_handler.cpp +++ b/src/paimon/core/manifest/index_manifest_file_handler.cpp @@ -197,7 +197,7 @@ IndexManifestFileHandler::GlobalDeletionVectorCombiner::Combine( Result IndexManifestFileHandler::Write( const std::optional& previous_index_manifest, - const std::vector& new_index_entries, int32_t bucket_mode, + const std::vector& new_index_entries, BucketMode bucket_mode, IndexManifestFile* index_manifest_file) { std::vector entries; if (previous_index_manifest != std::nullopt) { @@ -255,17 +255,16 @@ IndexManifestFileHandler::SeparateIndexEntries( Result> 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(); } - // `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(); } return std::make_unique(); diff --git a/src/paimon/core/manifest/index_manifest_file_handler.h b/src/paimon/core/manifest/index_manifest_file_handler.h index 8001e4740..df37e44e4 100644 --- a/src/paimon/core/manifest/index_manifest_file_handler.h +++ b/src/paimon/core/manifest/index_manifest_file_handler.h @@ -37,7 +37,8 @@ class IndexManifestFileHandler { static Result Write(const std::optional& previous_index_manifest, const std::vector& 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& index_entries); static Result> GetIndexManifestFileCombine( - const std::string& index_type, int32_t bucket_mode); + const std::string& index_type, BucketMode bucket_mode); }; } // namespace paimon diff --git a/src/paimon/core/manifest/index_manifest_file_handler_test.cpp b/src/paimon/core/manifest/index_manifest_file_handler_test.cpp index e0ce213a3..13045ed1c 100644 --- a/src/paimon/core/manifest/index_manifest_file_handler_test.cpp +++ b/src/paimon/core/manifest/index_manifest_file_handler_test.cpp @@ -54,7 +54,7 @@ class IndexManifestFileHandlerTest : public testing::Test { } Result> CreateManifestFile( - int32_t bucket_mode, const std::string& file_format_identifier = "avro", + BucketMode bucket_mode, const std::string& file_format_identifier = "avro", const std::shared_ptr& cache = nullptr) const { PAIMON_ASSIGN_OR_RAISE(std::shared_ptr file_format, FileFormatFactory::Get(file_format_identifier, {})); @@ -109,7 +109,7 @@ class IndexManifestFileHandlerTest : public testing::Test { TEST_F(IndexManifestFileHandlerTest, DecodedCacheReusesIpcAcrossReaders) { auto cache = std::make_shared(CacheKind::MANIFEST, 1024 * 1024); ASSERT_OK_AND_ASSIGN(std::unique_ptr writer, - CreateManifestFile(2, "avro", cache)); + CreateManifestFile(BucketMode::HASH_FIXED, "avro", cache)); const std::vector expected = { MakeDvEntry(FileKind::Add(), BinaryRow::EmptyRow(), 0, "dv-0", {"data-a", "data-b"}, 10), MakeEntry(FileKind::Add(), BinaryRow::EmptyRow(), 1, "BTREE", "index-1", 20)}; @@ -141,7 +141,7 @@ TEST_F(IndexManifestFileHandlerTest, DecodedCacheReusesIpcAcrossReaders) { writer->DeleteQuietly(written.first); writer.reset(); ASSERT_OK_AND_ASSIGN(std::unique_ptr reader, - CreateManifestFile(2, "avro", cache)); + CreateManifestFile(BucketMode::HASH_FIXED, "avro", cache)); std::vector warm; ASSERT_OK(reader->Read(written.first, nullptr, written.second, &warm)); ASSERT_EQ(expected, warm); @@ -152,7 +152,7 @@ TEST_F(IndexManifestFileHandlerTest, DecodedCacheReusesIpcAcrossReaders) { } TEST_F(IndexManifestFileHandlerTest, GlobalCombinerDeletesThenAddsByFileName) { - ASSERT_OK_AND_ASSIGN(auto index_manifest_file, CreateManifestFile(/*bucket_mode=*/4)); + ASSERT_OK_AND_ASSIGN(auto index_manifest_file, CreateManifestFile(BucketMode::HASH_FIXED)); auto partition = BinaryRow::EmptyRow(); std::vector previous_entries = { @@ -161,7 +161,7 @@ TEST_F(IndexManifestFileHandlerTest, GlobalCombinerDeletesThenAddsByFileName) { ASSERT_OK_AND_ASSIGN(std::string previous_manifest, IndexManifestFileHandler::Write( /*previous_index_manifest=*/std::nullopt, previous_entries, - /*bucket_mode=*/4, index_manifest_file.get())); + BucketMode::HASH_FIXED, index_manifest_file.get())); std::vector new_entries = { MakeEntry(FileKind::Delete(), partition, /*bucket=*/999, /*index_type=*/"BTREE", "global-0", @@ -170,8 +170,8 @@ TEST_F(IndexManifestFileHandlerTest, GlobalCombinerDeletesThenAddsByFileName) { ASSERT_OK_AND_ASSIGN( std::string current_manifest, - IndexManifestFileHandler::Write(previous_manifest, new_entries, - /*bucket_mode=*/4, index_manifest_file.get())); + IndexManifestFileHandler::Write(previous_manifest, new_entries, BucketMode::HASH_FIXED, + index_manifest_file.get())); std::vector written_entries; ASSERT_OK(index_manifest_file->Read(current_manifest, /*filter=*/nullptr, @@ -182,7 +182,7 @@ TEST_F(IndexManifestFileHandlerTest, GlobalCombinerDeletesThenAddsByFileName) { } TEST_F(IndexManifestFileHandlerTest, BucketedCombinerUsesPartitionBucketAndIndexType) { - ASSERT_OK_AND_ASSIGN(auto index_manifest_file, CreateManifestFile(/*bucket_mode=*/2)); + ASSERT_OK_AND_ASSIGN(auto index_manifest_file, CreateManifestFile(BucketMode::HASH_FIXED)); auto partition_10 = BinaryRowGenerator::GenerateRow({10}, pool_.get()); std::vector previous_entries = { @@ -194,7 +194,7 @@ TEST_F(IndexManifestFileHandlerTest, BucketedCombinerUsesPartitionBucketAndIndex ASSERT_OK_AND_ASSIGN(std::string previous_manifest, IndexManifestFileHandler::Write( /*previous_index_manifest=*/std::nullopt, previous_entries, - /*bucket_mode=*/2, index_manifest_file.get())); + BucketMode::HASH_FIXED, index_manifest_file.get())); std::vector new_entries = { MakeEntry(FileKind::Delete(), partition_10, /*bucket=*/0, @@ -204,8 +204,8 @@ TEST_F(IndexManifestFileHandlerTest, BucketedCombinerUsesPartitionBucketAndIndex ASSERT_OK_AND_ASSIGN( std::string current_manifest, - IndexManifestFileHandler::Write(previous_manifest, new_entries, - /*bucket_mode=*/2, index_manifest_file.get())); + IndexManifestFileHandler::Write(previous_manifest, new_entries, BucketMode::HASH_FIXED, + index_manifest_file.get())); std::vector written_entries; ASSERT_OK(index_manifest_file->Read(current_manifest, /*filter=*/nullptr, @@ -230,7 +230,7 @@ TEST_F(IndexManifestFileHandlerTest, BucketedCombinerUsesPartitionBucketAndIndex } TEST_F(IndexManifestFileHandlerTest, BucketedCombinerOverwritesDuplicateAddedEntries) { - ASSERT_OK_AND_ASSIGN(auto index_manifest_file, CreateManifestFile(/*bucket_mode=*/2)); + ASSERT_OK_AND_ASSIGN(auto index_manifest_file, CreateManifestFile(BucketMode::HASH_FIXED)); auto partition = BinaryRowGenerator::GenerateRow({10}, pool_.get()); std::vector new_entries = { @@ -242,7 +242,7 @@ TEST_F(IndexManifestFileHandlerTest, BucketedCombinerOverwritesDuplicateAddedEnt ASSERT_OK_AND_ASSIGN(std::string current_manifest, IndexManifestFileHandler::Write( /*previous_index_manifest=*/std::nullopt, new_entries, - /*bucket_mode=*/2, index_manifest_file.get())); + BucketMode::HASH_FIXED, index_manifest_file.get())); std::vector written_entries; ASSERT_OK(index_manifest_file->Read(current_manifest, /*filter=*/nullptr, @@ -253,7 +253,7 @@ TEST_F(IndexManifestFileHandlerTest, BucketedCombinerOverwritesDuplicateAddedEnt } TEST_F(IndexManifestFileHandlerTest, GlobalCombinerOverwritesDuplicateAddedEntries) { - ASSERT_OK_AND_ASSIGN(auto index_manifest_file, CreateManifestFile(/*bucket_mode=*/4)); + ASSERT_OK_AND_ASSIGN(auto index_manifest_file, CreateManifestFile(BucketMode::HASH_FIXED)); auto partition = BinaryRow::EmptyRow(); std::vector new_entries = { @@ -264,7 +264,7 @@ TEST_F(IndexManifestFileHandlerTest, GlobalCombinerOverwritesDuplicateAddedEntri ASSERT_OK_AND_ASSIGN(std::string current_manifest, IndexManifestFileHandler::Write( /*previous_index_manifest=*/std::nullopt, new_entries, - /*bucket_mode=*/4, index_manifest_file.get())); + BucketMode::HASH_FIXED, index_manifest_file.get())); std::vector written_entries; ASSERT_OK(index_manifest_file->Read(current_manifest, /*filter=*/nullptr, @@ -275,8 +275,9 @@ TEST_F(IndexManifestFileHandlerTest, GlobalCombinerOverwritesDuplicateAddedEntri } TEST_F(IndexManifestFileHandlerTest, LegacyManifestFormatIsReadOnly) { - ASSERT_OK_AND_ASSIGN(auto index_manifest_file, - CreateManifestFile(/*bucket_mode=*/4, /*file_format_identifier=*/"orc")); + ASSERT_OK_AND_ASSIGN( + auto index_manifest_file, + CreateManifestFile(BucketMode::HASH_FIXED, /*file_format_identifier=*/"orc")); auto partition = BinaryRow::EmptyRow(); std::vector entries = { MakeEntry(FileKind::Add(), partition, /*bucket=*/0, /*index_type=*/"BTREE", "global-0", 1)}; @@ -286,7 +287,7 @@ TEST_F(IndexManifestFileHandlerTest, LegacyManifestFormatIsReadOnly) { } TEST_F(IndexManifestFileHandlerTest, GlobalDvCombinerReplacesIndexFileInBucketUnawareMode) { - ASSERT_OK_AND_ASSIGN(auto index_manifest_file, CreateManifestFile(/*bucket_mode=*/-1)); + ASSERT_OK_AND_ASSIGN(auto index_manifest_file, CreateManifestFile(BucketMode::BUCKET_UNAWARE)); auto partition = BinaryRow::EmptyRow(); // an unaware bucket table writes every index file under bucket 0, so only the index file @@ -298,7 +299,7 @@ TEST_F(IndexManifestFileHandlerTest, GlobalDvCombinerReplacesIndexFileInBucketUn ASSERT_OK_AND_ASSIGN(std::string previous_manifest, IndexManifestFileHandler::Write( /*previous_index_manifest=*/std::nullopt, previous_entries, - /*bucket_mode=*/-1, index_manifest_file.get())); + BucketMode::BUCKET_UNAWARE, index_manifest_file.get())); // updating the vector of data-0.orc replaces its index file, and the data file it covers // moves to the new one @@ -308,8 +309,8 @@ TEST_F(IndexManifestFileHandlerTest, GlobalDvCombinerReplacesIndexFileInBucketUn ASSERT_OK_AND_ASSIGN( std::string current_manifest, - IndexManifestFileHandler::Write(previous_manifest, new_entries, - /*bucket_mode=*/-1, index_manifest_file.get())); + IndexManifestFileHandler::Write(previous_manifest, new_entries, BucketMode::BUCKET_UNAWARE, + index_manifest_file.get())); std::vector written_entries; ASSERT_OK(index_manifest_file->Read(current_manifest, /*filter=*/nullptr, @@ -331,7 +332,7 @@ TEST_F(IndexManifestFileHandlerTest, GlobalDvCombinerReplacesIndexFileInBucketUn ASSERT_OK_AND_ASSIGN( std::string reordered_manifest, IndexManifestFileHandler::Write(current_manifest, reordered_entries, - /*bucket_mode=*/-1, index_manifest_file.get())); + BucketMode::BUCKET_UNAWARE, index_manifest_file.get())); written_entries.clear(); ASSERT_OK(index_manifest_file->Read(reordered_manifest, /*filter=*/nullptr, @@ -344,8 +345,44 @@ TEST_F(IndexManifestFileHandlerTest, GlobalDvCombinerReplacesIndexFileInBucketUn ASSERT_EQ(written_file_names, (std::set{"dv-0-newer", "dv-1"})); } +TEST_F(IndexManifestFileHandlerTest, DynamicBucketDvUsesBucketedCombiner) { + ASSERT_OK_AND_ASSIGN(auto index_manifest_file, CreateManifestFile(BucketMode::HASH_DYNAMIC)); + + auto partition = BinaryRow::EmptyRow(); + std::vector previous_entries = { + MakeDvEntry(FileKind::Add(), partition, /*bucket=*/0, "dv-0", {"data-0.orc"}, 1), + MakeDvEntry(FileKind::Add(), partition, /*bucket=*/1, "dv-1", {"data-1.orc"}, 1)}; + ASSERT_OK_AND_ASSIGN(std::optional previous_manifest, + index_manifest_file->WriteIndexFiles( + /*previous_index_manifest=*/std::nullopt, previous_entries)); + ASSERT_TRUE(previous_manifest.has_value()); + + // a dynamic bucket table rewrites the whole index file of a bucket, so the added entry + // replaces dv-0 by (partition, bucket) although it is not deleted by name + std::vector new_entries = {MakeDvEntry( + FileKind::Add(), partition, /*bucket=*/0, "dv-0-new", {"data-0.orc", "data-2.orc"}, 2)}; + ASSERT_OK_AND_ASSIGN(std::optional current_manifest, + index_manifest_file->WriteIndexFiles(previous_manifest, new_entries)); + ASSERT_TRUE(current_manifest.has_value()); + + std::vector written_entries; + ASSERT_OK(index_manifest_file->Read(current_manifest.value(), /*filter=*/nullptr, + /*file_size=*/std::nullopt, &written_entries)); + std::set written_file_names; + for (const auto& entry : written_entries) { + written_file_names.insert(entry.index_file->FileName()); + } + ASSERT_EQ(written_file_names, (std::set{"dv-0-new", "dv-1"})); + + // the same delta in bucket unaware mode covers data-0.orc twice + ASSERT_OK_AND_ASSIGN(auto unaware_manifest_file, + CreateManifestFile(BucketMode::BUCKET_UNAWARE)); + ASSERT_NOK_WITH_MSG(unaware_manifest_file->WriteIndexFiles(previous_manifest, new_entries), + "Trying to add a second deletion vector for data file data-0.orc"); +} + TEST_F(IndexManifestFileHandlerTest, GlobalDvCombinerRejectsInconsistentDelta) { - ASSERT_OK_AND_ASSIGN(auto index_manifest_file, CreateManifestFile(/*bucket_mode=*/-1)); + ASSERT_OK_AND_ASSIGN(auto index_manifest_file, CreateManifestFile(BucketMode::BUCKET_UNAWARE)); auto partition = BinaryRow::EmptyRow(); std::vector previous_entries = { @@ -354,7 +391,7 @@ TEST_F(IndexManifestFileHandlerTest, GlobalDvCombinerRejectsInconsistentDelta) { ASSERT_OK_AND_ASSIGN(std::string previous_manifest, IndexManifestFileHandler::Write( /*previous_index_manifest=*/std::nullopt, previous_entries, - /*bucket_mode=*/-1, index_manifest_file.get())); + BucketMode::BUCKET_UNAWARE, index_manifest_file.get())); // every delta below was built against a base the previous manifest is not, so applying it // would leave the read either with two vectors for one data file or with none @@ -365,7 +402,7 @@ TEST_F(IndexManifestFileHandlerTest, GlobalDvCombinerRejectsInconsistentDelta) { ASSERT_NOK_WITH_MSG( IndexManifestFileHandler::Write(previous_manifest, new_entries, - /*bucket_mode=*/-1, index_manifest_file.get()), + BucketMode::BUCKET_UNAWARE, index_manifest_file.get()), "Trying to add a second deletion vector for data file data-0.orc"); } { @@ -375,7 +412,7 @@ TEST_F(IndexManifestFileHandlerTest, GlobalDvCombinerRejectsInconsistentDelta) { ASSERT_NOK_WITH_MSG( IndexManifestFileHandler::Write(previous_manifest, new_entries, - /*bucket_mode=*/-1, index_manifest_file.get()), + BucketMode::BUCKET_UNAWARE, index_manifest_file.get()), "Trying to add deletion vector index file dv-0 which is already added."); } { @@ -385,7 +422,7 @@ TEST_F(IndexManifestFileHandlerTest, GlobalDvCombinerRejectsInconsistentDelta) { ASSERT_NOK_WITH_MSG( IndexManifestFileHandler::Write(previous_manifest, new_entries, - /*bucket_mode=*/-1, index_manifest_file.get()), + BucketMode::BUCKET_UNAWARE, index_manifest_file.get()), "Trying to delete deletion vector index file dv-1 which does not exist."); } { @@ -397,7 +434,7 @@ TEST_F(IndexManifestFileHandlerTest, GlobalDvCombinerRejectsInconsistentDelta) { ASSERT_NOK_WITH_MSG( IndexManifestFileHandler::Write(previous_manifest, new_entries, - /*bucket_mode=*/-1, index_manifest_file.get()), + BucketMode::BUCKET_UNAWARE, index_manifest_file.get()), "Trying to delete the deletion vector of data file data-1.orc, which does not exist."); } } diff --git a/src/paimon/core/operation/commit/commit_scanner_test.cpp b/src/paimon/core/operation/commit/commit_scanner_test.cpp index 1e02849d0..a3e2e2537 100644 --- a/src/paimon/core/operation/commit/commit_scanner_test.cpp +++ b/src/paimon/core/operation/commit/commit_scanner_test.cpp @@ -132,7 +132,7 @@ class CommitScannerTest : public testing::Test { PAIMON_ASSIGN_OR_RAISE( std::unique_ptr index_manifest_file, IndexManifestFile::Create(dir_->GetFileSystem(), file_format, "zstd", path_factory, - /*bucket_mode=*/2, GetDefaultPool(), core_options_)); + BucketMode::HASH_FIXED, GetDefaultPool(), core_options_)); return std::shared_ptr(std::move(index_manifest_file)); } diff --git a/src/paimon/core/operation/expire_snapshots_test.cpp b/src/paimon/core/operation/expire_snapshots_test.cpp index f9ec4ec01..f64c0ff03 100644 --- a/src/paimon/core/operation/expire_snapshots_test.cpp +++ b/src/paimon/core/operation/expire_snapshots_test.cpp @@ -123,7 +123,7 @@ class ExpireSnapshotsTest : public testing::Test { ASSERT_OK_AND_ASSIGN(index_manifest_file_, IndexManifestFile::Create( fs_, options.GetManifestFormat(), options.GetManifestCompression(), - path_factory_, options.GetBucket(), mem_pool_, options)); + path_factory_, BucketMode::BUCKET_UNAWARE, mem_pool_, options)); } void TearDown() override {} diff --git a/src/paimon/core/operation/file_system_write_restore_test.cpp b/src/paimon/core/operation/file_system_write_restore_test.cpp index d4e8a9196..27091f8c9 100644 --- a/src/paimon/core/operation/file_system_write_restore_test.cpp +++ b/src/paimon/core/operation/file_system_write_restore_test.cpp @@ -90,8 +90,9 @@ Result> CreateRestore( options, CreateDefaultExecutor(), pool)); PAIMON_ASSIGN_OR_RAISE( std::unique_ptr 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_shared( fs, std::move(index_manifest_file), std::make_shared(path_factory), options.DeletionVectorsBitmap64(), pool); diff --git a/src/paimon/core/operation/orphan_files_cleaner.cpp b/src/paimon/core/operation/orphan_files_cleaner.cpp index 397ba68ea..2f7052501 100644 --- a/src/paimon/core/operation/orphan_files_cleaner.cpp +++ b/src/paimon/core/operation/orphan_files_cleaner.cpp @@ -210,7 +210,8 @@ Result> OrphanFilesCleaner::Create( std::shared_ptr index_manifest_file, IndexManifestFile::Create(options.GetFileSystem(), options.GetManifestFormat(), options.GetManifestCompression(), path_factory, - options.GetBucket(), ctx->GetMemoryPool(), options)); + ResolveBucketMode(options.GetBucket(), schema), + ctx->GetMemoryPool(), options)); return std::make_unique( ctx->GetMemoryPool(), ctx->GetExecutor(), arrow_schema, ctx->GetRootPath(), options, snapshot_manager, schema->PartitionKeys(), manifest_file, manifest_list, diff --git a/src/paimon/core/operation/orphan_files_cleaner_test.cpp b/src/paimon/core/operation/orphan_files_cleaner_test.cpp index 5ca386db1..6043e0d79 100644 --- a/src/paimon/core/operation/orphan_files_cleaner_test.cpp +++ b/src/paimon/core/operation/orphan_files_cleaner_test.cpp @@ -456,10 +456,12 @@ TEST(OrphanFilesCleanerTest, TestTableWithIndexManifest) { options.GetPartitionDefaultName(), options.GetFileFormat()->Identifier(), options.DataFilePrefix(), options.LegacyPartitionNameEnabled(), external_paths, global_index_external_path, options.IndexFileInDataFileDir(), memory_pool)); - ASSERT_OK_AND_ASSIGN(std::unique_ptr index_manifest_file, - IndexManifestFile::Create(file_system, options.GetManifestFormat(), - options.GetManifestCompression(), path_factory, - options.GetBucket(), memory_pool, options)); + ASSERT_OK_AND_ASSIGN( + std::unique_ptr index_manifest_file, + IndexManifestFile::Create(file_system, options.GetManifestFormat(), + options.GetManifestCompression(), path_factory, + ResolveBucketMode(options.GetBucket(), table_schema), memory_pool, + options)); ASSERT_OK_AND_ASSIGN(BinaryRow partition, path_factory->ToBinaryRow({{"p0", "0"}, {"p1", "0"}})); diff --git a/src/paimon/core/snapshot_file_scan.cpp b/src/paimon/core/snapshot_file_scan.cpp index c7f854b87..58dd539ce 100644 --- a/src/paimon/core/snapshot_file_scan.cpp +++ b/src/paimon/core/snapshot_file_scan.cpp @@ -395,11 +395,12 @@ Result> SnapshotFileScan::ListFiles( core_options.GetManifestCompression(), path_factory, core_options.GetManifestTargetFileSize(), pool, core_options, partition_schema)); - PAIMON_ASSIGN_OR_RAISE( - std::unique_ptr index_manifest_file, - IndexManifestFile::Create(core_options.GetFileSystem(), core_options.GetManifestFormat(), - core_options.GetManifestCompression(), path_factory, - core_options.GetBucket(), pool, core_options)); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr index_manifest_file, + IndexManifestFile::Create( + core_options.GetFileSystem(), core_options.GetManifestFormat(), + core_options.GetManifestCompression(), path_factory, + ResolveBucketMode(core_options.GetBucket(), latest_schema.value()), + pool, core_options)); auto index_file_handler = std::make_unique( core_options.GetFileSystem(), std::move(index_manifest_file), std::make_shared(path_factory), diff --git a/src/paimon/core/table/source/data_evolution_full_text_scan.cpp b/src/paimon/core/table/source/data_evolution_full_text_scan.cpp index 85b8540e2..a4dd53bfd 100644 --- a/src/paimon/core/table/source/data_evolution_full_text_scan.cpp +++ b/src/paimon/core/table/source/data_evolution_full_text_scan.cpp @@ -272,7 +272,8 @@ Result> DataEvolutionFullTextScan::CreatePla std::unique_ptr index_manifest_file, IndexManifestFile::Create(options_.GetFileSystem(), options_.GetManifestFormat(), options_.GetManifestCompression(), path_factory_, - options_.GetBucket(), pool_, options_)); + ResolveBucketMode(options_.GetBucket(), table_schema_), pool_, + options_)); IndexFileHandler index_file_handler(options_.GetFileSystem(), std::move(index_manifest_file), std::make_shared(path_factory_), options_.DeletionVectorsBitmap64(), pool_); diff --git a/src/paimon/core/table/table.cpp b/src/paimon/core/table/table.cpp index bba5aa150..e8341421d 100644 --- a/src/paimon/core/table/table.cpp +++ b/src/paimon/core/table/table.cpp @@ -423,7 +423,8 @@ Result> FileStoreWrite::Create(std::unique_ptr index_manifest_file, IndexManifestFile::Create(options.GetFileSystem(), options.GetManifestFormat(), options.GetManifestCompression(), file_store_path_factory, - options.GetBucket(), ctx->GetMemoryPool(), options)); + ResolveBucketMode(options.GetBucket(), schema), + ctx->GetMemoryPool(), options)); auto index_file_handler = std::make_shared( options.GetFileSystem(), std::move(index_manifest_file), std::make_shared(file_store_path_factory), @@ -524,7 +525,8 @@ Result> FileStoreWrite::Create(std::unique_ptr index_manifest_file, IndexManifestFile::Create(options.GetFileSystem(), options.GetManifestFormat(), options.GetManifestCompression(), file_store_path_factory, - options.GetBucket(), ctx->GetMemoryPool(), options)); + ResolveBucketMode(options.GetBucket(), schema), + ctx->GetMemoryPool(), options)); index_file_handler = std::make_shared( options.GetFileSystem(), std::move(index_manifest_file), std::make_shared(file_store_path_factory), @@ -664,13 +666,15 @@ class TableScanImpl { } static Result> CreateIndexFileHandler( - const CoreOptions& core_options, const std::shared_ptr& path_factory, + const std::shared_ptr& table_schema, const CoreOptions& core_options, + const std::shared_ptr& path_factory, const std::shared_ptr& memory_pool) { PAIMON_ASSIGN_OR_RAISE(std::unique_ptr index_manifest_file, 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)); return std::make_unique( core_options.GetFileSystem(), std::move(index_manifest_file), std::make_shared(path_factory), @@ -906,8 +910,8 @@ Result> NewDataTableScan( std::unique_ptr split_generator, TableScanImpl::CreateSplitGenerator(table_schema, core_options, context.get())); PAIMON_ASSIGN_OR_RAISE(std::unique_ptr index_file_handler, - TableScanImpl::CreateIndexFileHandler(core_options, path_factory, - context->GetMemoryPool())); + TableScanImpl::CreateIndexFileHandler( + table_schema, core_options, path_factory, context->GetMemoryPool())); auto snapshot_reader = std::make_shared( file_store_scan, path_factory, std::move(split_generator), std::move(index_file_handler)); const bool pk_table = !table_schema->PrimaryKeys().empty(); @@ -1413,7 +1417,8 @@ Result> FileStoreCommit::Create( std::shared_ptr index_manifest_file, IndexManifestFile::Create(options.GetFileSystem(), options.GetManifestFormat(), options.GetManifestCompression(), path_factory, - options.GetBucket(), ctx->GetMemoryPool(), options)); + ResolveBucketMode(options.GetBucket(), schema), + ctx->GetMemoryPool(), options)); auto expire_snapshots = std::make_shared( snapshot_manager, path_factory, manifest_list, manifest_file, index_manifest_file,