Skip to content

Commit a4c1fef

Browse files
lucasfanglucasfang
authored
fix(commit): fix changelogRecordCount nullopt and add ut (#428)
1 parent b22b3c4 commit a4c1fef

15 files changed

Lines changed: 875 additions & 90 deletions

‎src/paimon/core/catalog/commit_table_request_test.cpp‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -56,7 +56,6 @@ TEST(CommitTableRequestTest, TestSimple) {
5656
"baseManifestListSize": 291,
5757
"deltaManifestList": "manifest-list-3879e56f-2f27-49ae-a2f3-3dcbb8eb0beb-1",
5858
"deltaManifestListSize": 1342,
59-
"changelogManifestList": null,
6059
"commitUser": "commit_user_1",
6160
"commitIdentifier": 9223372036854775807,
6261
"commitKind": "APPEND",

‎src/paimon/core/manifest/manifest_entry.cpp‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,22 @@ const std::shared_ptr<arrow::DataType>& ManifestEntry::DataType() {
3333
return data_type;
3434
}
3535

36+
int64_t ManifestEntry::RecordCount(const std::vector<ManifestEntry>& entries) {
37+
int64_t record_count = 0;
38+
for (const auto& entry : entries) {
39+
record_count += entry.File()->row_count;
40+
}
41+
return record_count;
42+
}
43+
44+
std::optional<int64_t> ManifestEntry::NullableRecordCount(
45+
const std::vector<ManifestEntry>& entries) {
46+
if (entries.empty()) {
47+
return std::nullopt;
48+
}
49+
return RecordCount(entries);
50+
}
51+
3652
int64_t ManifestEntry::RecordCountAdd(const std::vector<ManifestEntry>& entries) {
3753
int64_t record_count = 0;
3854
for (const auto& entry : entries) {

‎src/paimon/core/manifest/manifest_entry.h‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,9 @@ namespace paimon {
3838
class ManifestEntry : public FileEntry {
3939
public:
4040
static const std::shared_ptr<arrow::DataType>& DataType();
41+
static int64_t RecordCount(const std::vector<ManifestEntry>& manifest_entries);
42+
static std::optional<int64_t> NullableRecordCount(
43+
const std::vector<ManifestEntry>& manifest_entries);
4144
static int64_t RecordCountAdd(const std::vector<ManifestEntry>& manifest_entries);
4245
static int64_t RecordCountDelete(const std::vector<ManifestEntry>& manifest_entries);
4346

‎src/paimon/core/manifest/manifest_entry_serializer_test.cpp‎

Lines changed: 21 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -33,12 +33,12 @@ namespace paimon::test {
3333

3434
class ManifestEntrySerializerTest : public testing::Test {
3535
private:
36-
std::shared_ptr<DataFileMeta> GetDataFileMeta() {
36+
std::shared_ptr<DataFileMeta> GetDataFileMeta(int64_t row_count) {
3737
return std::make_shared<DataFileMeta>(
38-
"some_file_name", 1024, 8, DataFileMeta::EmptyMinKey(), DataFileMeta::EmptyMaxKey(),
39-
SimpleStats::EmptyStats(), SimpleStats::EmptyStats(), /*min_seq_no=*/16,
40-
/*max_seq_no=*/32,
41-
/*schema_id=*/1, /*level=*/2, /*extra_files=*/std::vector<std::optional<std::string>>(),
38+
"some_file_name", 1024, row_count, DataFileMeta::EmptyMinKey(),
39+
DataFileMeta::EmptyMaxKey(), SimpleStats::EmptyStats(), SimpleStats::EmptyStats(),
40+
/*min_seq_no=*/16, /*max_seq_no=*/32, /*schema_id=*/1, /*level=*/2,
41+
/*extra_files=*/std::vector<std::optional<std::string>>(),
4242
/*creation_time=*/Timestamp(0, 0), /*delete_row_count=*/3,
4343
/*embedded_index=*/nullptr, /*file_source=*/std::nullopt,
4444
/*value_stats_cols=*/std::nullopt, /*external_path=*/std::optional<std::string>(),
@@ -48,9 +48,9 @@ class ManifestEntrySerializerTest : public testing::Test {
4848
TEST_F(ManifestEntrySerializerTest, TestToFromRow) {
4949
auto pool = GetDefaultPool();
5050
std::vector<ManifestEntry> entries = {
51-
ManifestEntry(FileKind::Add(), BinaryRow::EmptyRow(), 0, 2, GetDataFileMeta()),
51+
ManifestEntry(FileKind::Add(), BinaryRow::EmptyRow(), 0, 2, GetDataFileMeta(8)),
5252
ManifestEntry(FileKind::Add(), BinaryRowGenerator::GenerateRow({10}, pool.get()), 1, 2,
53-
GetDataFileMeta())};
53+
GetDataFileMeta(8))};
5454
ManifestEntrySerializer serializer(pool);
5555
for (const auto& entry : entries) {
5656
ASSERT_OK_AND_ASSIGN(auto row, serializer.ToRow(entry));
@@ -59,4 +59,18 @@ TEST_F(ManifestEntrySerializerTest, TestToFromRow) {
5959
ASSERT_EQ(entry.ToString(), result_entry.ToString());
6060
}
6161
}
62+
63+
TEST_F(ManifestEntrySerializerTest, TestNullableRecordCount) {
64+
std::vector<ManifestEntry> empty_entries;
65+
ASSERT_FALSE(ManifestEntry::NullableRecordCount(empty_entries).has_value());
66+
67+
std::vector<ManifestEntry> entries = {
68+
ManifestEntry(FileKind::Add(), BinaryRow::EmptyRow(), 0, 2, GetDataFileMeta(3)),
69+
ManifestEntry(FileKind::Delete(), BinaryRow::EmptyRow(), 0, 2, GetDataFileMeta(8))};
70+
ASSERT_EQ(11, ManifestEntry::RecordCount(entries));
71+
72+
std::optional<int64_t> record_count = ManifestEntry::NullableRecordCount(entries);
73+
ASSERT_TRUE(record_count.has_value());
74+
ASSERT_EQ(11, record_count.value());
75+
}
6276
} // namespace paimon::test

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

Lines changed: 179 additions & 30 deletions
Large diffs are not rendered by default.

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

Lines changed: 31 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
#include <functional>
2121
#include <memory>
2222
#include <optional>
23+
#include <string>
2324
#include <unordered_map>
2425
#include <vector>
2526

@@ -35,6 +36,7 @@ namespace paimon {
3536
class ManifestEntry;
3637
struct IndexManifestEntry;
3738
class CommitScanner;
39+
class FileStorePathFactory;
3840
class ManifestFile;
3941
class ManifestList;
4042
class RowIdColumnConflictChecker;
@@ -48,7 +50,9 @@ class ConflictDetection {
4850
std::shared_ptr<SnapshotManager> snapshot_manager,
4951
std::shared_ptr<ManifestList> manifest_list,
5052
std::shared_ptr<ManifestFile> manifest_file,
51-
std::shared_ptr<CommitScanner> commit_scanner);
53+
std::shared_ptr<CommitScanner> commit_scanner, const std::string& commit_user,
54+
const std::string& table_name,
55+
const std::shared_ptr<FileStorePathFactory>& path_factory);
5256

5357
Status CheckConflicts(const Snapshot& latest_snapshot,
5458
const std::vector<ManifestEntry>& base_entries,
@@ -75,17 +79,37 @@ class ConflictDetection {
7579

7680
private:
7781
Status CheckBucketKeepSame(const std::vector<ManifestEntry>& all_entries,
78-
const Snapshot::CommitKind& commit_kind) const;
82+
const Snapshot::CommitKind& commit_kind,
83+
const std::string& base_commit_user,
84+
const std::vector<ManifestEntry>& base_entries,
85+
const std::vector<ManifestEntry>& delta_entries) const;
7986

8087
Status BucketNumMismatch(const BinaryRow& partition, int32_t num_buckets,
8188
int32_t previous_num_buckets) const;
8289

90+
Status TotalBucketsChanged(const BinaryRow& partition, int32_t num_buckets,
91+
int32_t previous_num_buckets, const std::string& base_commit_user,
92+
const std::vector<ManifestEntry>& base_entries,
93+
const std::vector<ManifestEntry>& delta_entries) const;
94+
95+
std::string BuildConflictMessage(const std::string& message,
96+
const std::string& base_commit_user,
97+
const std::vector<ManifestEntry>& base_entries,
98+
const std::vector<ManifestEntry>& delta_entries,
99+
const std::string& cause = "") const;
100+
83101
void MarkBucketCheckedPartitions(
84102
const std::unordered_map<BinaryRow, int32_t>& total_buckets) const;
85103

86-
Status CheckDeleteInEntries(const std::vector<ManifestEntry>& merged_entries) const;
104+
Status CheckDeleteInEntries(const std::vector<ManifestEntry>& merged_entries,
105+
const std::string& base_commit_user,
106+
const std::vector<ManifestEntry>& base_entries,
107+
const std::vector<ManifestEntry>& delta_entries) const;
87108

88-
Status CheckKeyRange(const std::vector<ManifestEntry>& merged_entries) const;
109+
Status CheckKeyRange(const std::vector<ManifestEntry>& merged_entries,
110+
const std::string& base_commit_user,
111+
const std::vector<ManifestEntry>& base_entries,
112+
const std::vector<ManifestEntry>& delta_entries) const;
89113

90114
Status CheckRowIdExistence(const std::vector<ManifestEntry>& base_entries,
91115
const std::vector<ManifestEntry>& delta_entries,
@@ -121,6 +145,9 @@ class ConflictDetection {
121145
std::shared_ptr<ManifestList> manifest_list_;
122146
std::shared_ptr<ManifestFile> manifest_file_;
123147
std::shared_ptr<CommitScanner> commit_scanner_;
148+
std::shared_ptr<FileStorePathFactory> path_factory_;
149+
std::string commit_user_;
150+
std::string table_name_;
124151
mutable LinkedHashMap<BinaryRow, bool> same_bucket_checked_partitions_;
125152
};
126153

0 commit comments

Comments
 (0)