Skip to content

Commit 8596334

Browse files
lucasfanglucasfang
committed
fix
1 parent b4c3967 commit 8596334

11 files changed

Lines changed: 154 additions & 113 deletions

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

Lines changed: 15 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -24,36 +24,29 @@ namespace {
2424

2525
class FixedInputCommitChangesProvider final : public CommitChangesProvider {
2626
public:
27-
FixedInputCommitChangesProvider(const std::vector<ManifestEntry>& delta_files,
28-
const std::vector<ManifestEntry>& changelog_files,
29-
const std::vector<IndexManifestEntry>& index_entries)
30-
: delta_files_(delta_files),
31-
changelog_files_(changelog_files),
32-
index_entries_(index_entries) {}
33-
34-
Status Provide(const std::optional<Snapshot>&, std::vector<ManifestEntry>* delta_files,
35-
std::vector<ManifestEntry>* changelog_files,
36-
std::vector<IndexManifestEntry>* index_entries) const override {
37-
*delta_files = delta_files_;
38-
*changelog_files = changelog_files_;
39-
*index_entries = index_entries_;
40-
return Status::OK();
27+
FixedInputCommitChangesProvider(std::vector<ManifestEntry> delta_files,
28+
std::vector<ManifestEntry> changelog_files,
29+
std::vector<IndexManifestEntry> index_entries)
30+
: commit_changes_(std::make_shared<CommitChanges>(std::move(delta_files),
31+
std::move(changelog_files),
32+
std::move(index_entries))) {}
33+
34+
Result<std::shared_ptr<CommitChanges>> Provide(
35+
const std::optional<Snapshot>&) const override {
36+
return commit_changes_;
4137
}
4238

4339
private:
44-
const std::vector<ManifestEntry>& delta_files_;
45-
const std::vector<ManifestEntry>& changelog_files_;
46-
const std::vector<IndexManifestEntry>& index_entries_;
40+
std::shared_ptr<CommitChanges> commit_changes_;
4741
};
4842

4943
} // namespace
5044

5145
std::shared_ptr<CommitChangesProvider> CommitChangesProvider::Provider(
52-
const std::vector<ManifestEntry>& delta_files,
53-
const std::vector<ManifestEntry>& changelog_files,
54-
const std::vector<IndexManifestEntry>& index_entries) {
55-
return std::make_shared<FixedInputCommitChangesProvider>(delta_files, changelog_files,
56-
index_entries);
46+
std::vector<ManifestEntry> delta_files, std::vector<ManifestEntry> changelog_files,
47+
std::vector<IndexManifestEntry> index_entries) {
48+
return std::make_shared<FixedInputCommitChangesProvider>(
49+
std::move(delta_files), std::move(changelog_files), std::move(index_entries));
5750
}
5851

5952
} // namespace paimon

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

Lines changed: 20 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -29,19 +29,31 @@
2929

3030
namespace paimon {
3131

32+
struct CommitChanges {
33+
CommitChanges() = default;
34+
35+
CommitChanges(std::vector<ManifestEntry> delta, std::vector<ManifestEntry> changelog,
36+
std::vector<IndexManifestEntry> index)
37+
: delta_files(std::move(delta)),
38+
changelog_files(std::move(changelog)),
39+
index_entries(std::move(index)) {}
40+
41+
std::vector<ManifestEntry> delta_files;
42+
std::vector<ManifestEntry> changelog_files;
43+
std::vector<IndexManifestEntry> index_entries;
44+
};
45+
3246
class CommitChangesProvider {
3347
public:
3448
virtual ~CommitChangesProvider() = default;
3549

3650
static std::shared_ptr<CommitChangesProvider> Provider(
37-
const std::vector<ManifestEntry>& delta_files,
38-
const std::vector<ManifestEntry>& changelog_files,
39-
const std::vector<IndexManifestEntry>& index_entries);
40-
41-
virtual Status Provide(const std::optional<Snapshot>& latest_snapshot,
42-
std::vector<ManifestEntry>* delta_files,
43-
std::vector<ManifestEntry>* changelog_files,
44-
std::vector<IndexManifestEntry>* index_entries) const = 0;
51+
std::vector<ManifestEntry> delta_files,
52+
std::vector<ManifestEntry> changelog_files,
53+
std::vector<IndexManifestEntry> index_entries);
54+
55+
virtual Result<std::shared_ptr<CommitChanges>> Provide(
56+
const std::optional<Snapshot>& latest_snapshot) const = 0;
4557
};
4658

