Skip to content

Commit b2e5fba

Browse files
ZouxxyyZouxxyy
authored
feat(core): support per-column maximum sequence numbers (#247)
1 parent c73f008 commit b2e5fba

72 files changed

Lines changed: 817 additions & 324 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/CMakeLists.txt‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -275,6 +275,7 @@ set(PAIMON_CORE_SRCS
275275
core/io/data_file_meta_first_row_id_legacy_serializer.cpp
276276
core/io/data_file_meta.cpp
277277
core/io/data_file_meta_serializer.cpp
278+
core/io/data_file_meta_write_cols_legacy_serializer.cpp
278279
core/io/data_file_path_factory.cpp
279280
core/io/data_file_index_writer.cpp
280281
core/io/file_index_options.cpp

‎src/paimon/core/append/bucketed_append_compact_manager_test.cpp‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -115,7 +115,7 @@ class BucketedAppendCompactManagerTest : public testing::Test {
115115
/*embedded_index=*/nullptr, FileSource::Append(),
116116
/*value_stats_cols=*/std::nullopt, /*external_path=*/std::nullopt,
117117
/*first_row_id=*/std::nullopt,
118-
/*write_cols=*/std::nullopt);
118+
/*write_cols=*/std::nullopt, /*column_max_sequence_numbers=*/std::nullopt);
119119
}
120120

121121
std::shared_ptr<DataFileMeta> NewNamedFile(const std::string& file_name, int64_t file_size,
@@ -135,7 +135,7 @@ class BucketedAppendCompactManagerTest : public testing::Test {
135135
/*embedded_index=*/nullptr, FileSource::Append(),
136136
/*value_stats_cols=*/std::nullopt, /*external_path=*/std::nullopt,
137137
/*first_row_id=*/std::nullopt,
138-
/*write_cols=*/std::nullopt);
138+
/*write_cols=*/std::nullopt, /*column_max_sequence_numbers=*/std::nullopt);
139139
}
140140

141141
std::shared_ptr<BucketedDvMaintainer> CreateTestDvMaintainer(

‎src/paimon/core/deletionvectors/deletion_vector_test.cpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -54,7 +54,7 @@ std::shared_ptr<DataFileMeta> CreateDataFileMeta(const std::string& file_name) {
5454
/*min_sequence_number=*/0, /*max_sequence_number=*/0, /*schema_id=*/0,
5555
DataFileMeta::DUMMY_LEVEL, std::vector<std::optional<std::string>>{}, Timestamp(0, 0),
5656
std::nullopt, nullptr, FileSource::Append(), std::nullopt, std::nullopt, std::nullopt,
57-
std::nullopt);
57+
std::nullopt, /*column_max_sequence_numbers=*/std::nullopt);
5858
}
5959

6060
} // namespace

‎src/paimon/core/global_index/indexed_split_test.cpp‎

