Skip to content

Commit 25a1294

Browse files
authored
fix: align duplicate-key replacement semantics with Java Paimon (#389)
1 parent 84306cc commit 25a1294

10 files changed

Lines changed: 239 additions & 22 deletions

src/paimon/CMakeLists.txt

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -739,6 +739,7 @@ if(PAIMON_BUILD_TESTS)
739739
core/table/source/data_split_test.cpp
740740
core/table/source/deletion_file_test.cpp
741741
core/table/source/split_generator_test.cpp
742+
core/table/source/snapshot/snapshot_reader_test.cpp
742743
core/table/source/startup_mode_test.cpp
743744
core/table/source/table_scan_test.cpp
744745
core/table/system/system_table_test.cpp

src/paimon/core/deletionvectors/deletion_file_writer.cpp

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -42,8 +42,8 @@ Status DeletionFileWriter::Write(const std::string& key,
4242
}
4343
DataOutputStream output_stream(out_);
4444
PAIMON_ASSIGN_OR_RAISE(int32_t length, deletion_vector->SerializeTo(pool_, &output_stream));
45-
dv_metas_.insert(key, DeletionVectorMeta(key, static_cast<int32_t>(start), length,
46-
deletion_vector->GetCardinality()));
45+
dv_metas_.insert_or_assign(key, DeletionVectorMeta(key, static_cast<int32_t>(start), length,
46+
deletion_vector->GetCardinality()));
4747
return Status::OK();
4848
}
4949

src/paimon/core/deletionvectors/deletion_file_writer_test.cpp

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -95,6 +95,39 @@ TEST(DeletionFileWriterTest, GetResultWithoutCloseShouldFail) {
9595
ASSERT_NOK_WITH_MSG(writer->GetResult(), "Deletion file result length -1 out of int32 range");
9696
}
9797