4759
} // namespace paimon

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

Lines changed: 12 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -90,11 +90,11 @@ TEST(CommitChangesProviderTest, TestProvideReturnsGivenEntries) {
9090
std::shared_ptr<CommitChangesProvider> provider =
9191
CommitChangesProvider::Provider(delta_files, changelog_files, index_entries);
9292

93-
std::vector<ManifestEntry> provided_delta;
94-
std::vector<ManifestEntry> provided_changelog;
95-
std::vector<IndexManifestEntry> provided_index;
96-
ASSERT_OK(
97-
provider->Provide(std::nullopt, &provided_delta, &provided_changelog, &provided_index));
93+
ASSERT_OK_AND_ASSIGN(std::shared_ptr<CommitChanges> provided, provider->Provide(std::nullopt));
94+
95+
const std::vector<ManifestEntry>& provided_delta = provided->delta_files;
96+
const std::vector<ManifestEntry>& provided_changelog = provided->changelog_files;
97+
const std::vector<IndexManifestEntry>& provided_index = provided->index_entries;
9898

9999
ASSERT_EQ(delta_files.size(), provided_delta.size());
100100
ASSERT_EQ(changelog_files.size(), provided_changelog.size());
@@ -104,7 +104,7 @@ TEST(CommitChangesProviderTest, TestProvideReturnsGivenEntries) {
104104
EXPECT_EQ("index-1", provided_index[0].index_file->FileName());
105105
}
106106

107-
TEST(CommitChangesProviderTest, TestProvideReflectsReferencedInputs) {
107+
TEST(CommitChangesProviderTest, TestProvideUsesCopiedInputs) {
108108
std::vector<ManifestEntry> delta_files = {
109109
CreateManifestEntry("delta-1", FileKind::Add(), /*partition_value=*/1)};
110110
std::vector<ManifestEntry> changelog_files;
@@ -118,18 +118,12 @@ TEST(CommitChangesProviderTest, TestProvideReflectsReferencedInputs) {
118118
CreateManifestEntry("changelog-2", FileKind::Add(), /*partition_value=*/3));
119119
index_entries.push_back(CreateIndexEntry("index-2", /*partition_value=*/4));
120120

121-
std::vector<ManifestEntry> provided_delta;
122-
std::vector<ManifestEntry> provided_changelog;
123-
std::vector<IndexManifestEntry> provided_index;
124-
ASSERT_OK(
125-
provider->Provide(std::nullopt, &provided_delta, &provided_changelog, &provided_index));
126-
127-
ASSERT_EQ(2u, provided_delta.size());
128-
ASSERT_EQ(1u, provided_changelog.size());
129-
ASSERT_EQ(1u, provided_index.size());
130-
EXPECT_EQ("delta-2", provided_delta[1].FileName());
131-
EXPECT_EQ("changelog-2", provided_changelog[0].FileName());
132-
EXPECT_EQ("index-2", provided_index[0].index_file->FileName());
121+
ASSERT_OK_AND_ASSIGN(std::shared_ptr<CommitChanges> provided, provider->Provide(std::nullopt));
122+
123+
ASSERT_EQ(1u, provided->delta_files.size());
124+
ASSERT_EQ(0u, provided->changelog_files.size());
125+
ASSERT_EQ(0u, provided->index_entries.size());
126+
EXPECT_EQ("delta-1", provided->delta_files[0].FileName());
133127
}
134128

135129
} // namespace paimon::test

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

Lines changed: 25 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@
3333
#include "paimon/core/manifest/manifest_entry.h"
3434
#include "paimon/core/operation/commit/overwrite_changes_provider.h"
3535
#include "paimon/core/operation/file_store_scan.h"
36+
#include "paimon/core/table/bucket_mode.h"
3637
#include "paimon/scan_context.h"
3738

3839
namespace paimon {
@@ -90,13 +91,8 @@ Result<std::vector<ManifestEntry>> CommitScanner::ReadAllEntriesFromChangedParti
9091
std::vector<std::map<std::string, std::string>> partition_filters;
9192
PAIMON_ASSIGN_OR_RAISE(partition_filters, ToPartitionFilters(changed_partitions));
9293

93-
auto scan_filter = std::make_shared<ScanFilter>(/*predicate=*/nullptr, partition_filters,
94-
/*bucket_filter=*/std::nullopt);
95-
if (!scan_supplier_) {
96-
return Status::Invalid("CommitScanner requires non-empty scan supplier.");
97-
}
98-
99-
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileStoreScan> scan, scan_supplier_(scan_filter));
94+
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileStoreScan> scan,
95+
NewScan(partition_filters, /*for_overwrite=*/false));
10096
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FileStoreScan::RawPlan> plan,
10197
scan->WithSnapshot(snapshot)->WithKind(ScanMode::ALL)->CreatePlan());
10298
return plan->Files();
@@ -105,16 +101,27 @@ Result<std::vector<ManifestEntry>> CommitScanner::ReadAllEntriesFromChangedParti
105101
Result<std::vector<ManifestEntry>> CommitScanner::ReadAllEntriesFromPartitions(
106102
const Snapshot& snapshot,
107103
const std::vector<std::map<std::string, std::string>>& partitions) const {
104+
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileStoreScan> scan,
105+
NewScan(partitions, /*for_overwrite=*/false));
106+
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FileStoreScan::RawPlan> plan,
107+
scan->WithSnapshot(snapshot)->WithKind(ScanMode::ALL)->CreatePlan());
108+
return plan->Files();
109+
}
110+
111+
Result<std::unique_ptr<FileStoreScan>> CommitScanner::NewScan(
112+
const std::vector<std::map<std::string, std::string>>& partitions,
113+
bool for_overwrite) const {
108114
auto scan_filter = std::make_shared<ScanFilter>(/*predicate=*/nullptr, partitions,
109115
/*bucket_filter=*/std::nullopt);
110116
if (!scan_supplier_) {
111117
return Status::Invalid("CommitScanner requires non-empty scan supplier.");
112118
}
113119

114120
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileStoreScan> scan, scan_supplier_(scan_filter));
115-
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FileStoreScan::RawPlan> plan,
116-
scan->WithSnapshot(snapshot)->WithKind(ScanMode::ALL)->CreatePlan());
117-
return plan->Files();
121+
if (for_overwrite && core_options_.GetBucket() != BucketModeDefine::POSTPONE_BUCKET) {
122+
scan->OnlyReadRealBuckets();
123+
}
124+
return scan;
118125
}
119126