Lines changed: 24 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -56,17 +56,20 @@ TEST(IndexedSplitTest, TestSimple) {
5656
"file1.orc", 100l, 200l, BinaryRow::EmptyRow(), BinaryRow::EmptyRow(),
5757
SimpleStats::EmptyStats(), SimpleStats::EmptyStats(), 50l, 249l, 0, 0,
5858
std::vector<std::optional<std::string>>(), Timestamp(1765535214349l, 0), 0, nullptr,
59-
FileSource::Append(), std::nullopt, std::nullopt, 50l, std::nullopt);
59+
FileSource::Append(), std::nullopt, std::nullopt, 50l, std::nullopt,
60+
/*column_max_sequence_numbers=*/std::nullopt);
6061
auto meta2 = std::make_shared<DataFileMeta>(
6162
"file2.orc", 101l, 100l, BinaryRow::EmptyRow(), BinaryRow::EmptyRow(),
6263
SimpleStats::EmptyStats(), SimpleStats::EmptyStats(), 250l, 349l, 0, 0,
6364
std::vector<std::optional<std::string>>(), Timestamp(1765535214349l, 0), 0, nullptr,
64-
FileSource::Append(), std::nullopt, std::nullopt, 250l, std::nullopt);
65+
FileSource::Append(), std::nullopt, std::nullopt, 250l, std::nullopt,
66+
/*column_max_sequence_numbers=*/std::nullopt);
6567
auto meta3 = std::make_shared<DataFileMeta>(
6668
"file3.orc", 102l, 200l, BinaryRow::EmptyRow(), BinaryRow::EmptyRow(),
6769
SimpleStats::EmptyStats(), SimpleStats::EmptyStats(), 1000l, 1199l, 0, 0,
6870
std::vector<std::optional<std::string>>(), Timestamp(1765535214349l, 0), 0, nullptr,
69-
FileSource::Append(), std::nullopt, std::nullopt, 1000l, std::nullopt);
71+
FileSource::Append(), std::nullopt, std::nullopt, 1000l, std::nullopt,
72+
/*column_max_sequence_numbers=*/std::nullopt);
7073

7174
DataSplitImpl::Builder builder(
7275
/*partition=*/BinaryRow::EmptyRow(),
@@ -82,8 +85,11 @@ TEST(IndexedSplitTest, TestSimple) {
8285

8386
ASSERT_EQ(*result_indexed_split, *expected_indexed_split) << result_indexed_split->ToString();
8487
ASSERT_OK_AND_ASSIGN(std::string serialize_bytes, Split::Serialize(result_indexed_split, pool));
85-
ASSERT_EQ(serialize_bytes,
86-
std::string(reinterpret_cast<char*>(split_bytes.data()), split_bytes.size()));
88+
ASSERT_OK_AND_ASSIGN(std::shared_ptr<Split> roundtrip,
89+
Split::Deserialize(serialize_bytes.data(), serialize_bytes.size(), pool));
90+
auto roundtrip_indexed_split = std::dynamic_pointer_cast<IndexedSplitImpl>(roundtrip);
91+
ASSERT_EQ(*roundtrip_indexed_split, *expected_indexed_split)
92+
<< roundtrip_indexed_split->ToString();
8793
}
8894

8995
TEST(IndexedSplitTest, TestIndexedSplitWithScore) {
@@ -107,17 +113,20 @@ TEST(IndexedSplitTest, TestIndexedSplitWithScore) {
107113
"file1.orc", 100l, 200l, BinaryRow::EmptyRow(), BinaryRow::EmptyRow(),
108114
SimpleStats::EmptyStats(), SimpleStats::EmptyStats(), 50l, 249l, 0, 0,
109115
std::vector<std::optional<std::string>>(), Timestamp(1765549435648l, 0), 0, nullptr,
110-
FileSource::Append(), std::nullopt, std::nullopt, 50l, std::nullopt);
116+
FileSource::Append(), std::nullopt, std::nullopt, 50l, std::nullopt,
117+
/*column_max_sequence_numbers=*/std::nullopt);
111118
auto meta2 = std::make_shared<DataFileMeta>(
112119
"file2.orc", 101l, 100l, BinaryRow::EmptyRow(), BinaryRow::EmptyRow(),
113120
SimpleStats::EmptyStats(), SimpleStats::EmptyStats(), 250l, 349l, 0, 0,
114121
std::vector<std::optional<std::string>>(), Timestamp(1765549435649l, 0), 0, nullptr,
115-
FileSource::Append(), std::nullopt, std::nullopt, 250l, std::nullopt);
122+
FileSource::Append(), std::nullopt, std::nullopt, 250l, std::nullopt,
123+
/*column_max_sequence_numbers=*/std::nullopt);
116124
auto meta3 = std::make_shared<DataFileMeta>(
117125
"file3.orc", 102l, 200l, BinaryRow::EmptyRow(), BinaryRow::EmptyRow(),
118126
SimpleStats::EmptyStats(), SimpleStats::EmptyStats(), 1000l, 1199l, 0, 0,
119127
std::vector<std::optional<std::string>>(), Timestamp(1765549435649l, 0), 0, nullptr,
120-
FileSource::Append(), std::nullopt, std::nullopt, 1000l, std::nullopt);
128+
FileSource::Append(), std::nullopt, std::nullopt, 1000l, std::nullopt,
129+
/*column_max_sequence_numbers=*/std::nullopt);
121130