98+
TEST(DeletionFileWriterTest, WriteOverwritesDuplicateDataFileName) {
99+
auto dir = UniqueTestDirectory::Create();
100+
ASSERT_OK_AND_ASSIGN(std::shared_ptr<FileSystem> fs,
101+
FileSystemFactory::Get("local", dir->Str(), {}));
102+
auto path_factory = std::make_shared<MockIndexPathFactory>(dir->Str());
103+
auto pool = GetDefaultPool();
104+
105+
ASSERT_OK_AND_ASSIGN(auto writer, DeletionFileWriter::Create(path_factory, fs, pool));
106+
107+
RoaringBitmap32 roaring_1;
108+
roaring_1.Add(1);
109+
auto dv_1 = std::make_shared<BitmapDeletionVector>(roaring_1);
110+
111+
RoaringBitmap32 roaring_2;
112+
roaring_2.Add(2);
113+
roaring_2.Add(3);
114+
auto dv_2 = std::make_shared<BitmapDeletionVector>(roaring_2);
115+
116+
ASSERT_OK(writer->Write("data-file-1", dv_1));
117+
ASSERT_OK(writer->Write("data-file-1", dv_2));
118+
ASSERT_OK(writer->Close());
119+
120+
ASSERT_OK_AND_ASSIGN(auto meta, writer->GetResult());
121+
const auto& dv_ranges = meta->DvRanges();
122+
ASSERT_TRUE(dv_ranges.has_value());
123+
ASSERT_EQ(dv_ranges->size(), 1);
124+
125+
auto iter = dv_ranges->find("data-file-1");
126+
ASSERT_NE(iter, dv_ranges->end());
127+
ASSERT_GT(iter->second.GetOffset(), 1);
128+
ASSERT_EQ(iter->second.GetCardinality(), std::optional<int64_t>(2));
129+
}
130+
98131
TEST(DeletionFileWriterTest, ExternalPathInResult) {
99132
auto dir = UniqueTestDirectory::Create();
100133
ASSERT_OK_AND_ASSIGN(std::shared_ptr<FileSystem> fs,

src/paimon/core/deletionvectors/deletion_vector.cpp

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -70,7 +70,7 @@ std::unordered_map<std::string, DeletionFile> DeletionVector::CreateDeletionFile
7070
assert(deletion_files.size() == data_files.size());
7171
for (size_t i = 0; i < deletion_files.size(); i++) {
7272
if (deletion_files[i] != std::nullopt) {
73-
deletion_file_map.emplace(data_files[i]->file_name, deletion_files[i].value());
73+
deletion_file_map.insert_or_assign(data_files[i]->file_name, deletion_files[i].value());
7474
}
7575
}
7676
return deletion_file_map;

src/paimon/core/deletionvectors/deletion_vector_test.cpp

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -193,6 +193,18 @@ TEST(DeletionVectorTest, CreateDeletionFileMap) {
193193
ASSERT_EQ(deletion_file_map.at("file-0.orc"), deletion_file_0);
194194
ASSERT_EQ(deletion_file_map.count("file-1.orc"), 0);
195195
ASSERT_EQ(deletion_file_map.at("file-2.orc"), deletion_file_2);
196+
197+
DeletionFile deletion_file_0_new("dv-0-new", /*offset=*/50, /*length=*/60,
198+
/*cardinality=*/7);
199+
std::vector<std::shared_ptr<DataFileMeta>> duplicate_data_files = {
200+
CreateDataFileMeta("file-0.orc"), CreateDataFileMeta("file-0.orc")};
201+
std::vector<std::optional<DeletionFile>> duplicate_deletion_files = {deletion_file_0,
202+
deletion_file_0_new};
203+
204+
auto duplicate_deletion_file_map =
205+
DeletionVector::CreateDeletionFileMap(duplicate_data_files, duplicate_deletion_files);
206+
ASSERT_EQ(duplicate_deletion_file_map.size(), 1);
207+
ASSERT_EQ(duplicate_deletion_file_map.at("file-0.orc"), deletion_file_0_new);
196208
}
197209

198210
} // namespace paimon::test

src/paimon/core/manifest/index_manifest_file_handler.cpp

Lines changed: 11 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -30,33 +30,31 @@ std::vector<IndexManifestEntry> IndexManifestFileHandler::BucketedCombiner::Comb
3030
const std::vector<IndexManifestEntry>& new_index_files) const {
3131
std::unordered_map<BucketIdentifier, IndexManifestEntry> index_entries;
3232
for (const auto& entry : prev_index_files) {
33-
index_entries.emplace(
33+
index_entries.insert_or_assign(
3434
BucketIdentifier(entry.partition, entry.bucket, entry.index_file->IndexType()), entry);
3535
}
3636

37-
std::unordered_map<BucketIdentifier, IndexManifestEntry> removed;
37+
std::vector<IndexManifestEntry> removed;
3838
removed.reserve(new_index_files.size());
39-
std::unordered_map<BucketIdentifier, IndexManifestEntry> added;
39+
std::vector<IndexManifestEntry> added;
4040
added.reserve(new_index_files.size());
4141

4242
for (const auto& entry : new_index_files) {
4343
if (entry.kind == FileKind::Delete()) {
44-
removed.emplace(
45-
BucketIdentifier(entry.partition, entry.bucket, entry.index_file->IndexType()),
46-
entry);
44+
removed.push_back(entry);
4745
} else if (entry.kind == FileKind::Add()) {
48-
added.emplace(
49-
BucketIdentifier(entry.partition, entry.bucket, entry.index_file->IndexType()),
50-
entry);
46+
added.push_back(entry);
5147
}
5248
}
5349

5450
// The deleted entry is processed first to avoid overwriting a new entry.
5551
for (const auto& entry : removed) {
56-
index_entries.erase(entry.first);
52+
index_entries.erase(
53+
BucketIdentifier(entry.partition, entry.bucket, entry.index_file->IndexType()));
5754
}
5855
for (const auto& entry : added) {
59-
index_entries.emplace(entry.first, entry.second);
56+
index_entries.insert_or_assign(
57+
BucketIdentifier(entry.partition, entry.bucket, entry.index_file->IndexType()), entry);
6058
}
6159

6260
std::vector<IndexManifestEntry> result_entries;
@@ -72,7 +70,7 @@ std::vector<IndexManifestEntry> IndexManifestFileHandler::GlobalFileNameCombiner
7270
const std::vector<IndexManifestEntry>& new_index_files) const {
7371
std::map<std::string, IndexManifestEntry> index_entries;
7472
for (const auto& entry : prev_index_files) {
75-
index_entries.emplace(entry.index_file->FileName(), entry);
73+
index_entries.insert_or_assign(entry.index_file->FileName(), entry);
7674
}
7775

7876
std::vector<IndexManifestEntry> removed;
@@ -93,7 +91,7 @@ std::vector<IndexManifestEntry> IndexManifestFileHandler::GlobalFileNameCombiner
9391
index_entries.erase(entry.index_file->FileName());
9492
}
9593
for (const auto& entry : added) {
96-
index_entries.emplace(entry.index_file->FileName(), entry);
94+
index_entries.insert_or_assign(entry.index_file->FileName(), entry);
9795
}
9896

9997
std::vector<IndexManifestEntry> result_entries;

src/paimon/core/manifest/index_manifest_file_handler.h

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -46,15 +46,15 @@ class IndexManifestFileHandler {
4646
const std::vector<IndexManifestEntry>& new_index_files) const = 0;
4747
};
4848

