Skip to content

feat(realtime): support shredding-aware realtime reads & refactor rt read #269

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
lxy-9602 wants to merge 3 commits into
base: main
Choose a base branch
from
Open

feat(realtime): support shredding-aware realtime reads & refactor rt read #269

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
08b8709
refactor(realtime): move offset handling to Paimon readers
lxy-9602 Aug 31, 2026
b5dd7dc
support shredding
lxy-9602 Sep 1, 2026
997d4b1
rebase main
lxy-9602 Sep 1, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 18 additions & 30 deletions include/paimon/realtime/realtime_store.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,8 @@ enum class PAIMON_EXPORT RealtimeStoreMode {
/// Parameters used by a `RealtimeStoreFactory` to create a store.
struct PAIMON_EXPORT RealtimeStoreCreateRequest {
/// Schema whose ownership is transferred to the factory. Append mode receives the complete
/// table write schema. Primary-key mode receives the realtime primary-key transport schema:
/// append transport schema: [_REALTIME_OFFSET, table write fields]. Primary-key mode receives
/// the realtime primary-key transport schema:
/// [_VALUE_KIND, _SEQUENCE_NUMBER, _REALTIME_OFFSET, table write fields].
std::unique_ptr<::ArrowSchema> write_schema;
/// Table options available to the store implementation.
Expand All @@ -65,10 +66,10 @@ struct PAIMON_EXPORT RealtimeStoreCreateRequest {

/// A record batch and its framework-assigned contiguous offset range.
///
/// Append-mode batches contain table write fields, and row `i` has offset
/// `offset_range.begin + i`. Primary-key batches use the realtime primary-key transport schema,
/// are sorted by full primary key then sequence number, and retain the original offset in
/// `_REALTIME_OFFSET`.
/// Append-mode batches use the append transport schema [_REALTIME_OFFSET, table write fields], and
/// row `i` has offset `offset_range.begin + i`. Primary-key batches use the realtime primary-key
/// transport schema, are sorted by full primary key then sequence number, and retain the original
/// offset in `_REALTIME_OFFSET`.
struct PAIMON_EXPORT RealtimeWriteBatch {
/// Input batch whose ownership is transferred to `RealtimeStore::Write`.
std::unique_ptr<RecordBatch> batch;
Expand Down Expand Up @@ -103,20 +104,15 @@ class PAIMON_EXPORT RealtimeReadView {

/// Parameters used by a `RealtimeStore` to create readers for a query.
struct PAIMON_EXPORT RealtimeQueryContext {
/// Append mode receives the requested output fields before the mandatory leading
/// `_VALUE_KIND` field is added. Primary-key mode receives the requested realtime primary-key
/// transport schema.
/// Physical source schema the store must materialize. Query readers must include the mandatory
/// `_VALUE_KIND` field in returned batches. Paimon may subsequently convert physical fields
/// into the query's logical output schema, for example for selected-key MAP or VARIANT access.
/// This schema is borrowed and remains valid only during `CreateQueryReaders`; plugins must
/// import or copy it synchronously.
::ArrowSchema* read_schema;
/// Predicate using field indexes from `read_schema`.
/// Optional predicate using field indexes from `read_schema`. A non-null predicate allows the
/// plugin to prune candidate rows. Exact filtering is applied by the Paimon read framework.
std::shared_ptr<Predicate> predicate;
/// Whether the plugin may use `predicate` to prune candidate rows.
///
/// Keep this disabled for primary-key merge-on-read. Pruning memory before PK merge may remove
/// the newest row and incorrectly expose an older disk row. Exact predicate filtering, when
/// requested, is applied by the Paimon read framework after plugin reader creation.
bool enable_predicate_pushdown;
};

/// Customizable plugin interface for storing and querying real-time rows before Paimon data-file
Expand Down Expand Up @@ -145,9 +141,9 @@ class PAIMON_EXPORT RealtimeStore {
/// Creates readers that expose all rows in a sealed segment for Paimon file writing.
///
/// The returned readers collectively expose every sealed row exactly once. Append-mode readers
/// preserve write order and contain `_VALUE_KIND` followed by table write fields. Primary-key
/// readers use the realtime primary-key transport schema; each reader's complete stream is
/// sorted by full primary key then sequence number.
/// preserve write order and contain `_VALUE_KIND`, `_REALTIME_OFFSET`, and table write fields.
/// Primary-key readers contain the realtime primary-key transport fields; each reader's
/// complete stream is sorted by full primary key then sequence number.
virtual Result<std::vector<std::unique_ptr<BatchReader>>> CreateCommitReaders(
const std::shared_ptr<RealtimeSegmentHandle>& segment) = 0;

Expand All @@ -157,19 +153,11 @@ class PAIMON_EXPORT RealtimeStore {
/// also provide a consistent snapshot when a write or seal is in progress.
virtual Result<std::shared_ptr<RealtimeReadView>> AcquireReadView() = 0;

/// Creates readers over rows in `view`. Append mode returns rows whose offsets are greater than
/// or equal to `offset_begin`; primary-key mode ignores `offset_begin`.
///
/// Append-mode batches contain `_VALUE_KIND` followed by the requested fields except a
/// duplicate `_VALUE_KIND`, and collectively expose every matching row exactly once.
/// Primary-key batches use the requested realtime primary-key transport schema, including
/// nested field-ID alignment, and may contain multiple mutations per key; each reader's
/// complete stream is sorted by full primary key then sequence number, and the readers
/// collectively expose every raw mutation exactly once. Paimon retains `view` for the lifetime
/// of the resulting framework reader.
/// Creates readers over rows in `view`. The readers collectively expose every candidate row
/// exactly once. Primary-key reader streams are sorted by full primary key then sequence
/// number. Paimon retains `view` for the lifetime of the resulting framework reader.
virtual Result<std::vector<std::unique_ptr<BatchReader>>> CreateQueryReaders(
const std::shared_ptr<RealtimeReadView>& view, int64_t offset_begin,
const RealtimeQueryContext& context) = 0;
const std::shared_ptr<RealtimeReadView>& view, const RealtimeQueryContext& context) = 0;

/// Notifies the store that its partition-bucket committed end offset has advanced.
///
Expand Down
4 changes: 4 additions & 0 deletions src/paimon/CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -385,11 +385,13 @@ set(PAIMON_CORE_SRCS
core/operation/write_restore.cpp
core/realtime/arrow_realtime_store.cpp
core/realtime/arrow_realtime_store_factory.cpp
core/realtime/realtime_offset_batch_reader.cpp
core/realtime/realtime_primary_key_reader.cpp
core/realtime/primary_key_realtime_store.cpp
core/realtime/realtime_append_only_writer.cpp
core/realtime/realtime_context.cpp
core/realtime/realtime_context_impl.cpp
core/realtime/realtime_store_read_pipeline.cpp
core/realtime/realtime_primary_key_writer.cpp
core/postpone/postpone_bucket_writer.cpp
core/schema/arrow_schema_validator.cpp
Expand Down Expand Up @@ -795,6 +797,8 @@ if(PAIMON_BUILD_TESTS)
core/memory/writer_memory_manager_test.cpp
core/realtime/arrow_realtime_store_test.cpp
core/realtime/primary_key_realtime_store_test.cpp
core/realtime/realtime_offset_batch_reader_test.cpp
core/realtime/realtime_store_read_pipeline_test.cpp
core/realtime/realtime_primary_key_reader_test.cpp
core/realtime/realtime_context_test.cpp
core/realtime/realtime_reader_test.cpp
Expand Down
101 changes: 4 additions & 97 deletions src/paimon/core/io/field_mapping_reader.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,100 +45,6 @@
namespace paimon {
class MemoryPool;

Result<bool> FieldMappingReader::HasMapSelectedKeysRecursively(
const std::shared_ptr<arrow::Field>& read_field) const {
if (!read_field) {
return false;
}
auto type_id = read_field->type()->id();
if (NestedProjectionUtils::IsMapSharedShreddingAccessField(read_field)) {
PAIMON_ASSIGN_OR_RAISE(std::vector<std::string> selected_keys,
NestedProjectionUtils::GetMapSelectedKeys(read_field));
auto read_struct = checked_pointer_cast<arrow::StructType>(read_field->type());
if (selected_keys.size() != static_cast<size_t>(read_struct->num_fields())) {
return Status::Invalid(fmt::format(
"selected-key metadata size {} does not match STRUCT field count {} for {}",
selected_keys.size(), read_struct->num_fields(), read_field->name()));
}
return true;
}
if (type_id == arrow::Type::MAP) {
PAIMON_ASSIGN_OR_RAISE(std::vector<std::string> selected_keys,
NestedProjectionUtils::GetMapSelectedKeys(read_field));
return !selected_keys.empty();
}
if (type_id == arrow::Type::STRUCT) {
for (const auto& child : read_field->type()->fields()) {
PAIMON_ASSIGN_OR_RAISE(bool has_selected_keys, HasMapSelectedKeysRecursively(child));
if (has_selected_keys) {
return true;
}
}
}
return false;
}

Result<std::shared_ptr<arrow::Array>> FieldMappingReader::FilterMapSelectedKeysRecursively(
const std::shared_ptr<arrow::Array>& array,
const std::shared_ptr<arrow::Field>& read_field) const {
if (!array || !read_field) {
return array;
}

auto type_id = read_field->type()->id();
if (NestedProjectionUtils::IsMapSharedShreddingAccessField(read_field)) {
// The shared-shredding wrapper (including its default MAP fallback) has already
// materialized this projection as a STRUCT.
return array;
}
if (type_id == arrow::Type::MAP) {
PAIMON_ASSIGN_OR_RAISE(std::vector<std::string> selected_keys,
NestedProjectionUtils::GetMapSelectedKeys(read_field));
if (selected_keys.empty()) {
return array;
}
return NestedProjectionUtils::FilterMapArrayBySelectedKeys(array, selected_keys,
arrow_pool_.get());
}

if (type_id == arrow::Type::STRUCT) {
if (array->type_id() != arrow::Type::STRUCT) {
return Status::Invalid(
fmt::format("FilterMapSelectedKeysRecursively requires struct array for read "
"field '{}', got {}",
read_field->name(), array->type()->ToString()));
}
auto struct_array = checked_pointer_cast<arrow::StructArray>(array);
auto read_struct_type = checked_pointer_cast<arrow::StructType>(read_field->type());
if (struct_array->num_fields() != read_struct_type->num_fields()) {
return Status::Invalid(fmt::format(
"FilterMapSelectedKeysRecursively struct field count mismatch for '{}': "
"array {} vs read {}",
read_field->name(), struct_array->num_fields(), read_struct_type->num_fields()));
}

arrow::ArrayVector filtered_children;
std::vector<std::shared_ptr<arrow::ArrayData>> filtered_child_data;
filtered_children.reserve(struct_array->num_fields());
filtered_child_data.reserve(struct_array->num_fields());
for (int32_t i = 0; i < struct_array->num_fields(); ++i) {
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> filtered_child,
FilterMapSelectedKeysRecursively(struct_array->field(i),
read_struct_type->field(i)));
filtered_child_data.push_back(filtered_child->data());
filtered_children.push_back(std::move(filtered_child));
}

// Preserve parent struct null semantics after filtering children.
auto filtered_struct_data = arrow::ArrayData::Make(
read_struct_type, struct_array->length(), {struct_array->null_bitmap()},
std::move(filtered_child_data), struct_array->null_count(), struct_array->offset());
return arrow::MakeArray(std::move(filtered_struct_data));
}

return array;
}

Result<std::unique_ptr<FieldMappingReader>> FieldMappingReader::Create(
int32_t field_count, std::unique_ptr<FileBatchReader>&& reader, const BinaryRow& partition,
std::unique_ptr<FieldMapping>&& mapping,
Expand Down Expand Up @@ -186,7 +92,7 @@ Result<std::unique_ptr<FieldMappingReader>> FieldMappingReader::Create(
// FilterMapArrayBySelectedKeys can filter out unwanted entries.
PAIMON_ASSIGN_OR_RAISE(
bool has_map_selected_keys,
mapping_reader->HasMapSelectedKeysRecursively(
NestedProjectionUtils::HasMapSelectedKeysRecursively(
mapping_reader->non_partition_info_.non_partition_read_schema[i].ArrowField()));
if (has_map_selected_keys &&
mapping_reader->skip_map_selected_keys_filter_field_ids_.count(
Expand Down Expand Up @@ -443,8 +349,9 @@ Status FieldMappingReader::MappingFields(const std::shared_ptr<arrow::Array>& da

// Filter map entries by selected keys recursively (supports MAP nested in STRUCT).
if (skip_map_selected_keys_filter_field_ids_.count(read_field.Id()) == 0) {
PAIMON_ASSIGN_OR_RAISE(field_array, FilterMapSelectedKeysRecursively(
field_array, read_field.ArrowField()));
PAIMON_ASSIGN_OR_RAISE(field_array,
NestedProjectionUtils::FilterMapArrayBySelectedKeysRecursively(
field_array, read_field.ArrowField(), arrow_pool_.get()));
}

(*target_array)[idx_in_target_schema[i]] = std::move(field_array);
Expand Down
7 changes: 0 additions & 7 deletions src/paimon/core/io/field_mapping_reader.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,13 +111,6 @@ class FieldMappingReader : public FileBatchReader {
arrow::ArrayVector* target_array,
std::vector<std::string>* target_field_names);

Result<bool> HasMapSelectedKeysRecursively(
const std::shared_ptr<arrow::Field>& read_field) const;

Result<std::shared_ptr<arrow::Array>> FilterMapSelectedKeysRecursively(
const std::shared_ptr<arrow::Array>& array,
const std::shared_ptr<arrow::Field>& read_field) const;

private:
bool need_mapping_ = false;
bool need_casting_ = false;
Expand Down
47 changes: 30 additions & 17 deletions src/paimon/core/io/key_value_data_file_record_reader.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,20 +35,20 @@
#include "paimon/common/utils/arrow/arrow_utils.h"
#include "paimon/common/utils/arrow/status_utils.h"
#include "paimon/common/utils/checked_cast.h"
#include "paimon/reader/file_batch_reader.h"
#include "paimon/status.h"
namespace paimon {
class MemoryPool;

KeyValueDataFileRecordReader::KeyValueDataFileRecordReader(
std::unique_ptr<FileBatchReader>&& reader, const std::shared_ptr<arrow::Schema>& key_schema,
std::unique_ptr<BatchReader>&& reader, const std::shared_ptr<arrow::Schema>& key_schema,
const std::shared_ptr<arrow::Schema>& value_schema, int32_t level,
const std::shared_ptr<MemoryPool>& pool)
: level_(level),
pool_(pool),
reader_(std::move(reader)),
key_schema_(key_schema),
value_schema_(value_schema),
value_names_(value_schema_->field_names()) {}
value_schema_(value_schema) {}

Result<bool> KeyValueDataFileRecordReader::Iterator::HasNext() const {
int64_t array_length = reader_->row_kind_array_->length();
Expand Down Expand Up @@ -83,8 +83,11 @@ Result<KeyValue> KeyValueDataFileRecordReader::Iterator::Next() {

Result<std::pair<int64_t, KeyValue>> KeyValueDataFileRecordReader::Iterator::NextWithFilePos() {
PAIMON_ASSIGN_OR_RAISE(KeyValue kv, Next());
if (!reader_->file_reader_) {
return Status::Invalid("KeyValueRecordReader does not support file row positions");
}
PAIMON_ASSIGN_OR_RAISE(uint64_t global_row_id,
reader_->reader_->GetPreviousBatchFileRowId(cursor_ - 1));
reader_->file_reader_->GetPreviousBatchFileRowId(cursor_ - 1));
return std::make_pair(static_cast<int64_t>(global_row_id), std::move(kv));
}

Expand All @@ -107,26 +110,34 @@ Result<std::unique_ptr<KeyValueRecordReader::Iterator>> KeyValueDataFileRecordRe
return Status::Invalid("cannot cast data batch to StructArray");
}
auto data_batch = checked_pointer_cast<arrow::StructArray>(arrow_array);
if (data_batch->num_fields() < SpecialFields::KEY_VALUE_SPECIAL_FIELD_COUNT) {
return Status::Invalid(
fmt::format("data batch field count {} is less than required special field count {}",
data_batch->num_fields(), SpecialFields::KEY_VALUE_SPECIAL_FIELD_COUNT));
}
if (!data_batch->field(0) || data_batch->field(0)->type_id() != arrow::Type::INT64) {
std::shared_ptr<arrow::Array> sequence_number =
data_batch->GetFieldByName(SpecialFields::SequenceNumber().Name());
if (!sequence_number || sequence_number->type_id() != arrow::Type::INT64) {
return Status::Invalid("cannot cast SEQUENCE_NUMBER column to int64 arrow array");
}
sequence_number_array_ =
checked_pointer_cast<arrow::NumericArray<arrow::Int64Type>>(data_batch->field(0));
if (!data_batch->field(1) || data_batch->field(1)->type_id() != arrow::Type::INT8) {
checked_pointer_cast<arrow::NumericArray<arrow::Int64Type>>(sequence_number);
if (sequence_number_array_->null_count() != 0) {
return Status::Invalid("SEQUENCE_NUMBER column contains null");
}
std::shared_ptr<arrow::Array> row_kind =
data_batch->GetFieldByName(SpecialFields::ValueKind().Name());
if (!row_kind || row_kind->type_id() != arrow::Type::INT8) {
return Status::Invalid("cannot cast VALUE_KIND column to int8 arrow array");
}
row_kind_array_ =
checked_pointer_cast<arrow::NumericArray<arrow::Int8Type>>(data_batch->field(1));
row_kind_array_ = checked_pointer_cast<arrow::NumericArray<arrow::Int8Type>>(row_kind);
if (row_kind_array_->null_count() != 0) {
return Status::Invalid("VALUE_KIND column contains null");
}
arrow::ArrayVector key_fields;
key_fields.reserve(key_schema_->num_fields());
for (const auto& key_field : key_schema_->fields()) {
// skip special fields
key_fields.emplace_back(data_batch->GetFieldByName(key_field->name()));
std::shared_ptr<arrow::Array> field_array = data_batch->GetFieldByName(key_field->name());
if (!field_array) {
return Status::Invalid(
fmt::format("cannot find field {} in data batch", key_field->name()));
}
key_fields.emplace_back(std::move(field_array));
}
// e.g., file schema: seq, kind, key1, key2, s1, s2, v1, v2
// user raw read schema: key1, v1, s1
Expand All @@ -140,17 +151,19 @@ Result<std::unique_ptr<KeyValueRecordReader::Iterator>> KeyValueDataFileRecordRe
return Status::Invalid(
fmt::format("cannot find field {} in data batch", value_field->name()));
}
value_fields.emplace_back(field_array);
value_fields.emplace_back(std::move(field_array));
}

selection_bitmap_ = std::move(bitmap);
file_reader_ = dynamic_cast<FileBatchReader*>(reader_.get());
key_ctx_ = std::make_shared<ColumnarBatchContext>(key_fields, pool_);
value_ctx_ = std::make_shared<ColumnarBatchContext>(value_fields, pool_);
ArrowUtils::TraverseArray(data_batch);
return std::make_unique<KeyValueDataFileRecordReader::Iterator>(this);
}

void KeyValueDataFileRecordReader::Reset() {
file_reader_ = nullptr;
selection_bitmap_ = RoaringBitmap32();
key_ctx_.reset();
value_ctx_.reset();
Expand Down
Loading
Loading