122131
DataSplitImpl::Builder builder(
123132
/*partition=*/BinaryRow::EmptyRow(),
@@ -139,16 +148,20 @@ TEST(IndexedSplitTest, TestIndexedSplitWithScore) {
139148
"rowRanges=[[55, 56],[270, 270],[1001, 1002]], scores=[1.01,2.1,-1.32,4.23,50.74]") !=
140149
std::string::npos);
141150
ASSERT_OK_AND_ASSIGN(std::string serialize_bytes, Split::Serialize(result_indexed_split, pool));
142-
ASSERT_EQ(serialize_bytes,
143-
std::string(reinterpret_cast<char*>(split_bytes.data()), split_bytes.size()));
151+
ASSERT_OK_AND_ASSIGN(std::shared_ptr<Split> roundtrip,
152+
Split::Deserialize(serialize_bytes.data(), serialize_bytes.size(), pool));
153+
auto roundtrip_indexed_split = std::dynamic_pointer_cast<IndexedSplitImpl>(roundtrip);
154+
ASSERT_EQ(*roundtrip_indexed_split, *expected_indexed_split)
155+
<< roundtrip_indexed_split->ToString();
144156
}
145157

146158
TEST(IndexedSplitTest, TestValidate) {
147159
auto meta = std::make_shared<DataFileMeta>(
148160
"file.orc", 1l, 200l, BinaryRow::EmptyRow(), BinaryRow::EmptyRow(),
149161
SimpleStats::EmptyStats(), SimpleStats::EmptyStats(), 1000l, 1199l, 0, 0,
150162
std::vector<std::optional<std::string>>(), Timestamp(0l, 0), 0, nullptr,
151-
FileSource::Append(), std::nullopt, std::nullopt, 1000l, std::nullopt);
163+
FileSource::Append(), std::nullopt, std::nullopt, 1000l, std::nullopt,
164+
/*column_max_sequence_numbers=*/std::nullopt);
152165

153166
DataSplitImpl::Builder builder(
154167
/*partition=*/BinaryRow::EmptyRow(),

‎src/paimon/core/index/pksorted/pk_sorted_bucket_index_state_test.cpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -53,7 +53,7 @@ class PkSortedBucketIndexStateTest : public ::testing::Test {
5353
/*creation_time=*/Timestamp(0, 0), /*delete_row_count=*/std::nullopt,
5454
/*embedded_index=*/nullptr, file_source, /*value_stats_cols=*/std::nullopt,
5555
/*external_path=*/std::nullopt, /*first_row_id=*/std::nullopt,
56-
/*write_cols=*/std::nullopt);
56+
/*write_cols=*/std::nullopt, /*column_max_sequence_numbers=*/std::nullopt);
5757
}
5858

5959
/// Builds a payload whose source metadata lists the given sources in the given order.

‎src/paimon/core/io/data_file_meta.cpp‎

Lines changed: 31 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -72,7 +72,8 @@ Result<std::shared_ptr<DataFileMeta>> DataFileMeta::ForAppend(
7272
file_name, file_size, row_count, EmptyMinKey(), EmptyMaxKey(), SimpleStats::EmptyStats(),
7373
row_stats, min_sequence_number, max_sequence_number, schema_id, DUMMY_LEVEL, extra_files,
7474
Timestamp(/*millisecond=*/local_micro / 1000, /*nano_of_millisecond=*/0), 0ll,
75-
embedded_index, file_source, value_stats_cols, external_path, first_row_id, write_cols);
75+
embedded_index, file_source, value_stats_cols, external_path, first_row_id, write_cols,
76+
/*column_max_sequence_numbers=*/std::nullopt);
7677
}
7778