49-
/// Combine previous and new global index files by file `BucketIdentifier`.
49+
/// Combine previous and new index files by partition, bucket and index type.
5050
class BucketedCombiner : public IndexManifestFileCombiner {
5151
public:
5252
std::vector<IndexManifestEntry> Combine(
5353
const std::vector<IndexManifestEntry>& prev_index_files,
5454
const std::vector<IndexManifestEntry>& new_index_files) const override;
5555
};
5656

57-
/// Combine previous and new global index files by file name.
57+
/// Combine previous and new index files by file name.
5858
class GlobalFileNameCombiner : public IndexManifestFileCombiner {
5959
public:
6060
std::vector<IndexManifestEntry> Combine(

src/paimon/core/manifest/index_manifest_file_handler_test.cpp

Lines changed: 63 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -74,6 +74,25 @@ class IndexManifestFileHandlerTest : public testing::Test {
7474
/*external_path=*/std::nullopt));
7575
}
7676

77+
static IndexManifestEntry MakeDvEntry(const FileKind& kind, const BinaryRow& partition,
78+
int32_t bucket, const std::string& file_name,
79+
const std::vector<std::string>& data_file_names,
80+
int64_t row_count) {
81+
LinkedHashMap<std::string, DeletionVectorMeta> dv_ranges;
82+
int32_t offset = 0;
83+
for (const auto& data_file_name : data_file_names) {
84+
dv_ranges.insert(data_file_name,
85+
DeletionVectorMeta(data_file_name, offset, /*length=*/10,
86+
/*cardinality=*/std::nullopt));
87+
offset += 10;
88+
}
89+
return IndexManifestEntry(
90+
kind, partition, bucket,
91+
std::make_shared<IndexFileMeta>(DeletionVectorsIndexFile::DELETION_VECTORS_INDEX,
92+
file_name, /*file_size=*/row_count * 10, row_count,
93+
dv_ranges, /*external_path=*/std::nullopt));
94+
}
95+
7796
std::shared_ptr<MemoryPool> pool_;
7897
std::unique_ptr<UniqueTestDirectory> dir_;
7998
};
@@ -154,13 +173,55 @@ TEST_F(IndexManifestFileHandlerTest, BucketedCombinerUsesPartitionBucketAndIndex
154173
ASSERT_TRUE(found_bucket1);
155174
}
156175