120127
Result<std::vector<IndexManifestEntry>> CommitScanner::ReadAllIndexEntriesFromPartitions(
@@ -165,8 +172,14 @@ std::shared_ptr<CommitChangesProvider> CommitScanner::OverwriteChangesProvider(
165172
const std::vector<IndexManifestEntry>& index_entries) const {
166173
return std::make_shared<paimon::OverwriteChangesProvider>(
167174
changes, index_entries,
168-
[this, partitions](const Snapshot& snapshot) {
169-
return ReadAllEntriesFromPartitions(snapshot, partitions);
175+
[this, partitions](const Snapshot& snapshot) -> Result<std::vector<ManifestEntry>> {
176+
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileStoreScan> scan,
177+
NewScan(partitions, /*for_overwrite=*/true));
178+
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FileStoreScan::RawPlan> plan,
179+
scan->WithSnapshot(snapshot)
180+
->WithKind(ScanMode::ALL)
181+
->CreatePlan());
182+
return plan->Files();
170183
},
171184
[this, partitions](const Snapshot& snapshot) {
172185
return ReadAllIndexEntriesFromPartitions(snapshot, partitions);

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

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -90,6 +90,10 @@ class CommitScanner {
9090
Result<std::vector<std::map<std::string, std::string>>> ToPartitionFilters(
9191
const std::vector<BinaryRow>& changed_partitions) const;
9292

93+
Result<std::unique_ptr<FileStoreScan>> NewScan(
94+
const std::vector<std::map<std::string, std::string>>& partitions,
95+
bool for_overwrite) const;
96+
9397
private:
9498
std::shared_ptr<SnapshotManager> snapshot_manager_;
9599
std::shared_ptr<SchemaManager> schema_manager_;

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

Lines changed: 18 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -25,24 +25,23 @@
2525
namespace paimon {
2626

2727
OverwriteChangesProvider::OverwriteChangesProvider(
28-
const std::vector<ManifestEntry>& changes, const std::vector<IndexManifestEntry>& index_entries,
28+
std::vector<ManifestEntry> changes, std::vector<IndexManifestEntry> index_entries,
2929
ManifestScan manifest_scan, IndexScan index_scan)
30-
: changes_(changes),
31-
index_entries_(index_entries),
30+
: changes_(std::move(changes)),
31+
index_entries_(std::move(index_entries)),
3232
manifest_scan_(std::move(manifest_scan)),
3333
index_scan_(std::move(index_scan)) {}
3434

35-
Status OverwriteChangesProvider::Provide(const std::optional<Snapshot>& latest_snapshot,
36-
std::vector<ManifestEntry>* delta_files,
37-
std::vector<ManifestEntry>* changelog_files,
38-
std::vector<IndexManifestEntry>* index_entries) const {
39-
changelog_files->clear();
40-
*index_entries = index_entries_;
41-
delta_files->clear();
35+
Result<std::shared_ptr<CommitChanges>> OverwriteChangesProvider::Provide(
36+
const std::optional<Snapshot>& latest_snapshot) const {
37+
std::vector<ManifestEntry> delta_files;
38+
std::vector<ManifestEntry> changelog_files;
39+
std::vector<IndexManifestEntry> index_entries = index_entries_;
4240

4341
if (!latest_snapshot) {
44-
delta_files->insert(delta_files->end(), changes_.begin(), changes_.end());
45-
return Status::OK();
42+
delta_files.insert(delta_files.end(), changes_.begin(), changes_.end());
43+
return std::make_shared<CommitChanges>(std::move(delta_files), std::move(changelog_files),
44+
std::move(index_entries));
4645
}
4746

4847
PAIMON_ASSIGN_OR_RAISE(std::vector<ManifestEntry> entries,
@@ -51,26 +50,27 @@ Status OverwriteChangesProvider::Provide(const std::optional<Snapshot>& latest_s
5150
existing_identifiers.reserve(entries.size());
5251
for (const auto& entry : entries) {
5352
existing_identifiers.insert(entry.CreateIdentifier());
54-
delta_files->emplace_back(FileKind::Delete(), entry.Partition(), entry.Bucket(),
55-
entry.TotalBuckets(), entry.File());
53+
delta_files.emplace_back(FileKind::Delete(), entry.Partition(), entry.Bucket(),
54+
entry.TotalBuckets(), entry.File());
5655
}
5756

5857
for (const auto& change : changes_) {
5958
if (change.Kind() == FileKind::Add() &&
6059
existing_identifiers.find(change.CreateIdentifier()) != existing_identifiers.end()) {
6160
continue;
6261
}
63-
delta_files->push_back(change);
62+
delta_files.push_back(change);
6463
}
6564

6665
PAIMON_ASSIGN_OR_RAISE(std::vector<IndexManifestEntry> previous_index_entries,
6766
index_scan_(latest_snapshot.value()));
6867
for (const auto& entry : previous_index_entries) {
69-
index_entries->emplace_back(FileKind::Delete(), entry.partition, entry.bucket,
70-
entry.index_file);
68+
index_entries.emplace_back(FileKind::Delete(), entry.partition, entry.bucket,
69+
entry.index_file);
7170
}
7271

73-
return Status::OK();
72+
return std::make_shared<CommitChanges>(std::move(delta_files), std::move(changelog_files),
73+
std::move(index_entries));
7474
}
7575

7676
} // namespace paimon

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

Lines changed: 6 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -30,18 +30,16 @@ class OverwriteChangesProvider final : public CommitChangesProvider {
3030
using IndexScan =
3131
std::function<Result<std::vector<IndexManifestEntry>>(const Snapshot& snapshot)>;
3232

33-
OverwriteChangesProvider(const std::vector<ManifestEntry>& changes,
34-
const std::vector<IndexManifestEntry>& index_entries,
33+
OverwriteChangesProvider(std::vector<ManifestEntry> changes,
34+
std::vector<IndexManifestEntry> index_entries,
3535
ManifestScan manifest_scan, IndexScan index_scan);
3636

37-
Status Provide(const std::optional<Snapshot>& latest_snapshot,
38-
std::vector<ManifestEntry>* delta_files,
39-
std::vector<ManifestEntry>* changelog_files,
40-
std::vector<IndexManifestEntry>* index_entries) const override;
37+
Result<std::shared_ptr<CommitChanges>> Provide(
38+
const std::optional<Snapshot>& latest_snapshot) const override;
4139

4240
private:
43-
const std::vector<ManifestEntry>& changes_;
44-
const std::vector<IndexManifestEntry>& index_entries_;
41+
std::vector<ManifestEntry> changes_;
42+
std::vector<IndexManifestEntry> index_entries_;
4543
ManifestScan manifest_scan_;
4644
IndexScan index_scan_;
4745
};

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

Lines changed: 13 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -122,11 +122,11 @@ TEST(OverwriteChangesProviderTest, TestProvideWithoutLatestSnapshotUsesChangesOn
122122
return Status::Invalid("should not call index_scan");
123123
});
124124

125-
std::vector<ManifestEntry> delta_files;
126-
std::vector<ManifestEntry> changelog_files;
127-
std::vector<IndexManifestEntry> provided_index_entries;
128-
ASSERT_OK(
129-
provider.Provide(std::nullopt, &delta_files, &changelog_files, &provided_index_entries));
125+
ASSERT_OK_AND_ASSIGN(std::shared_ptr<CommitChanges> provided, provider.Provide(std::nullopt));
126+
127+
const std::vector<ManifestEntry>& delta_files = provided->delta_files;
128+
const std::vector<ManifestEntry>& changelog_files = provided->changelog_files;
129+
const std::vector<IndexManifestEntry>& provided_index_entries = provided->index_entries;
130130

131131
ASSERT_EQ(0, manifest_scan_calls);
132132
ASSERT_EQ(0, index_scan_calls);
@@ -158,11 +158,12 @@ TEST(OverwriteChangesProviderTest, TestProvideWithLatestSnapshotAddsDeletesAndDe
158158
CreateIndexEntry("index-old", /*partition_value=*/1)};
159159
});
160160

161-
std::vector<ManifestEntry> delta_files;
162-
std::vector<ManifestEntry> changelog_files;
163-
std::vector<IndexManifestEntry> provided_index_entries;
164-
ASSERT_OK(provider.Provide(std::optional<Snapshot>(MakeSnapshot()), &delta_files,
165-
&changelog_files, &provided_index_entries));
161+
ASSERT_OK_AND_ASSIGN(std::shared_ptr<CommitChanges> provided,
162+
provider.Provide(std::optional<Snapshot>(MakeSnapshot())));
163+
164+
const std::vector<ManifestEntry>& delta_files = provided->delta_files;
165+
const std::vector<ManifestEntry>& changelog_files = provided->changelog_files;
166+
const std::vector<IndexManifestEntry>& provided_index_entries = provided->index_entries;
166167

167168
ASSERT_TRUE(changelog_files.empty());
168169

@@ -199,12 +200,8 @@ TEST(OverwriteChangesProviderTest, TestProvidePropagatesScanErrors) {
199200
return std::vector<IndexManifestEntry>{};
200201
});
201202

202-
std::vector<ManifestEntry> delta_files;
203-
std::vector<ManifestEntry> changelog_files;
204-
std::vector<IndexManifestEntry> provided_index_entries;
205203
ASSERT_NOK_WITH_MSG(
206-
provider_manifest_fail.Provide(std::optional<Snapshot>(MakeSnapshot()), &delta_files,
207-
&changelog_files, &provided_index_entries),
204+
provider_manifest_fail.Provide(std::optional<Snapshot>(MakeSnapshot())),
208205
"manifest scan failed");
209206

210207
OverwriteChangesProvider provider_index_fail(
@@ -217,8 +214,7 @@ TEST(OverwriteChangesProviderTest, TestProvidePropagatesScanErrors) {
217214
});
218215

219216
ASSERT_NOK_WITH_MSG(
220-
provider_index_fail.Provide(std::optional<Snapshot>(MakeSnapshot()), &delta_files,
221-
&changelog_files, &provided_index_entries),
217+
provider_index_fail.Provide(std::optional<Snapshot>(MakeSnapshot())),
222218
"index scan failed");
223219
}
224220

0 commit comments

Comments
 (0)