SteNicholas commented on code in PR #490:
URL: https://github.com/apache/paimon-cpp/pull/490#discussion_r4476170949
##########
src/paimon/core/operation/data_evolution_split_read.h:
##########
@@ -97,8 +97,7 @@ class DataEvolutionSplitRead : public AbstractSplitRead {
const std::shared_ptr<Predicate>& predicate, DeletionVector::Factory
dv_factory,
const std::optional<std::vector<Range>>& row_ranges,
const std::shared_ptr<DataFilePathFactory>& data_file_path_factory,
- const std::map<std::string, std::string>& file_index_options,
- std::vector<float>* index_scores) const override;
+ const std::map<std::string, std::string>& file_index_options) const
override;
Review Comment:
The merge build still has a caller passing ten arguments. The test added on
`main` by `eb0dd421` passes `/*index_scores=*/nullptr` after
`file_index_options`. Please sync with `main` and remove that trailing argument
there too so `paimon-core-test` can compile. Both
[Clang](https://github.com/apache/paimon-cpp/actions/runs/41948662626/job/126987074702)
and [GCC
8](https://github.com/apache/paimon-cpp/actions/runs/41948662640/job/126987074094)
report the same mismatch.
##########
include/paimon/file_index/file_indexer.h:
##########
@@ -47,6 +47,10 @@ class PAIMON_EXPORT FileIndexer {
return false;
}
+ /// Create a reader for an empty index entry without serialized bytes.
+ /// Scalar indexers use the default reader. Search indexers must preserve
their capabilities.
+ virtual std::shared_ptr<FileIndexReader> CreateEmptyReader() const;
Review Comment:
Inserting `CreateEmptyReader` here changes dispatch for already-built index
plugins, including scalar plugins that never use search. Clang's vtable dump
puts `CreateReader` in slot 4 before this change and slot 5 afterwards. Slot 5
in an old plugin is `CreateWriter`, which has a different signature. Could we
expose this through a separate capability interface, as with
`WrittenFileStatsProvider`, or version the plugin ABI and reject incompatible
binaries at load time? An ABI test built entirely with the new headers misses
this case.
##########
src/paimon/core/io/file_index_evaluator.cpp:
##########
@@ -52,64 +54,165 @@ namespace paimon {
class MemoryPool;
enum class FieldType;
-Result<std::shared_ptr<FileIndexResult>> FileIndexEvaluator::Evaluate(
+FileIndexEvaluator::FileIndexEvaluator(std::shared_ptr<arrow::Schema>
data_schema,
+ std::shared_ptr<DataFileMeta> file_meta,
+
std::unique_ptr<FileIndexFormat::Reader> format_reader)
+ : data_schema_(std::move(data_schema)),
+ file_meta_(std::move(file_meta)),
+ format_reader_(std::move(format_reader)) {}
+
+Result<std::unique_ptr<FileIndexEvaluator>> FileIndexEvaluator::Create(
const std::shared_ptr<arrow::Schema>& data_schema,
- const std::map<std::string, std::string>& options, const
std::shared_ptr<Predicate>& predicate,
- const std::shared_ptr<DataFilePathFactory>& data_file_path_factory,
+ const std::map<std::string, std::string>& options,
+ const std::shared_ptr<DataFilePathFactory>& path_factory,
const std::shared_ptr<DataFileMeta>& file_meta, const
std::shared_ptr<FileSystem>& file_system,
const std::shared_ptr<MemoryPool>& pool) {
- return Evaluate(/*only_use_embedding_index=*/false, data_schema, options,
predicate,
- data_file_path_factory, file_meta, file_system, pool);
+ std::unique_ptr<FileIndexFormat::Reader> format_reader;
+ if (file_meta->row_count >= 0 && file_meta->row_count <=
RoaringBitmap32::MAX_VALUE) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InputStream> input_stream,
+ ExtractIndexInputStream(path_factory,
file_meta, file_system));
+ if (input_stream) {
+ PAIMON_ASSIGN_OR_RAISE(format_reader,
+ FileIndexFormat::CreateReader(input_stream,
pool, options));
+ }
+ }
+ return std::unique_ptr<FileIndexEvaluator>(
+ new FileIndexEvaluator(data_schema, file_meta,
std::move(format_reader)));
}
-Result<std::shared_ptr<FileIndexResult>> FileIndexEvaluator::Evaluate(
- const std::shared_ptr<arrow::Schema>& data_schema,
- const std::map<std::string, std::string>& options, const
std::shared_ptr<Predicate>& predicate,
- const std::shared_ptr<DataFileMeta>& file_meta, const
std::shared_ptr<MemoryPool>& pool) {
- return Evaluate(/*only_use_embedding_index=*/true, data_schema, options,
predicate,
- /*data_file_path_factory=*/nullptr, file_meta,
/*file_system=*/nullptr, pool);
+Result<std::vector<std::shared_ptr<FileIndexReader>>>
FileIndexEvaluator::GetReaders(
+ const std::string& field_name) {
+ auto iter = field_readers_.find(field_name);
+ if (iter != field_readers_.end()) {
+ return iter->second;
+ }
+ std::shared_ptr<arrow::Field> field =
data_schema_->GetFieldByName(field_name);
+ if (!field) {
+ return Status::Invalid(fmt::format("field '{}' does not exist in data
schema", field_name));
+ }
+ std::vector<std::shared_ptr<FileIndexReader>> readers;
+ if (format_reader_) {
+ ::ArrowSchema c_schema;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(
+ arrow::ExportSchema(*arrow::schema({std::move(field)}),
&c_schema));
+ PAIMON_ASSIGN_OR_RAISE(readers,
format_reader_->ReadColumnIndex(field_name, &c_schema));
+ }
+ field_readers_.emplace(field_name, readers);
+ return readers;
}
-Result<std::shared_ptr<FileIndexResult>> FileIndexEvaluator::Evaluate(
- bool only_use_embedding_index, const std::shared_ptr<arrow::Schema>&
data_schema,
- const std::map<std::string, std::string>& options, const
std::shared_ptr<Predicate>& predicate,
- const std::shared_ptr<DataFilePathFactory>& data_file_path_factory,
- const std::shared_ptr<DataFileMeta>& file_meta, const
std::shared_ptr<FileSystem>& file_system,
- const std::shared_ptr<MemoryPool>& pool) {
- if (predicate == nullptr || file_meta->row_count >
RoaringBitmap32::MAX_VALUE) {
- return FileIndexResult::Remain();
+Result<std::shared_ptr<FileIndexReader>> FileIndexEvaluator::GetSearchReader(
+ const std::string& field_name, FileIndexReader::SearchType search_type) {
+ if (!data_schema_->GetFieldByName(field_name)) {
+ return Status::Invalid(
+ fmt::format("Search field '{}' does not exist in data schema",
field_name));
+ }
+ if (!format_reader_) {
+ return Status::Invalid(fmt::format("Data file '{}' has no File Index
for search field '{}'",
+ file_meta_->file_name, field_name));
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<FileIndexReader>>
readers,
+ GetReaders(field_name));
+ std::shared_ptr<FileIndexReader> selected;
+ for (const auto& reader : readers) {
+ if (!reader->SupportsSearch(search_type)) {
+ continue;
+ }
+ if (selected) {
+ return Status::Invalid(fmt::format(
+ "Multiple File Index readers support requested search for
field '{}'", field_name));
+ }
+ selected = reader;
}
- PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InputStream> input_stream,
- ExtractIndexInputStream(only_use_embedding_index,
data_file_path_factory,
- file_meta, file_system));
- if (!input_stream) {
- // no index files, just return REMAIN
+ if (!selected) {
+ return Status::NotImplemented(fmt::format(
+ "No File Index reader supports requested search for field '{}'",
field_name));
+ }
+ return selected;
+}
+
+Result<std::shared_ptr<FileIndexResult>> FileIndexEvaluator::EvaluateLeaf(
+ const std::shared_ptr<LeafPredicate>& predicate, bool needs_scores) {
+ if (predicate->GetFunction().GetType() == Function::Type::FULL_TEXT_MATCH)
{
+ PAIMON_ASSIGN_OR_RAISE(
+ std::shared_ptr<FileIndexReader> reader,
+ GetSearchReader(predicate->FieldName(),
FileIndexReader::SearchType::FULL_TEXT_MATCH));
+ auto full_text =
std::dynamic_pointer_cast<FullTextMatchPredicate>(predicate);
+ if (!full_text) {
+ return Status::Invalid("FullTextMatch requires a full-text match
predicate");
+ }
+ return reader->VisitFullTextMatch(full_text->GetFullTextMatch(),
needs_scores);
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<FileIndexReader>>
readers,
+ GetReaders(predicate->FieldName()));
+ std::shared_ptr<FileIndexResult> result = FileIndexResult::Remain();
+ for (const auto& reader : readers) {
+ PAIMON_ASSIGN_OR_RAISE(
+ std::shared_ptr<FileIndexResult> child,
+
PredicateUtils::VisitPredicate<std::shared_ptr<FileIndexResult>>(predicate,
reader));
+ PAIMON_ASSIGN_OR_RAISE(result, result->And(child));
+ PAIMON_ASSIGN_OR_RAISE(bool remain, result->IsRemain());
Review Comment:
An empty `IN` or `NOT IN` list can crash here on a Bloom-indexed column. The
Bloom reader inherits both default visitors. With an empty list they return
success with a null pointer, and `Remain()->And(child)` passes it through to
`result->IsRemain()`. I reproduced both null results with this revision. Please
handle the empty list in the default visitors and add a Bloom-backed regression
with `[1, 2, null]`: `IN {}` should keep zero rows and `NOT IN {}` should keep
`1` and `2`.
##########
docs/source/user_guide/file_index.rst:
##########
@@ -45,7 +59,9 @@ positions in that schema. For example, the following
constructs
std::shared_ptr<paimon::Predicate> filter,
paimon::PredicateBuilder::Or({text, price}));
paimon::ReadContextBuilder builder(table_path);
- builder.SetReadFieldNames({"body", "price"}).SetPredicate(filter);
+ builder.EnablePreciseFileIndexEvaluate(true)
+ .SetReadFieldNames({"body", "price"})
+ .SetPredicate(filter);
Review Comment:
Could the example show where to attach this mixed predicate when planning
splits? Passing the same predicate to `ScanContextBuilder::SetPredicate` fails
in `FileStoreScan` as soon as it sees `FullTextMatch`. The working public-entry
test plans the splits first and attaches the full predicate to
`ReadContextBuilder`. Please also mention that the example requires a plugin
offering file-level full-text matching; enabling the global full-text index
alone does not supply that reader.
##########
docs/source/user_guide/file_index.rst:
##########
@@ -21,6 +21,20 @@ File Index
Full-text predicates and root TopK
----------------------------------
+Enable ``ReadContextBuilder::EnablePreciseFileIndexEvaluate(true)`` to
precisely
+filter rows at the file-index-query stage, before reading all columns
requested by
+the query from each data file. It evaluates the complete predicate to compute
+exact matching row positions within that file.
+
+For a root full-text or vector TopK request, predicate filtering, row-range
+restrictions and deletion vectors are applied before the TopK limit. Filtering
after
Review Comment:
Please spell out that `limit` applies independently to each file. With two
matching files and `limit=1`, a caller can receive two rows. Within each file
the reader emits selected rows in physical order, so consumers that need a
table-wide TopK must merge by score. A two-file regression that checks the row
count and score alignment would make this contract explicit.
##########
src/paimon/core/operation/precise_file_index_read.cpp:
##########
@@ -0,0 +1,573 @@
+/*
+ * 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/operation/precise_file_index_read.h"
+
+#include <algorithm>
+#include <cmath>
+#include <map>
+#include <memory>
+#include <optional>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "arrow/c/abi.h"
+#include "arrow/type.h"
+#include "fmt/format.h"
+#include "paimon/common/reader/complete_index_score_file_batch_reader.h"
+#include "paimon/common/reader/predicate_batch_reader.h"
+#include "paimon/common/reader/reader_utils.h"
+#include "paimon/common/types/data_field.h"
+#include "paimon/common/utils/object_utils.h"
+#include "paimon/common/utils/scope_guard.h"
+#include "paimon/core/deletionvectors/bitmap64_deletion_vector.h"
+#include "paimon/core/deletionvectors/bitmap_deletion_vector.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/io/data_file_path_factory.h"
+#include "paimon/core/io/file_index_evaluator.h"
+#include "paimon/core/io/file_index_options.h"
+#include "paimon/core/operation/internal_read_context.h"
+#include "paimon/core/operation/raw_file_split_read.h"
+#include "paimon/core/schema/schema_manager.h"
+#include "paimon/core/schema/table_schema.h"
+#include "paimon/core/utils/field_mapping.h"
+#include "paimon/file_index/bitmap_index_result.h"
+#include "paimon/file_index/file_index_result.h"
+#include "paimon/file_index/scored_file_index_result.h"
+#include "paimon/format/reader_builder.h"
+#include "paimon/predicate/compound_predicate.h"
+#include "paimon/predicate/full_text_search.h"
+#include "paimon/predicate/function.h"
+#include "paimon/predicate/leaf_predicate.h"
+#include "paimon/predicate/vector_search.h"
+#include "paimon/utils/roaring_bitmap64.h"
+
+namespace paimon {
+namespace {
+
+Result<std::string> ResolveSearchFieldName(
+ const std::string& field_name, const std::map<std::string, std::string>&
field_name_mapping,
+ const std::string& file_name) {
+ auto iter = field_name_mapping.find(field_name);
+ if (iter == field_name_mapping.end()) {
+ return Status::Invalid(
+ fmt::format("Search field '{}' has no matching field ID in data
schema for file '{}'",
+ field_name, file_name));
+ }
+ return iter->second;
+}
+
+Status PreparePredicate(
+ const std::shared_ptr<Predicate>& predicate,
+ const std::map<std::string, DataField>& read_by_name,
+ const std::map<int32_t, std::pair<int32_t, DataField>>& data_by_id,
+ const std::vector<std::string>& partition_keys, FileIndexEvaluator*
indexes,
+ std::map<std::shared_ptr<LeafPredicate>, std::shared_ptr<LeafPredicate>>*
mapped_leaves) {
+ if (!predicate) {
+ return Status::OK();
+ }
+ if (auto compound =
std::dynamic_pointer_cast<CompoundPredicate>(predicate)) {
+ const Function::Type type = compound->GetFunction().GetType();
+ if (type != Function::Type::AND && type != Function::Type::OR) {
+ return Status::Invalid(
+ fmt::format("unsupported compound file predicate {}",
predicate->ToString()));
+ }
+ for (const auto& child : compound->Children()) {
+ PAIMON_RETURN_NOT_OK(PreparePredicate(child, read_by_name,
data_by_id, partition_keys,
+ indexes, mapped_leaves));
+ }
+ return Status::OK();
+ }
+ auto leaf = std::dynamic_pointer_cast<LeafPredicate>(predicate);
+ if (!leaf) {
+ return Status::Invalid("File predicate must contain leaf or compound
predicates");
+ }
+ bool full_text = leaf->GetFunction().GetType() ==
Function::Type::FULL_TEXT_MATCH;
+ if (!full_text && ObjectUtils::Contains(partition_keys,
leaf->FieldName())) {
+ mapped_leaves->emplace(leaf, nullptr);
+ return Status::OK();
+ }
+ PAIMON_ASSIGN_OR_RAISE(
+ std::optional<std::shared_ptr<Predicate>> mapped,
+ FieldMappingBuilder::ReconstructPredicateWithDataFields(leaf,
read_by_name, data_by_id));
+ std::shared_ptr<LeafPredicate> mapped_leaf;
+ if (mapped) {
+ mapped_leaf = std::dynamic_pointer_cast<LeafPredicate>(*mapped);
+ if (!mapped_leaf) {
+ return Status::Invalid("Mapped file predicate must be a leaf");
+ }
+ if (full_text) {
+ PAIMON_ASSIGN_OR_RAISE(
+ [[maybe_unused]] std::shared_ptr<FileIndexReader> reader,
+ indexes->GetSearchReader(mapped_leaf->FieldName(),
+
FileIndexReader::SearchType::FULL_TEXT_MATCH));
+ }
+ }
+ mapped_leaves->emplace(leaf, std::move(mapped_leaf));
+ return Status::OK();
+}
+
+Status PrepareSearch(const std::shared_ptr<InternalReadContext>& context,
+ const std::map<std::string, std::string>& field_names,
+ const std::string& file_name, FileIndexEvaluator* indexes,
+ std::shared_ptr<VectorSearch>* vector_search,
+ std::shared_ptr<FullTextSearch>* full_text_search) {
+ if (context->GetVectorSearch()) {
+ *vector_search =
std::make_shared<VectorSearch>(*context->GetVectorSearch());
+ PAIMON_ASSIGN_OR_RAISE(
+ (*vector_search)->field_name,
+ ResolveSearchFieldName((*vector_search)->field_name, field_names,
file_name));
+ PAIMON_ASSIGN_OR_RAISE(
+ [[maybe_unused]] std::shared_ptr<FileIndexReader> reader,
+ indexes->GetSearchReader((*vector_search)->field_name,
+
FileIndexReader::SearchType::VECTOR_SEARCH));
+ }
+ if (context->GetFullTextSearch()) {
+ *full_text_search =
std::make_shared<FullTextSearch>(*context->GetFullTextSearch());
+ PAIMON_ASSIGN_OR_RAISE(
+ (*full_text_search)->field_name,
+ ResolveSearchFieldName((*full_text_search)->field_name,
field_names, file_name));
+ PAIMON_ASSIGN_OR_RAISE(
+ [[maybe_unused]] std::shared_ptr<FileIndexReader> reader,
+ indexes->GetSearchReader((*full_text_search)->field_name,
+
FileIndexReader::SearchType::FULL_TEXT_SEARCH));
+ }
+ return Status::OK();
+}
+
+} // namespace
+
+Result<std::unique_ptr<PreciseFileIndexRead>> PreciseFileIndexRead::Create(
+ const RawFileSplitRead& split_read, const BinaryRow& partition,
+ const std::shared_ptr<DataFileMeta>& file, const FieldMappingBuilder*
field_mapping_builder,
+ const std::shared_ptr<DataFilePathFactory>& path_factory,
+ const std::map<std::string, std::string>& extra_format_options) {
+ std::shared_ptr<TableSchema> table_schema =
split_read.context_->GetTableSchema();
+ if (file->schema_id != table_schema->Id()) {
+ PAIMON_ASSIGN_OR_RAISE(table_schema,
+
split_read.schema_manager_->ReadSchema(file->schema_id));
+ }
+ const auto data_schema =
DataField::ConvertDataFieldsToArrowSchema(table_schema->Fields());
+ if (file->row_count < 0 || file->row_count > RoaringBitmap32::MAX_VALUE) {
+ return Status::NotImplemented(
+ "Precise file-index evaluation requires bitmap32 physical row
positions");
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::vector<DataField> data_fields,
+
DataField::ConvertArrowSchemaToDataFields(data_schema));
+ PAIMON_ASSIGN_OR_RAISE(
+ std::vector<DataField> read_fields,
+
DataField::ConvertArrowSchemaToDataFields(split_read.context_->GetReadSchema()));
+ auto field_names = FieldMappingBuilder::CreateFieldNameMapping(
+ split_read.context_->GetTableSchema()->Fields(), data_fields);
+ std::map<std::string, std::string> index_options;
+ PAIMON_ASSIGN_OR_RAISE(index_options,
+ FileIndexOptions::RewriteFieldNames(
+ split_read.context_->GetCoreOptions().ToMap(),
field_names));
+ PAIMON_ASSIGN_OR_RAISE(
+ std::unique_ptr<FileIndexEvaluator> indexes,
+ FileIndexEvaluator::Create(data_schema, index_options, path_factory,
file,
+
split_read.context_->GetCoreOptions().GetFileSystem(),
+ split_read.pool_));
+
+ std::map<std::string, DataField> read_by_name;
+ for (const auto& field : read_fields) {
+ read_by_name.emplace(field.Name(), field);
+ }
+ // Field ID -> (zero-based position in data_fields, field definition from
the file's schema).
+ std::map<int32_t, std::pair<int32_t, DataField>> data_by_id;
+ for (size_t i = 0; i < data_fields.size(); ++i) {
+ data_by_id.emplace(data_fields[i].Id(), std::make_pair(i,
data_fields[i]));
+ }
+ std::map<std::shared_ptr<LeafPredicate>, std::shared_ptr<LeafPredicate>>
mapped_leaves;
+ // Precheck the entire tree before an empty candidate set can
short-circuit it.
+ PAIMON_RETURN_NOT_OK(PreparePredicate(split_read.context_->GetPredicate(),
read_by_name,
+ data_by_id,
split_read.context_->GetPartitionKeys(),
+ indexes.get(), &mapped_leaves));
+ std::shared_ptr<VectorSearch> vector_search;
+ std::shared_ptr<FullTextSearch> full_text_search;
+ PAIMON_RETURN_NOT_OK(PrepareSearch(split_read.context_, field_names,
file->file_name,
+ indexes.get(), &vector_search,
&full_text_search));
+ return std::unique_ptr<PreciseFileIndexRead>(new PreciseFileIndexRead(
+ split_read, partition, file, field_mapping_builder, path_factory,
extra_format_options,
+ std::move(indexes), mapped_leaves, vector_search, full_text_search));
+}
+
+PreciseFileIndexRead::PreciseFileIndexRead(
+ const RawFileSplitRead& split_read, const BinaryRow& partition,
+ const std::shared_ptr<DataFileMeta>& file, const FieldMappingBuilder*
field_mapping_builder,
+ const std::shared_ptr<DataFilePathFactory>& path_factory,
+ const std::map<std::string, std::string>& extra_format_options,
+ std::unique_ptr<FileIndexEvaluator> indexes,
+ const std::map<std::shared_ptr<LeafPredicate>,
std::shared_ptr<LeafPredicate>>& mapped_leaves,
+ const std::shared_ptr<VectorSearch>& vector_search,
+ const std::shared_ptr<FullTextSearch>& full_text_search)
+ : split_read_(split_read),
+ partition_(partition),
+ file_(file),
+ field_mapping_builder_(field_mapping_builder),
+ path_factory_(path_factory),
+ extra_format_options_(extra_format_options),
+ indexes_(std::move(indexes)),
+ mapped_leaves_(mapped_leaves),
+ vector_search_(vector_search),
+ full_text_search_(full_text_search) {}
+
+Result<RoaringBitmap32> PreciseFileIndexRead::CreateCandidates(
+ const DeletionVector::Factory& dv_factory,
+ const std::optional<std::vector<Range>>& row_ranges) const {
+ RoaringBitmap32 candidates;
+ if (row_ranges) {
+ for (const auto& range : *row_ranges) {
+ candidates.AddRange(static_cast<int32_t>(range.from),
+ static_cast<int32_t>(range.to + 1));
+ }
+ } else {
+ candidates.AddRange(0, static_cast<int32_t>(file_->row_count));
+ }
+ if (dv_factory) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<DeletionVector> dv,
dv_factory(file_->file_name));
+ if (auto bitmap = dynamic_cast<BitmapDeletionVector*>(dv.get())) {
+ candidates -= *bitmap->GetBitmap();
+ } else if (dv) {
+ return Status::NotImplemented(
+ "Precise file-index evaluation requires bitmap32 deletion
vectors");
+ }
+ }
+ return candidates;
+}
+
+Result<PreciseFileIndexRead::Evaluation> PreciseFileIndexRead::Evaluate(
+ const std::shared_ptr<Predicate>& predicate, const RoaringBitmap32&
candidates) const {
+ if (!predicate || candidates.IsEmpty()) {
+ return Evaluation{candidates, {}};
+ }
+ if (auto leaf = std::dynamic_pointer_cast<LeafPredicate>(predicate)) {
+ return EvaluateLeaf(leaf, candidates);
+ }
+ if (auto compound =
std::dynamic_pointer_cast<CompoundPredicate>(predicate)) {
+ return EvaluateCompound(compound, candidates);
+ }
+ return Status::Invalid(fmt::format("unsupported file predicate {}",
predicate->ToString()));
+}
+
+Result<PreciseFileIndexRead::Evaluation>
PreciseFileIndexRead::EvaluateCompound(
+ const std::shared_ptr<CompoundPredicate>& predicate, const
RoaringBitmap32& candidates) const {
+ const Function::Type type = predicate->GetFunction().GetType();
+ if (type == Function::Type::OR) {
+ Evaluation result;
+ for (const auto& child : predicate->Children()) {
+ // Every branch must see the same input, including rows matched by
earlier branches.
+ PAIMON_ASSIGN_OR_RAISE(Evaluation child_result, Evaluate(child,
candidates));
+ result.rows |= child_result.rows;
+ result.scores.insert(child_result.scores.begin(),
child_result.scores.end());
+ }
+ return result;
+ }
+ if (type != Function::Type::AND) {
+ return Status::Invalid(
+ fmt::format("unsupported compound file predicate {}",
predicate->ToString()));
+ }
+
+ struct Child {
+ std::shared_ptr<Predicate> predicate;
+ std::optional<double> selectivity;
+ };
+ std::vector<Child> children;
+ children.reserve(predicate->Children().size());
+ for (const auto& child : predicate->Children()) {
+ PAIMON_ASSIGN_OR_RAISE(std::optional<double> selectivity,
EstimateSelectivity(child));
+ children.push_back({child, selectivity});
+ }
+ std::stable_sort(children.begin(), children.end(), [](const Child& lhs,
const Child& rhs) {
+ if (!lhs.selectivity) {
+ return false;
+ }
+ return !rhs.selectivity || *lhs.selectivity < *rhs.selectivity;
+ });
+
+ Evaluation result{candidates, {}};
+ for (const auto& child : children) {
+ PAIMON_ASSIGN_OR_RAISE(Evaluation child_result,
Evaluate(child.predicate, result.rows));
+ result.rows = std::move(child_result.rows);
+ result.scores.insert(child_result.scores.begin(),
child_result.scores.end());
+ if (result.rows.IsEmpty()) {
+ break;
+ }
+ }
+ return result;
+}
+
+Result<std::optional<double>> PreciseFileIndexRead::EstimateSelectivity(
+ const std::shared_ptr<Predicate>& predicate) const {
+ if (!predicate) {
+ return std::optional<double>();
+ }
+ if (auto leaf = std::dynamic_pointer_cast<LeafPredicate>(predicate)) {
+ PAIMON_ASSIGN_OR_RAISE(std::optional<double> estimate,
EstimateLeafSelectivity(leaf));
+ if (estimate && (!std::isfinite(*estimate) || *estimate < 0.0 ||
*estimate > 1.0)) {
+ return std::optional<double>();
+ }
+ return estimate;
+ }
+ auto compound = std::dynamic_pointer_cast<CompoundPredicate>(predicate);
+ if (!compound) {
+ return Status::Invalid(fmt::format("unsupported file predicate {}",
predicate->ToString()));
+ }
+ const Function::Type type = compound->GetFunction().GetType();
+ if (type != Function::Type::AND && type != Function::Type::OR) {
+ return Status::Invalid(
+ fmt::format("unsupported compound file predicate {}",
predicate->ToString()));
+ }
+ std::optional<double> estimate =
+ type == Function::Type::AND ? std::nullopt : std::make_optional(0.0);
+ // These are scheduling hints. Continue visiting all children to propagate
errors.
+ for (const auto& child : compound->Children()) {
+ PAIMON_ASSIGN_OR_RAISE(std::optional<double> child_estimate,
EstimateSelectivity(child));
+ if (type == Function::Type::AND) {
+ if (child_estimate) {
+ if (estimate.has_value()) {
+ estimate = std::min(*estimate, *child_estimate);
+ } else {
+ estimate = *child_estimate;
+ }
+ }
+ } else if (estimate && child_estimate) {
+ estimate = std::min(1.0, *estimate + *child_estimate);
+ } else {
+ estimate = std::nullopt;
+ }
+ }
+ return estimate;
+}
+
+Result<std::optional<double>> PreciseFileIndexRead::EstimateLeafSelectivity(
+ const std::shared_ptr<LeafPredicate>& predicate) const {
+ auto iter = mapped_leaves_.find(predicate);
+ if (iter == mapped_leaves_.end()) {
+ return Status::Invalid("File predicate mapping was not prepared");
+ }
+ if (!iter->second) {
+ return std::optional<double>();
+ }
+ return indexes_->EstimateSelectivity(iter->second);
+}
+
+Result<PreciseFileIndexRead::Evaluation> PreciseFileIndexRead::EvaluateLeaf(
+ const std::shared_ptr<LeafPredicate>& predicate, const RoaringBitmap32&
candidates) const {
+ bool is_full_text = predicate->GetFunction().GetType() ==
Function::Type::FULL_TEXT_MATCH;
+ bool needs_scores =
+ is_full_text &&
+ split_read_.context_->NeedsScores(
+ {InternalReadContext::IndexScoreSource::Type::FULL_TEXT_MATCH,
predicate->FieldName()});
+ std::shared_ptr<FileIndexResult> index_result = FileIndexResult::Remain();
+ auto iter = mapped_leaves_.find(predicate);
+ if (iter == mapped_leaves_.end()) {
+ return Status::Invalid("File predicate mapping was not prepared");
+ }
+ if (iter->second) {
+ // TODO(lisizhuo.lsz): Pass candidates through
FileIndexEvaluator::EvaluateLeaf to index
+ // readers so they can skip rows excluded by earlier predicates, row
ranges, or deletion
+ // vectors.
+ PAIMON_ASSIGN_OR_RAISE(index_result,
indexes_->EvaluateLeaf(iter->second, needs_scores));
+ }
+ if (!index_result) {
+ return Status::Invalid("file predicate index returned a null result");
+ }
+ Evaluation result{candidates, {}};
+ if (needs_scores) {
+ auto scored =
std::dynamic_pointer_cast<ScoredFileIndexResult>(index_result);
+ if (!scored) {
+ return Status::Invalid(
+ fmt::format("full-text index for field '{}' did not return
requested scores",
+ predicate->FieldName()));
+ }
+ result.rows &= scored->GetRowPositions();
+ result.scores.emplace(predicate->FieldName(), std::move(scored));
+ return result;
+ }
+
+ if (auto bitmap =
std::dynamic_pointer_cast<BitmapIndexResult>(index_result)) {
+ PAIMON_ASSIGN_OR_RAISE(const RoaringBitmap32* rows,
bitmap->GetBitmap());
+ result.rows &= *rows;
Review Comment:
This intersection can discard a true match before `ScanScalar` can recheck
it. For a millisecond timestamp column, the row `1000 ms` satisfies `ts <
1000.5 ms`. `RangeBitmapTypeAdapter` converts both operands to the key `1000`,
so the index tests `1000 < 1000` and removes the row. I reproduced the key
conversion on this revision. Please bypass that index when the bound loses
precision, or adjust the bound for the comparison operator. A regression
comparing indexed and unindexed reads at this boundary would also protect TopK
from ranking the wrong rows.
##########
src/paimon/core/operation/precise_file_index_read.cpp:
##########
@@ -0,0 +1,573 @@
+/*
+ * 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/operation/precise_file_index_read.h"
+
+#include <algorithm>
+#include <cmath>
+#include <map>
+#include <memory>
+#include <optional>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "arrow/c/abi.h"
+#include "arrow/type.h"
+#include "fmt/format.h"
+#include "paimon/common/reader/complete_index_score_file_batch_reader.h"
+#include "paimon/common/reader/predicate_batch_reader.h"
+#include "paimon/common/reader/reader_utils.h"
+#include "paimon/common/types/data_field.h"
+#include "paimon/common/utils/object_utils.h"
+#include "paimon/common/utils/scope_guard.h"
+#include "paimon/core/deletionvectors/bitmap64_deletion_vector.h"
+#include "paimon/core/deletionvectors/bitmap_deletion_vector.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/io/data_file_path_factory.h"
+#include "paimon/core/io/file_index_evaluator.h"
+#include "paimon/core/io/file_index_options.h"
+#include "paimon/core/operation/internal_read_context.h"
+#include "paimon/core/operation/raw_file_split_read.h"
+#include "paimon/core/schema/schema_manager.h"
+#include "paimon/core/schema/table_schema.h"
+#include "paimon/core/utils/field_mapping.h"
+#include "paimon/file_index/bitmap_index_result.h"
+#include "paimon/file_index/file_index_result.h"
+#include "paimon/file_index/scored_file_index_result.h"
+#include "paimon/format/reader_builder.h"
+#include "paimon/predicate/compound_predicate.h"
+#include "paimon/predicate/full_text_search.h"
+#include "paimon/predicate/function.h"
+#include "paimon/predicate/leaf_predicate.h"
+#include "paimon/predicate/vector_search.h"
+#include "paimon/utils/roaring_bitmap64.h"
+
+namespace paimon {
+namespace {
+
+Result<std::string> ResolveSearchFieldName(
+ const std::string& field_name, const std::map<std::string, std::string>&
field_name_mapping,
+ const std::string& file_name) {
+ auto iter = field_name_mapping.find(field_name);
+ if (iter == field_name_mapping.end()) {
+ return Status::Invalid(
+ fmt::format("Search field '{}' has no matching field ID in data
schema for file '{}'",
+ field_name, file_name));
+ }
+ return iter->second;
+}
+
+Status PreparePredicate(
+ const std::shared_ptr<Predicate>& predicate,
+ const std::map<std::string, DataField>& read_by_name,
+ const std::map<int32_t, std::pair<int32_t, DataField>>& data_by_id,
+ const std::vector<std::string>& partition_keys, FileIndexEvaluator*
indexes,
+ std::map<std::shared_ptr<LeafPredicate>, std::shared_ptr<LeafPredicate>>*
mapped_leaves) {
+ if (!predicate) {
+ return Status::OK();
+ }
+ if (auto compound =
std::dynamic_pointer_cast<CompoundPredicate>(predicate)) {
+ const Function::Type type = compound->GetFunction().GetType();
+ if (type != Function::Type::AND && type != Function::Type::OR) {
+ return Status::Invalid(
+ fmt::format("unsupported compound file predicate {}",
predicate->ToString()));
+ }
+ for (const auto& child : compound->Children()) {
+ PAIMON_RETURN_NOT_OK(PreparePredicate(child, read_by_name,
data_by_id, partition_keys,
+ indexes, mapped_leaves));
+ }
+ return Status::OK();
+ }
+ auto leaf = std::dynamic_pointer_cast<LeafPredicate>(predicate);
+ if (!leaf) {
+ return Status::Invalid("File predicate must contain leaf or compound
predicates");
+ }
+ bool full_text = leaf->GetFunction().GetType() ==
Function::Type::FULL_TEXT_MATCH;
+ if (!full_text && ObjectUtils::Contains(partition_keys,
leaf->FieldName())) {
+ mapped_leaves->emplace(leaf, nullptr);
+ return Status::OK();
+ }
+ PAIMON_ASSIGN_OR_RAISE(
+ std::optional<std::shared_ptr<Predicate>> mapped,
+ FieldMappingBuilder::ReconstructPredicateWithDataFields(leaf,
read_by_name, data_by_id));
+ std::shared_ptr<LeafPredicate> mapped_leaf;
+ if (mapped) {
+ mapped_leaf = std::dynamic_pointer_cast<LeafPredicate>(*mapped);
+ if (!mapped_leaf) {
+ return Status::Invalid("Mapped file predicate must be a leaf");
+ }
+ if (full_text) {
+ PAIMON_ASSIGN_OR_RAISE(
+ [[maybe_unused]] std::shared_ptr<FileIndexReader> reader,
+ indexes->GetSearchReader(mapped_leaf->FieldName(),
+
FileIndexReader::SearchType::FULL_TEXT_MATCH));
+ }
+ }
+ mapped_leaves->emplace(leaf, std::move(mapped_leaf));
+ return Status::OK();
+}
+
+Status PrepareSearch(const std::shared_ptr<InternalReadContext>& context,
+ const std::map<std::string, std::string>& field_names,
+ const std::string& file_name, FileIndexEvaluator* indexes,
+ std::shared_ptr<VectorSearch>* vector_search,
+ std::shared_ptr<FullTextSearch>* full_text_search) {
+ if (context->GetVectorSearch()) {
+ *vector_search =
std::make_shared<VectorSearch>(*context->GetVectorSearch());
+ PAIMON_ASSIGN_OR_RAISE(
+ (*vector_search)->field_name,
+ ResolveSearchFieldName((*vector_search)->field_name, field_names,
file_name));
+ PAIMON_ASSIGN_OR_RAISE(
+ [[maybe_unused]] std::shared_ptr<FileIndexReader> reader,
+ indexes->GetSearchReader((*vector_search)->field_name,
+
FileIndexReader::SearchType::VECTOR_SEARCH));
+ }
+ if (context->GetFullTextSearch()) {
+ *full_text_search =
std::make_shared<FullTextSearch>(*context->GetFullTextSearch());
+ PAIMON_ASSIGN_OR_RAISE(
+ (*full_text_search)->field_name,
+ ResolveSearchFieldName((*full_text_search)->field_name,
field_names, file_name));
+ PAIMON_ASSIGN_OR_RAISE(
+ [[maybe_unused]] std::shared_ptr<FileIndexReader> reader,
+ indexes->GetSearchReader((*full_text_search)->field_name,
+
FileIndexReader::SearchType::FULL_TEXT_SEARCH));
+ }
+ return Status::OK();
+}
+
+} // namespace
+
+Result<std::unique_ptr<PreciseFileIndexRead>> PreciseFileIndexRead::Create(
+ const RawFileSplitRead& split_read, const BinaryRow& partition,
+ const std::shared_ptr<DataFileMeta>& file, const FieldMappingBuilder*
field_mapping_builder,
+ const std::shared_ptr<DataFilePathFactory>& path_factory,
+ const std::map<std::string, std::string>& extra_format_options) {
+ std::shared_ptr<TableSchema> table_schema =
split_read.context_->GetTableSchema();
+ if (file->schema_id != table_schema->Id()) {
+ PAIMON_ASSIGN_OR_RAISE(table_schema,
+
split_read.schema_manager_->ReadSchema(file->schema_id));
+ }
+ const auto data_schema =
DataField::ConvertDataFieldsToArrowSchema(table_schema->Fields());
+ if (file->row_count < 0 || file->row_count > RoaringBitmap32::MAX_VALUE) {
+ return Status::NotImplemented(
+ "Precise file-index evaluation requires bitmap32 physical row
positions");
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::vector<DataField> data_fields,
+
DataField::ConvertArrowSchemaToDataFields(data_schema));
+ PAIMON_ASSIGN_OR_RAISE(
+ std::vector<DataField> read_fields,
+
DataField::ConvertArrowSchemaToDataFields(split_read.context_->GetReadSchema()));
+ auto field_names = FieldMappingBuilder::CreateFieldNameMapping(
+ split_read.context_->GetTableSchema()->Fields(), data_fields);
+ std::map<std::string, std::string> index_options;
+ PAIMON_ASSIGN_OR_RAISE(index_options,
+ FileIndexOptions::RewriteFieldNames(
+ split_read.context_->GetCoreOptions().ToMap(),
field_names));
+ PAIMON_ASSIGN_OR_RAISE(
+ std::unique_ptr<FileIndexEvaluator> indexes,
+ FileIndexEvaluator::Create(data_schema, index_options, path_factory,
file,
+
split_read.context_->GetCoreOptions().GetFileSystem(),
+ split_read.pool_));
+
+ std::map<std::string, DataField> read_by_name;
+ for (const auto& field : read_fields) {
+ read_by_name.emplace(field.Name(), field);
+ }
+ // Field ID -> (zero-based position in data_fields, field definition from
the file's schema).
+ std::map<int32_t, std::pair<int32_t, DataField>> data_by_id;
+ for (size_t i = 0; i < data_fields.size(); ++i) {
+ data_by_id.emplace(data_fields[i].Id(), std::make_pair(i,
data_fields[i]));
+ }
+ std::map<std::shared_ptr<LeafPredicate>, std::shared_ptr<LeafPredicate>>
mapped_leaves;
+ // Precheck the entire tree before an empty candidate set can
short-circuit it.
+ PAIMON_RETURN_NOT_OK(PreparePredicate(split_read.context_->GetPredicate(),
read_by_name,
+ data_by_id,
split_read.context_->GetPartitionKeys(),
+ indexes.get(), &mapped_leaves));
+ std::shared_ptr<VectorSearch> vector_search;
+ std::shared_ptr<FullTextSearch> full_text_search;
+ PAIMON_RETURN_NOT_OK(PrepareSearch(split_read.context_, field_names,
file->file_name,
+ indexes.get(), &vector_search,
&full_text_search));
+ return std::unique_ptr<PreciseFileIndexRead>(new PreciseFileIndexRead(
+ split_read, partition, file, field_mapping_builder, path_factory,
extra_format_options,
+ std::move(indexes), mapped_leaves, vector_search, full_text_search));
+}
+
+PreciseFileIndexRead::PreciseFileIndexRead(
+ const RawFileSplitRead& split_read, const BinaryRow& partition,
+ const std::shared_ptr<DataFileMeta>& file, const FieldMappingBuilder*
field_mapping_builder,
+ const std::shared_ptr<DataFilePathFactory>& path_factory,
+ const std::map<std::string, std::string>& extra_format_options,
+ std::unique_ptr<FileIndexEvaluator> indexes,
+ const std::map<std::shared_ptr<LeafPredicate>,
std::shared_ptr<LeafPredicate>>& mapped_leaves,
+ const std::shared_ptr<VectorSearch>& vector_search,
+ const std::shared_ptr<FullTextSearch>& full_text_search)
+ : split_read_(split_read),
+ partition_(partition),
+ file_(file),
+ field_mapping_builder_(field_mapping_builder),
+ path_factory_(path_factory),
+ extra_format_options_(extra_format_options),
+ indexes_(std::move(indexes)),
+ mapped_leaves_(mapped_leaves),
+ vector_search_(vector_search),
+ full_text_search_(full_text_search) {}
+
+Result<RoaringBitmap32> PreciseFileIndexRead::CreateCandidates(
+ const DeletionVector::Factory& dv_factory,
+ const std::optional<std::vector<Range>>& row_ranges) const {
+ RoaringBitmap32 candidates;
+ if (row_ranges) {
+ for (const auto& range : *row_ranges) {
+ candidates.AddRange(static_cast<int32_t>(range.from),
+ static_cast<int32_t>(range.to + 1));
+ }
+ } else {
+ candidates.AddRange(0, static_cast<int32_t>(file_->row_count));
+ }
+ if (dv_factory) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<DeletionVector> dv,
dv_factory(file_->file_name));
+ if (auto bitmap = dynamic_cast<BitmapDeletionVector*>(dv.get())) {
+ candidates -= *bitmap->GetBitmap();
+ } else if (dv) {
+ return Status::NotImplemented(
+ "Precise file-index evaluation requires bitmap32 deletion
vectors");
+ }
+ }
+ return candidates;
+}
+
+Result<PreciseFileIndexRead::Evaluation> PreciseFileIndexRead::Evaluate(
+ const std::shared_ptr<Predicate>& predicate, const RoaringBitmap32&
candidates) const {
+ if (!predicate || candidates.IsEmpty()) {
+ return Evaluation{candidates, {}};
+ }
+ if (auto leaf = std::dynamic_pointer_cast<LeafPredicate>(predicate)) {
+ return EvaluateLeaf(leaf, candidates);
+ }
+ if (auto compound =
std::dynamic_pointer_cast<CompoundPredicate>(predicate)) {
+ return EvaluateCompound(compound, candidates);
+ }
+ return Status::Invalid(fmt::format("unsupported file predicate {}",
predicate->ToString()));
+}
+
+Result<PreciseFileIndexRead::Evaluation>
PreciseFileIndexRead::EvaluateCompound(
+ const std::shared_ptr<CompoundPredicate>& predicate, const
RoaringBitmap32& candidates) const {
+ const Function::Type type = predicate->GetFunction().GetType();
+ if (type == Function::Type::OR) {
+ Evaluation result;
+ for (const auto& child : predicate->Children()) {
+ // Every branch must see the same input, including rows matched by
earlier branches.
+ PAIMON_ASSIGN_OR_RAISE(Evaluation child_result, Evaluate(child,
candidates));
+ result.rows |= child_result.rows;
+ result.scores.insert(child_result.scores.begin(),
child_result.scores.end());
+ }
+ return result;
+ }
+ if (type != Function::Type::AND) {
+ return Status::Invalid(
+ fmt::format("unsupported compound file predicate {}",
predicate->ToString()));
+ }
+
+ struct Child {
+ std::shared_ptr<Predicate> predicate;
+ std::optional<double> selectivity;
+ };
+ std::vector<Child> children;
+ children.reserve(predicate->Children().size());
+ for (const auto& child : predicate->Children()) {
+ PAIMON_ASSIGN_OR_RAISE(std::optional<double> selectivity,
EstimateSelectivity(child));
+ children.push_back({child, selectivity});
+ }
+ std::stable_sort(children.begin(), children.end(), [](const Child& lhs,
const Child& rhs) {
+ if (!lhs.selectivity) {
+ return false;
+ }
+ return !rhs.selectivity || *lhs.selectivity < *rhs.selectivity;
+ });
+
+ Evaluation result{candidates, {}};
+ for (const auto& child : children) {
+ PAIMON_ASSIGN_OR_RAISE(Evaluation child_result,
Evaluate(child.predicate, result.rows));
+ result.rows = std::move(child_result.rows);
+ result.scores.insert(child_result.scores.begin(),
child_result.scores.end());
+ if (result.rows.IsEmpty()) {
+ break;
+ }
+ }
+ return result;
+}
+
+Result<std::optional<double>> PreciseFileIndexRead::EstimateSelectivity(
+ const std::shared_ptr<Predicate>& predicate) const {
+ if (!predicate) {
+ return std::optional<double>();
+ }
+ if (auto leaf = std::dynamic_pointer_cast<LeafPredicate>(predicate)) {
+ PAIMON_ASSIGN_OR_RAISE(std::optional<double> estimate,
EstimateLeafSelectivity(leaf));
+ if (estimate && (!std::isfinite(*estimate) || *estimate < 0.0 ||
*estimate > 1.0)) {
+ return std::optional<double>();
+ }
+ return estimate;
+ }
+ auto compound = std::dynamic_pointer_cast<CompoundPredicate>(predicate);
+ if (!compound) {
+ return Status::Invalid(fmt::format("unsupported file predicate {}",
predicate->ToString()));
+ }
+ const Function::Type type = compound->GetFunction().GetType();
+ if (type != Function::Type::AND && type != Function::Type::OR) {
+ return Status::Invalid(
+ fmt::format("unsupported compound file predicate {}",
predicate->ToString()));
+ }
+ std::optional<double> estimate =
+ type == Function::Type::AND ? std::nullopt : std::make_optional(0.0);
+ // These are scheduling hints. Continue visiting all children to propagate
errors.
+ for (const auto& child : compound->Children()) {
+ PAIMON_ASSIGN_OR_RAISE(std::optional<double> child_estimate,
EstimateSelectivity(child));
+ if (type == Function::Type::AND) {
+ if (child_estimate) {
+ if (estimate.has_value()) {
+ estimate = std::min(*estimate, *child_estimate);
+ } else {
+ estimate = *child_estimate;
+ }
+ }
+ } else if (estimate && child_estimate) {
+ estimate = std::min(1.0, *estimate + *child_estimate);
+ } else {
+ estimate = std::nullopt;
+ }
+ }
+ return estimate;
+}
+
+Result<std::optional<double>> PreciseFileIndexRead::EstimateLeafSelectivity(
+ const std::shared_ptr<LeafPredicate>& predicate) const {
+ auto iter = mapped_leaves_.find(predicate);
+ if (iter == mapped_leaves_.end()) {
+ return Status::Invalid("File predicate mapping was not prepared");
+ }
+ if (!iter->second) {
+ return std::optional<double>();
+ }
+ return indexes_->EstimateSelectivity(iter->second);
+}
+
+Result<PreciseFileIndexRead::Evaluation> PreciseFileIndexRead::EvaluateLeaf(
+ const std::shared_ptr<LeafPredicate>& predicate, const RoaringBitmap32&
candidates) const {
+ bool is_full_text = predicate->GetFunction().GetType() ==
Function::Type::FULL_TEXT_MATCH;
+ bool needs_scores =
+ is_full_text &&
+ split_read_.context_->NeedsScores(
+ {InternalReadContext::IndexScoreSource::Type::FULL_TEXT_MATCH,
predicate->FieldName()});
+ std::shared_ptr<FileIndexResult> index_result = FileIndexResult::Remain();
+ auto iter = mapped_leaves_.find(predicate);
+ if (iter == mapped_leaves_.end()) {
+ return Status::Invalid("File predicate mapping was not prepared");
+ }
+ if (iter->second) {
+ // TODO(lisizhuo.lsz): Pass candidates through
FileIndexEvaluator::EvaluateLeaf to index
+ // readers so they can skip rows excluded by earlier predicates, row
ranges, or deletion
+ // vectors.
+ PAIMON_ASSIGN_OR_RAISE(index_result,
indexes_->EvaluateLeaf(iter->second, needs_scores));
+ }
+ if (!index_result) {
+ return Status::Invalid("file predicate index returned a null result");
+ }
+ Evaluation result{candidates, {}};
+ if (needs_scores) {
+ auto scored =
std::dynamic_pointer_cast<ScoredFileIndexResult>(index_result);
+ if (!scored) {
+ return Status::Invalid(
+ fmt::format("full-text index for field '{}' did not return
requested scores",
+ predicate->FieldName()));
+ }
+ result.rows &= scored->GetRowPositions();
+ result.scores.emplace(predicate->FieldName(), std::move(scored));
+ return result;
+ }
+
+ if (auto bitmap =
std::dynamic_pointer_cast<BitmapIndexResult>(index_result)) {
+ PAIMON_ASSIGN_OR_RAISE(const RoaringBitmap32* rows,
bitmap->GetBitmap());
+ result.rows &= *rows;
+ } else if (is_full_text) {
+ return Status::Invalid(fmt::format("full-text index for field '{}' did
not return a bitmap",
+ predicate->FieldName()));
+ } else if (std::dynamic_pointer_cast<Skip>(index_result)) {
+ return Evaluation{};
+ } else if (!std::dynamic_pointer_cast<Remain>(index_result)) {
+ return Status::Invalid(
+ fmt::format("invalid index result for predicate {}",
predicate->ToString()));
+ }
+ // TODO(lisizhuo.lsz): Whether to ScanScalar depends on whether the index
returns exact results.
+ if (!is_full_text && !result.rows.IsEmpty()) {
+ // Scalar indexes may contain false positives, even if they return a
bitmap.
+ PAIMON_ASSIGN_OR_RAISE(RoaringBitmap32 exact_rows,
ScanScalar(predicate, result.rows));
+ result.rows &= exact_rows;
+ }
+ return result;
+}
+
+Result<std::unique_ptr<FileBatchReader>>
PreciseFileIndexRead::CreateSelectedFileReader(
+ const RoaringBitmap32& selection, const FieldMappingBuilder*
field_mapping_builder) const {
+ PAIMON_ASSIGN_OR_RAISE(std::string format, file_->FileFormat());
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ReaderBuilder> builder,
+ split_read_.PrepareReaderBuilder(format,
extra_format_options_));
+ return split_read_.CreateFieldMappingReader(path_factory_->ToPath(file_),
file_, partition_,
+ std::move(builder),
field_mapping_builder, nullptr,
+ std::nullopt, path_factory_,
&selection);
Review Comment:
When all physical rows are eligible, passing `&selection` still makes the
Lance reader expand the bitmap into one `uint64_t` per row. A 100-million-row
file needs about 800 MB just for those IDs before it can return the first
batch. This happens even with no predicate, TopK, deletion or range
restriction. Please pass an absent bitmap to the format reader when the
selection covers the whole file, while preserving the precise path's validation
and filtering for partial selections.
##########
src/paimon/core/operation/precise_file_index_read.cpp:
##########
@@ -0,0 +1,573 @@
+/*
+ * 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/operation/precise_file_index_read.h"
+
+#include <algorithm>
+#include <cmath>
+#include <map>
+#include <memory>
+#include <optional>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "arrow/c/abi.h"
+#include "arrow/type.h"
+#include "fmt/format.h"
+#include "paimon/common/reader/complete_index_score_file_batch_reader.h"
+#include "paimon/common/reader/predicate_batch_reader.h"
+#include "paimon/common/reader/reader_utils.h"
+#include "paimon/common/types/data_field.h"
+#include "paimon/common/utils/object_utils.h"
+#include "paimon/common/utils/scope_guard.h"
+#include "paimon/core/deletionvectors/bitmap64_deletion_vector.h"
+#include "paimon/core/deletionvectors/bitmap_deletion_vector.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/io/data_file_path_factory.h"
+#include "paimon/core/io/file_index_evaluator.h"
+#include "paimon/core/io/file_index_options.h"
+#include "paimon/core/operation/internal_read_context.h"
+#include "paimon/core/operation/raw_file_split_read.h"
+#include "paimon/core/schema/schema_manager.h"
+#include "paimon/core/schema/table_schema.h"
+#include "paimon/core/utils/field_mapping.h"
+#include "paimon/file_index/bitmap_index_result.h"
+#include "paimon/file_index/file_index_result.h"
+#include "paimon/file_index/scored_file_index_result.h"
+#include "paimon/format/reader_builder.h"
+#include "paimon/predicate/compound_predicate.h"
+#include "paimon/predicate/full_text_search.h"
+#include "paimon/predicate/function.h"
+#include "paimon/predicate/leaf_predicate.h"
+#include "paimon/predicate/vector_search.h"
+#include "paimon/utils/roaring_bitmap64.h"
+
+namespace paimon {
+namespace {
+
+Result<std::string> ResolveSearchFieldName(
+ const std::string& field_name, const std::map<std::string, std::string>&
field_name_mapping,
+ const std::string& file_name) {
+ auto iter = field_name_mapping.find(field_name);
+ if (iter == field_name_mapping.end()) {
+ return Status::Invalid(
+ fmt::format("Search field '{}' has no matching field ID in data
schema for file '{}'",
+ field_name, file_name));
+ }
+ return iter->second;
+}
+
+Status PreparePredicate(
+ const std::shared_ptr<Predicate>& predicate,
+ const std::map<std::string, DataField>& read_by_name,
+ const std::map<int32_t, std::pair<int32_t, DataField>>& data_by_id,
+ const std::vector<std::string>& partition_keys, FileIndexEvaluator*
indexes,
+ std::map<std::shared_ptr<LeafPredicate>, std::shared_ptr<LeafPredicate>>*
mapped_leaves) {
+ if (!predicate) {
+ return Status::OK();
+ }
+ if (auto compound =
std::dynamic_pointer_cast<CompoundPredicate>(predicate)) {
+ const Function::Type type = compound->GetFunction().GetType();
+ if (type != Function::Type::AND && type != Function::Type::OR) {
+ return Status::Invalid(
+ fmt::format("unsupported compound file predicate {}",
predicate->ToString()));
+ }
+ for (const auto& child : compound->Children()) {
+ PAIMON_RETURN_NOT_OK(PreparePredicate(child, read_by_name,
data_by_id, partition_keys,
+ indexes, mapped_leaves));
+ }
+ return Status::OK();
+ }
+ auto leaf = std::dynamic_pointer_cast<LeafPredicate>(predicate);
+ if (!leaf) {
+ return Status::Invalid("File predicate must contain leaf or compound
predicates");
+ }
+ bool full_text = leaf->GetFunction().GetType() ==
Function::Type::FULL_TEXT_MATCH;
+ if (!full_text && ObjectUtils::Contains(partition_keys,
leaf->FieldName())) {
+ mapped_leaves->emplace(leaf, nullptr);
+ return Status::OK();
+ }
+ PAIMON_ASSIGN_OR_RAISE(
+ std::optional<std::shared_ptr<Predicate>> mapped,
+ FieldMappingBuilder::ReconstructPredicateWithDataFields(leaf,
read_by_name, data_by_id));
+ std::shared_ptr<LeafPredicate> mapped_leaf;
+ if (mapped) {
+ mapped_leaf = std::dynamic_pointer_cast<LeafPredicate>(*mapped);
+ if (!mapped_leaf) {
+ return Status::Invalid("Mapped file predicate must be a leaf");
+ }
+ if (full_text) {
+ PAIMON_ASSIGN_OR_RAISE(
+ [[maybe_unused]] std::shared_ptr<FileIndexReader> reader,
+ indexes->GetSearchReader(mapped_leaf->FieldName(),
+
FileIndexReader::SearchType::FULL_TEXT_MATCH));
+ }
+ }
+ mapped_leaves->emplace(leaf, std::move(mapped_leaf));
+ return Status::OK();
+}
+
+Status PrepareSearch(const std::shared_ptr<InternalReadContext>& context,
+ const std::map<std::string, std::string>& field_names,
+ const std::string& file_name, FileIndexEvaluator* indexes,
+ std::shared_ptr<VectorSearch>* vector_search,
+ std::shared_ptr<FullTextSearch>* full_text_search) {
+ if (context->GetVectorSearch()) {
+ *vector_search =
std::make_shared<VectorSearch>(*context->GetVectorSearch());
+ PAIMON_ASSIGN_OR_RAISE(
+ (*vector_search)->field_name,
+ ResolveSearchFieldName((*vector_search)->field_name, field_names,
file_name));
+ PAIMON_ASSIGN_OR_RAISE(
+ [[maybe_unused]] std::shared_ptr<FileIndexReader> reader,
+ indexes->GetSearchReader((*vector_search)->field_name,
+
FileIndexReader::SearchType::VECTOR_SEARCH));
+ }
+ if (context->GetFullTextSearch()) {
+ *full_text_search =
std::make_shared<FullTextSearch>(*context->GetFullTextSearch());
+ PAIMON_ASSIGN_OR_RAISE(
+ (*full_text_search)->field_name,
+ ResolveSearchFieldName((*full_text_search)->field_name,
field_names, file_name));
+ PAIMON_ASSIGN_OR_RAISE(
+ [[maybe_unused]] std::shared_ptr<FileIndexReader> reader,
+ indexes->GetSearchReader((*full_text_search)->field_name,
+
FileIndexReader::SearchType::FULL_TEXT_SEARCH));
+ }
+ return Status::OK();
+}
+
+} // namespace
+
+Result<std::unique_ptr<PreciseFileIndexRead>> PreciseFileIndexRead::Create(
+ const RawFileSplitRead& split_read, const BinaryRow& partition,
+ const std::shared_ptr<DataFileMeta>& file, const FieldMappingBuilder*
field_mapping_builder,
+ const std::shared_ptr<DataFilePathFactory>& path_factory,
+ const std::map<std::string, std::string>& extra_format_options) {
+ std::shared_ptr<TableSchema> table_schema =
split_read.context_->GetTableSchema();
Review Comment:
Could this local be named `file_schema`? For an older file it holds the
historical schema loaded on lines 163–164, while `CreateFieldNameMapping` below
still uses the current table schema. The distinction matters when a column has
been renamed.
##########
src/paimon/core/operation/precise_file_index_read.cpp:
##########
@@ -0,0 +1,573 @@
+/*
+ * 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/operation/precise_file_index_read.h"
+
+#include <algorithm>
+#include <cmath>
+#include <map>
+#include <memory>
+#include <optional>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "arrow/c/abi.h"
+#include "arrow/type.h"
+#include "fmt/format.h"
+#include "paimon/common/reader/complete_index_score_file_batch_reader.h"
+#include "paimon/common/reader/predicate_batch_reader.h"
+#include "paimon/common/reader/reader_utils.h"
+#include "paimon/common/types/data_field.h"
+#include "paimon/common/utils/object_utils.h"
+#include "paimon/common/utils/scope_guard.h"
+#include "paimon/core/deletionvectors/bitmap64_deletion_vector.h"
+#include "paimon/core/deletionvectors/bitmap_deletion_vector.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/io/data_file_path_factory.h"
+#include "paimon/core/io/file_index_evaluator.h"
+#include "paimon/core/io/file_index_options.h"
+#include "paimon/core/operation/internal_read_context.h"
+#include "paimon/core/operation/raw_file_split_read.h"
+#include "paimon/core/schema/schema_manager.h"
+#include "paimon/core/schema/table_schema.h"
+#include "paimon/core/utils/field_mapping.h"
+#include "paimon/file_index/bitmap_index_result.h"
+#include "paimon/file_index/file_index_result.h"
+#include "paimon/file_index/scored_file_index_result.h"
+#include "paimon/format/reader_builder.h"
+#include "paimon/predicate/compound_predicate.h"
+#include "paimon/predicate/full_text_search.h"
+#include "paimon/predicate/function.h"
+#include "paimon/predicate/leaf_predicate.h"
+#include "paimon/predicate/vector_search.h"
+#include "paimon/utils/roaring_bitmap64.h"
+
+namespace paimon {
+namespace {
+
+Result<std::string> ResolveSearchFieldName(
+ const std::string& field_name, const std::map<std::string, std::string>&
field_name_mapping,
+ const std::string& file_name) {
+ auto iter = field_name_mapping.find(field_name);
+ if (iter == field_name_mapping.end()) {
+ return Status::Invalid(
+ fmt::format("Search field '{}' has no matching field ID in data
schema for file '{}'",
+ field_name, file_name));
+ }
+ return iter->second;
+}
+
+Status PreparePredicate(
+ const std::shared_ptr<Predicate>& predicate,
+ const std::map<std::string, DataField>& read_by_name,
+ const std::map<int32_t, std::pair<int32_t, DataField>>& data_by_id,
+ const std::vector<std::string>& partition_keys, FileIndexEvaluator*
indexes,
+ std::map<std::shared_ptr<LeafPredicate>, std::shared_ptr<LeafPredicate>>*
mapped_leaves) {
+ if (!predicate) {
+ return Status::OK();
+ }
+ if (auto compound =
std::dynamic_pointer_cast<CompoundPredicate>(predicate)) {
+ const Function::Type type = compound->GetFunction().GetType();
+ if (type != Function::Type::AND && type != Function::Type::OR) {
+ return Status::Invalid(
+ fmt::format("unsupported compound file predicate {}",
predicate->ToString()));
+ }
+ for (const auto& child : compound->Children()) {
+ PAIMON_RETURN_NOT_OK(PreparePredicate(child, read_by_name,
data_by_id, partition_keys,
+ indexes, mapped_leaves));
+ }
+ return Status::OK();
+ }
+ auto leaf = std::dynamic_pointer_cast<LeafPredicate>(predicate);
+ if (!leaf) {
+ return Status::Invalid("File predicate must contain leaf or compound
predicates");
+ }
+ bool full_text = leaf->GetFunction().GetType() ==
Function::Type::FULL_TEXT_MATCH;
+ if (!full_text && ObjectUtils::Contains(partition_keys,
leaf->FieldName())) {
+ mapped_leaves->emplace(leaf, nullptr);
+ return Status::OK();
+ }
+ PAIMON_ASSIGN_OR_RAISE(
+ std::optional<std::shared_ptr<Predicate>> mapped,
+ FieldMappingBuilder::ReconstructPredicateWithDataFields(leaf,
read_by_name, data_by_id));
+ std::shared_ptr<LeafPredicate> mapped_leaf;
+ if (mapped) {
+ mapped_leaf = std::dynamic_pointer_cast<LeafPredicate>(*mapped);
+ if (!mapped_leaf) {
+ return Status::Invalid("Mapped file predicate must be a leaf");
+ }
+ if (full_text) {
+ PAIMON_ASSIGN_OR_RAISE(
+ [[maybe_unused]] std::shared_ptr<FileIndexReader> reader,
+ indexes->GetSearchReader(mapped_leaf->FieldName(),
+
FileIndexReader::SearchType::FULL_TEXT_MATCH));
+ }
+ }
+ mapped_leaves->emplace(leaf, std::move(mapped_leaf));
+ return Status::OK();
+}
+
+Status PrepareSearch(const std::shared_ptr<InternalReadContext>& context,
+ const std::map<std::string, std::string>& field_names,
+ const std::string& file_name, FileIndexEvaluator* indexes,
+ std::shared_ptr<VectorSearch>* vector_search,
+ std::shared_ptr<FullTextSearch>* full_text_search) {
+ if (context->GetVectorSearch()) {
+ *vector_search =
std::make_shared<VectorSearch>(*context->GetVectorSearch());
+ PAIMON_ASSIGN_OR_RAISE(
+ (*vector_search)->field_name,
+ ResolveSearchFieldName((*vector_search)->field_name, field_names,
file_name));
+ PAIMON_ASSIGN_OR_RAISE(
+ [[maybe_unused]] std::shared_ptr<FileIndexReader> reader,
+ indexes->GetSearchReader((*vector_search)->field_name,
+
FileIndexReader::SearchType::VECTOR_SEARCH));
+ }
+ if (context->GetFullTextSearch()) {
+ *full_text_search =
std::make_shared<FullTextSearch>(*context->GetFullTextSearch());
+ PAIMON_ASSIGN_OR_RAISE(
+ (*full_text_search)->field_name,
+ ResolveSearchFieldName((*full_text_search)->field_name,
field_names, file_name));
+ PAIMON_ASSIGN_OR_RAISE(
+ [[maybe_unused]] std::shared_ptr<FileIndexReader> reader,
+ indexes->GetSearchReader((*full_text_search)->field_name,
+
FileIndexReader::SearchType::FULL_TEXT_SEARCH));
+ }
+ return Status::OK();
+}
+
+} // namespace
+
+Result<std::unique_ptr<PreciseFileIndexRead>> PreciseFileIndexRead::Create(
+ const RawFileSplitRead& split_read, const BinaryRow& partition,
+ const std::shared_ptr<DataFileMeta>& file, const FieldMappingBuilder*
field_mapping_builder,
+ const std::shared_ptr<DataFilePathFactory>& path_factory,
+ const std::map<std::string, std::string>& extra_format_options) {
+ std::shared_ptr<TableSchema> table_schema =
split_read.context_->GetTableSchema();
+ if (file->schema_id != table_schema->Id()) {
+ PAIMON_ASSIGN_OR_RAISE(table_schema,
+
split_read.schema_manager_->ReadSchema(file->schema_id));
+ }
+ const auto data_schema =
DataField::ConvertDataFieldsToArrowSchema(table_schema->Fields());
+ if (file->row_count < 0 || file->row_count > RoaringBitmap32::MAX_VALUE) {
+ return Status::NotImplemented(
+ "Precise file-index evaluation requires bitmap32 physical row
positions");
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::vector<DataField> data_fields,
+
DataField::ConvertArrowSchemaToDataFields(data_schema));
+ PAIMON_ASSIGN_OR_RAISE(
+ std::vector<DataField> read_fields,
+
DataField::ConvertArrowSchemaToDataFields(split_read.context_->GetReadSchema()));
+ auto field_names = FieldMappingBuilder::CreateFieldNameMapping(
+ split_read.context_->GetTableSchema()->Fields(), data_fields);
+ std::map<std::string, std::string> index_options;
+ PAIMON_ASSIGN_OR_RAISE(index_options,
+ FileIndexOptions::RewriteFieldNames(
+ split_read.context_->GetCoreOptions().ToMap(),
field_names));
+ PAIMON_ASSIGN_OR_RAISE(
+ std::unique_ptr<FileIndexEvaluator> indexes,
+ FileIndexEvaluator::Create(data_schema, index_options, path_factory,
file,
+
split_read.context_->GetCoreOptions().GetFileSystem(),
+ split_read.pool_));
+
+ std::map<std::string, DataField> read_by_name;
+ for (const auto& field : read_fields) {
+ read_by_name.emplace(field.Name(), field);
+ }
+ // Field ID -> (zero-based position in data_fields, field definition from
the file's schema).
+ std::map<int32_t, std::pair<int32_t, DataField>> data_by_id;
+ for (size_t i = 0; i < data_fields.size(); ++i) {
+ data_by_id.emplace(data_fields[i].Id(), std::make_pair(i,
data_fields[i]));
+ }
+ std::map<std::shared_ptr<LeafPredicate>, std::shared_ptr<LeafPredicate>>
mapped_leaves;
+ // Precheck the entire tree before an empty candidate set can
short-circuit it.
+ PAIMON_RETURN_NOT_OK(PreparePredicate(split_read.context_->GetPredicate(),
read_by_name,
+ data_by_id,
split_read.context_->GetPartitionKeys(),
+ indexes.get(), &mapped_leaves));
+ std::shared_ptr<VectorSearch> vector_search;
+ std::shared_ptr<FullTextSearch> full_text_search;
+ PAIMON_RETURN_NOT_OK(PrepareSearch(split_read.context_, field_names,
file->file_name,
+ indexes.get(), &vector_search,
&full_text_search));
+ return std::unique_ptr<PreciseFileIndexRead>(new PreciseFileIndexRead(
+ split_read, partition, file, field_mapping_builder, path_factory,
extra_format_options,
+ std::move(indexes), mapped_leaves, vector_search, full_text_search));
+}
+
+PreciseFileIndexRead::PreciseFileIndexRead(
+ const RawFileSplitRead& split_read, const BinaryRow& partition,
+ const std::shared_ptr<DataFileMeta>& file, const FieldMappingBuilder*
field_mapping_builder,
+ const std::shared_ptr<DataFilePathFactory>& path_factory,
+ const std::map<std::string, std::string>& extra_format_options,
+ std::unique_ptr<FileIndexEvaluator> indexes,
+ const std::map<std::shared_ptr<LeafPredicate>,
std::shared_ptr<LeafPredicate>>& mapped_leaves,
+ const std::shared_ptr<VectorSearch>& vector_search,
+ const std::shared_ptr<FullTextSearch>& full_text_search)
+ : split_read_(split_read),
+ partition_(partition),
+ file_(file),
+ field_mapping_builder_(field_mapping_builder),
+ path_factory_(path_factory),
+ extra_format_options_(extra_format_options),
+ indexes_(std::move(indexes)),
+ mapped_leaves_(mapped_leaves),
+ vector_search_(vector_search),
+ full_text_search_(full_text_search) {}
+
+Result<RoaringBitmap32> PreciseFileIndexRead::CreateCandidates(
+ const DeletionVector::Factory& dv_factory,
+ const std::optional<std::vector<Range>>& row_ranges) const {
+ RoaringBitmap32 candidates;
+ if (row_ranges) {
+ for (const auto& range : *row_ranges) {
+ candidates.AddRange(static_cast<int32_t>(range.from),
+ static_cast<int32_t>(range.to + 1));
+ }
+ } else {
+ candidates.AddRange(0, static_cast<int32_t>(file_->row_count));
+ }
+ if (dv_factory) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<DeletionVector> dv,
dv_factory(file_->file_name));
+ if (auto bitmap = dynamic_cast<BitmapDeletionVector*>(dv.get())) {
+ candidates -= *bitmap->GetBitmap();
+ } else if (dv) {
+ return Status::NotImplemented(
+ "Precise file-index evaluation requires bitmap32 deletion
vectors");
+ }
+ }
+ return candidates;
+}
+
+Result<PreciseFileIndexRead::Evaluation> PreciseFileIndexRead::Evaluate(
+ const std::shared_ptr<Predicate>& predicate, const RoaringBitmap32&
candidates) const {
+ if (!predicate || candidates.IsEmpty()) {
+ return Evaluation{candidates, {}};
+ }
+ if (auto leaf = std::dynamic_pointer_cast<LeafPredicate>(predicate)) {
+ return EvaluateLeaf(leaf, candidates);
+ }
+ if (auto compound =
std::dynamic_pointer_cast<CompoundPredicate>(predicate)) {
+ return EvaluateCompound(compound, candidates);
+ }
+ return Status::Invalid(fmt::format("unsupported file predicate {}",
predicate->ToString()));
+}
+
+Result<PreciseFileIndexRead::Evaluation>
PreciseFileIndexRead::EvaluateCompound(
+ const std::shared_ptr<CompoundPredicate>& predicate, const
RoaringBitmap32& candidates) const {
+ const Function::Type type = predicate->GetFunction().GetType();
+ if (type == Function::Type::OR) {
+ Evaluation result;
+ for (const auto& child : predicate->Children()) {
+ // Every branch must see the same input, including rows matched by
earlier branches.
+ PAIMON_ASSIGN_OR_RAISE(Evaluation child_result, Evaluate(child,
candidates));
+ result.rows |= child_result.rows;
+ result.scores.insert(child_result.scores.begin(),
child_result.scores.end());
+ }
+ return result;
+ }
+ if (type != Function::Type::AND) {
+ return Status::Invalid(
+ fmt::format("unsupported compound file predicate {}",
predicate->ToString()));
+ }
+
+ struct Child {
+ std::shared_ptr<Predicate> predicate;
+ std::optional<double> selectivity;
+ };
+ std::vector<Child> children;
+ children.reserve(predicate->Children().size());
+ for (const auto& child : predicate->Children()) {
+ PAIMON_ASSIGN_OR_RAISE(std::optional<double> selectivity,
EstimateSelectivity(child));
+ children.push_back({child, selectivity});
+ }
+ std::stable_sort(children.begin(), children.end(), [](const Child& lhs,
const Child& rhs) {
+ if (!lhs.selectivity) {
+ return false;
+ }
+ return !rhs.selectivity || *lhs.selectivity < *rhs.selectivity;
+ });
+
+ Evaluation result{candidates, {}};
+ for (const auto& child : children) {
+ PAIMON_ASSIGN_OR_RAISE(Evaluation child_result,
Evaluate(child.predicate, result.rows));
+ result.rows = std::move(child_result.rows);
+ result.scores.insert(child_result.scores.begin(),
child_result.scores.end());
+ if (result.rows.IsEmpty()) {
+ break;
+ }
+ }
+ return result;
+}
+
+Result<std::optional<double>> PreciseFileIndexRead::EstimateSelectivity(
+ const std::shared_ptr<Predicate>& predicate) const {
+ if (!predicate) {
+ return std::optional<double>();
+ }
+ if (auto leaf = std::dynamic_pointer_cast<LeafPredicate>(predicate)) {
+ PAIMON_ASSIGN_OR_RAISE(std::optional<double> estimate,
EstimateLeafSelectivity(leaf));
+ if (estimate && (!std::isfinite(*estimate) || *estimate < 0.0 ||
*estimate > 1.0)) {
+ return std::optional<double>();
+ }
+ return estimate;
+ }
+ auto compound = std::dynamic_pointer_cast<CompoundPredicate>(predicate);
+ if (!compound) {
+ return Status::Invalid(fmt::format("unsupported file predicate {}",
predicate->ToString()));
+ }
+ const Function::Type type = compound->GetFunction().GetType();
+ if (type != Function::Type::AND && type != Function::Type::OR) {
+ return Status::Invalid(
+ fmt::format("unsupported compound file predicate {}",
predicate->ToString()));
+ }
+ std::optional<double> estimate =
+ type == Function::Type::AND ? std::nullopt : std::make_optional(0.0);
+ // These are scheduling hints. Continue visiting all children to propagate
errors.
+ for (const auto& child : compound->Children()) {
+ PAIMON_ASSIGN_OR_RAISE(std::optional<double> child_estimate,
EstimateSelectivity(child));
+ if (type == Function::Type::AND) {
+ if (child_estimate) {
+ if (estimate.has_value()) {
+ estimate = std::min(*estimate, *child_estimate);
+ } else {
+ estimate = *child_estimate;
+ }
+ }
+ } else if (estimate && child_estimate) {
+ estimate = std::min(1.0, *estimate + *child_estimate);
+ } else {
+ estimate = std::nullopt;
+ }
+ }
+ return estimate;
+}
+
+Result<std::optional<double>> PreciseFileIndexRead::EstimateLeafSelectivity(
+ const std::shared_ptr<LeafPredicate>& predicate) const {
+ auto iter = mapped_leaves_.find(predicate);
+ if (iter == mapped_leaves_.end()) {
+ return Status::Invalid("File predicate mapping was not prepared");
+ }
+ if (!iter->second) {
+ return std::optional<double>();
+ }
+ return indexes_->EstimateSelectivity(iter->second);
+}
+
+Result<PreciseFileIndexRead::Evaluation> PreciseFileIndexRead::EvaluateLeaf(
+ const std::shared_ptr<LeafPredicate>& predicate, const RoaringBitmap32&
candidates) const {
+ bool is_full_text = predicate->GetFunction().GetType() ==
Function::Type::FULL_TEXT_MATCH;
+ bool needs_scores =
+ is_full_text &&
+ split_read_.context_->NeedsScores(
+ {InternalReadContext::IndexScoreSource::Type::FULL_TEXT_MATCH,
predicate->FieldName()});
+ std::shared_ptr<FileIndexResult> index_result = FileIndexResult::Remain();
+ auto iter = mapped_leaves_.find(predicate);
+ if (iter == mapped_leaves_.end()) {
+ return Status::Invalid("File predicate mapping was not prepared");
+ }
+ if (iter->second) {
+ // TODO(lisizhuo.lsz): Pass candidates through
FileIndexEvaluator::EvaluateLeaf to index
+ // readers so they can skip rows excluded by earlier predicates, row
ranges, or deletion
+ // vectors.
+ PAIMON_ASSIGN_OR_RAISE(index_result,
indexes_->EvaluateLeaf(iter->second, needs_scores));
+ }
+ if (!index_result) {
+ return Status::Invalid("file predicate index returned a null result");
+ }
+ Evaluation result{candidates, {}};
+ if (needs_scores) {
+ auto scored =
std::dynamic_pointer_cast<ScoredFileIndexResult>(index_result);
+ if (!scored) {
+ return Status::Invalid(
+ fmt::format("full-text index for field '{}' did not return
requested scores",
+ predicate->FieldName()));
+ }
+ result.rows &= scored->GetRowPositions();
+ result.scores.emplace(predicate->FieldName(), std::move(scored));
+ return result;
+ }
+
+ if (auto bitmap =
std::dynamic_pointer_cast<BitmapIndexResult>(index_result)) {
+ PAIMON_ASSIGN_OR_RAISE(const RoaringBitmap32* rows,
bitmap->GetBitmap());
+ result.rows &= *rows;
+ } else if (is_full_text) {
+ return Status::Invalid(fmt::format("full-text index for field '{}' did
not return a bitmap",
+ predicate->FieldName()));
+ } else if (std::dynamic_pointer_cast<Skip>(index_result)) {
+ return Evaluation{};
+ } else if (!std::dynamic_pointer_cast<Remain>(index_result)) {
+ return Status::Invalid(
+ fmt::format("invalid index result for predicate {}",
predicate->ToString()));
+ }
+ // TODO(lisizhuo.lsz): Whether to ScanScalar depends on whether the index
returns exact results.
+ if (!is_full_text && !result.rows.IsEmpty()) {
+ // Scalar indexes may contain false positives, even if they return a
bitmap.
+ PAIMON_ASSIGN_OR_RAISE(RoaringBitmap32 exact_rows,
ScanScalar(predicate, result.rows));
+ result.rows &= exact_rows;
+ }
+ return result;
+}
+
+Result<std::unique_ptr<FileBatchReader>>
PreciseFileIndexRead::CreateSelectedFileReader(
+ const RoaringBitmap32& selection, const FieldMappingBuilder*
field_mapping_builder) const {
+ PAIMON_ASSIGN_OR_RAISE(std::string format, file_->FileFormat());
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<ReaderBuilder> builder,
+ split_read_.PrepareReaderBuilder(format,
extra_format_options_));
+ return split_read_.CreateFieldMappingReader(path_factory_->ToPath(file_),
file_, partition_,
+ std::move(builder),
field_mapping_builder, nullptr,
+ std::nullopt, path_factory_,
&selection);
+}
+
+Result<RoaringBitmap32> PreciseFileIndexRead::ScanScalar(
+ const std::shared_ptr<LeafPredicate>& predicate, const RoaringBitmap32&
input) const {
+ auto field =
split_read_.context_->GetReadSchema()->GetFieldByName(predicate->FieldName());
+ if (!field) {
+ return Status::Invalid(fmt::format(
+ "Scalar predicate field '{}' is missing from the read schema",
predicate->FieldName()));
+ }
+ // TODO(lisizhuo.lsz): Push the scalar predicate to use min/max statistics
for pruning row
+ // groups/pages.
+ PAIMON_ASSIGN_OR_RAISE(
+ std::unique_ptr<FieldMappingBuilder> field_mapping_builder,
+ FieldMappingBuilder::Create(arrow::schema({field}),
+ split_read_.context_->GetPartitionKeys(),
nullptr));
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileBatchReader> reader,
+ CreateSelectedFileReader(input,
field_mapping_builder.get()));
+ FileBatchReader& file_reader = *reader;
+ PAIMON_ASSIGN_OR_RAISE(
+ std::unique_ptr<PredicateBatchReader> filtered_reader,
+ PredicateBatchReader::Create(std::move(reader), predicate,
+
split_read_.context_->GetArrowMemoryPool()));
+ ScopeGuard close([&]() { filtered_reader->Close(); });
Review Comment:
The scalar recheck reader can scan an entire column and then be closed here
without its I/O metrics reaching `GetReaderMetrics()` on the returned reader.
If the predicate eliminates every row, the caller can observe zero file-read
I/O after a full scan. Please collect the temporary reader's metrics and carry
them into the split-level metrics, including the case where no output reader is
created. An all-filtered regression would cover the easiest case to miss.
##########
src/paimon/core/io/file_index_evaluator.cpp:
##########
@@ -52,64 +54,165 @@ namespace paimon {
class MemoryPool;
enum class FieldType;
-Result<std::shared_ptr<FileIndexResult>> FileIndexEvaluator::Evaluate(
+FileIndexEvaluator::FileIndexEvaluator(std::shared_ptr<arrow::Schema>
data_schema,
+ std::shared_ptr<DataFileMeta> file_meta,
+
std::unique_ptr<FileIndexFormat::Reader> format_reader)
+ : data_schema_(std::move(data_schema)),
+ file_meta_(std::move(file_meta)),
+ format_reader_(std::move(format_reader)) {}
+
+Result<std::unique_ptr<FileIndexEvaluator>> FileIndexEvaluator::Create(
const std::shared_ptr<arrow::Schema>& data_schema,
- const std::map<std::string, std::string>& options, const
std::shared_ptr<Predicate>& predicate,
- const std::shared_ptr<DataFilePathFactory>& data_file_path_factory,
+ const std::map<std::string, std::string>& options,
+ const std::shared_ptr<DataFilePathFactory>& path_factory,
const std::shared_ptr<DataFileMeta>& file_meta, const
std::shared_ptr<FileSystem>& file_system,
const std::shared_ptr<MemoryPool>& pool) {
- return Evaluate(/*only_use_embedding_index=*/false, data_schema, options,
predicate,
- data_file_path_factory, file_meta, file_system, pool);
+ std::unique_ptr<FileIndexFormat::Reader> format_reader;
+ if (file_meta->row_count >= 0 && file_meta->row_count <=
RoaringBitmap32::MAX_VALUE) {
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InputStream> input_stream,
+ ExtractIndexInputStream(path_factory,
file_meta, file_system));
+ if (input_stream) {
+ PAIMON_ASSIGN_OR_RAISE(format_reader,
+ FileIndexFormat::CreateReader(input_stream,
pool, options));
+ }
+ }
+ return std::unique_ptr<FileIndexEvaluator>(
+ new FileIndexEvaluator(data_schema, file_meta,
std::move(format_reader)));
}
-Result<std::shared_ptr<FileIndexResult>> FileIndexEvaluator::Evaluate(
- const std::shared_ptr<arrow::Schema>& data_schema,
- const std::map<std::string, std::string>& options, const
std::shared_ptr<Predicate>& predicate,
- const std::shared_ptr<DataFileMeta>& file_meta, const
std::shared_ptr<MemoryPool>& pool) {
- return Evaluate(/*only_use_embedding_index=*/true, data_schema, options,
predicate,
- /*data_file_path_factory=*/nullptr, file_meta,
/*file_system=*/nullptr, pool);
+Result<std::vector<std::shared_ptr<FileIndexReader>>>
FileIndexEvaluator::GetReaders(
+ const std::string& field_name) {
+ auto iter = field_readers_.find(field_name);
+ if (iter != field_readers_.end()) {
+ return iter->second;
+ }
+ std::shared_ptr<arrow::Field> field =
data_schema_->GetFieldByName(field_name);
+ if (!field) {
+ return Status::Invalid(fmt::format("field '{}' does not exist in data
schema", field_name));
+ }
+ std::vector<std::shared_ptr<FileIndexReader>> readers;
+ if (format_reader_) {
+ ::ArrowSchema c_schema;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(
+ arrow::ExportSchema(*arrow::schema({std::move(field)}),
&c_schema));
+ PAIMON_ASSIGN_OR_RAISE(readers,
format_reader_->ReadColumnIndex(field_name, &c_schema));
+ }
+ field_readers_.emplace(field_name, readers);
+ return readers;
}
-Result<std::shared_ptr<FileIndexResult>> FileIndexEvaluator::Evaluate(
- bool only_use_embedding_index, const std::shared_ptr<arrow::Schema>&
data_schema,
- const std::map<std::string, std::string>& options, const
std::shared_ptr<Predicate>& predicate,
- const std::shared_ptr<DataFilePathFactory>& data_file_path_factory,
- const std::shared_ptr<DataFileMeta>& file_meta, const
std::shared_ptr<FileSystem>& file_system,
- const std::shared_ptr<MemoryPool>& pool) {
- if (predicate == nullptr || file_meta->row_count >
RoaringBitmap32::MAX_VALUE) {
- return FileIndexResult::Remain();
+Result<std::shared_ptr<FileIndexReader>> FileIndexEvaluator::GetSearchReader(
+ const std::string& field_name, FileIndexReader::SearchType search_type) {
+ if (!data_schema_->GetFieldByName(field_name)) {
+ return Status::Invalid(
+ fmt::format("Search field '{}' does not exist in data schema",
field_name));
+ }
+ if (!format_reader_) {
+ return Status::Invalid(fmt::format("Data file '{}' has no File Index
for search field '{}'",
+ file_meta_->file_name, field_name));
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<FileIndexReader>>
readers,
+ GetReaders(field_name));
+ std::shared_ptr<FileIndexReader> selected;
+ for (const auto& reader : readers) {
+ if (!reader->SupportsSearch(search_type)) {
+ continue;
+ }
+ if (selected) {
+ return Status::Invalid(fmt::format(
+ "Multiple File Index readers support requested search for
field '{}'", field_name));
+ }
+ selected = reader;
}
- PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InputStream> input_stream,
- ExtractIndexInputStream(only_use_embedding_index,
data_file_path_factory,
- file_meta, file_system));
- if (!input_stream) {
- // no index files, just return REMAIN
+ if (!selected) {
+ return Status::NotImplemented(fmt::format(
+ "No File Index reader supports requested search for field '{}'",
field_name));
Review Comment:
Please include `file_meta_->file_name` and the requested search operation in
this error and the multiple-reader error above. If one file in a multi-file
split lacks the required backend, the field name alone leaves the caller unable
to identify the file or distinguish `FULL_TEXT_MATCH` from `FULL_TEXT_SEARCH`.
--
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]