176+
TEST_F(IndexManifestFileHandlerTest, BucketedCombinerOverwritesDuplicateAddedEntries) {
177+
ASSERT_OK_AND_ASSIGN(auto index_manifest_file, CreateManifestFile(/*bucket_mode=*/2));
178+
179+
auto partition = BinaryRowGenerator::GenerateRow({10}, pool_.get());
180+
std::vector<IndexManifestEntry> new_entries = {
181+
MakeEntry(FileKind::Add(), partition, /*bucket=*/0,
182+
DeletionVectorsIndexFile::DELETION_VECTORS_INDEX, "dv-0-old", 10),
183+
MakeEntry(FileKind::Add(), partition, /*bucket=*/0,
184+
DeletionVectorsIndexFile::DELETION_VECTORS_INDEX, "dv-0-new", 20)};
185+
186+
ASSERT_OK_AND_ASSIGN(std::string current_manifest,
187+
IndexManifestFileHandler::Write(
188+
/*previous_index_manifest=*/std::nullopt, new_entries,
189+
/*bucket_mode=*/2, index_manifest_file.get()));
190+
191+
std::vector<IndexManifestEntry> written_entries;
192+
ASSERT_OK(index_manifest_file->Read(current_manifest, /*filter=*/nullptr, &written_entries));
193+
ASSERT_EQ(written_entries.size(), 1);
194+
ASSERT_EQ(written_entries[0].index_file->FileName(), "dv-0-new");
195+
ASSERT_EQ(written_entries[0].index_file->RowCount(), 20);
196+
}
197+
198+
TEST_F(IndexManifestFileHandlerTest, GlobalCombinerOverwritesDuplicateAddedEntries) {
199+
ASSERT_OK_AND_ASSIGN(auto index_manifest_file, CreateManifestFile(/*bucket_mode=*/4));
200+
201+
auto partition = BinaryRow::EmptyRow();
202+
std::vector<IndexManifestEntry> new_entries = {
203+
MakeEntry(FileKind::Add(), partition, /*bucket=*/0, /*index_type=*/"BTREE", "global-0", 10),
204+
MakeEntry(FileKind::Add(), partition, /*bucket=*/0, /*index_type=*/"BTREE", "global-0",
205+
20)};
206+
207+
ASSERT_OK_AND_ASSIGN(std::string current_manifest,
208+
IndexManifestFileHandler::Write(
209+
/*previous_index_manifest=*/std::nullopt, new_entries,
210+
/*bucket_mode=*/4, index_manifest_file.get()));
211+
212+
std::vector<IndexManifestEntry> written_entries;
213+
ASSERT_OK(index_manifest_file->Read(current_manifest, /*filter=*/nullptr, &written_entries));
214+
ASSERT_EQ(written_entries.size(), 1);
215+
ASSERT_EQ(written_entries[0].index_file->FileName(), "global-0");
216+
ASSERT_EQ(written_entries[0].index_file->RowCount(), 20);
217+
}
218+
157219
TEST_F(IndexManifestFileHandlerTest, DvWithBucketUnawareModeReturnsNotImplemented) {
158220
ASSERT_OK_AND_ASSIGN(auto index_manifest_file, CreateManifestFile(/*bucket_mode=*/-1));
159221

160222
auto partition = BinaryRow::EmptyRow();
161223
std::vector<IndexManifestEntry> new_entries = {
162-
MakeEntry(FileKind::Add(), partition, /*bucket=*/0,
163-
DeletionVectorsIndexFile::DELETION_VECTORS_INDEX, "dv-0", 1)};
224+
MakeDvEntry(FileKind::Add(), partition, /*bucket=*/0, "dv-0", {"data-0.orc"}, 1)};
164225

