Skip to content

Commit 775bc8c

Browse files
jichen20210919jichen20210919
authored
feat:Support specify field ids for table read context (alibaba#75)
1 parent 4288002 commit 775bc8c

9 files changed

Lines changed: 232 additions & 46 deletions

File tree

‎include/paimon/read_context.h‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
#include "paimon/predicate/predicate.h"
2727
#include "paimon/result.h"
2828
#include "paimon/type_fwd.h"
29+
#include "paimon/utils/special_field_ids.h"
2930
#include "paimon/visibility.h"
3031

3132
namespace paimon {
@@ -43,6 +44,7 @@ class PAIMON_EXPORT ReadContext {
4344
public:
4445
ReadContext(const std::string& path, const std::string& branch,
4546
const std::vector<std::string>& read_schema,
47+
const std::vector<int32_t>& read_field_ids,
4648
const std::shared_ptr<Predicate>& predicate, bool enable_predicate_filter,
4749
bool enable_prefetch, uint32_t prefetch_batch_count,
4850
uint32_t prefetch_max_parallel_num, bool enable_multi_thread_row_to_batch,
@@ -74,6 +76,10 @@ class PAIMON_EXPORT ReadContext {
7476
return read_schema_;
7577
}
7678

79+
const std::vector<int32_t>& GetReadFieldIds() const {
80+
return read_field_ids_;
81+
}
82+
7783
const std::shared_ptr<Predicate>& GetPredicate() const {
7884
return predicate_;
7985
}
@@ -113,6 +119,7 @@ class PAIMON_EXPORT ReadContext {
113119
std::string path_;
114120
std::string branch_;
115121
std::vector<std::string> read_schema_;
122+
std::vector<int32_t> read_field_ids_;
116123
std::shared_ptr<Predicate> predicate_;
117124
bool enable_predicate_filter_;
118125
bool enable_prefetch_;
@@ -151,6 +158,19 @@ class PAIMON_EXPORT ReadContextBuilder {
151158
/// @note Currently supports top-level field selection. Future versions may support
152159
/// nested field selection using ArrowSchema for more granular projection
153160
ReadContextBuilder& SetReadSchema(const std::vector<std::string>& read_field_names);
161+
/// Set the schema fields to read from the table.
162+
///
163+
/// If not set, all fields from the table schema will be read. This is useful for
164+
/// projection pushdown to reduce I/O and improve performance by reading only
165+
/// the required columns.
166+
///
167+
/// @param read_field_ids Vector of field ids to read from the table.
168+
/// @return Reference to this builder for method chaining.
169+
/// @note Currently supports top-level field selection. Future versions may support
170+
/// nested field selection using ArrowSchema for more granular projection,
171+
/// If SetReadFieldIds() call and SetReadSchema() are natually are mutually
172+
/// exclusive. Calling both will ignore the read schema set by SetReadSchema().
173+
ReadContextBuilder& SetReadFieldIds(const std::vector<int32_t>& read_field_ids);
154174

155175
/// Set a configuration options map to set some option entries which are not defined in the
156176
/// table schema or whose values you want to overwrite.

‎include/paimon/utils/special_field_ids.h‎

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,42 @@
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+
#pragma once
18+
19+
#include <cstdint>
20+
#include <limits>
21+
22+
namespace paimon {
23+
24+
/// A utility class for accessing special field IDs used in metadata.
25+
class SpecialFieldIds {
26+
protected:
27+
/// System defined constant for field id boundary. Value: INT32_MAX - 10000
28+
static const int32_t CPP_FIELD_ID_END;
29+
30+
public:
31+
/// Special field ID reserved for sequence number. Value: INT32_MAX - 1
32+
static const int32_t SEQUENCE_NUMBER;
33+
/// Special field ID reserved for value kind. Value: INT32_MAX - 2
34+
static const int32_t VALUE_KIND;
35+
/// Special field ID reserved for row ID. Value: INT32_MAX - 5
36+
static const int32_t ROW_ID;
37+
38+
/// Special field ID reserved for index score. Value: CPP_FIELD_ID_END - 1
39+
static const int32_t INDEX_SCORE;
40+
};
41+
42+
} // namespace paimon

‎src/paimon/CMakeLists.txt‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -267,7 +267,8 @@ set(PAIMON_CORE_SRCS
267267
core/utils/manifest_meta_reader.cpp
268268
core/utils/partition_path_utils.cpp
269269
core/utils/primary_key_table_utils.cpp
270-
core/utils/snapshot_manager.cpp)
270+
core/utils/snapshot_manager.cpp
271+
core/utils/special_field_ids.cpp)
271272

272273
add_paimon_lib(paimon
273274
SOURCES

‎src/paimon/common/table/special_fields.h‎

Lines changed: 8 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -22,36 +22,35 @@
2222

2323
#include "arrow/type_fwd.h"
2424
#include "paimon/common/types/data_field.h"
25+
#include "paimon/utils/special_field_ids.h"
2526

2627
namespace paimon {
2728

2829
struct SpecialFields {
2930
static constexpr char KEY_FIELD_PREFIX[] = "_KEY_";
3031
static constexpr int32_t KEY_VALUE_SPECIAL_FIELD_COUNT = 2;
31-
static constexpr int32_t CPP_FIELD_ID_END = std::numeric_limits<int32_t>::max() - 10000;
3232

3333
static const DataField& SequenceNumber() {
34-
static const DataField data_field =
35-
DataField(std::numeric_limits<int32_t>::max() - 1,
36-
arrow::field("_SEQUENCE_NUMBER", arrow::int64()));
34+
static const DataField data_field = DataField(
35+
SpecialFieldIds::SEQUENCE_NUMBER, arrow::field("_SEQUENCE_NUMBER", arrow::int64()));
3736
return data_field;
3837
}
3938

4039
static const DataField& ValueKind() {
41-
static const DataField data_field = DataField(std::numeric_limits<int32_t>::max() - 2,
42-
arrow::field("_VALUE_KIND", arrow::int8()));
40+
static const DataField data_field =
41+
DataField(SpecialFieldIds::VALUE_KIND, arrow::field("_VALUE_KIND", arrow::int8()));
4342
return data_field;
4443
}
4544

4645
static const DataField& RowId() {
47-
static const DataField data_field = DataField(std::numeric_limits<int32_t>::max() - 5,
48-
arrow::field("_ROW_ID", arrow::int64()));
46+
static const DataField data_field =
47+
DataField(SpecialFieldIds::ROW_ID, arrow::field("_ROW_ID", arrow::int64()));
4948
return data_field;
5049
}
5150

5251
static const DataField& IndexScore() {
5352
static const DataField data_field =
54-
DataField(CPP_FIELD_ID_END - 1, arrow::field("_INDEX_SCORE", arrow::float32()));
53+
DataField(SpecialFieldIds::INDEX_SCORE, arrow::field("_INDEX_SCORE", arrow::float32()));
5554
return data_field;
5655
}
5756

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

Lines changed: 40 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -38,26 +38,48 @@ Result<std::unique_ptr<InternalReadContext>> InternalReadContext::Create(
3838
context->GetSpecificFileSystem()));
3939
// prepare read schema
4040
std::vector<DataField> read_data_fields;
41-
read_data_fields.reserve(context->GetReadSchema().size());
42-
for (const auto& name : context->GetReadSchema()) {
43-
// if enable row tracking or data evolution, check special fields
44-
if (core_options.RowTrackingEnabled() && name == SpecialFields::RowId().Name()) {
45-
read_data_fields.push_back(SpecialFields::RowId());
46-
continue;
41+
if (!context->GetReadFieldIds().empty()) {
42+
read_data_fields.reserve(context->GetReadFieldIds().size());
43+
for (const auto& field_id : context->GetReadFieldIds()) {
44+
// if enable row tracking or data evolution, check special fields
45+
if (core_options.RowTrackingEnabled() && field_id == SpecialFields::RowId().Id()) {
46+
read_data_fields.push_back(SpecialFields::RowId());
47+
continue;
48+
}
49+
if (core_options.RowTrackingEnabled() &&
50+
field_id == SpecialFields::SequenceNumber().Id()) {
51+
read_data_fields.push_back(SpecialFields::SequenceNumber());
52+
continue;
53+
}
54+
if (core_options.DataEvolutionEnabled() &&
55+
field_id == SpecialFields::IndexScore().Id()) {
56+
read_data_fields.push_back(SpecialFields::IndexScore());
57+
continue;
58+
}
59+
PAIMON_ASSIGN_OR_RAISE(DataField field, table_schema->GetField(field_id));
60+
read_data_fields.push_back(field);
4761
}
48-
if (core_options.RowTrackingEnabled() && name == SpecialFields::SequenceNumber().Name()) {
49-
read_data_fields.push_back(SpecialFields::SequenceNumber());
50-
continue;
62+
} else if (!context->GetReadSchema().empty()) {
63+
read_data_fields.reserve(context->GetReadSchema().size());
64+
for (const auto& name : context->GetReadSchema()) {
65+
// if enable row tracking or data evolution, check special fields
66+
if (core_options.RowTrackingEnabled() && name == SpecialFields::RowId().Name()) {
67+
read_data_fields.push_back(SpecialFields::RowId());
68+
continue;
69+
}
70+
if (core_options.RowTrackingEnabled() &&
71+
name == SpecialFields::SequenceNumber().Name()) {
72+
read_data_fields.push_back(SpecialFields::SequenceNumber());
73+
continue;
74+
}
75+
if (core_options.DataEvolutionEnabled() && name == SpecialFields::IndexScore().Name()) {
76+
read_data_fields.push_back(SpecialFields::IndexScore());
77+
continue;
78+
}
79+
PAIMON_ASSIGN_OR_RAISE(DataField field, table_schema->GetField(name));
80+
read_data_fields.push_back(field);
5181
}
52-
if (core_options.DataEvolutionEnabled() && name == SpecialFields::IndexScore().Name()) {
53-
read_data_fields.push_back(SpecialFields::IndexScore());
54-
continue;
55-
}
56-
PAIMON_ASSIGN_OR_RAISE(DataField field, table_schema->GetField(name));
57-
read_data_fields.push_back(field);
58-
}
59-
60-
if (read_data_fields.empty()) {
82+
} else {
6183
// if field names not set, read all fields
6284
read_data_fields = table_schema->Fields();
6385
}

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

Lines changed: 63 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -63,6 +63,41 @@ TEST(InternalReadContext, TestReadWithSpecifiedSchema) {
6363
ASSERT_TRUE(internal_context->GetReadSchema()->Equals(expected_schema));
6464
}
6565

66+
TEST(InternalReadContext, TestReadWithSpecifiedFieldId) {
67+
std::string path = paimon::test::GetDataDir() + "/orc/append_09.db/append_09";
68+
ReadContextBuilder context_builder(path);
69+
context_builder.SetReadFieldIds({3, 0});
70+
ASSERT_OK_AND_ASSIGN(auto read_context, context_builder.Finish());
71+
SchemaManager schema_manager(std::make_shared<LocalFileSystem>(), read_context->GetPath());
72+
ASSERT_OK_AND_ASSIGN(auto table_schema, schema_manager.ReadSchema(0));
73+
ASSERT_OK_AND_ASSIGN(auto internal_context,
74+
InternalReadContext::Create(std::move(read_context), table_schema,
75+
table_schema->Options()));
76+
std::vector<DataField> read_fields = {DataField(3, arrow::field("f3", arrow::float64())),
77+
DataField(0, arrow::field("f0", arrow::utf8()))};
78+
auto expected_schema = DataField::ConvertDataFieldsToArrowSchema(read_fields);
79+
ASSERT_TRUE(internal_context->GetReadSchema()->Equals(expected_schema));
80+
}
81+
82+
TEST(InternalReadContext, TestReadWithSpecifiedFieldIdAndSchema) {
83+
std::string path = paimon::test::GetDataDir() + "/orc/append_09.db/append_09";
84+
ReadContextBuilder context_builder(path);
85+
// read schema is specified, read fields in schema
86+
// will use field ids instead of field names.
87+
context_builder.SetReadSchema({"f0"});
88+
context_builder.SetReadFieldIds({3, 0});
89+
ASSERT_OK_AND_ASSIGN(auto read_context, context_builder.Finish());
90+
SchemaManager schema_manager(std::make_shared<LocalFileSystem>(), read_context->GetPath());
91+
ASSERT_OK_AND_ASSIGN(auto table_schema, schema_manager.ReadSchema(0));
92+
ASSERT_OK_AND_ASSIGN(auto internal_context,
93+
InternalReadContext::Create(std::move(read_context), table_schema,
94+
table_schema->Options()));
95+
std::vector<DataField> read_fields = {DataField(3, arrow::field("f3", arrow::float64())),
96+
DataField(0, arrow::field("f0", arrow::utf8()))};
97+
auto expected_schema = DataField::ConvertDataFieldsToArrowSchema(read_fields);
98+
ASSERT_TRUE(internal_context->GetReadSchema()->Equals(expected_schema));
99+
}
100+
66101
TEST(InternalReadContext, TestReadWithRowTrackingAndScoreFields) {
67102
{
68103
// test simple
@@ -111,4 +146,32 @@ TEST(InternalReadContext, TestReadWithRowTrackingAndScoreFields) {
111146
}
112147
}
113148

149+
TEST(InternalReadContext, TestReadWithFieldIdsAndSpecialFields) {
150+
{
151+
// test simple
152+
std::string path = paimon::test::GetDataDir() + "/orc/append_09.db/append_09";
153+
ReadContextBuilder context_builder(path);
154+
// here we use field ids instead of field names, and specify special ids for row id,
155+
// sequence number and index score.
156+
context_builder.SetReadFieldIds({3, 0, SpecialFieldIds::ROW_ID,
157+
SpecialFieldIds::SEQUENCE_NUMBER,
158+
SpecialFieldIds::INDEX_SCORE});
159+
ASSERT_OK_AND_ASSIGN(auto read_context, context_builder.Finish());
160+
SchemaManager schema_manager(std::make_shared<LocalFileSystem>(), read_context->GetPath());
161+
ASSERT_OK_AND_ASSIGN(auto table_schema, schema_manager.ReadSchema(0));
162+
auto new_options = table_schema->Options();
163+
new_options[Options::ROW_TRACKING_ENABLED] = "true";
164+
new_options[Options::DATA_EVOLUTION_ENABLED] = "true";
165+
ASSERT_OK_AND_ASSIGN(
166+
auto internal_context,
167+
InternalReadContext::Create(std::move(read_context), table_schema, new_options));
168+
std::vector<DataField> read_fields = {
169+
DataField(3, arrow::field("f3", arrow::float64())),
170+
DataField(0, arrow::field("f0", arrow::utf8())), SpecialFields::RowId(),
171+
SpecialFields::SequenceNumber(), SpecialFields::IndexScore()};
172+
auto expected_schema = DataField::ConvertDataFieldsToArrowSchema(read_fields);
173+
ASSERT_TRUE(internal_context->GetReadSchema()->Equals(expected_schema));
174+
}
175+
}
176+
114177
} // namespace paimon::test

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

Lines changed: 25 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -27,21 +27,20 @@
2727
namespace paimon {
2828
class Predicate;
2929

30-
ReadContext::ReadContext(const std::string& path, const std::string& branch,
31-
const std::vector<std::string>& read_schema,
32-
const std::shared_ptr<Predicate>& predicate, bool enable_predicate_filter,
33-
bool enable_prefetch, uint32_t prefetch_batch_count,
34-
uint32_t prefetch_max_parallel_num, bool enable_multi_thread_row_to_batch,
35-
uint32_t row_to_batch_thread_number,
36-
const std::optional<std::string>& table_schema,
37-
const std::shared_ptr<MemoryPool>& memory_pool,
38-
const std::shared_ptr<Executor>& executor,
39-
const std::shared_ptr<FileSystem>& specific_file_system,
40-
const std::map<std::string, std::string>& fs_scheme_to_identifier_map,
41-
const std::map<std::string, std::string>& options)
30+
ReadContext::ReadContext(
31+
const std::string& path, const std::string& branch, const std::vector<std::string>& read_schema,
32+
const std::vector<int32_t>& read_field_ids, const std::shared_ptr<Predicate>& predicate,
33+
bool enable_predicate_filter, bool enable_prefetch, uint32_t prefetch_batch_count,
34+
uint32_t prefetch_max_parallel_num, bool enable_multi_thread_row_to_batch,
35+
uint32_t row_to_batch_thread_number, const std::optional<std::string>& table_schema,
36+
const std::shared_ptr<MemoryPool>& memory_pool, const std::shared_ptr<Executor>& executor,
37+
const std::shared_ptr<FileSystem>& specific_file_system,
38+
const std::map<std::string, std::string>& fs_scheme_to_identifier_map,
39+
const std::map<std::string, std::string>& options)
4240
: path_(path),
4341
branch_(branch),
4442
read_schema_(read_schema),
43+
read_field_ids_(read_field_ids),
4544
predicate_(predicate),
4645
enable_predicate_filter_(enable_predicate_filter),
4746
enable_prefetch_(enable_prefetch),
@@ -64,6 +63,7 @@ class ReadContextBuilder::Impl {
6463
void Reset() {
6564
branch_ = BranchManager::DEFAULT_MAIN_BRANCH;
6665
read_field_names_.clear();
66+
read_field_ids_.clear();
6767
fs_scheme_to_identifier_map_.clear();
6868
options_.clear();
6969
predicate_.reset();
@@ -83,6 +83,7 @@ class ReadContextBuilder::Impl {
8383
std::string path_;
8484
std::string branch_ = BranchManager::DEFAULT_MAIN_BRANCH;
8585
std::vector<std::string> read_field_names_;
86+
std::vector<int32_t> read_field_ids_;
8687
std::map<std::string, std::string> fs_scheme_to_identifier_map_;
8788
std::map<std::string, std::string> options_;
8889
std::shared_ptr<Predicate> predicate_;
@@ -125,6 +126,12 @@ ReadContextBuilder& ReadContextBuilder::SetReadSchema(
125126
return *this;
126127
}
127128

129+
ReadContextBuilder& ReadContextBuilder::SetReadFieldIds(
130+
const std::vector<int32_t>& read_field_ids) {
131+
impl_->read_field_ids_ = read_field_ids;
132+
return *this;
133+
}
134+
128135
ReadContextBuilder& ReadContextBuilder::SetPredicate(const std::shared_ptr<Predicate>& predicate) {
129136
impl_->predicate_ = predicate;
130137
return *this;
@@ -216,12 +223,12 @@ Result<std::unique_ptr<ReadContext>> ReadContextBuilder::Finish() {
216223
return Status::Invalid("row to batch thread number should be greater than 0");
217224
}
218225
auto ctx = std::make_unique<ReadContext>(
219-
impl_->path_, impl_->branch_, impl_->read_field_names_, impl_->predicate_,
220-
impl_->enable_predicate_filter_, impl_->enable_prefetch_, impl_->prefetch_batch_count_,
221-
impl_->prefetch_max_parallel_num_, impl_->enable_multi_thread_row_to_batch_,
222-
impl_->row_to_batch_thread_number_, impl_->table_schema_, impl_->memory_pool_,
223-
impl_->executor_, impl_->specific_file_system_, impl_->fs_scheme_to_identifier_map_,
224-
impl_->options_);
226+
impl_->path_, impl_->branch_, impl_->read_field_names_, impl_->read_field_ids_,
227+
impl_->predicate_, impl_->enable_predicate_filter_, impl_->enable_prefetch_,
228+
impl_->prefetch_batch_count_, impl_->prefetch_max_parallel_num_,
229+
impl_->enable_multi_thread_row_to_batch_, impl_->row_to_batch_thread_number_,
230+
impl_->table_schema_, impl_->memory_pool_, impl_->executor_, impl_->specific_file_system_,
231+
impl_->fs_scheme_to_identifier_map_, impl_->options_);
225232
impl_->Reset();
226233
return ctx;
227234
}

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

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@ TEST(ReadContextTest, TestSimple) {
3333
ASSERT_TRUE(ctx->GetMemoryPool());
3434
ASSERT_TRUE(ctx->GetExecutor());
3535
ASSERT_TRUE(ctx->GetReadSchema().empty());
36+
ASSERT_TRUE(ctx->GetReadFieldIds().empty());
3637
ASSERT_TRUE(ctx->GetOptions().empty());
3738
ASSERT_FALSE(ctx->GetPredicate());
3839
ASSERT_FALSE(ctx->EnablePredicateFilter());
@@ -50,6 +51,7 @@ TEST(ReadContextTest, TestSetContent) {
5051
ReadContextBuilder builder("table_root_path");
5152
builder.AddOption("key", "value");
5253
builder.SetReadSchema({"f1", "f2"});
54+
builder.SetReadFieldIds({0, 1});
5355
auto predicate =
5456
PredicateBuilder::IsNull(/*field_index=*/0, /*field_name=*/"f1", FieldType::INT);
5557
builder.SetPredicate(predicate);
@@ -70,6 +72,7 @@ TEST(ReadContextTest, TestSetContent) {
7072
ASSERT_TRUE(ctx->GetMemoryPool());
7173
ASSERT_TRUE(ctx->GetExecutor());
7274
ASSERT_EQ(ctx->GetReadSchema(), std::vector<std::string>({"f1", "f2"}));
75+
ASSERT_EQ(ctx->GetReadFieldIds(), std::vector<int32_t>({0, 1}));
7376
ASSERT_EQ(*predicate, *(ctx->GetPredicate()));
7477
ASSERT_TRUE(ctx->EnablePredicateFilter());
7578
ASSERT_TRUE(ctx->EnablePrefetch());

0 commit comments

Comments
 (0)