7879
Result<std::shared_ptr<DataFileMeta>> DataFileMeta::Upgrade(int32_t new_level) const {
@@ -84,7 +85,7 @@ Result<std::shared_ptr<DataFileMeta>> DataFileMeta::Upgrade(int32_t new_level) c
8485
file_name, file_size, row_count, min_key, max_key, key_stats, value_stats,
8586
min_sequence_number, max_sequence_number, schema_id, new_level, extra_files, creation_time,
8687
delete_row_count, embedded_index, file_source, value_stats_cols, external_path,
87-
first_row_id, write_cols);
88+
first_row_id, write_cols, column_max_sequence_numbers);
8889
}
8990

9091
std::shared_ptr<DataFileMeta> DataFileMeta::CopyWithExtraFiles(
@@ -93,15 +94,24 @@ std::shared_ptr<DataFileMeta> DataFileMeta::CopyWithExtraFiles(
9394
file_name, file_size, row_count, min_key, max_key, key_stats, value_stats,
9495
min_sequence_number, max_sequence_number, schema_id, level, new_extra_files, creation_time,
9596
delete_row_count, embedded_index, file_source, value_stats_cols, external_path,
96-
first_row_id, write_cols);
97+
first_row_id, write_cols, column_max_sequence_numbers);
98+
}
99+
100+
std::shared_ptr<DataFileMeta> DataFileMeta::CopyWithColumnMaxSequenceNumbers(
101+
const std::optional<std::vector<int64_t>>& new_column_max_sequence_numbers) const {
102+
return std::make_shared<DataFileMeta>(
103+
file_name, file_size, row_count, min_key, max_key, key_stats, value_stats,
104+
min_sequence_number, max_sequence_number, schema_id, level, extra_files, creation_time,
105+
delete_row_count, embedded_index, file_source, value_stats_cols, external_path,
106+
first_row_id, write_cols, new_column_max_sequence_numbers);
97107
}
98108

99109
std::shared_ptr<DataFileMeta> DataFileMeta::CopyWithoutStats() const {
100110
return std::make_shared<DataFileMeta>(
101111
file_name, file_size, row_count, min_key, max_key, key_stats, SimpleStats::EmptyStats(),
102112
min_sequence_number, max_sequence_number, schema_id, level, extra_files, creation_time,
103113
delete_row_count, embedded_index, file_source, std::vector<std::string>(), external_path,
104-
first_row_id, write_cols);
114+
first_row_id, write_cols, column_max_sequence_numbers);
105115
}
106116

