Skip to content

Commit c8ad743

Browse files
lszskye子懿
lszskye
authored and
子懿
committed
feat: update commit message to version 12 (#179)
* feat: update commit message to version 12 * add compatibility test * fix pre-commit
1 parent 32f095f commit c8ad743

42 files changed

Lines changed: 390 additions & 9 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎src/paimon/core/index/global_index_meta.cpp‎

Lines changed: 34 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -33,11 +33,20 @@ GlobalIndexMeta::GlobalIndexMeta(int64_t _row_range_start, int64_t _row_range_en
3333
int32_t _index_field_id,
3434
const std::optional<std::vector<int32_t>>& _extra_field_ids,
3535
const std::shared_ptr<Bytes>& _index_meta)
36+
: GlobalIndexMeta(_row_range_start, _row_range_end, _index_field_id, _extra_field_ids,
37+
_index_meta, nullptr) {}
38+
39+
GlobalIndexMeta::GlobalIndexMeta(int64_t _row_range_start, int64_t _row_range_end,
40+
int32_t _index_field_id,
41+
const std::optional<std::vector<int32_t>>& _extra_field_ids,
42+
const std::shared_ptr<Bytes>& _index_meta,
43+
const std::shared_ptr<Bytes>& _source_meta)
3644
: row_range_start(_row_range_start),
3745
row_range_end(_row_range_end),
3846
index_field_id(_index_field_id),
3947
extra_field_ids(_extra_field_ids),
40-
index_meta(_index_meta) {}
48+
index_meta(_index_meta),
49+
source_meta(_source_meta) {}
4150

4251
bool GlobalIndexMeta::operator==(const GlobalIndexMeta& other) const {
4352
if (this == &other) {
@@ -49,6 +58,12 @@ bool GlobalIndexMeta::operator==(const GlobalIndexMeta& other) const {
4958
if (index_meta && other.index_meta && !(*index_meta == *other.index_meta)) {
5059
return false;
5160
}
61+
if ((source_meta && !other.source_meta) || (!source_meta && other.source_meta)) {
62+
return false;
63+
}
64+
if (source_meta && other.source_meta && !(*source_meta == *other.source_meta)) {
65+
return false;
66+
}
5267
return row_range_start == other.row_range_start && row_range_end == other.row_range_end &&
5368
index_field_id == other.index_field_id && extra_field_ids == other.extra_field_ids;
5469
}
@@ -61,14 +76,17 @@ std::string GlobalIndexMeta::ToString() const {
6176

6277
std::string index_meta_str =
6378
index_meta == nullptr ? "null" : std::string(index_meta->data(), index_meta->size());
79+
std::string source_meta_str =
80+
source_meta == nullptr ? "null" : std::string(source_meta->data(), source_meta->size());
6481
return fmt::format(
6582
"{{row_range_start={}, row_range_end={}, index_field_id={}, extra_field_ids={}, "
66-
"index_meta={}}}",
67-
row_range_start, row_range_end, index_field_id, extra_field_ids_str, index_meta_str);
83+
"index_meta={}, source_meta={}}}",
84+
row_range_start, row_range_end, index_field_id, extra_field_ids_str, index_meta_str,
85+
source_meta_str);
6886
}
6987

7088
BinaryRow GlobalIndexMeta::ToRow(MemoryPool* pool) const {
71-
BinaryRow row(5);
89+
BinaryRow row(6);
7290
BinaryRowWriter writer(&row, 32 * 1024, pool);
7391
writer.WriteLong(0, row_range_start);
7492
writer.WriteLong(1, row_range_end);
@@ -83,6 +101,11 @@ BinaryRow GlobalIndexMeta::ToRow(MemoryPool* pool) const {
83101
} else {
84102
writer.WriteBinary(4, *index_meta);
85103
}
104+
if (source_meta == nullptr) {
105+
writer.SetNullAt(5);
106+
} else {
107+
writer.WriteBinary(5, *source_meta);
108+
}
86109
writer.Complete();
87110
return row;
88111
}
@@ -104,8 +127,13 @@ Result<GlobalIndexMeta> GlobalIndexMeta::FromRow(const InternalRow& row) {
104127
index_meta = row.GetBinary(4);
105128
assert(index_meta);
106129
}
130+
std::shared_ptr<Bytes> source_meta;
131+
if (!row.IsNullAt(5)) {
132+
source_meta = row.GetBinary(5);
133+
assert(source_meta);
134+
}
107135
return GlobalIndexMeta(row_range_start, row_range_end, index_field_id, extra_field_ids,
108-
index_meta);
136+
index_meta, source_meta);
109137
}
110138

