Skip to content

Commit 87ea98f

Browse files
lucasfanglucasfang
committed
feat(commit): support TruncateTable and Abort and add ut
1 parent a4c1fef commit 87ea98f

6 files changed

Lines changed: 823 additions & 0 deletions

File tree

‎include/paimon/file_store_commit.h‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -135,6 +135,20 @@ class PAIMON_EXPORT FileStoreCommit {
135135
virtual Status DropPartition(const std::vector<std::map<std::string, std::string>>& partitions,
136136
int64_t commit_identifier) = 0;
137137

138+
/// Truncate the whole table by overwriting all partitions with empty data. The generated
139+
/// snapshot has commit kind OVERWRITE.
140+
///
141+
/// @param commit_identifier An identifier for the commit operation.
142+
/// @return Status indicating the success or failure of the truncate operation.
143+
virtual Status TruncateTable(int64_t commit_identifier) = 0;
144+
145+
/// Abort an unsuccessful commit. The data and index files described by the given commit
146+
/// messages will be deleted on a best-effort basis (delete failures are ignored).
147+
///
148+
/// @param commit_messages A vector of commit messages whose files should be cleaned up.
149+
/// @return Status indicating the success or failure of the abort operation.
150+
virtual Status Abort(const std::vector<std::shared_ptr<CommitMessage>>& commit_messages) = 0;
151+
138152
/// Configure row-id conflict checking from a specific snapshot id.
139153
///
140154
/// If set to a snapshot id, commit conflict detection will additionally validate row-id

‎src/paimon/CMakeLists.txt‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -720,6 +720,7 @@ if(PAIMON_BUILD_TESTS)
720720
core/operation/commit/overwrite_changes_provider_test.cpp
721721
core/operation/commit/row_id_column_conflict_checker_test.cpp
722722
core/operation/commit/row_tracking_commit_utils_test.cpp
723+
core/operation/commit/sequence_snapshot_properties_test.cpp
723724
core/operation/commit/retry_waiter_test.cpp
724725
core/operation/key_value_file_store_write_test.cpp
725726
core/operation/internal_read_context_test.cpp

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

Lines changed: 176 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,176 @@
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/operation/commit/sequence_snapshot_properties.h"
18+
19+
#include <cstdint>
20+
#include <map>
21+
#include <optional>
22+
#include <string>
23+
#include <vector>
24+
25+
#include "gtest/gtest.h"
26+
#include "paimon/common/data/binary_row.h"
27+
#include "paimon/core/io/data_file_meta.h"
28+
#include "paimon/core/manifest/file_kind.h"
29+
#include "paimon/core/manifest/manifest_entry.h"
30+
#include "paimon/core/snapshot.h"
31+
#include "paimon/core/stats/simple_stats.h"
32+
#include "paimon/data/timestamp.h"
33+
#include "paimon/testing/utils/testharness.h"
34+
35+
namespace paimon::test {
36+
37+
class SequenceSnapshotPropertiesTest : public testing::Test {
38+
protected:
39+
Snapshot MakeSnapshot(
40+
const std::optional<std::map<std::string, std::string>>& properties) const {
41+
return Snapshot(
42+
/*id=*/1,
43+
/*schema_id=*/1,
44+
/*base_manifest_list=*/"base-manifest-list",
45+
/*base_manifest_list_size=*/std::nullopt,
46+
/*delta_manifest_list=*/"delta-manifest-list",
47+
/*delta_manifest_list_size=*/std::nullopt,
48+
/*changelog_manifest_list=*/std::nullopt,
49+
/*changelog_manifest_list_size=*/std::nullopt,
50+
/*index_manifest=*/std::nullopt,
51+
/*commit_user=*/"test-user",
52+
/*commit_identifier=*/1, Snapshot::CommitKind::Append(),
53+
/*time_millis=*/0,
54+
/*total_record_count=*/0,
55+
/*delta_record_count=*/0,
56+
/*changelog_record_count=*/std::nullopt,
57+
/*watermark=*/std::nullopt,
58+
/*statistics=*/std::nullopt, properties,
59+
/*next_row_id=*/std::nullopt);
60+
}
61+
62+
std::shared_ptr<DataFileMeta> CreateDataFileMeta(int64_t max_sequence_number) const {
63+
return std::make_shared<DataFileMeta>(
64+
"data-file", 1024, 8, DataFileMeta::EmptyMinKey(), DataFileMeta::EmptyMaxKey(),
65+
SimpleStats::EmptyStats(), SimpleStats::EmptyStats(), /*min_seq_no=*/0,
66+
/*max_seq_no=*/max_sequence_number,
67+
/*schema_id=*/1, /*level=*/0,
68+
/*extra_files=*/std::vector<std::optional<std::string>>(),
69+
/*creation_time=*/Timestamp(0, 0),
70+
/*delete_row_count=*/std::nullopt,
71+
/*embedded_index=*/nullptr, /*file_source=*/std::nullopt,
72+
/*external_path=*/std::nullopt,
73+
/*value_stats_cols=*/std::nullopt, /*first_row_id=*/std::nullopt,
74+
/*write_cols=*/std::nullopt);
75+
}
76+
77+
ManifestEntry CreateEntry(const FileKind& kind, int64_t max_sequence_number) const {
78+
return ManifestEntry(kind, BinaryRow(0), /*bucket=*/0, /*total_buckets=*/1,
79+
CreateDataFileMeta(max_sequence_number));
80+
}
81+
};
82+
83+
TEST_F(SequenceSnapshotPropertiesTest, MaxSequenceNumberEmptySnapshot) {
84+
ASSERT_OK_AND_ASSIGN(std::optional<int64_t> result,
85+
SequenceSnapshotProperties::MaxSequenceNumber(std::nullopt));
86+
EXPECT_FALSE(result.has_value());
87+
}
88+
89+
TEST_F(SequenceSnapshotPropertiesTest, MaxSequenceNumberSnapshotWithoutProperties) {
90+
ASSERT_OK_AND_ASSIGN(std::optional<int64_t> result,
91+
SequenceSnapshotProperties::MaxSequenceNumber(MakeSnapshot(std::nullopt)));
92+
EXPECT_FALSE(result.has_value());
93+
}
94+
95+
TEST_F(SequenceSnapshotPropertiesTest, MaxSequenceNumberKeyMissing) {
96+
std::map<std::string, std::string> properties{{"other-key", "42"}};
97+
ASSERT_OK_AND_ASSIGN(std::optional<int64_t> result,
98+
SequenceSnapshotProperties::MaxSequenceNumber(MakeSnapshot(properties)));
99+
EXPECT_FALSE(result.has_value());
100+
}
101+
102+
TEST_F(SequenceSnapshotPropertiesTest, MaxSequenceNumberValid) {
103+
std::map<std::string, std::string> properties{
104+
{SequenceSnapshotProperties::kMaxSequenceNumberKey, "123"}};
105+
ASSERT_OK_AND_ASSIGN(std::optional<int64_t> result,
106+
SequenceSnapshotProperties::MaxSequenceNumber(MakeSnapshot(properties)));
107+
ASSERT_TRUE(result.has_value());
108+
EXPECT_EQ(123, result.value());
109+
}
110+
111+
TEST_F(SequenceSnapshotPropertiesTest, MaxSequenceNumberTrailingCharacters) {
112+
std::map<std::string, std::string> properties{
113+
{SequenceSnapshotProperties::kMaxSequenceNumberKey, "123abc"}};
114+
ASSERT_NOK_WITH_MSG(SequenceSnapshotProperties::MaxSequenceNumber(MakeSnapshot(properties)),
115+
"trailing characters are not allowed");
116+
}
117+
118+
TEST_F(SequenceSnapshotPropertiesTest, MaxSequenceNumberNotANumber) {
119+
std::map<std::string, std::string> properties{
120+
{SequenceSnapshotProperties::kMaxSequenceNumberKey, "not-a-number"}};
121+
ASSERT_NOK_WITH_MSG(SequenceSnapshotProperties::MaxSequenceNumber(MakeSnapshot(properties)),
122+
"Invalid");
123+
}
124+
125+
TEST_F(SequenceSnapshotPropertiesTest, MaxSequenceNumberFromFilesEmpty) {
126+
EXPECT_FALSE(SequenceSnapshotProperties::MaxSequenceNumberFromFiles({}).has_value());
127+
}
128+
129+
TEST_F(SequenceSnapshotPropertiesTest, MaxSequenceNumberFromFilesOnlyDelete) {
130+
std::vector<ManifestEntry> files{CreateEntry(FileKind::Delete(), 100)};
131+
EXPECT_FALSE(SequenceSnapshotProperties::MaxSequenceNumberFromFiles(files).has_value());
132+
}
133+
134+
TEST_F(SequenceSnapshotPropertiesTest, MaxSequenceNumberFromFilesSkipsDelete) {
135+
std::vector<ManifestEntry> files{CreateEntry(FileKind::Add(), 10),
136+
CreateEntry(FileKind::Delete(), 999),
137+
CreateEntry(FileKind::Add(), 42)};
138+
std::optional<int64_t> result = SequenceSnapshotProperties::MaxSequenceNumberFromFiles(files);
139+
ASSERT_TRUE(result.has_value());
140+
EXPECT_EQ(42, result.value());
141+
}
142+
143+
TEST_F(SequenceSnapshotPropertiesTest, MergeMaxSequenceNumberNoInput) {
144+
std::map<std::string, std::string> properties{{"existing", "value"}};
145+
std::map<std::string, std::string> merged = SequenceSnapshotProperties::MergeMaxSequenceNumber(
146+
properties, /*latest_max_sequence_number=*/std::nullopt, /*delta_files=*/{});
147+
EXPECT_EQ(properties, merged);
148+
EXPECT_EQ(0u, merged.count(SequenceSnapshotProperties::kMaxSequenceNumberKey));
149+
}
150+
151+
TEST_F(SequenceSnapshotPropertiesTest, MergeMaxSequenceNumberLatestOnly) {
152+
std::map<std::string, std::string> merged = SequenceSnapshotProperties::MergeMaxSequenceNumber(
153+
/*properties=*/{}, /*latest_max_sequence_number=*/50, /*delta_files=*/{});
154+
EXPECT_EQ("50", merged.at(SequenceSnapshotProperties::kMaxSequenceNumberKey));
155+
}
156+
157+
TEST_F(SequenceSnapshotPropertiesTest, MergeMaxSequenceNumberDeltaOnly) {
158+
std::vector<ManifestEntry> delta_files{CreateEntry(FileKind::Add(), 77)};
159+
std::map<std::string, std::string> merged = SequenceSnapshotProperties::MergeMaxSequenceNumber(
160+
/*properties=*/{}, /*latest_max_sequence_number=*/std::nullopt, delta_files);
161+
EXPECT_EQ("77", merged.at(SequenceSnapshotProperties::kMaxSequenceNumberKey));
162+
}
163+
164+
TEST_F(SequenceSnapshotPropertiesTest, MergeMaxSequenceNumberTakesMaximum) {
165+
std::vector<ManifestEntry> delta_files{CreateEntry(FileKind::Add(), 30)};
166+
std::map<std::string, std::string> merged = SequenceSnapshotProperties::MergeMaxSequenceNumber(
167+
/*properties=*/{}, /*latest_max_sequence_number=*/90, delta_files);
168+
EXPECT_EQ("90", merged.at(SequenceSnapshotProperties::kMaxSequenceNumberKey));
169+
170+
std::vector<ManifestEntry> larger_delta{CreateEntry(FileKind::Add(), 150)};
171+
std::map<std::string, std::string> merged2 = SequenceSnapshotProperties::MergeMaxSequenceNumber(
172+
/*properties=*/{}, /*latest_max_sequence_number=*/90, larger_delta);
173+
EXPECT_EQ("150", merged2.at(SequenceSnapshotProperties::kMaxSequenceNumberKey));
174+
}
175+
176+
} // namespace paimon::test

‎src/paimon/core/operation/file_store_commit_impl.cpp‎

Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -202,6 +202,62 @@ Status FileStoreCommitImpl::DropPartition(
202202
return Status::OK();
203203
}
204204

205+
Status FileStoreCommitImpl::TruncateTable(int64_t commit_identifier) {
206+
// An empty partition list means "all partitions", so this overwrites the whole table with
207+
// no new files, effectively truncating it. Mirrors Java tryOverwritePartition(null, ...).
208+
PAIMON_ASSIGN_OR_RAISE([[maybe_unused]] int32_t attempt,
209+
TryOverwrite(/*partition=*/{}, /*changes=*/{}, /*index_entries=*/{},
210+
commit_identifier, std::nullopt, /*properties=*/{}));
211+
return Status::OK();
212+
}
213+
214+
Status FileStoreCommitImpl::Abort(
215+
const std::vector<std::shared_ptr<CommitMessage>>& commit_messages) {
216+
for (const auto& message : commit_messages) {
217+
auto* msg = dynamic_cast<CommitMessageImpl*>(message.get());
218+
if (msg == nullptr) {
219+
return Status::Invalid("fail to cast commit message to impl");
220+
}
221+
PAIMON_ASSIGN_OR_RAISE(
222+
std::shared_ptr<DataFilePathFactory> data_file_path_factory,
223+
path_factory_->CreateDataFilePathFactory(msg->Partition(), msg->Bucket()));
224+
PAIMON_ASSIGN_OR_RAISE(
225+
std::unique_ptr<IndexPathFactory> index_file_path_factory,
226+
path_factory_->CreateIndexFileFactory(msg->Partition(), msg->Bucket()));
227+
228+
const DataIncrement& new_files_increment = msg->GetNewFilesIncrement();
229+
const CompactIncrement& compact_increment = msg->GetCompactIncrement();
230+
231+
std::vector<std::shared_ptr<DataFileMeta>> data_files_to_delete;
232+
auto append_data_files =
233+
[&data_files_to_delete](const std::vector<std::shared_ptr<DataFileMeta>>& files) {
234+
data_files_to_delete.insert(data_files_to_delete.end(), files.begin(), files.end());
235+
};
236+
append_data_files(new_files_increment.NewFiles());
237+
append_data_files(new_files_increment.ChangelogFiles());
238+
append_data_files(compact_increment.CompactAfter());
239+
append_data_files(compact_increment.ChangelogFiles());
240+
for (const auto& file : data_files_to_delete) {
241+
// Best-effort cleanup: ignore delete failures, aligning with Java deleteQuietly.
242+
[[maybe_unused]] Status status =
243+
fs_->Delete(data_file_path_factory->ToPath(file), /*recursive=*/false);
244+
}
245+
246+
std::vector<std::shared_ptr<IndexFileMeta>> index_files_to_delete;
247+
auto append_index_files = [&index_files_to_delete](
248+
const std::vector<std::shared_ptr<IndexFileMeta>>& files) {
249+
index_files_to_delete.insert(index_files_to_delete.end(), files.begin(), files.end());
250+
};
251+
append_index_files(new_files_increment.NewIndexFiles());
252+
append_index_files(compact_increment.NewIndexFiles());
253+
for (const auto& file : index_files_to_delete) {
254+
[[maybe_unused]] Status status =
255+
fs_->Delete(index_file_path_factory->ToPath(file), /*recursive=*/false);
256+
}
257+
}
258+
return Status::OK();
259+
}
260+
205261
FileStoreCommit& FileStoreCommitImpl::RowIdCheckConflict(
206262
std::optional<int64_t> row_id_check_from_snapshot) {
207263
conflict_detection_.SetRowIdCheckFromSnapshot(row_id_check_from_snapshot);

‎src/paimon/core/operation/file_store_commit_impl.h‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -125,6 +125,10 @@ class FileStoreCommitImpl : public FileStoreCommit {
125125
Status DropPartition(const std::vector<std::map<std::string, std::string>>& partitions,
126126
int64_t commit_identifier) override;
127127

128+
Status TruncateTable(int64_t commit_identifier) override;
129+
130+
Status Abort(const std::vector<std::shared_ptr<CommitMessage>>& commit_messages) override;
131+
128132
FileStoreCommit& RowIdCheckConflict(std::optional<int64_t> row_id_check_from_snapshot) override;
129133

130134
std::shared_ptr<Metrics> GetCommitMetrics() const override {

0 commit comments

Comments
 (0)