Skip to content

Commit 0a31601

Browse files
authored
feat: add compact and postpone bucket writer utilities (#114)
1 parent 805b01e commit 0a31601

15 files changed

Lines changed: 2292 additions & 0 deletions
Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,43 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
19+
#pragma once
20+
21+
#include <atomic>
22+
23+
namespace paimon {
24+
25+
class CancellationController {
26+
public:
27+
void Cancel() {
28+
cancelled_.store(true, std::memory_order_relaxed);
29+
}
30+
31+
void Reset() {
32+
cancelled_.store(false, std::memory_order_relaxed);
33+
}
34+
35+
bool IsCancelled() const {
36+
return cancelled_.load(std::memory_order_relaxed);
37+
}
38+
39+
private:
40+
std::atomic_bool cancelled_{false};
41+
};
42+
43+
} // namespace paimon
Lines changed: 142 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,142 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
19+
#pragma once
20+
21+
#include <memory>
22+
#include <optional>
23+
24+
#include "paimon/core/deletionvectors/bucketed_dv_maintainer.h"
25+
#include "paimon/core/deletionvectors/deletion_vectors_index_file.h"
26+
#include "paimon/core/index/index_file_meta.h"
27+
#include "paimon/result.h"
28+
29+
namespace paimon {
30+
31+
/// Deletion File from compaction.
32+
class CompactDeletionFile {
33+
public:
34+
virtual ~CompactDeletionFile() = default;
35+
36+
/// Used by async compaction, when compaction task is completed, deletions file will be
37+
/// generated immediately, so when updateCompactResult, we need to merge old deletion files
38+
/// (just delete them).
39+
static Result<std::shared_ptr<CompactDeletionFile>> GenerateFiles(
40+
const std::shared_ptr<BucketedDvMaintainer>& maintainer);
41+
42+
/// For sync compaction, only create deletion files when prepare commit.
43+
static Result<std::shared_ptr<CompactDeletionFile>> LazyGeneration(
44+
const std::shared_ptr<BucketedDvMaintainer>& maintainer);
45+
46+
virtual Result<std::optional<std::shared_ptr<IndexFileMeta>>> GetOrCompute() = 0;
47+
48+
virtual Result<std::shared_ptr<CompactDeletionFile>> MergeOldFile(
49+
const std::shared_ptr<CompactDeletionFile>& old) = 0;
50+
51+
virtual void Clean() = 0;
52+
};
53+
54+
/// A generated files implementation of `CompactDeletionFile`.
55+
class GeneratedDeletionFile : public CompactDeletionFile,
56+
public std::enable_shared_from_this<GeneratedDeletionFile> {
57+
public:
58+
GeneratedDeletionFile(const std::shared_ptr<IndexFileMeta>& deletion_file,
59+
const std::shared_ptr<DeletionVectorsIndexFile>& dv_index_file)
60+
: deletion_file_(deletion_file), dv_index_file_(dv_index_file) {}
61+
62+
Result<std::optional<std::shared_ptr<IndexFileMeta>>> GetOrCompute() override {
63+
get_invoked_ = true;
64+
return deletion_file_ ? std::optional<std::shared_ptr<IndexFileMeta>>(deletion_file_)
65+
: std::nullopt;
66+
}
67+
68+
Result<std::shared_ptr<CompactDeletionFile>> MergeOldFile(
69+
const std::shared_ptr<CompactDeletionFile>& old) override {
70+
auto derived = dynamic_cast<GeneratedDeletionFile*>(old.get());
71+
if (derived == nullptr) {
72+
return Status::Invalid("old should be a GeneratedDeletionFile, but it is not");
73+
}
74+
if (derived->get_invoked_) {
75+
return Status::Invalid("old should not be get, this is a bug.");
76+
}
77+
if (deletion_file_ == nullptr) {
78+
return old;
79+
}
80+
old->Clean();
81+
return shared_from_this();
82+
}
83+
84+
void Clean() override {
85+
if (deletion_file_ != nullptr) {
86+
dv_index_file_->Delete(deletion_file_);
87+
}
88+
}
89+
90+
private:
91+
std::shared_ptr<IndexFileMeta> deletion_file_;
92+
std::shared_ptr<DeletionVectorsIndexFile> dv_index_file_;
93+
bool get_invoked_ = false;
94+
};
95+
96+
/// A lazy generation implementation of `CompactDeletionFile`.
97+
class LazyCompactDeletionFile : public CompactDeletionFile,
98+
public std::enable_shared_from_this<LazyCompactDeletionFile> {
99+
public:
100+
explicit LazyCompactDeletionFile(const std::shared_ptr<BucketedDvMaintainer>& maintainer)
101+
: maintainer_(maintainer) {}
102+
103+
Result<std::optional<std::shared_ptr<IndexFileMeta>>> GetOrCompute() override {
104+
generated_ = true;
105+
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<CompactDeletionFile> generated,
106+
CompactDeletionFile::GenerateFiles(maintainer_));
107+
return generated->GetOrCompute();
108+
}
109+
110+
Result<std::shared_ptr<CompactDeletionFile>> MergeOldFile(
111+
const std::shared_ptr<CompactDeletionFile>& old) override {
112+
auto derived = dynamic_cast<LazyCompactDeletionFile*>(old.get());
113+
if (derived == nullptr) {
114+
return Status::Invalid("old should be a LazyCompactDeletionFile, but it is not");
115+
}
116+
if (derived->generated_) {
117+
return Status::Invalid("old should not be generated, this is a bug.");
118+
}
119+
return shared_from_this();
120+
}
121+
122+
void Clean() override {}
123+
124+
private:
125+
std::shared_ptr<BucketedDvMaintainer> maintainer_;
126+
bool generated_ = false;
127+
};
128+
129+
inline Result<std::shared_ptr<CompactDeletionFile>> CompactDeletionFile::GenerateFiles(
130+
const std::shared_ptr<BucketedDvMaintainer>& maintainer) {
131+
PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<IndexFileMeta>> file,
132+
maintainer->WriteDeletionVectorsIndex());
133+
return std::make_shared<GeneratedDeletionFile>(file.value_or(nullptr),
134+
maintainer->DvIndexFile());
135+
}
136+
137+
inline Result<std::shared_ptr<CompactDeletionFile>> CompactDeletionFile::LazyGeneration(
138+
const std::shared_ptr<BucketedDvMaintainer>& maintainer) {
139+
return std::make_shared<LazyCompactDeletionFile>(maintainer);
140+
}
141+
142+
} // namespace paimon

0 commit comments

Comments
 (0)