Skip to content

Commit a22663a

Browse files
lucasfanglucasfang
authored
feat(manifest): add manifest entry, file metadata and serialization utilities (#86)
1 parent 8447abf commit a22663a

21 files changed

Lines changed: 1930 additions & 0 deletions

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

Lines changed: 160 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,160 @@
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+
#pragma once
19+
20+
#include <cstddef>
21+
#include <cstdint>
22+
#include <optional>
23+
#include <string>
24+
#include <variant>
25+
#include <vector>
26+
27+
#include "fmt/format.h"
28+
#include "paimon/common/data/binary_row.h"
29+
#include "paimon/common/utils/linked_hash_map.h"
30+
#include "paimon/common/utils/murmurhash_utils.h"
31+
#include "paimon/core/manifest/file_kind.h"
32+
#include "paimon/status.h"
33+
34+
namespace paimon {
35+
36+
/// Entry representing a file.
37+
class FileEntry {
38+
public:
39+
/// The same `Identifier` indicates that the `ManifestEntry` refers to the same data
40+
/// file.
41+
struct Identifier {
42+
Identifier(const BinaryRow& _partition, int32_t _bucket, int32_t _level,
43+
const std::string& _file_name, const std::optional<std::string>& _external_path)
44+
: partition(_partition),
45+
bucket(_bucket),
46+
level(_level),
47+
file_name(_file_name),
48+
external_path(_external_path) {}
49+
50+
bool operator==(const Identifier& other) const {
51+
return partition == other.partition && bucket == other.bucket && level == other.level &&
52+
file_name == other.file_name && external_path == other.external_path;
53+
}
54+
55+
size_t HashCode() const {
56+
if (hash_ == static_cast<size_t>(-1)) {
57+
hash_ = partition.HashCode();
58+
hash_ =
59+
MurmurHashUtils::HashUnsafeBytes(reinterpret_cast<const void*>(&bucket),
60+
/*offset=*/0, sizeof(bucket), /*seed=*/hash_);
61+
hash_ = MurmurHashUtils::HashUnsafeBytes(reinterpret_cast<const void*>(&level),
62+
/*offset=*/0, sizeof(level),
63+
/*seed=*/hash_);
64+
hash_ = MurmurHashUtils::HashUnsafeBytes(
65+
reinterpret_cast<const void*>(file_name.data()),
66+
/*offset=*/0, file_name.size(),
67+
/*seed=*/hash_);
68+
if (external_path) {
69+
hash_ = MurmurHashUtils::HashUnsafeBytes(
70+
reinterpret_cast<const void*>(external_path.value().data()),
71+
/*offset=*/0, external_path.value().size(),
72+
/*seed=*/hash_);
73+
}
74+
}
75+
return hash_;
76+
}
77+
78+
std::string ToString() const {
79+
std::string external_path_str =
80+
external_path == std::nullopt ? "null" : external_path.value();
81+
return fmt::format("{{{}, {}, {}, {}, {}}}", partition.ToString(), bucket, level,
82+
file_name, external_path_str);
83+
}
84+
85+
BinaryRow partition;
86+
int32_t bucket;
87+
int32_t level;
88+
std::string file_name;
89+
std::optional<std::string> external_path;
90+
91+
private:
92+
mutable size_t hash_ = -1;
93+
};
94+
95+
public:
96+
virtual ~FileEntry() = default;
97+
virtual const FileKind& Kind() const = 0;
98+
virtual const BinaryRow& Partition() const = 0;
99+
virtual int32_t Bucket() const = 0;
100+
virtual int32_t Level() const = 0;
101+
virtual const std::string& FileName() const = 0;
102+
virtual const std::optional<std::string>& ExternalPath() const = 0;
103+
virtual Identifier CreateIdentifier() const = 0;
104+
virtual const BinaryRow& MinKey() const = 0;
105+
virtual const BinaryRow& MaxKey() const = 0;
106+
107+
template <typename T>
108+
static Status MergeEntries(const std::vector<T>& unmerged_entries,
109+
std::vector<T>* merged_entries) {
110+
LinkedHashMap<Identifier, T> merged_map;
111+
PAIMON_RETURN_NOT_OK(MergeEntries(unmerged_entries, &merged_map));
112+
for (const auto& [identifier, entry] : merged_map) {
113+
merged_entries->emplace_back(entry);
114+
}
115+
return Status::OK();
116+
}
117+
118+
template <typename T>
119+
static Status MergeEntries(const std::vector<T>& unmerged_entries,
120+
LinkedHashMap<Identifier, T>* merged_map_ptr) {
121+
auto& merged_map = *merged_map_ptr;
122+
for (const auto& entry : unmerged_entries) {
123+
Identifier identifier = entry.CreateIdentifier();
124+
const auto& kind = entry.Kind();
125+
auto iter = merged_map.find(identifier);
126+
if (kind == FileKind::Add()) {
127+
if (iter != merged_map.end()) {
128+
return Status::Invalid(fmt::format(
129+
"Trying to add file {} which is already added.", identifier.ToString()));
130+
}
131+
merged_map.insert(identifier, entry);
132+
} else if (kind == FileKind::Delete()) {
133+
// each dataFile will only be added once and deleted once,
134+
// if we know that it is added before then both add and delete entry can be
135+
// removed because there won't be further operations on this file,
136+
// otherwise we have to keep the delete entry because the add entry must be
137+
// in the previous manifest files
138+
if (iter != merged_map.end()) {
139+
merged_map.erase(identifier);
140+
} else {
141+
merged_map.insert(identifier, entry);
142+
}
143+
} else {
144+
return Status::Invalid("Unknown value kind ",
145+
std::to_string(static_cast<int32_t>(kind.ToByteValue())));
146+
}
147+
}
148+
return Status::OK();
149+
}
150+
};
151+
} // namespace paimon
152+
153+
namespace std {
154+
template <>
155+
struct hash<paimon::FileEntry::Identifier> {
156+
size_t operator()(const paimon::FileEntry::Identifier& identifier) const {
157+
return identifier.HashCode();
158+
}
159+
};
160+
} // namespace std

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

Lines changed: 193 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,193 @@
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+
#include "paimon/core/manifest/file_entry.h"
19+
20+
#include <cassert>
21+
#include <list>
22+
#include <memory>
23+
#include <string>
24+
#include <vector>
25+
26+
#include "gtest/gtest.h"
27+
#include "paimon/common/data/binary_row_writer.h"
28+
#include "paimon/common/data/binary_string.h"
29+
#include "paimon/core/manifest/file_kind.h"
30+
#include "paimon/memory/memory_pool.h"
31+
#include "paimon/testing/utils/testharness.h"
32+
33+
namespace paimon::test {
34+
35+
class MockFileEntry : public FileEntry {
36+
public:
37+
MockFileEntry(FileKind kind, BinaryRow partition, int32_t bucket, int32_t level,
38+
const std::string& file_name,
39+
const std::optional<std::string>& external_path = std::nullopt)
40+
: kind_(kind),
41+
partition_(partition),
42+
bucket_(bucket),
43+
level_(level),
44+
file_name_(file_name),
45+
external_path_(external_path),
46+
min_key_(BinaryRow::EmptyRow()),
47+
max_key_(BinaryRow::EmptyRow()) {}
48+
49+
const FileKind& Kind() const override {
50+
return kind_;
51+
}
52+
const BinaryRow& Partition() const override {
53+
return partition_;
54+
}
55+
int32_t Bucket() const override {
56+
return bucket_;
57+
}
58+
int32_t Level() const override {
59+
return level_;
60+
}
61+
const std::string& FileName() const override {
62+
return file_name_;
63+
}
64+
const std::optional<std::string>& ExternalPath() const override {
65+
return external_path_;
66+
}
67+
Identifier CreateIdentifier() const override {
68+
return Identifier(partition_, bucket_, level_, file_name_, external_path_);
69+
}
70+
const BinaryRow& MinKey() const override {
71+
assert(false);
72+
return min_key_;
73+
}
74+
const BinaryRow& MaxKey() const override {
75+
assert(false);
76+
return max_key_;
77+
}
78+
79+
private:
80+
FileKind kind_;
81+
BinaryRow partition_;
82+
int32_t bucket_;
83+
int32_t level_;
84+
std::string file_name_;
85+
std::optional<std::string> external_path_;
86+
BinaryRow min_key_;
87+
BinaryRow max_key_;
88+
};
89+
90+
class FileEntryTest : public testing::Test {
91+
public:
92+
void SetUp() override {
93+
pool_ = GetDefaultPool();
94+
}
95+
96+
BinaryRow GetPartition(const std::string& part_str) {
97+
BinaryRow part(/*arity=*/1);
98+
BinaryRowWriter writer(&part, /*initial_size=*/20, pool_.get());
99+
writer.WriteString(0, BinaryString::FromString(part_str, pool_.get()));
100+
return part;
101+
}
102+
103+
private:
104+
std::shared_ptr<MemoryPool> pool_;
105+
};
106+
107+
TEST_F(FileEntryTest, TestMergeEntriesSimple) {
108+
std::vector<MockFileEntry> entries;
109+
entries.emplace_back(FileKind::Add(), GetPartition("0"), /*bucket=*/0, /*level=*/0, "file1");
110+
entries.emplace_back(FileKind::Add(), GetPartition("0"), /*bucket=*/0, /*level=*/0, "file2");
111+
entries.emplace_back(FileKind::Add(), GetPartition("0"), /*bucket=*/0, /*level=*/0, "file3");
112+
entries.emplace_back(FileKind::Delete(), GetPartition("0"), /*bucket=*/0, /*level=*/0, "file1");
113+
114+
std::vector<MockFileEntry> merged_entries;
115+
ASSERT_OK(FileEntry::MergeEntries(entries, &merged_entries));
116+
ASSERT_EQ(2u, merged_entries.size());
117+
ASSERT_EQ("file2", merged_entries[0].FileName());
118+
ASSERT_EQ("file3", merged_entries[1].FileName());
119+
}
120+
121+
TEST_F(FileEntryTest, TestMergeEntriesAddSameFile) {
122+
std::vector<MockFileEntry> entries;
123+
entries.emplace_back(FileKind::Add(), GetPartition("0"), /*bucket=*/0, /*level=*/0, "file1");
124+
entries.emplace_back(FileKind::Add(), GetPartition("0"), /*bucket=*/0, /*level=*/0, "file1");
125+
std::vector<MockFileEntry> merged_entries;
126+
ASSERT_NOK_WITH_MSG(FileEntry::MergeEntries(entries, &merged_entries),
127+
"which is already added.");
128+
}
129+
130+
TEST_F(FileEntryTest, TestMergeEntriesAddSameFileWithDiffPart) {
131+
std::vector<MockFileEntry> entries;
132+
entries.emplace_back(FileKind::Add(), GetPartition("0"), /*bucket=*/0, /*level=*/0, "file1");
133+
entries.emplace_back(FileKind::Add(), GetPartition("1"), /*bucket=*/0, /*level=*/0, "file1");
134+
std::vector<MockFileEntry> merged_entries;
135+
ASSERT_OK(FileEntry::MergeEntries(entries, &merged_entries));
136+
ASSERT_EQ(2u, merged_entries.size());
137+
ASSERT_EQ("file1", merged_entries[0].FileName());
138+
ASSERT_EQ("file1", merged_entries[1].FileName());
139+
}
140+
141+
TEST_F(FileEntryTest, TestMergeEntriesAddSameFileWithDiffBucket) {
142+
std::vector<MockFileEntry> entries;
143+
entries.emplace_back(FileKind::Add(), GetPartition("0"), /*bucket=*/0, /*level=*/0, "file1");
144+
entries.emplace_back(FileKind::Add(), GetPartition("0"), /*bucket=*/1, /*level=*/0, "file1");
145+
std::vector<MockFileEntry> merged_entries;
146+
ASSERT_OK(FileEntry::MergeEntries(entries, &merged_entries));
147+
ASSERT_EQ(2u, merged_entries.size());
148+
ASSERT_EQ("file1", merged_entries[0].FileName());
149+
ASSERT_EQ("file1", merged_entries[1].FileName());
150+
}
151+
152+
TEST_F(FileEntryTest, TestMergeEntriesAddSameFileWithDiffLevel) {
153+
std::vector<MockFileEntry> entries;
154+
entries.emplace_back(FileKind::Add(), GetPartition("0"), /*bucket=*/0, /*level=*/0, "file1");
155+
entries.emplace_back(FileKind::Add(), GetPartition("0"), /*bucket=*/0, /*level=*/1, "file1");
156+
std::vector<MockFileEntry> merged_entries;
157+
ASSERT_OK(FileEntry::MergeEntries(entries, &merged_entries));
158+
ASSERT_EQ(2u, merged_entries.size());
159+
ASSERT_EQ("file1", merged_entries[0].FileName());
160+
ASSERT_EQ("file1", merged_entries[1].FileName());
161+
}
162+
163+
TEST_F(FileEntryTest, TestMergeEntriesAddSameFileWithDiffExternalPath) {
164+
std::vector<MockFileEntry> entries;
165+
entries.emplace_back(FileKind::Add(), GetPartition("0"), /*bucket=*/0, /*level=*/0, "file1",
166+
"/tmp/external_path1");
167+
entries.emplace_back(FileKind::Add(), GetPartition("0"), /*bucket=*/0, /*level=*/0, "file1",
168+
"/tmp/external_path2");
169+
std::vector<MockFileEntry> merged_entries;
170+
ASSERT_OK(FileEntry::MergeEntries(entries, &merged_entries));
171+
ASSERT_EQ(2u, merged_entries.size());
172+
ASSERT_EQ("file1", merged_entries[0].FileName());
173+
ASSERT_EQ("file1", merged_entries[1].FileName());
174+
}
175+
176+
TEST_F(FileEntryTest, TestMergeEntriesAddAndDelete) {
177+
std::vector<MockFileEntry> entries;
178+
entries.emplace_back(FileKind::Add(), GetPartition("0"), /*bucket=*/0, /*level=*/0, "file1");
179+
entries.emplace_back(FileKind::Delete(), GetPartition("0"), /*bucket=*/0, /*level=*/0, "file1");
180+
std::vector<MockFileEntry> merged_entries;
181+
ASSERT_OK(FileEntry::MergeEntries(entries, &merged_entries));
182+
ASSERT_EQ(0u, merged_entries.size());
183+
}
184+
185+
TEST_F(FileEntryTest, TestMergeEntriesDeleteAndAdd) {
186+
std::vector<MockFileEntry> entries;
187+
entries.emplace_back(FileKind::Delete(), GetPartition("0"), /*bucket=*/0, /*level=*/0, "file1");
188+
entries.emplace_back(FileKind::Add(), GetPartition("0"), /*bucket=*/0, /*level=*/0, "file1");
189+
std::vector<MockFileEntry> merged_entries;
190+
ASSERT_NOK(FileEntry::MergeEntries(entries, &merged_entries));
191+
}
192+
193+
} // namespace paimon::test

0 commit comments

Comments
 (0)