SteNicholas commented on code in PR #194:
URL: https://github.com/apache/paimon-cpp/pull/194#discussion_r3809950879
##########
src/paimon/core/table/source/key_value_table_read.cpp:
##########
@@ -74,7 +75,33 @@ void KeyValueTableRead::ForceKeepDelete(bool
force_keep_delete) {
Result<std::unique_ptr<BatchReader>> KeyValueTableRead::CreateReader(
const std::shared_ptr<Split>& split) {
- auto data_split = std::dynamic_pointer_cast<DataSplit>(split);
+ std::shared_ptr<Split> dispatch_split = split;
+ if (auto indexed_split =
std::dynamic_pointer_cast<IndexedSplitImpl>(split)) {
+ PAIMON_RETURN_NOT_OK(indexed_split->Validate());
+ if (!indexed_split->Scores().empty()) {
+ // TODO(wangyong9999): Propagate indexed scores through the
primary-key
+ // physical-position read path.
+ return Status::NotImplemented(
+ "Primary-key reads do not support scored indexed splits yet.");
+ }
+ // Primary-key indexed splits carry physical positions and are routed
independently
+ // of the inner split's raw-convertible marker, matching Java's
dedicated provider.
+ const std::shared_ptr<DataSplit>& inner_split =
indexed_split->GetDataSplit();
+ if (!force_keep_delete_) {
+ for (const auto& read : split_reads_) {
+ if (dynamic_cast<RawFileSplitRead*>(read.get()) != nullptr) {
+ return read->CreateReader(indexed_split);
Review Comment:
**[P1] Preserve the legacy raw-read safety check**
This direct dispatch bypasses `RawFileSplitRead::Match`, whose primary-key
path deliberately rejects files with an unknown `delete_row_count` because
legacy metadata cannot prove that they contain no DELETE rows.
`MergeTreeSplitGenerator` still treats a missing count as raw-convertible,
while the new index-source policy checks only the COMPACT source and level, so
an indexed legacy file can reach this path and expose tombstones or obsolete
rows instead of using merge semantics. Require a known zero delete count for
every indexed file, or reuse the existing match guard and fall back to the
inner merge read.
##########
src/paimon/core/table/source/primary_key_sorted_index_scan.cpp:
##########
@@ -0,0 +1,549 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "paimon/core/table/source/primary_key_sorted_index_scan.h"
+
+#include <cassert>
+#include <set>
+#include <unordered_map>
+#include <utility>
+
+#include "arrow/api.h"
+#include "arrow/c/bridge.h"
+#include "arrow/c/helpers.h"
+#include "fmt/format.h"
+#include "paimon/common/types/data_field.h"
+#include "paimon/common/utils/arrow/status_utils.h"
+#include "paimon/common/utils/scope_guard.h"
+#include "paimon/core/global_index/global_index_evaluator_impl.h"
+#include "paimon/core/index/pk/primary_key_index_definitions.h"
+#include "paimon/core/index/pksorted/pk_sorted_bucket_index_state.h"
+#include "paimon/core/manifest/file_kind.h"
+#include "paimon/global_index/bitmap_global_index_result.h"
+#include "paimon/global_index/global_index_io_meta.h"
+#include "paimon/global_index/global_indexer.h"
+#include "paimon/global_index/global_indexer_factory.h"
+#include "paimon/global_index/io/global_index_file_reader.h"
+#include "paimon/logging.h"
+#include "paimon/predicate/predicate_utils.h"
+
+namespace paimon {
+namespace {
+using BucketKey = std::pair<BinaryRow, int32_t>;
+
+struct QueryKey {
+ Function::Type operation;
+ std::vector<Literal> literals;
+
+ bool operator==(const QueryKey& other) const {
+ return operation == other.operation && literals == other.literals;
+ }
+};
+
+/// Shares one group payload reader and its group-scope query results across
all source
+/// files of the group; localizes group ordinals to file-local physical
positions using the
+/// ordered source row-count prefix.
+class SharedGroupReader {
+ public:
+ using UnderlyingReaderFactory =
std::function<Result<std::shared_ptr<GlobalIndexReader>>()>;
+
+ SharedGroupReader(const std::shared_ptr<PkSortedIndexGroup>& group,
+ UnderlyingReaderFactory reader_factory)
+ : group_(group), reader_factory_(std::move(reader_factory)) {
+ const std::vector<PrimaryKeyIndexSourceFile>& source_files =
group->SourceFiles();
+ source_offsets_.reserve(source_files.size() + 1);
+ source_offsets_.push_back(0);
+ for (const PrimaryKeyIndexSourceFile& source_file : source_files) {
+ source_offsets_.push_back(source_offsets_.back() +
source_file.row_count);
+ }
+ }
+
+ const std::shared_ptr<PkSortedIndexGroup>& Group() const {
+ return group_;
+ }
+
+ /// Runs one group-scope query with caching; equal queries evaluate
exactly once.
+ Result<std::shared_ptr<GlobalIndexResult>> Query(
+ const QueryKey& key,
+ const
std::function<Result<std::shared_ptr<GlobalIndexResult>>(GlobalIndexReader*)>&
+ query) {
+ for (const auto& cached : query_cache_) {
+ if (cached.first == key) {
+ if (!cached.second.status.ok()) {
+ return cached.second.status;
+ }
+ return cached.second.result;
+ }
+ }
+ Result<std::shared_ptr<GlobalIndexResult>> result = RunQuery(query);
+ CachedQuery cached_query;
+ if (result.ok()) {
+ cached_query.result = result.value();
+ } else {
+ cached_query.status = result.status();
+ }
+ query_cache_.emplace_back(key, cached_query);
+ return result;
+ }
+
+ /// Restricts one group-scope result to the local row positions of
`source_index`.
+ /// Any out-of-range group ordinal fails the localization so that every
covered file
+ /// of this group falls back to a normal scan; a poison marker would not
survive the
+ /// AND/OR combination of results from other indexes.
+ Result<std::shared_ptr<GlobalIndexResult>> Localize(
+ const std::shared_ptr<GlobalIndexResult>& result, size_t source_index)
{
+ if (result == nullptr) {
+ return std::shared_ptr<GlobalIndexResult>(nullptr);
+ }
+ assert(source_index + 1 < source_offsets_.size());
+ auto localized = localized_cache_.find(result.get());
+ if (localized == localized_cache_.end()) {
+
PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<GlobalIndexResult>>
partitions,
+ PartitionBySource(result));
+ localized = localized_cache_.emplace(result.get(),
std::move(partitions)).first;
+ }
+ return localized->second[source_index];
+ }
+
+ private:
+ struct CachedQuery {
+ Status status;
+ std::shared_ptr<GlobalIndexResult> result;
+ };
+
+ Result<std::shared_ptr<GlobalIndexResult>> RunQuery(
+ const
std::function<Result<std::shared_ptr<GlobalIndexResult>>(GlobalIndexReader*)>&
+ query) {
+ if (!reader_status_.ok()) {
+ return reader_status_;
+ }
+ if (reader_ == nullptr) {
+ Result<std::shared_ptr<GlobalIndexReader>> reader_result =
reader_factory_();
+ if (!reader_result.ok()) {
+ reader_status_ = reader_result.status();
+ return reader_status_;
+ }
+ reader_ = reader_result.value();
+ if (reader_ == nullptr) {
+ // The index type has no usable reader; keep normal scan
semantics.
+ return std::shared_ptr<GlobalIndexResult>(nullptr);
+ }
+ }
+ return query(reader_.get());
+ }
+
+ Result<std::vector<std::shared_ptr<GlobalIndexResult>>> PartitionBySource(
+ const std::shared_ptr<GlobalIndexResult>& result) {
+ size_t source_count = source_offsets_.size() - 1;
+ std::vector<RoaringBitmap64> partitions(source_count);
+ int64_t total_row_count = source_offsets_.back();
+ size_t source_index = 0;
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<GlobalIndexResult::Iterator>
iterator,
+ result->CreateIterator());
+ while (iterator->HasNext()) {
+ int64_t position = iterator->Next();
+ if (position < 0 || position >= total_row_count) {
+ return Status::Invalid(fmt::format(
+ "Sorted index returned group ordinal {} outside the source
row range "
+ "[0, {}).",
+ position, total_row_count));
+ }
+ while (position >= source_offsets_[source_index + 1]) {
+ source_index++;
+ }
+ partitions[source_index].Add(position -
source_offsets_[source_index]);
+ }
+ std::vector<std::shared_ptr<GlobalIndexResult>> localized;
+ localized.reserve(source_count);
+ for (RoaringBitmap64& partition : partitions) {
+ auto bitmap =
std::make_shared<RoaringBitmap64>(std::move(partition));
+ localized.push_back(std::make_shared<BitmapGlobalIndexResult>(
+ [bitmap]() -> Result<RoaringBitmap64> { return *bitmap; }));
+ }
+ return localized;
+ }
+
+ std::shared_ptr<PkSortedIndexGroup> group_;
+ UnderlyingReaderFactory reader_factory_;
+ std::vector<int64_t> source_offsets_;
+ std::vector<std::pair<QueryKey, CachedQuery>> query_cache_;
+ std::unordered_map<const GlobalIndexResult*,
std::vector<std::shared_ptr<GlobalIndexResult>>>
+ localized_cache_;
+ std::shared_ptr<GlobalIndexReader> reader_;
+ Status reader_status_;
+};
+
+/// Restricts merged source-group ordinals to one source file's local row
positions.
+class FileLocalGroupReader : public GlobalIndexReader {
+ public:
+ FileLocalGroupReader(std::shared_ptr<SharedGroupReader> shared_reader,
size_t source_index)
+ : shared_reader_(std::move(shared_reader)),
source_index_(source_index) {}
+
+ Result<std::shared_ptr<GlobalIndexResult>> VisitIsNotNull() override {
+ return Query({Function::Type::IS_NOT_NULL, {}},
+ [](GlobalIndexReader* reader) { return
reader->VisitIsNotNull(); });
+ }
+
+ Result<std::shared_ptr<GlobalIndexResult>> VisitIsNull() override {
+ return Query({Function::Type::IS_NULL, {}},
+ [](GlobalIndexReader* reader) { return
reader->VisitIsNull(); });
+ }
+
+ Result<std::shared_ptr<GlobalIndexResult>> VisitEqual(const Literal&
literal) override {
+ return Query({Function::Type::EQUAL, {literal}},
+ [&literal](GlobalIndexReader* reader) { return
reader->VisitEqual(literal); });
+ }
+
+ Result<std::shared_ptr<GlobalIndexResult>> VisitNotEqual(const Literal&
literal) override {
+ return Query({Function::Type::NOT_EQUAL, {literal}},
[&literal](GlobalIndexReader* reader) {
+ return reader->VisitNotEqual(literal);
+ });
+ }
+
+ Result<std::shared_ptr<GlobalIndexResult>> VisitLessThan(const Literal&
literal) override {
+ return Query({Function::Type::LESS_THAN, {literal}},
[&literal](GlobalIndexReader* reader) {
+ return reader->VisitLessThan(literal);
+ });
+ }
+
+ Result<std::shared_ptr<GlobalIndexResult>> VisitLessOrEqual(const Literal&
literal) override {
+ return Query(
+ {Function::Type::LESS_OR_EQUAL, {literal}},
+ [&literal](GlobalIndexReader* reader) { return
reader->VisitLessOrEqual(literal); });
+ }
+
+ Result<std::shared_ptr<GlobalIndexResult>> VisitGreaterThan(const Literal&
literal) override {
+ return Query(
+ {Function::Type::GREATER_THAN, {literal}},
+ [&literal](GlobalIndexReader* reader) { return
reader->VisitGreaterThan(literal); });
+ }
+
+ Result<std::shared_ptr<GlobalIndexResult>> VisitGreaterOrEqual(
+ const Literal& literal) override {
+ return Query(
+ {Function::Type::GREATER_OR_EQUAL, {literal}},
+ [&literal](GlobalIndexReader* reader) { return
reader->VisitGreaterOrEqual(literal); });
+ }
+
+ Result<std::shared_ptr<GlobalIndexResult>> VisitIn(
+ const std::vector<Literal>& literals) override {
+ return Query({Function::Type::IN, literals},
+ [&literals](GlobalIndexReader* reader) { return
reader->VisitIn(literals); });
+ }
+
+ Result<std::shared_ptr<GlobalIndexResult>> VisitNotIn(
+ const std::vector<Literal>& literals) override {
+ return Query({Function::Type::NOT_IN, literals},
[&literals](GlobalIndexReader* reader) {
+ return reader->VisitNotIn(literals);
+ });
+ }
+
+ Result<std::shared_ptr<GlobalIndexResult>> VisitStartsWith(const Literal&
prefix) override {
+ return Query({Function::Type::STARTS_WITH, {prefix}},
[&prefix](GlobalIndexReader* reader) {
+ return reader->VisitStartsWith(prefix);
+ });
+ }
+
+ Result<std::shared_ptr<GlobalIndexResult>> VisitEndsWith(const Literal&
suffix) override {
+ return Query({Function::Type::ENDS_WITH, {suffix}},
[&suffix](GlobalIndexReader* reader) {
+ return reader->VisitEndsWith(suffix);
+ });
+ }
+
+ Result<std::shared_ptr<GlobalIndexResult>> VisitContains(const Literal&
literal) override {
+ return Query({Function::Type::CONTAINS, {literal}},
[&literal](GlobalIndexReader* reader) {
+ return reader->VisitContains(literal);
+ });
+ }
+
+ Result<std::shared_ptr<GlobalIndexResult>> VisitLike(const Literal&
literal) override {
+ return Query({Function::Type::LIKE, {literal}},
+ [&literal](GlobalIndexReader* reader) { return
reader->VisitLike(literal); });
+ }
+
+ Result<std::shared_ptr<ScoredGlobalIndexResult>> VisitVectorSearch(
+ const std::shared_ptr<VectorSearch>& vector_search) override {
+ return Status::Invalid("Primary-key sorted index does not support
vector search.");
+ }
+
+ Result<std::shared_ptr<GlobalIndexResult>> VisitFullTextSearch(
+ const std::shared_ptr<FullTextSearch>& full_text_search) override {
+ return Status::Invalid("Primary-key sorted index does not support full
text search.");
+ }
+
+ bool IsThreadSafe() const override {
+ return false;
+ }
+
+ std::string GetIndexType() const override {
+ return shared_reader_->Group()->Payload()->IndexType();
+ }
+
+ private:
+ Result<std::shared_ptr<GlobalIndexResult>> Query(
+ QueryKey key,
+ const
std::function<Result<std::shared_ptr<GlobalIndexResult>>(GlobalIndexReader*)>&
+ query) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<GlobalIndexResult> group_result,
+ shared_reader_->Query(key, query));
+ return shared_reader_->Localize(group_result, source_index_);
+ }
+
+ std::shared_ptr<SharedGroupReader> shared_reader_;
+ size_t source_index_;
+};
+
+Result<size_t> FindSourceIndex(const PkSortedIndexGroup& group, const
DataFileMeta& data_file) {
+ const std::vector<PrimaryKeyIndexSourceFile>& source_files =
group.SourceFiles();
+ for (size_t i = 0; i < source_files.size(); i++) {
+ if (source_files[i].file_name == data_file.file_name &&
+ source_files[i].row_count == data_file.row_count) {
+ return i;
+ }
+ }
+ return Status::Invalid(fmt::format(
+ "Data file {} is not covered by its sorted-index source group.",
data_file.file_name));
+}
+} // namespace
+
+Result<PrimaryKeySortedIndexScan::Plan> PrimaryKeySortedIndexScan::CreatePlan(
+ int64_t snapshot_id, const std::vector<std::shared_ptr<DataSplitImpl>>&
data_splits,
+ const std::vector<PrimaryKeyIndexDefinition>& definitions,
+ const std::vector<IndexManifestEntry>& index_entries) {
+ std::unordered_map<BucketKey, std::vector<std::shared_ptr<IndexFileMeta>>>
payloads_by_bucket;
+ for (const IndexManifestEntry& entry : index_entries) {
+ const std::shared_ptr<IndexFileMeta>& payload = entry.index_file;
+ if (payload == nullptr || !(entry.kind == FileKind::Add())) {
+ continue;
+ }
+ const std::optional<GlobalIndexMeta>& meta =
payload->GetGlobalIndexMeta();
+ if (meta == std::nullopt || meta.value().source_meta == nullptr) {
+ continue;
+ }
+ payloads_by_bucket[BucketKey(entry.partition,
entry.bucket)].push_back(payload);
+ }
+
+ std::vector<PrimaryKeyIndexDefinition> scalar_definitions =
+ PrimaryKeyIndexDefinitions::ScalarDefinitions(definitions);
+
+ std::unordered_map<BucketKey, std::vector<std::shared_ptr<DataFileMeta>>>
data_files_by_bucket;
+ for (const std::shared_ptr<DataSplitImpl>& split : data_splits) {
+ if (split == nullptr) {
+ return Status::Invalid("Primary-key sorted-index scan received a
null data split.");
+ }
+ if (split->SnapshotId() != snapshot_id) {
+ return Status::Invalid(
+ fmt::format("Data split snapshot {} does not match
sorted-index scan snapshot {}.",
+ split->SnapshotId(), snapshot_id));
+ }
+ if (split->IsStreaming()) {
+ return Status::Invalid("Primary-key sorted-index scan requires
batch splits.");
+ }
+ if (!split->DeletionFiles().empty() &&
+ split->DeletionFiles().size() != split->DataFiles().size()) {
+ return Status::Invalid(
+ "Deletion files must align with data files in a sorted-index
split.");
+ }
+ std::vector<std::shared_ptr<DataFileMeta>>& data_files =
+ data_files_by_bucket[BucketKey(split->Partition(),
split->Bucket())];
+ data_files.insert(data_files.end(), split->DataFiles().begin(),
split->DataFiles().end());
+ }
+
+ // file name -> field id -> validated group, per bucket.
+ std::unordered_map<
+ BucketKey, std::map<std::string, std::map<int32_t,
std::shared_ptr<PkSortedIndexGroup>>>>
+ groups_by_bucket;
+ for (const auto& bucket_entry : data_files_by_bucket) {
+ const BucketKey& bucket = bucket_entry.first;
+ std::vector<std::shared_ptr<IndexFileMeta>> bucket_payloads;
+ auto payloads_iter = payloads_by_bucket.find(bucket);
+ if (payloads_iter != payloads_by_bucket.end()) {
+ bucket_payloads = payloads_iter->second;
+ }
+ std::set<std::pair<std::string, int64_t>> active_source_files;
+ for (const std::shared_ptr<DataFileMeta>& data_file :
bucket_entry.second) {
+ active_source_files.emplace(data_file->file_name,
data_file->row_count);
+ }
+ std::map<std::string, std::map<int32_t,
std::shared_ptr<PkSortedIndexGroup>>>
+ groups_by_source;
+ for (const PrimaryKeyIndexDefinition& definition : scalar_definitions)
{
+ std::vector<std::shared_ptr<IndexFileMeta>> definition_payloads;
+ for (const std::shared_ptr<IndexFileMeta>& payload :
bucket_payloads) {
+ const std::optional<GlobalIndexMeta>& meta =
payload->GetGlobalIndexMeta();
+ if (meta != std::nullopt && definition.IndexType() ==
payload->IndexType() &&
+ definition.FieldId() == meta.value().index_field_id) {
+ definition_payloads.push_back(payload);
+ }
+ }
+ PkSortedBucketIndexState state =
PkSortedBucketIndexState::FromActiveDataFiles(
+ definition.FieldId(), definition.IndexType(),
bucket_entry.second,
+ definition_payloads);
+ for (const std::shared_ptr<PkSortedIndexGroup>& group :
state.Groups()) {
+ for (const PrimaryKeyIndexSourceFile& source_file :
group->SourceFiles()) {
+ if (active_source_files.count({source_file.file_name,
source_file.row_count}) ==
+ 0) {
+ continue;
+ }
+
groups_by_source[source_file.file_name][definition.FieldId()] = group;
+ }
+ }
+ }
+ groups_by_bucket[bucket] = std::move(groups_by_source);
+ }
+
+ std::vector<FilePlan> files;
+ for (const std::shared_ptr<DataSplitImpl>& split : data_splits) {
+ auto bucket_groups =
groups_by_bucket.find(BucketKey(split->Partition(), split->Bucket()));
+ for (size_t file_index = 0; file_index < split->DataFiles().size();
file_index++) {
+ const std::shared_ptr<DataFileMeta>& data_file =
split->DataFiles()[file_index];
+ std::map<int32_t, std::shared_ptr<PkSortedIndexGroup>> groups;
+ if (bucket_groups != groups_by_bucket.end() && data_file !=
nullptr) {
+ auto source_groups =
bucket_groups->second.find(data_file->file_name);
+ if (source_groups != bucket_groups->second.end()) {
+ groups = source_groups->second;
+ }
+ }
+ files.emplace_back(split, static_cast<int32_t>(file_index),
std::move(groups));
+ }
+ }
+ return Plan(snapshot_id, std::move(files));
+}
+
+Result<PrimaryKeySortedIndexScan::EvaluatedPlan>
PrimaryKeySortedIndexScan::Evaluate(
+ const Plan& plan, const std::shared_ptr<TableSchema>& table_schema,
+ const std::shared_ptr<Predicate>& predicate,
+ const std::vector<PrimaryKeyIndexDefinition>& definitions,
+ const ReaderFactory& reader_factory) {
+ std::map<int32_t, PrimaryKeyIndexDefinition> definitions_by_field;
+ for (const PrimaryKeyIndexDefinition& definition :
+ PrimaryKeyIndexDefinitions::ScalarDefinitions(definitions)) {
+ definitions_by_field.emplace(definition.FieldId(), definition);
+ }
+
+ std::unordered_map<const PkSortedIndexGroup*,
std::shared_ptr<SharedGroupReader>>
+ shared_readers;
+ std::vector<EvaluatedFile> files;
+ files.reserve(plan.Files().size());
+ for (const FilePlan& file : plan.Files()) {
+ GlobalIndexEvaluatorImpl::IndexReadersCreator create_readers =
+ [&file, &definitions_by_field, &shared_readers, &reader_factory](
+ int32_t field_id) ->
Result<std::vector<std::shared_ptr<GlobalIndexReader>>> {
+ auto definition_iter = definitions_by_field.find(field_id);
+ std::shared_ptr<PkSortedIndexGroup> group = file.Group(field_id);
+ if (definition_iter == definitions_by_field.end() || group ==
nullptr) {
+ return std::vector<std::shared_ptr<GlobalIndexReader>>();
+ }
+ auto shared_iter = shared_readers.find(group.get());
+ if (shared_iter == shared_readers.end()) {
+ const PrimaryKeyIndexDefinition& definition =
definition_iter->second;
+ // The shared reader outlives this file plan, so the factory
owns a copy of
+ // the file plan instead of referencing the loop variable.
+ SharedGroupReader::UnderlyingReaderFactory underlying_factory =
+ [file_copy = file, definition, group,
+ &reader_factory]() ->
Result<std::shared_ptr<GlobalIndexReader>> {
+ return reader_factory(file_copy, definition, *group);
+ };
+ shared_iter = shared_readers
+ .emplace(group.get(),
std::make_shared<SharedGroupReader>(
+ group,
std::move(underlying_factory)))
+ .first;
+ }
+ PAIMON_ASSIGN_OR_RAISE(size_t source_index,
FindSourceIndex(*group, *file.DataFile()));
+ std::vector<std::shared_ptr<GlobalIndexReader>> readers;
+ readers.push_back(
+ std::make_shared<FileLocalGroupReader>(shared_iter->second,
source_index));
+ return readers;
+ };
+ GlobalIndexEvaluatorImpl evaluator(table_schema, create_readers);
+ Result<std::shared_ptr<GlobalIndexResult>> result =
evaluator.Evaluate(predicate);
+ if (result.ok()) {
+ files.emplace_back(file, result.value());
+ } else {
+ // Evaluation failures degrade to a normal scan for this file only.
+ static auto logger =
Logger::GetLogger("PrimaryKeySortedIndexScan");
+ PAIMON_LOG_WARN(logger,
+ "Failed to evaluate primary-key sorted index for
data file %s; "
+ "falling back to a normal scan for this file: %s",
+ file.DataFile()->file_name.c_str(),
result.status().ToString().c_str());
+ files.emplace_back(file, nullptr);
+ }
+ }
+ return EvaluatedPlan(plan.SnapshotId(), std::move(files));
+}
+
+namespace {
+class FsGlobalIndexFileReader : public GlobalIndexFileReader {
+ public:
+ explicit FsGlobalIndexFileReader(std::shared_ptr<FileSystem> file_system)
+ : file_system_(std::move(file_system)) {}
+
+ Result<std::unique_ptr<InputStream>> GetInputStream(
+ const std::string& file_path) const override {
+ return file_system_->Open(file_path);
+ }
+
+ private:
+ std::shared_ptr<FileSystem> file_system_;
+};
+} // namespace
+
+PrimaryKeySortedIndexScan::ReaderFactory
PrimaryKeySortedIndexScan::MakeReaderFactory(
+ const std::shared_ptr<FileSystem>& file_system,
+ const std::shared_ptr<IndexFilePathFactories>& path_factories,
+ const std::shared_ptr<TableSchema>& table_schema, const
std::shared_ptr<MemoryPool>& pool) {
+ auto file_reader = std::make_shared<FsGlobalIndexFileReader>(file_system);
+ return [path_factories, table_schema, pool, file_reader](
+ const FilePlan& file, const PrimaryKeyIndexDefinition&
definition,
+ const PkSortedIndexGroup& group) ->
Result<std::shared_ptr<GlobalIndexReader>> {
+ if (definition.GetFamily() !=
PrimaryKeyIndexDefinition::Family::BTREE) {
+ // Only the BTree payload reader is wired up; other families keep
normal scan
+ // semantics until their dedicated readers are supported.
+ return std::shared_ptr<GlobalIndexReader>(nullptr);
+ }
+ PAIMON_ASSIGN_OR_RAISE(
+ std::unique_ptr<GlobalIndexer> indexer,
+ GlobalIndexerFactory::Get(definition.IndexType(),
definition.Options()));
+ if (indexer == nullptr) {
+ return std::shared_ptr<GlobalIndexReader>(nullptr);
+ }
+ const std::shared_ptr<DataSplitImpl>& split = file.SourceSplit();
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<IndexPathFactory> path_factory,
+ path_factories->Get(split->Partition(),
split->Bucket()));
+ const std::shared_ptr<IndexFileMeta>& payload = group.Payload();
+ const std::optional<GlobalIndexMeta>& payload_meta =
payload->GetGlobalIndexMeta();
+ if (payload_meta == std::nullopt) {
+ // Group validation guarantees the metadata; degrade to a normal
scan of the
+ // covered files if it is ever violated, like the Java reader
factory.
+ return Status::Invalid(fmt::format(
+ "Sorted index payload {} has no global index metadata.",
payload->FileName()));
+ }
+ std::vector<GlobalIndexIOMeta> io_metas;
+ io_metas.emplace_back(path_factory->ToPath(payload),
payload->FileSize(),
Review Comment:
**[P1] Reject missing or malformed BTree metadata**
`_INDEX_META` is nullable, and the group validation currently accepts a null
payload, but this value is passed directly to `BTreeFileMetaSelector`. Its
eager deserialization dereferences null metadata and reads fields without
checking the buffer length, so an incomplete persisted index can crash the
process instead of triggering the intended per-file fallback. Validate the
BTree metadata before creating the reader—ideally by making deserialization
return a checked `Result`—and mark the index group unusable when validation
fails.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