165226
ASSERT_NOK_WITH_MSG(IndexManifestFileHandler::Write(
166227
/*previous_index_manifest=*/std::nullopt, new_entries,

src/paimon/core/table/source/snapshot/snapshot_reader.cpp

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -120,8 +120,8 @@ Result<std::vector<std::optional<DeletionFile>>> SnapshotReader::GetDeletionFile
120120
if (dv_metas != std::nullopt) {
121121
for (const auto& dv_meta_iter : dv_metas.value()) {
122122
const auto& dv_meta = dv_meta_iter.second;
123-
data_file_to_index_file_meta.insert(
124-
std::make_pair(dv_meta.GetDataFileName(), index_file_meta));
123+
data_file_to_index_file_meta.insert_or_assign(dv_meta.GetDataFileName(),
124+
index_file_meta);
125125
}
126126
}
127127
}
Lines changed: 112 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,112 @@
1+
/*
2+
* Copyright 2026-present Alibaba Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
#include "paimon/core/table/source/snapshot/snapshot_reader.h"
18+
19+
#include <memory>
20+
#include <optional>
21+
#include <string>
22+
#include <vector>
23+
24+
#include "arrow/api.h"
25+
#include "gtest/gtest.h"
26+
#include "paimon/core/deletionvectors/deletion_vectors_index_file.h"
27+
#include "paimon/core/index/index_file_handler.h"
28+
#include "paimon/core/index/index_file_meta.h"
29+
#include "paimon/core/io/data_file_meta.h"
30+
#include "paimon/core/utils/file_store_path_factory.h"
31+
#include "paimon/core/utils/index_file_path_factories.h"
32+
#include "paimon/fs/local/local_file_system.h"
33+
#include "paimon/memory/memory_pool.h"
34+
#include "paimon/testing/utils/testharness.h"
35+
36+
namespace paimon::test {
37+
class SnapshotReaderTest : public testing::Test {
38+
protected:
39+
void SetUp() override {
40+
pool_ = GetDefaultPool();
41+
dir_ = UniqueTestDirectory::Create();
42+
ASSERT_TRUE(dir_ != nullptr);
43+
}
44+
45+
std::shared_ptr<DataFileMeta> CreateDataFileMeta(const std::string& file_name) const {
46+
return std::make_shared<DataFileMeta>(
47+
file_name, /*file_size=*/100, /*row_count=*/10, DataFileMeta::EmptyMinKey(),
48+
DataFileMeta::EmptyMaxKey(), SimpleStats::EmptyStats(), SimpleStats::EmptyStats(),
49+
/*min_sequence_number=*/0, /*max_sequence_number=*/0, /*schema_id=*/0,
50+
DataFileMeta::DUMMY_LEVEL, std::vector<std::optional<std::string>>{}, Timestamp(0, 0),
51+
std::nullopt, nullptr, FileSource::Append(), std::nullopt, std::nullopt, std::nullopt,
52+
std::nullopt);
53+
}
54+
55+
std::shared_ptr<IndexFileMeta> CreateIndexFileMeta(const std::string& index_file_name,
56+
const std::string& data_file_name,
57+
int64_t offset, int64_t length,
58+
std::optional<int64_t> cardinality) const {
59+
LinkedHashMap<std::string, DeletionVectorMeta> dv_ranges;
60+
dv_ranges.insert(data_file_name,
61+
DeletionVectorMeta(data_file_name, offset, length, cardinality));
62+
return std::make_shared<IndexFileMeta>(DeletionVectorsIndexFile::DELETION_VECTORS_INDEX,
63+
index_file_name, /*file_size=*/100, /*row_count=*/10,
64+
dv_ranges, /*external_path=*/std::nullopt);
65+
}
66+
67+
Result<std::unique_ptr<IndexFileHandler>> CreateIndexFileHandler() const {
68+
auto schema = arrow::schema({arrow::field("f0", arrow::int32())});
69+
PAIMON_ASSIGN_OR_RAISE(
70+
std::shared_ptr<FileStorePathFactory> path_factory,
71+
FileStorePathFactory::Create(
72+
dir_->Str(), schema, /*partition_keys=*/{}, /*default_part_value=*/"", "orc",
73+
/*data_file_prefix=*/"data-", /*legacy_partition_name_enabled=*/true,
74+
/*external_paths=*/{}, /*global_index_external_path=*/std::nullopt,
75+
/*index_file_in_data_file_dir=*/false, pool_));
76+
auto path_factories = std::make_shared<IndexFilePathFactories>(path_factory);
77+
return std::make_unique<IndexFileHandler>(std::make_shared<LocalFileSystem>(),
78+
std::unique_ptr<IndexManifestFile>(),
79+
path_factories, /*dv_bitmap64=*/false, pool_);
80+
}
81+
82+
std::shared_ptr<MemoryPool> pool_;
83+
std::unique_ptr<UniqueTestDirectory> dir_;
84+
};
85+
86+
TEST_F(SnapshotReaderTest, GetDeletionFilesOverwritesDuplicateDataFileName) {
87+
ASSERT_OK_AND_ASSIGN(std::unique_ptr<IndexFileHandler> index_file_handler,
88+
CreateIndexFileHandler());
89+
SnapshotReader snapshot_reader(/*scan=*/nullptr, /*path_factory=*/nullptr,
90+
/*split_generator=*/nullptr, std::move(index_file_handler));
91+
92+
const std::string data_file_name = "data-0.orc";
93+
std::vector<std::shared_ptr<DataFileMeta>> data_files = {CreateDataFileMeta(data_file_name)};
94+
std::vector<std::shared_ptr<IndexFileMeta>> index_file_metas = {
95+
CreateIndexFileMeta("index-first", data_file_name, /*offset=*/1, /*length=*/11,
96+
/*cardinality=*/3),
97+
CreateIndexFileMeta("index-second", data_file_name, /*offset=*/2, /*length=*/22,
98+
/*cardinality=*/4)};
99+
100+
ASSERT_OK_AND_ASSIGN(std::vector<std::optional<DeletionFile>> deletion_files,
101+
snapshot_reader.GetDeletionFiles(BinaryRow::EmptyRow(), /*bucket=*/0,
102+
data_files, index_file_metas));
103+
104+
ASSERT_EQ(deletion_files.size(), 1);
105+
ASSERT_TRUE(deletion_files[0].has_value());
106+
EXPECT_EQ(deletion_files[0]->path, dir_->Str() + "/index/index-second");
107+
EXPECT_EQ(deletion_files[0]->offset, 2);
108+
EXPECT_EQ(deletion_files[0]->length, 22);
109+
EXPECT_EQ(deletion_files[0]->cardinality, std::optional<int64_t>(4));
110+
}
111+
112+
} // namespace paimon::test

0 commit comments

Comments
 (0)