111139
const std::shared_ptr<arrow::DataType>& GlobalIndexMeta::DataType() {
@@ -117,6 +145,7 @@ const std::shared_ptr<arrow::DataType>& GlobalIndexMeta::DataType() {
117145
arrow::list(arrow::field("item", arrow::int32(), /*nullable=*/false)),
118146
/*nullable=*/true),
119147
arrow::field("_INDEX_META", arrow::binary(), /*nullable=*/true),
148+
arrow::field("_SOURCE_META", arrow::binary(), /*nullable=*/true),
120149
});
121150
return schema;
122151
}

‎src/paimon/core/index/global_index_meta.h‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,12 +31,17 @@
3131
namespace paimon {
3232
/// Schema for global index.
3333
struct GlobalIndexMeta {
34-
static constexpr int32_t NUM_FIELDS = 5;
34+
static constexpr int32_t NUM_FIELDS = 6;
3535

3636
GlobalIndexMeta(int64_t _row_range_start, int64_t _row_range_end, int32_t _index_field_id,
3737
const std::optional<std::vector<int32_t>>& _extra_field_ids,
3838
const std::shared_ptr<Bytes>& _index_meta);
3939

40+
GlobalIndexMeta(int64_t _row_range_start, int64_t _row_range_end, int32_t _index_field_id,
41+
const std::optional<std::vector<int32_t>>& _extra_field_ids,
42+
const std::shared_ptr<Bytes>& _index_meta,
43+
const std::shared_ptr<Bytes>& _source_meta);
44+
4045
bool operator==(const GlobalIndexMeta& other) const;
4146

4247
std::string ToString() const;
@@ -52,6 +57,7 @@ struct GlobalIndexMeta {
5257
int32_t index_field_id;
5358
std::optional<std::vector<int32_t>> extra_field_ids;
5459
std::shared_ptr<Bytes> index_meta;
60+
std::shared_ptr<Bytes> source_meta;
5561
};
5662

5763
} // namespace paimon

‎src/paimon/core/index/index_file_meta_serializer_test.cpp‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -127,6 +127,37 @@ TEST_F(IndexFileMetaSerializerTest, TestToFromRowWithGlobalIndex) {
127127
}
128128
}
129129

