Skip to content

Commit 12f95c5

Browse files
lucasfanglucasfang
committed
fix
1 parent f8c3ad2 commit 12f95c5

9 files changed

Lines changed: 128 additions & 203 deletions

‎src/paimon/common/utils/linked_hash_map.h‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -70,12 +70,13 @@ class LinkedHashMap {
7070
}
7171

7272
IteratorType erase(const K& key) {
73-
if (!map_.count(key)) {
73+
auto map_iter = map_.find(key);
74+
if (map_iter == map_.end()) {
7475
return order_.end();
7576
}
76-
auto iter = order_.erase(map_[key]);
77-
map_.erase(key);
78-
return iter;
77+
IteratorType order_iter = map_iter->second;
78+
map_.erase(map_iter);
79+
return order_.erase(order_iter);
7980
}
8081

8182
IteratorType insert(const K& key, const V& value) {

‎src/paimon/core/core_options.h‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,7 @@ class Cache;
4848

4949
class PAIMON_EXPORT CoreOptions {
5050
public:
51-
// Specifies how to initialize the next sequence number for primary key table writers.
51+
/// Specifies how to initialize the next sequence number for primary key table writers.
5252
enum class SequenceNumberInitMode {
5353
// initialize by scanning existing file metadata.
5454
SCAN,

‎src/paimon/core/operation/commit/conflict_detection.cpp‎

Lines changed: 31 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@
3131
#include "paimon/common/data/blob_utils.h"
3232
#include "paimon/common/types/data_field.h"
3333
#include "paimon/common/utils/fields_comparator.h"
34+
#include "paimon/common/utils/range_helper.h"
3435
#include "paimon/core/deletionvectors/deletion_vectors_index_file.h"
3536
#include "paimon/core/manifest/file_entry.h"
3637
#include "paimon/core/manifest/file_kind.h"
@@ -351,56 +352,48 @@ Status ConflictDetection::CheckRowIdRangeConflicts(
351352
return Status::OK();
352353
}
353354

354-
std::vector<std::pair<Range, ManifestEntry>> entries_with_ranges;
355+
std::vector<ManifestEntry> entries_with_ranges;
355356
entries_with_ranges.reserve(merged_entries.size());
356357
for (const ManifestEntry& entry : merged_entries) {
357-
if (!entry.File()->first_row_id) {
358-
continue;
358+
if (entry.File()->first_row_id) {
359+
entries_with_ranges.push_back(entry);
359360
}
360-
int64_t range_from = entry.File()->first_row_id.value();
361-
int64_t range_to = range_from + entry.File()->row_count - 1;
362-
entries_with_ranges.emplace_back(Range(range_from, range_to), entry);
363361
}
364362
if (entries_with_ranges.empty()) {
365363
return Status::OK();
366364
}
367365

368-
std::sort(entries_with_ranges.begin(), entries_with_ranges.end(),
369-
[](const auto& lhs, const auto& rhs) { return lhs.first.from < rhs.first.from; });
370-
371-
size_t group_start = 0;
372-
int64_t group_max_to = entries_with_ranges[0].first.to;
373-
for (size_t i = 1; i <= entries_with_ranges.size(); ++i) {
374-
bool overlap_group_end =
375-
(i == entries_with_ranges.size()) || (entries_with_ranges[i].first.from > group_max_to);
376-
if (!overlap_group_end) {
377-
group_max_to = std::max(group_max_to, entries_with_ranges[i].first.to);
378-
continue;
379-
}
380-
381-
std::optional<std::pair<int64_t, int64_t>> expected_range;
382-
for (size_t j = group_start; j < i; ++j) {
383-
const ManifestEntry& entry = entries_with_ranges[j].second;
384-
if (IsDedicatedStorageFile(entry.FileName())) {
385-
continue;
386-
}
387-
std::pair<int64_t, int64_t> current = {entries_with_ranges[j].first.from,
388-
entries_with_ranges[j].first.to};
389-
if (!expected_range) {
390-
expected_range = current;
391-
} else if (expected_range.value() != current) {
392-
return Status::Invalid(
393-
"For Data Evolution table, multiple MERGE INTO/COMPACT operations have "
394-
"encountered row-id range conflicts.");
366+
RangeHelper<ManifestEntry> range_helper(
367+
[](const ManifestEntry& entry) -> Result<int64_t> {
368+
return entry.File()->first_row_id.value();
369+
},
370+
[](const ManifestEntry& entry) -> Result<int64_t> {
371+
return entry.File()->first_row_id.value() + entry.File()->row_count - 1;
372+
});
373+
PAIMON_ASSIGN_OR_RAISE(std::vector<std::vector<ManifestEntry>> merged_groups,
374+
range_helper.MergeOverlappingRanges(std::move(entries_with_ranges)));
375+
376+
for (const auto& group : merged_groups) {
377+
std::vector<ManifestEntry> data_files;
378+
for (const ManifestEntry& entry : group) {
379+
// release-1.4 parity: row-id range conflict check excludes blob files only.
380+
if (!BlobUtils::IsBlobFile(entry.FileName())) {
381+
data_files.push_back(entry);
395382
}
396383
}
397384

398-
if (i < entries_with_ranges.size()) {
399-
group_start = i;
400-
group_max_to = entries_with_ranges[i].first.to;
385+
PAIMON_ASSIGN_OR_RAISE(bool all_data_ranges_same, range_helper.AreAllRangesSame(data_files));
386+
if (!all_data_ranges_same) {
387+
return Status::Invalid(
388+
"For Data Evolution table, multiple MERGE INTO/COMPACT operations have "
389+
"encountered row-id range conflicts.");
401390
}
402391
}
403392

393+
// TODO(yonghao.fyh): release-1.4 parity keeps dedicated-storage row-id containment unchecked here.
394+
// Consider adding a strict-mode validation that each dedicated-file range is fully
395+
// covered by one data-file range to catch dangling/cross-file mappings.
396+
404397
return Status::OK();
405398
}
406399

@@ -423,7 +416,8 @@ Status ConflictDetection::CheckForRowIdFromSnapshot(
423416
PAIMON_ASSIGN_OR_RAISE(Snapshot check_snapshot,
424417
snapshot_manager_->LoadSnapshot(row_id_check_from_snapshot_.value()));
425418
if (!check_snapshot.NextRowId()) {
426-
return Status::OK();
419+
return Status::Invalid(fmt::format("Next row id cannot be null for snapshot {}.",
420+
row_id_check_from_snapshot_.value()));
427421
}
428422
int64_t check_next_row_id = check_snapshot.NextRowId().value();
429423

‎src/paimon/core/operation/commit/conflict_detection_test.cpp‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -243,6 +243,38 @@ TEST_F(ConflictDetectionTest, TestGlobalIndexRowIdExistenceConflicts) {
243243
"Global index row ID existence conflict");
244244
}
245245

246+
TEST_F(ConflictDetectionTest, TestDedicatedStorageRowIdRangeParityWithRelease14) {
247+
ASSERT_OK_AND_ASSIGN(
248+
std::shared_ptr<TableSchema> table_schema,
249+
TableSchema::Create(/*schema_id=*/0, arrow::schema(fields_), /*partition_keys=*/{"f1"},
250+
/*primary_keys=*/{}, /*options=*/{}));
251+
ASSERT_OK_AND_ASSIGN(CoreOptions core_options,
252+
CoreOptions::FromMap({{Options::DATA_EVOLUTION_ENABLED, "true"}}));
253+
ConflictDetection detection(table_schema, core_options, nullptr, nullptr, nullptr);
254+
255+
const BinaryRow partition = CreateIntRow(10);
256+
std::vector<ManifestEntry> base_entries;
257+
base_entries.push_back(CreateManifestEntryWithFirstRowId("data-0.orc", partition,
258+
FileKind::Add(), /*bucket=*/0,
259+
/*first_row_id=*/0,
260+
/*row_count=*/10));
261+
262+
std::vector<ManifestEntry> out_of_range_dedicated_entries;
263+
out_of_range_dedicated_entries.push_back(CreateManifestEntryWithFirstRowId(
264+
"blob-0.blob", partition, FileKind::Add(), /*bucket=*/0, /*first_row_id=*/5,
265+
/*row_count=*/10));
266+
// release-1.4 parity: CheckRowIdRangeConflicts validates only data-file range consistency.
267+
ASSERT_OK(CheckConflicts(detection, base_entries, out_of_range_dedicated_entries,
268+
Snapshot::CommitKind::Compact()));
269+
270+
std::vector<ManifestEntry> contained_dedicated_entries;
271+
contained_dedicated_entries.push_back(CreateManifestEntryWithFirstRowId(
272+
"blob-1.blob", partition, FileKind::Add(), /*bucket=*/0, /*first_row_id=*/5,
273+
/*row_count=*/4));
274+
ASSERT_OK(CheckConflicts(detection, base_entries, contained_dedicated_entries,
275+
Snapshot::CommitKind::Compact()));
276+
}
277+
246278
TEST_F(ConflictDetectionTest, TestBucketKeepSame) {
247279
ASSERT_OK_AND_ASSIGN(
248280
std::shared_ptr<TableSchema> table_schema,

‎src/paimon/core/operation/commit/retry_waiter.cpp‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -32,8 +32,16 @@ RetryWaiter::RetryWaiter(int64_t min_retry_wait_ms, int64_t max_retry_wait_ms)
3232
void RetryWaiter::RetryWait(int32_t retry_count) const {
3333
int32_t non_negative_retry_count = std::max<int32_t>(0, retry_count);
3434
double exponential = std::pow(2.0, static_cast<double>(non_negative_retry_count));
35-
int64_t retry_wait = static_cast<int64_t>(min_retry_wait_ms_ * exponential);
36-
retry_wait = std::min(retry_wait, max_retry_wait_ms_);
35+
int64_t retry_wait = 0;
36+
if (min_retry_wait_ms_ > 0 && max_retry_wait_ms_ > 0) {
37+
double max_safe_exponential =
38+
static_cast<double>(max_retry_wait_ms_) / static_cast<double>(min_retry_wait_ms_);
39+
if (!std::isfinite(exponential) || exponential >= max_safe_exponential) {
40+
retry_wait = max_retry_wait_ms_;
41+
} else {
42+
retry_wait = static_cast<int64_t>(min_retry_wait_ms_ * exponential);
43+
}
44+
}
3745

3846
int64_t jitter_upper = std::max<int64_t>(1, static_cast<int64_t>(retry_wait * 0.2));
3947
std::mt19937 rng(std::random_device{}()); // NOLINT(whitespace/braces)

‎src/paimon/core/operation/commit/retry_waiter_test.cpp‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,4 +33,15 @@ TEST(RetryWaiterTest, TestRetryWaitWithZeroBoundsReturnsQuickly) {
3333
ASSERT_LT(elapsed.count(), 20);
3434
}
3535

36+
TEST(RetryWaiterTest, TestRetryWaitWithLargeRetryCountIsClamped) {
37+
RetryWaiter waiter(/*min_retry_wait_ms=*/10, /*max_retry_wait_ms=*/10);
38+
39+
auto begin = std::chrono::steady_clock::now();
40+
waiter.RetryWait(/*retry_count=*/1024);
41+
auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
42+
std::chrono::steady_clock::now() - begin);
43+
44+
ASSERT_GE(elapsed.count(), 8);
45+
}
46+
3647
} // namespace paimon::test

‎src/paimon/core/operation/commit/row_tracking_commit_utils_test.cpp‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -171,4 +171,36 @@ TEST_F(RowTrackingCommitUtilsTest, TestAssignRowTrackingDoesNotMutateInputEntrie
171171
EXPECT_NE(input_file_1.get(), assigned.assigned_entries[1].File().get());
172172
}
173173

174+
TEST_F(RowTrackingCommitUtilsTest, TestAssignRowTrackingReassignsOnRetryWithAdvancedRowId) {
175+
std::vector<ManifestEntry> input;
176+
input.push_back(CreateEntry("retry-file", /*row_count=*/10, /*min_seq=*/0, /*max_seq=*/0,
177+
FileSource::Append(), std::vector<std::string>{"f0"}));
178+
179+
ASSERT_OK_AND_ASSIGN(RowTrackingCommitUtils::RowTrackingAssigned first_attempt,
180+
RowTrackingCommitUtils::AssignRowTracking(
181+
/*new_snapshot_id=*/100, /*first_row_id_start=*/1000, input));
182+
ASSERT_EQ(1u, first_attempt.assigned_entries.size());
183+
EXPECT_EQ(100, first_attempt.assigned_entries[0].File()->min_sequence_number);
184+
EXPECT_EQ(100, first_attempt.assigned_entries[0].File()->max_sequence_number);
185+
ASSERT_TRUE(first_attempt.assigned_entries[0].File()->first_row_id.has_value());
186+
EXPECT_EQ(1000, first_attempt.assigned_entries[0].File()->first_row_id.value());
187+
EXPECT_EQ(1010, first_attempt.next_row_id_start);
188+
189+
// Simulate CAS retry with a newer latest snapshot and advanced next_row_id.
190+
ASSERT_OK_AND_ASSIGN(RowTrackingCommitUtils::RowTrackingAssigned second_attempt,
191+
RowTrackingCommitUtils::AssignRowTracking(
192+
/*new_snapshot_id=*/101, /*first_row_id_start=*/2000, input));
193+
ASSERT_EQ(1u, second_attempt.assigned_entries.size());
194+
EXPECT_EQ(101, second_attempt.assigned_entries[0].File()->min_sequence_number);
195+
EXPECT_EQ(101, second_attempt.assigned_entries[0].File()->max_sequence_number);
196+
ASSERT_TRUE(second_attempt.assigned_entries[0].File()->first_row_id.has_value());
197+
EXPECT_EQ(2000, second_attempt.assigned_entries[0].File()->first_row_id.value());
198+
EXPECT_EQ(2010, second_attempt.next_row_id_start);
199+
200+
// Input remains immutable across attempts; retry assignment always starts from fresh metadata.
201+
EXPECT_EQ(0, input[0].File()->min_sequence_number);
202+
EXPECT_EQ(0, input[0].File()->max_sequence_number);
203+
EXPECT_EQ(std::nullopt, input[0].File()->first_row_id);
204+
}
205+
174206
} // namespace paimon::test

0 commit comments

Comments
 (0)