107117
DataFileMeta::DataFileMeta(
@@ -113,7 +123,8 @@ DataFileMeta::DataFileMeta(
113123
const std::shared_ptr<Bytes>& _embedded_index, const std::optional<FileSource>& _file_source,
114124
const std::optional<std::vector<std::string>>& _value_stats_cols,
115125
const std::optional<std::string>& _external_path, const std::optional<int64_t>& _first_row_id,
116-
const std::optional<std::vector<std::string>>& _write_cols)
126+
const std::optional<std::vector<std::string>>& _write_cols,
127+
const std::optional<std::vector<int64_t>>& _column_max_sequence_numbers)
117128
: file_name(_file_name),
118129
file_size(_file_size),
119130
row_count(_row_count),
@@ -133,7 +144,8 @@ DataFileMeta::DataFileMeta(
133144
value_stats_cols(_value_stats_cols),
134145
external_path(_external_path),
135146
first_row_id(_first_row_id),
136-
write_cols(_write_cols) {}
147+
write_cols(_write_cols),
148+
column_max_sequence_numbers(_column_max_sequence_numbers) {}
137149

138150
Result<std::string> DataFileMeta::FileFormat() const {
139151
size_t last_dot_index = file_name.find_last_of(".");
@@ -198,7 +210,8 @@ bool DataFileMeta::operator==(const DataFileMeta& other) const {
198210
creation_time == other.creation_time && delete_row_count == other.delete_row_count &&
199211
file_source == other.file_source && value_stats_cols == other.value_stats_cols &&
200212
external_path == other.external_path && first_row_id == other.first_row_id &&
201-
write_cols == other.write_cols;
213+
write_cols == other.write_cols &&
214+
column_max_sequence_numbers == other.column_max_sequence_numbers;
202215
}
203216

204217
bool DataFileMeta::operator!=(const DataFileMeta& other) const {
@@ -243,7 +256,8 @@ bool DataFileMeta::TEST_Equal(const DataFileMeta& other) const {
243256
level == other.level && delete_row_count == other.delete_row_count &&
244257
file_source == other.file_source && value_stats_cols == other.value_stats_cols &&
245258
compare_optional_ignore_name(external_path, other.external_path) &&
246-
first_row_id == other.first_row_id && write_cols == other.write_cols;
259+
first_row_id == other.first_row_id && write_cols == other.write_cols &&
260+
column_max_sequence_numbers == other.column_max_sequence_numbers;
247261
}
248262

249263
std::string DataFileMeta::ToString() const {
@@ -261,7 +275,8 @@ std::string DataFileMeta::ToString() const {
261275
"{}, "
262276
"keyStats: {}, valueStats: {}, minSequenceNumber: {}, maxSequenceNumber: {}, schemaId: "
263277
"{}, level: {}, extraFiles: {}, creationTime: {}, deleteRowCount: {}, fileSource: {}, "
264-
"valueStatsCols: {}, externalPath: {}, firstRowId: {}, writeCols: {}}}",
278+
"valueStatsCols: {}, externalPath: {}, firstRowId: {}, writeCols: {}, "
279+
"columnMaxSequenceNumbers: {}}}",
265280
file_name, file_size, row_count,
266281
embedded_index == nullptr ? "null"
267282
: std::string(embedded_index->data(), embedded_index->size()),
@@ -275,7 +290,10 @@ std::string DataFileMeta::ToString() const {
275290
: fmt::format("{}", fmt::join(value_stats_cols.value(), ", ")),
276291
external_path == std::nullopt ? "null" : external_path.value(),
277292
first_row_id == std::nullopt ? "null" : std::to_string(first_row_id.value()),
278-
write_cols == std::nullopt ? "null" : fmt::format("{}", write_cols.value()));
293+
write_cols == std::nullopt ? "null" : fmt::format("{}", write_cols.value()),
294+
column_max_sequence_numbers == std::nullopt
295+
? "null"
296+
: fmt::format("{}", column_max_sequence_numbers.value()));
279297
}
280298

281299
int64_t DataFileMeta::GetMaxSequenceNumber(
@@ -315,6 +333,9 @@ const std::shared_ptr<arrow::DataType>& DataFileMeta::DataType() {
315333
arrow::field("_FIRST_ROW_ID", arrow::int64(), /*nullable=*/true),
316334
arrow::field("_WRITE_COLS",
317335
arrow::list(arrow::field("item", arrow::utf8(), /*nullable=*/false)),
336+
/*nullable=*/true),
337+
arrow::field("_WRITE_COLS_SEQUENCES",
338+
arrow::list(arrow::field("item", arrow::int64(), /*nullable=*/false)),
318339
/*nullable=*/true)});
319340
return schema;
320341
}

‎src/paimon/core/io/data_file_meta.h‎

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -58,7 +58,8 @@ struct DataFileMeta {
5858
const std::optional<std::vector<std::string>>& _value_stats_cols,
5959
const std::optional<std::string>& _external_path,
6060
const std::optional<int64_t>& _first_row_id,
61-
const std::optional<std::vector<std::string>>& _write_cols);
61+
const std::optional<std::vector<std::string>>& _write_cols,
62+
const std::optional<std::vector<int64_t>>& _column_max_sequence_numbers);
6263

6364
static Result<std::shared_ptr<DataFileMeta>> ForAppend(
6465
const std::string& file_name, int64_t file_size, int64_t row_count,
@@ -83,6 +84,9 @@ struct DataFileMeta {
8384
std::shared_ptr<DataFileMeta> CopyWithExtraFiles(
8485
const std::vector<std::optional<std::string>>& new_extra_files) const;
8586

87+
std::shared_ptr<DataFileMeta> CopyWithColumnMaxSequenceNumbers(
88+
const std::optional<std::vector<int64_t>>& new_column_max_sequence_numbers) const;
89+
8690
/// Create a copy without value statistics. All other metadata is preserved.
8791
///
8892
/// @return A new metadata object with empty value statistics and value-stat columns.
@@ -167,5 +171,12 @@ struct DataFileMeta {
167171
std::optional<int64_t> first_row_id;
168172

169173
std::optional<std::vector<std::string>> write_cols;
174+
175+
/// Maximum sequence number per physical table field after data-evolution compaction.
176+
///
177+
/// Values follow the table-field order selected by `write_cols` when it is non-null (system
178+
/// fields are ignored), or the file schema field order otherwise. A null value means that only
179+
/// the file-level sequence range is available.
180+
std::optional<std::vector<int64_t>> column_max_sequence_numbers;
170181
};
171182
} // namespace paimon

‎src/paimon/core/io/data_file_meta_09_serializer.cpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -114,7 +114,7 @@ Result<std::shared_ptr<DataFileMeta>> DataFileMeta09Serializer::FromRow(
114114
embedded_file_index, file_source,
115115
/*value_stats_cols=*/std::nullopt, /*external_path=*/std::nullopt,
116116
/*first_row_id=*/std::nullopt,
117-
/*write_cols=*/std::nullopt);
117+
/*write_cols=*/std::nullopt, /*column_max_sequence_numbers=*/std::nullopt);
118118
}
119119