130+
TEST_F(IndexFileMetaSerializerTest, TestToFromRowWithGlobalIndexWithSourceMeta) {
131+
auto index_meta_bytes = std::make_shared<Bytes>("apple", memory_pool_.get());
132+
auto source_meta_bytes = std::make_shared<Bytes>("banana", memory_pool_.get());
133+
IndexFileMetaSerializer serializer(memory_pool_);
134+
GlobalIndexMeta global_index_meta(
135+
/*row_range_start=*/10, /*row_range_end=*/50,
136+
/*index_field_id=*/5, /*extra_field_ids=*/std::optional<std::vector<int32_t>>({0, 1}),
137+
index_meta_bytes, source_meta_bytes);
138+
{
139+
auto expected =
140+
std::make_shared<IndexFileMeta>("bitmap", "bitmap_index_file_0", /*file_size=*/10,
141+
/*row_count=*/41, /*dv_ranges=*/std::nullopt,
142+
/*external_path=*/std::nullopt, global_index_meta);
143+
ASSERT_OK_AND_ASSIGN(BinaryRow row, serializer.ToRow(expected));
144+
ASSERT_OK_AND_ASSIGN(std::shared_ptr<IndexFileMeta> actual, serializer.FromRow(row));
145+
ASSERT_EQ(expected->ToString(), actual->ToString());
146+
ASSERT_EQ(*expected, *actual);
147+
}
148+
{
149+
// test external path
150+
auto expected = std::make_shared<IndexFileMeta>(
151+
"bitmap", "bitmap_index_file_0", /*file_size=*/10,
152+
/*row_count=*/41, /*dv_ranges=*/std::nullopt,
153+
/*external_path=*/"FILE:/tmp/external/bitmap_index_file_0", global_index_meta);
154+
ASSERT_OK_AND_ASSIGN(BinaryRow row, serializer.ToRow(expected));
155+
ASSERT_OK_AND_ASSIGN(std::shared_ptr<IndexFileMeta> actual, serializer.FromRow(row));
156+
ASSERT_EQ(expected->ToString(), actual->ToString());
157+
ASSERT_EQ(*expected, *actual);
158+
}
159+
}
160+
130161
TEST_F(IndexFileMetaSerializerTest, TestSerialize) {
131162
IndexFileMetaSerializer serializer(memory_pool_);
132163
auto expected = GetRandomDeletionVectorIndexFile();

‎src/paimon/core/index/index_file_meta_v4_deserializer.h‎

Lines changed: 125 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,125 @@
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,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
#pragma once
21+
22+
#include <cassert>
23+
#include <cstdint>
24+
#include <list>
25+
#include <memory>
26+
#include <optional>
27+
#include <string>
28+
#include <utility>
29+
#include <vector>
30+
31+
#include "paimon/common/data/internal_array.h"
32+
#include "paimon/common/data/internal_row.h"
33+
#include "paimon/common/utils/linked_hash_map.h"
34+
#include "paimon/core/index/deletion_vector_meta.h"
35+
#include "paimon/core/index/index_file_meta.h"
36+
#include "paimon/core/index/index_file_meta_v2_deserializer.h"
37+
#include "paimon/core/utils/object_serializer.h"
38+
#include "paimon/result.h"
39+
40+
namespace paimon {
41+
class MemoryPool;
42+
43+
class IndexFileMetaV4Deserializer : public ObjectSerializer<std::shared_ptr<IndexFileMeta>> {
44+
public:
45+
static const std::shared_ptr<arrow::DataType>& GlobalIndexMetaDataType() {
46+
static std::shared_ptr<arrow::DataType> schema = arrow::struct_({
47+
arrow::field("_ROW_RANGE_START", arrow::int64(), /*nullable=*/false),
48+
arrow::field("_ROW_RANGE_END", arrow::int64(), /*nullable=*/false),
49+
arrow::field("_INDEX_FIELD_ID", arrow::int32(), /*nullable=*/false),
50+
arrow::field("_EXTRA_FIELD_IDS",
51+
arrow::list(arrow::field("item", arrow::int32(), /*nullable=*/false)),
52+
/*nullable=*/true),
53+
arrow::field("_INDEX_META", arrow::binary(), /*nullable=*/true),
54+
});
55+
return schema;
56+
}
57+
58+
static const std::shared_ptr<arrow::DataType>& DataType() {
59+
static std::shared_ptr<arrow::DataType> schema = arrow::struct_({
60+
arrow::field("_INDEX_TYPE", arrow::utf8(), false),
61+
arrow::field("_FILE_NAME", arrow::utf8(), false),
62+
arrow::field("_FILE_SIZE", arrow::int64(), false),
63+
arrow::field("_ROW_COUNT", arrow::int64(), false),
64+
arrow::field("_DELETIONS_VECTORS_RANGES",
65+
arrow::list(arrow::field("item", DeletionVectorMeta::DataType(), true)),
66+
true),
67+
arrow::field("_EXTERNAL_PATH", arrow::utf8(), true),
68+
arrow::field("_GLOBAL_INDEX", GlobalIndexMetaDataType(), true),
69+
});
70+
return schema;
71+
}
72+
73+
explicit IndexFileMetaV4Deserializer(const std::shared_ptr<MemoryPool>& pool)
74+
: ObjectSerializer<std::shared_ptr<IndexFileMeta>>(DataType(), pool) {}
75+
76+
Result<BinaryRow> ToRow(const std::shared_ptr<IndexFileMeta>& meta) const override {
77+
assert(false);
78+
return Status::Invalid("IndexFileMetaV4Deserializer to row is not valid");
79+
}
80+
81+
Result<std::shared_ptr<IndexFileMeta>> FromRow(const InternalRow& row) const override {
82+
auto file_type = row.GetString(0);
83+
auto file_name = row.GetString(1);
84+
auto file_size = row.GetLong(2);
85+
auto row_count = row.GetLong(3);
86+
std::optional<LinkedHashMap<std::string, DeletionVectorMeta>> dv_ranges;
87+
if (!row.IsNullAt(4)) {
88+
dv_ranges = IndexFileMetaV2Deserializer::RowArrayDataToDvRanges(row.GetArray(4).get());
89+
}
90+
std::optional<std::string> external_path;
91+
if (!row.IsNullAt(5)) {
92+
external_path = row.GetString(5).ToString();
93+
}
94+
std::optional<GlobalIndexMeta> global_index_meta;
95+
if (!row.IsNullAt(6)) {
96+
std::shared_ptr<InternalRow> global_index_meta_row =
97+
row.GetRow(6, GlobalIndexMetaDataType()->num_fields());
98+
assert(global_index_meta_row);
99+
int64_t row_range_start = global_index_meta_row->GetLong(0);
100+
int64_t row_range_end = global_index_meta_row->GetLong(1);
101+
int32_t index_field_id = global_index_meta_row->GetInt(2);
102+
std::optional<std::vector<int32_t>> extra_field_ids;
103+
if (!global_index_meta_row->IsNullAt(3)) {
104+
std::shared_ptr<InternalArray> array = global_index_meta_row->GetArray(3);
105+
if (!array) {
106+
return Status::Invalid(
107+
"GlobalIndexMeta FromRow failed with nullptr extra field ids");
108+
}
109+
PAIMON_ASSIGN_OR_RAISE(extra_field_ids, array->ToIntArray());
110+
}
111+
std::shared_ptr<Bytes> index_meta;
112+
if (!global_index_meta_row->IsNullAt(4)) {
113+
index_meta = global_index_meta_row->GetBinary(4);
114+
assert(index_meta);
115+
}
116+
global_index_meta = GlobalIndexMeta(row_range_start, row_range_end, index_field_id,
117+
extra_field_ids, index_meta);
118+
}
119+
return std::make_shared<IndexFileMeta>(file_type.ToString(), file_name.ToString(),
120+
file_size, row_count, dv_ranges, external_path,
121+
global_index_meta);
122+
}
123+
};
124+
125+
} // namespace paimon

‎src/paimon/core/table/sink/commit_message_serializer.cpp‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@
3131
#include "paimon/core/index/index_file_meta_v1_deserializer.h"
3232
#include "paimon/core/index/index_file_meta_v2_deserializer.h"
3333
#include "paimon/core/index/index_file_meta_v3_deserializer.h"
34+
#include "paimon/core/index/index_file_meta_v4_deserializer.h"
3435
#include "paimon/core/io/compact_increment.h"
3536
#include "paimon/core/io/data_file_meta_09_serializer.h"
3637
#include "paimon/core/io/data_file_meta_10_serializer.h"
@@ -45,7 +46,7 @@
4546
namespace paimon {
4647
class MemoryPool;
4748

48-
const int32_t CommitMessageSerializer::CURRENT_VERSION = 11;
49+
const int32_t CommitMessageSerializer::CURRENT_VERSION = 12;
4950

5051
CommitMessageSerializer::CommitMessageSerializer(const std::shared_ptr<MemoryPool>& pool)
5152
: memory_pool_(pool),
@@ -203,6 +204,11 @@ Result<std::shared_ptr<CommitMessage>> CommitMessageSerializer::Deserialize(int3
203204
DataInputStream* in) {
204205
if (version == CURRENT_VERSION) {
205206
return Deserialize(version, data_file_serializer_.get(), index_entry_serializer_.get(), in);
207+
} else if (version == 11) {
208+
auto index_entry_v4_deserializer =
209+
std::make_unique<IndexFileMetaV4Deserializer>(memory_pool_);
210+
return Deserialize(version, data_file_serializer_.get(), index_entry_v4_deserializer.get(),
211+
in);
206212
} else if (version == 9 || version == 10) {
207213
auto index_entry_v3_deserializer =
208214
std::make_unique<IndexFileMetaV3Deserializer>(memory_pool_);

‎src/paimon/core/table/sink/commit_message_test.cpp‎

Lines changed: 33 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -65,6 +65,39 @@ TEST(CommitMessageTest, TestCurrentVersion) {
6565
ASSERT_EQ(CommitMessageSerializer::CURRENT_VERSION, CommitMessage::CurrentVersion());
6666
}
6767

68+
TEST(CommitMessageTest, TestCompatibleWithVersion12) {
69+
// index file meta: add global index meta source meta
70+
int32_t version = 12;
71+
std::string data_path = paimon::test::GetDataDir() +
72+
"orc/pk_btree_source_meta.db/pk_btree_source_meta/"
73+
"commit_messages/commit_messages-01";
74+
auto file_system = std::make_shared<LocalFileSystem>();
75+
auto buffer_length = file_system->GetFileStatus(data_path).value()->GetLen();
76+
77+
std::vector<char> buffer(buffer_length, 0);
78+
ASSERT_OK_AND_ASSIGN(auto in_stream, file_system->Open(data_path));
79+
ASSERT_OK(in_stream->Read(reinterpret_cast<char*>(buffer.data()), buffer.size()));
80+
ASSERT_OK(in_stream->Close());
81+
82+
auto pool = GetDefaultPool();
83+
ASSERT_OK_AND_ASSIGN(std::shared_ptr<CommitMessage> ret,
84+
CommitMessage::Deserialize(version, reinterpret_cast<char*>(buffer.data()),
85+
buffer.size(), pool));
86+
auto res_msg = std::dynamic_pointer_cast<CommitMessageImpl>(ret);
87+
ASSERT_NE(res_msg, nullptr);
88+
89+
// check source_meta exists
90+
const auto& new_indexes = res_msg->GetCompactIncrement().NewIndexFiles();
91+
ASSERT_EQ(new_indexes.size(), 1);
92+
const auto& global_index_meta = new_indexes[0]->GetGlobalIndexMeta();
93+
ASSERT_TRUE(global_index_meta.has_value());
94+
ASSERT_NE(global_index_meta->source_meta, nullptr);
95+
96+
// check result
97+
ASSERT_OK_AND_ASSIGN(std::string serialized_bytes, CommitMessage::Serialize(ret, pool));
98+
ASSERT_EQ(serialized_bytes, std::string(reinterpret_cast<char*>(buffer.data()), buffer.size()));
99+
}
100+
68101
TEST(CommitMessageTest, TestCompatibleWithVersion11) {
69102
// index file meta: add global index meta
70103
int32_t version = 11;
@@ -106,8 +139,6 @@ TEST(CommitMessageTest, TestCompatibleWithVersion11) {
106139

107140
// check result
108141
ASSERT_EQ(res_msgs, expected_msgs);
109-
ASSERT_OK_AND_ASSIGN(std::string serialized_bytes, CommitMessage::SerializeList(ret, pool));
110-
ASSERT_EQ(serialized_bytes, std::string(reinterpret_cast<char*>(buffer.data()), buffer.size()));
111142
}
112143

113144
TEST(CommitMessageTest, TestCompatibleWithVersion10) {

‎test/test_data/orc/pk_btree_source_meta.db/pk_btree_source_meta/README‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
id:int score:int tag:string
2+
primary key: id
3+
no partition key
4+
no bucket key
5+
bucket count: 1
6+
7+
file format: orc
8+
manifest format: orc
9+
deletion-vectors.enabled: true
10+
primary-key btree index: score
11+
compaction.force-rewrite-all-files: true
12+
13+
Msgs:
14+
snapshot-1: APPEND
15+
Add: 1, 10, first
16+
Add: 2, 20, second
17+
18+
snapshot-2: COMPACT
19+
Build the primary-key btree index for score.
20+
21+
snapshot-3: APPEND
22+
Add: 3, 30, third
23+
Add: 4, 40, fourth
24+
25+
snapshot-4: COMPACT
26+
Rebuild the primary-key btree index for score.
27+
28+
snapshot-5: COMPACT
29+
Force a full compaction and rebuild the primary-key btree index for score.
30+
31+
commit_messages-01:
32+
Single CommitMessage serialized by Java CommitMessageSerializer version 12 for snapshot-5.

‎test/test_data/orc/pk_btree_source_meta.db/pk_btree_source_meta/bucket-0/data-022e32ce-1d75-4b9a-8496-a3311c44bad0-0.orc‎

694 Bytes
Binary file not shown.

‎test/test_data/orc/pk_btree_source_meta.db/pk_btree_source_meta/bucket-0/data-293d6705-bd19-4986-9aa0-3a24443e7e8e-0.orc‎

722 Bytes
Binary file not shown.

‎test/test_data/orc/pk_btree_source_meta.db/pk_btree_source_meta/bucket-0/data-b4949ff6-78f4-4e58-86ad-33eb3e5e6749-0.orc‎

694 Bytes
Binary file not shown.

0 commit comments

Comments
 (0)