120120
} // namespace paimon

‎src/paimon/core/io/data_file_meta_10_serializer.cpp‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -123,7 +123,8 @@ Result<std::shared_ptr<DataFileMeta>> DataFileMeta10Serializer::FromRow(
123123
min_sequence_number, max_sequence_number, schema_id, level,
124124
InternalRowUtils::FromStringArrayData(extra_files.get()), creation_time, delete_row_count,
125125
embedded_file_index, file_source, std::optional<std::vector<std::string>>(value_stats_cols),
126-
/*external_path=*/std::nullopt, /*first_row_id=*/std::nullopt, /*write_cols=*/std::nullopt);
126+
/*external_path=*/std::nullopt, /*first_row_id=*/std::nullopt, /*write_cols=*/std::nullopt,
127+
/*column_max_sequence_numbers=*/std::nullopt);
127128
}
128129

129130
} // namespace paimon

‎src/paimon/core/io/data_file_meta_12_serializer.cpp‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -128,7 +128,8 @@ Result<std::shared_ptr<DataFileMeta>> DataFileMeta12Serializer::FromRow(
128128
min_sequence_number, max_sequence_number, schema_id, level,
129129
InternalRowUtils::FromStringArrayData(extra_files.get()), creation_time, delete_row_count,
130130
embedded_file_index, file_source, std::optional<std::vector<std::string>>(value_stats_cols),
131-
external_path, /*first_row_id=*/std::nullopt, /*write_cols=*/std::nullopt);
131+
external_path, /*first_row_id=*/std::nullopt, /*write_cols=*/std::nullopt,
132+
/*column_max_sequence_numbers=*/std::nullopt);
132133
}
133134

134135
} // namespace paimon

0 commit comments

Comments
 (0)