lxy-9602 commented on code in PR #490:
URL: https://github.com/apache/paimon-cpp/pull/490#discussion_r4442932277
##########
docs/source/user_guide/file_index.rst:
##########
@@ -21,6 +21,11 @@ File Index
Full-text predicates and root TopK
---------------------------------
+Enable ``ReadContextBuilder::EnablePreciseFileIndexEvaluate(true)`` to evaluate
+complete predicates within each data file. The option defaults to ``false``.
+Full-text predicates and root TopK requests require this option; configuring
+them while it is disabled is rejected by ``ReadContextBuilder::Finish()``.
+
Review Comment:
This doc doesn’t clearly explain what this option does. It’s used to
precisely filter data at the index-query stage—before reading all required
columns from data file—to ensure the correctness of Full-text/vector top-k
queries.
##########
docs/source/user_guide/file_index.rst:
##########
@@ -80,4 +91,4 @@ An ordinary scalar predicate on the same field is allowed.
For example, when
projecting the score of ``match``, ``match AND (p OR q)`` is valid, while
``match OR p`` is rejected. Without that score projection, both OR branches and
repeated full-text search fields are allowed. These checks use the effective
-read projection, after resolving the read schema.
+read projection, after resolving the read schema.
Review Comment:
It seems like this just removes the trailing newline. Please change it back.
##########
include/paimon/file_index/file_index_reader.h:
##########
@@ -42,8 +44,18 @@ struct FullTextMatch;
/// of the indexed data; otherwise, index filtering results may be incorrect.
class PAIMON_EXPORT FileIndexReader : public
FunctionVisitor<std::shared_ptr<FileIndexResult>> {
public:
+ enum class SearchType { FULL_TEXT_MATCH, FULL_TEXT_SEARCH, VECTOR_SEARCH };
+
Review Comment:
Why is `FULL_TEXT_MATCH` placed into `SearchType`? `FULL_TEXT_MATCH` is
already a function in the predicate. Putting it into the `search` layer is a
bit ambiguous—at least please rename it.
##########
src/paimon/common/file_index/file_index_format.cpp:
##########
@@ -108,13 +108,14 @@ class FileIndexFormatReaderImpl : public
FileIndexFormat::Reader {
Result<std::shared_ptr<FileIndexReader>> GetFileIndexReader(
const std::string& column_name, const std::shared_ptr<arrow::Schema>&
arrow_schema,
const std::string& index_type, const std::pair<int64_t, int64_t>&
offset_and_length) const {
- if (offset_and_length.first == FileIndexFormat::EMPTY_INDEX_FLAG) {
- return std::make_shared<EmptyFileIndexReader>();
- }
const std::map<std::string, std::string>& indexer_options =
options_.GetIndexerOptions(column_name, index_type);
PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileIndexer> file_indexer,
FileIndexerFactory::Get(index_type,
indexer_options));
+ if (offset_and_length.first == FileIndexFormat::EMPTY_INDEX_FLAG) {
+ return file_indexer ? file_indexer->CreateEmptyReader()
+ : std::make_shared<EmptyFileIndexReader>();
Review Comment:
I’d like to understand: are there any scenarios where `file_indexer` can
benull?
##########
src/paimon/core/operation/file_index_read_executor.h:
##########
@@ -0,0 +1,121 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <map>
+#include <memory>
+#include <optional>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "paimon/core/deletionvectors/deletion_vector.h"
+#include "paimon/reader/file_batch_reader.h"
+#include "paimon/result.h"
+#include "paimon/status.h"
+#include "paimon/utils/range.h"
+#include "paimon/utils/roaring_bitmap32.h"
+
+namespace paimon {
+class BinaryRow;
+class CompoundPredicate;
+class DataField;
+class DataFilePathFactory;
+class FieldMappingBuilder;
+class FileIndexEvaluator;
+class LeafPredicate;
+class Predicate;
+class RawFileSplitRead;
+class ScoredFileIndexResult;
+struct DataFileMeta;
+struct FullTextSearch;
+struct VectorSearch;
+
+/// Evaluates a file predicate and root search before opening the selected
output rows.
+class FileIndexReadExecutor {
Review Comment:
Please rename the class name—“executor” is too easy to confuse with the
current thread-pool concept. It seems like it’s doing something like parallel
queries for the file index.
##########
src/paimon/core/operation/raw_file_split_read.h:
##########
@@ -46,19 +46,17 @@ class FileStorePathFactory;
class InternalReadContext;
class MemoryPool;
class Predicate;
-class ScoredFileIndexResult;
struct DataFileMeta;
struct DeletionFile;
-struct FullTextSearch;
-struct VectorSearch;
/// If the class name below is enclosed in parentheses, it might be present in
the read path;
/// otherwise, it must be present in the read path.
///
/// Readers Overview: (ConcatBatchReader across splits) ->
CompleteRowKindBatchReader ->
/// (PredicateBatchReader) -> ConcatBatchReader across files ->
(CompleteIndexScoreFileBatchReader)
-/// -> FieldMappingReader -> (ApplyBitmapIndexBatchReader) ->
(CompleteRowTrackingFieldsBatchReader)
-/// -> (ShreddingFileReader) -> (VectorFileBatchReader) ->
(DelegatingPrefetchReader) ->
+/// -> FieldMappingReader -> (ApplyDeletionVectorBatchReader) ->
(ApplyBitmapIndexBatchReader)
+/// -> (CompleteRowTrackingFieldsBatchReader) -> (ShreddingFileReader) ->
(VectorFileBatchReader) ->
+/// (DelegatingPrefetchReader) ->
/// (PrefetchFileBatchReader) -> (LateMaterializingFileBatchReader) ->
FormatReader
Review Comment:
The formatting at this stage seems a bit off.
##########
src/paimon/core/io/file_index_evaluator.cpp:
##########
@@ -52,60 +54,163 @@ 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);
+ bool only_use_embedding_index, const std::shared_ptr<MemoryPool>& 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(only_use_embedding_index, 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;
+ }
+ 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());
+ if (!remain) {
+ break;
+ }
}
- 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
+ return result;
+}
+
+Result<std::optional<double>> FileIndexEvaluator::EstimateSelectivity(
+ const std::shared_ptr<LeafPredicate>& predicate) {
+ 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));
+ return reader->EstimateSelectivity(predicate);
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<FileIndexReader>>
readers,
+ GetReaders(predicate->FieldName()));
+ std::optional<double> estimate;
+ for (const auto& reader : readers) {
+ auto value = reader->EstimateSelectivity(predicate);
+ if (value && std::isfinite(*value) && *value >= 0.0 && *value <= 1.0 &&
+ (!estimate || *value < *estimate)) {
+ estimate = value;
+ }
+ }
+ return estimate;
+}
+
+Result<std::shared_ptr<ScoredFileIndexResult>>
FileIndexEvaluator::SearchVector(
+ const std::shared_ptr<VectorSearch>& search) {
+ PAIMON_ASSIGN_OR_RAISE(
+ std::shared_ptr<FileIndexReader> reader,
+ GetSearchReader(search->field_name,
FileIndexReader::SearchType::VECTOR_SEARCH));
+ return reader->VisitVectorSearch(search);
+}
+
+Result<std::shared_ptr<ScoredFileIndexResult>>
FileIndexEvaluator::SearchFullText(
+ const std::shared_ptr<FullTextSearch>& search) {
+ PAIMON_ASSIGN_OR_RAISE(
+ std::shared_ptr<FileIndexReader> reader,
+ GetSearchReader(search->field_name,
FileIndexReader::SearchType::FULL_TEXT_SEARCH));
+ return reader->VisitFullTextSearch(search);
+}
+
+Result<std::shared_ptr<FileIndexResult>> FileIndexEvaluator::Evaluate(
+ const std::shared_ptr<Predicate>& predicate) {
+ if (!predicate || file_meta_->row_count < 0 ||
+ file_meta_->row_count > RoaringBitmap32::MAX_VALUE || !format_reader_)
{
return FileIndexResult::Remain();
}
Review Comment:
It seems that on the original path, regardless of whether there is a
predicate, whenever precise filtering is needed we always open an
`format_reader` in `FileIndexEvaluator::Create`, which triggers I/O. Please
optimize the corresponding logic.
##########
src/paimon/core/operation/file_index_read_executor.cpp:
##########
@@ -0,0 +1,553 @@
+/*
+ * 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/file_index_read_executor.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/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;
+}
+
+} // namespace
+
+Result<std::unique_ptr<FileIndexReadExecutor>> FileIndexReadExecutor::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::unique_ptr<FileIndexReadExecutor> executor(new FileIndexReadExecutor(
+ split_read, partition, file, field_mapping_builder, path_factory,
extra_format_options));
+ PAIMON_RETURN_NOT_OK(executor->Prepare());
+ return executor;
Review Comment:
Why do we need to create a `FileIndexReadExecutor` object first and then
call `Prepare`? Can’t we initialize all required members inside `Create`, then
construct it directly? The current flow is inconsistent with other classes.
##########
src/paimon/core/operation/abstract_split_read.h:
##########
@@ -79,6 +80,17 @@ class AbstractSplitRead : public SplitRead {
const std::map<std::string, std::string>& extra_format_options) const;
protected:
+ virtual bool EnablePreciseFileIndexEvaluate() const {
+ return false;
Review Comment:
Why do we still need this function as a virtual method? It seems every
`read` can obtain what it needs from `context_`.
##########
include/paimon/read_context.h:
##########
@@ -352,39 +360,49 @@ class PAIMON_EXPORT ReadContextBuilder {
///
/// All referenced fields, including full-text match fields, must be
projected in the read
/// schema.
- /// TODO(lisizhuo.lsz): The behavior here will be aligned with Java Paimon
in a follow-up
- /// change: fields referenced in predicates will no longer be required to
appear in the
- /// projection. Full-text matches can be combined with scalar predicates
using
- /// PredicateBuilder::And/Or. Full-text predicate execution is not
implemented yet; reading such
- /// a predicate returns NotImplemented.
+ /// TODO(lisizhuo.lsz): The behavior here will be aligned with Java Paimon.
+ /// Full-text matches can be combined with scalar predicates using
PredicateBuilder::And/Or.
Review Comment:
Why was the comment “fields referenced in predicates will no longer be
required to appear in the projection” removed? The current alignment
description makes it hard to understand.
##########
src/paimon/core/operation/file_index_read_executor.cpp:
##########
@@ -0,0 +1,553 @@
+/*
+ * 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/file_index_read_executor.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/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;
+}
+
+} // namespace
+
+Result<std::unique_ptr<FileIndexReadExecutor>> FileIndexReadExecutor::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::unique_ptr<FileIndexReadExecutor> executor(new FileIndexReadExecutor(
+ split_read, partition, file, field_mapping_builder, path_factory,
extra_format_options));
+ PAIMON_RETURN_NOT_OK(executor->Prepare());
+ return executor;
+}
+
+FileIndexReadExecutor::FileIndexReadExecutor(
+ 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)
+ : split_read_(split_read),
+ partition_(partition),
+ file_(file),
+ field_mapping_builder_(field_mapping_builder),
+ path_factory_(path_factory),
+ extra_format_options_(extra_format_options) {}
+
+Status FileIndexReadExecutor::Prepare() {
+ 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(indexes_, FileIndexEvaluator::Create(
+ data_schema, index_options,
path_factory_, file_,
+
split_read_.context_->GetCoreOptions().GetFileSystem(),
+ /*only_use_embedding_index=*/false,
split_read_.pool_));
+
+ std::map<std::string, DataField> read_by_name;
+ for (const auto& field : read_fields) {
+ read_by_name.emplace(field.Name(), field);
+ }
+ 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]));
+ }
+ // 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));
+ return PrepareSearch(field_names);
+}
+
+Status FileIndexReadExecutor::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) {
+ 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));
+ }
+ 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(split_read_.context_->GetPartitionKeys(),
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 FileIndexReadExecutor::PrepareSearch(const std::map<std::string,
std::string>& field_names) {
+ if (split_read_.context_->GetVectorSearch()) {
+ vector_search_ =
std::make_shared<VectorSearch>(*split_read_.context_->GetVectorSearch());
+ PAIMON_ASSIGN_OR_RAISE(
+ vector_search_->field_name,
+ ResolveSearchFieldName(vector_search_->field_name, field_names,
file_->file_name));
+ PAIMON_ASSIGN_OR_RAISE(
+ [[maybe_unused]] std::shared_ptr<FileIndexReader> reader,
+ indexes_->GetSearchReader(vector_search_->field_name,
+
FileIndexReader::SearchType::VECTOR_SEARCH));
+ }
+ if (split_read_.context_->GetFullTextSearch()) {
+ full_text_search_ =
+
std::make_shared<FullTextSearch>(*split_read_.context_->GetFullTextSearch());
+ PAIMON_ASSIGN_OR_RAISE(
+ full_text_search_->field_name,
+ ResolveSearchFieldName(full_text_search_->field_name, field_names,
file_->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();
+}
+
+Result<RoaringBitmap32> FileIndexReadExecutor::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 (dynamic_cast<Bitmap64DeletionVector*>(dv.get())) {
+ return Status::NotImplemented(
+ "Precise file-index evaluation does not support bitmap64
deletion vectors");
+ }
+ 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<FileIndexReadExecutor::Evaluation> FileIndexReadExecutor::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<FileIndexReadExecutor::Evaluation>
FileIndexReadExecutor::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>> FileIndexReadExecutor::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>();
Review Comment:
Please update the comment to note: “Non-finite estimates or values outside
[0, 1] are treated as unknown.” Alternatively, you can directly reject
selectivity values that fall outside the [0, 1] range.
##########
test/inte/write_and_read_inte_with_index_test.cpp:
##########
@@ -1088,13 +1095,107 @@ class WriteAndReadInteWithIndexTest
ASSERT_EQ(1, (*split)->DataFiles()[1]->schema_id);
}
+ private:
+ std::string PkDvTablePath() const {
+ return data_source_ == IndexDataSource::Java
+ ? GetDataDir() + "/" + file_format_ +
"/pk_with_bitmap_dv.db/pk_with_bitmap_dv/"
+ : dir_->Str() + "/foo.db/bar";
+ }
+
+ void CreatePkDvSplit(std::shared_ptr<DataSplitImpl>* split,
+ const std::shared_ptr<Bytes>& embedded_index =
nullptr) const {
+ std::string path = PkDvTablePath();
+ if (data_source_ != IndexDataSource::Java) {
+ std::vector<DataField> read_fields = {
+ SpecialFields::ValueKind(),
+ DataField(0, arrow::field("key", arrow::int32())),
+ DataField(1, arrow::field("f0", arrow::utf8())),
+ DataField(2, arrow::field("f1", arrow::int32())),
+ DataField(3, arrow::field("f2", arrow::int32())),
+ DataField(4, arrow::field("f3", arrow::float64()))};
+ ASSERT_NO_FATAL_FAILURE(WriteCppTableWithDv(read_fields, &path,
split));
+ if (embedded_index) {
+ (*split)->DataFiles()[0]->embedded_index = embedded_index;
+ }
+ return;
+ }
+ const bool is_orc = file_format_ == "orc";
+ const std::string file_name = is_orc
+ ?
"data-0da0dc51-797c-4c82-9b08-269b2ce400dc-0.orc"
+ :
"data-bfa36b28-40e7-4026-893b-d23116973174-0.parquet";
+ const std::string deletion_file_path =
+ path + (is_orc ?
"index/index-73ec72e9-489e-49d9-a96c-f6db9b56fa07-0"
+ :
"index/index-daed25d2-8e68-45ba-aaad-ff342666ef27-0");
+ auto file = std::make_shared<DataFileMeta>(
+ file_name, /*file_size=*/1001, /*row_count=*/8,
BinaryRow::EmptyRow(),
+ BinaryRow::EmptyRow(), SimpleStats::EmptyStats(),
SimpleStats::EmptyStats(),
+ /*min_sequence_number=*/0, /*max_sequence_number=*/7,
/*schema_id=*/0, /*level=*/5,
+ std::vector<std::optional<std::string>>{file_name + ".index"},
Timestamp(0ll, 0),
+ /*delete_row_count=*/0, embedded_index, FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt, /*external_path=*/std::nullopt,
+ /*first_row_id=*/std::nullopt, /*write_cols=*/std::nullopt,
+ /*column_max_sequence_numbers=*/std::nullopt);
+ // The fixture's deletion vector removes physical rows 0 (Alice) and 4
(Lucy).
+ DeletionFile deletion_file(deletion_file_path, /*offset=*/1,
/*length=*/24,
+ /*cardinality=*/2);
+ DataSplitImpl::Builder builder(BinaryRow::EmptyRow(), /*bucket=*/0,
path + "bucket-0/",
+ {file});
+ ASSERT_OK_AND_ASSIGN(*split, builder.WithSnapshot(4)
+
.WithDataDeletionFiles({deletion_file})
+ .IsStreaming(false)
+ .RawConvertible(true)
+ .Build());
+ }
+
+ void CheckPkDvPreciseResult(
+ const std::shared_ptr<DataSplit>& split, const arrow::FieldVector&
fields,
+ const std::string& json, const std::shared_ptr<Predicate>& predicate,
+ const std::shared_ptr<VectorSearch>& vector_search = nullptr,
+ const std::shared_ptr<FullTextSearch>& full_text_search = nullptr)
const {
+ const std::string path = PkDvTablePath();
+ arrow::FieldVector output_fields =
{SpecialFields::ValueKind().ArrowField()};
+ output_fields.insert(output_fields.end(), fields.begin(),
fields.end());
+ std::shared_ptr<arrow::ChunkedArray> expected;
+
ASSERT_TRUE(arrow::ipc::internal::json::ChunkedArrayFromJSON(arrow::struct_(output_fields),
+ {json},
&expected)
+ .ok());
+ std::vector<std::string> field_names;
+ for (const auto& field : fields) {
+ field_names.push_back(field->name());
+ }
+ ReadContextBuilder builder(path);
+ builder.EnablePreciseFileIndexEvaluate(true)
+ .EnablePredicateFilter(false)
+ .SetReadFieldNames(field_names)
+ .SetPredicate(predicate)
+ .SetVectorSearch(vector_search)
+ .SetFullTextSearch(full_text_search)
+ .AddOption("read.batch-size", "1")
+ .AddOption("test.enable-adaptive-prefetch-strategy", "false")
+ .EnableLateMaterializing(false);
+ if (enable_prefetch_) {
+ builder.EnablePrefetch(true).SetPrefetchBatchCount(3);
+ }
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<ReadContext> context,
builder.Finish());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableRead> read,
+ TableRead::Create(std::move(context)));
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<BatchReader> reader,
read->CreateReader(split));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> actual,
+
ReadResultCollector::CollectResult(std::move(reader)));
+ ASSERT_TRUE(actual);
+ ASSERT_TRUE(actual->Equals(expected)) << actual->ToString();
+ }
+
protected:
std::string file_format_;
bool enable_prefetch_ = false;
bool embedded_index_ = false;
IndexDataSource data_source_ = IndexDataSource::Java;
std::shared_ptr<MemoryPool> pool_;
std::unique_ptr<UniqueTestDirectory> dir_;
+ std::string statistics_index_type_ = "read-inte-statistics-full-text";
+ std::shared_ptr<MockSearchFileIndex::Statistics> statistics_ =
+ std::make_shared<MockSearchFileIndex::Statistics>();
Review Comment:
I found that it seems possible to configure queries like
`FullTextMatch(body, query) AND _body_INDEX_SCORE > 0.5`. Single-result results
are incorrect—please guard against this in the query execution path.
##########
src/paimon/core/operation/file_index_read_executor.cpp:
##########
@@ -0,0 +1,553 @@
+/*
+ * 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/file_index_read_executor.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/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;
+}
+
+} // namespace
+
+Result<std::unique_ptr<FileIndexReadExecutor>> FileIndexReadExecutor::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::unique_ptr<FileIndexReadExecutor> executor(new FileIndexReadExecutor(
+ split_read, partition, file, field_mapping_builder, path_factory,
extra_format_options));
+ PAIMON_RETURN_NOT_OK(executor->Prepare());
+ return executor;
+}
+
+FileIndexReadExecutor::FileIndexReadExecutor(
+ 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)
+ : split_read_(split_read),
+ partition_(partition),
+ file_(file),
+ field_mapping_builder_(field_mapping_builder),
+ path_factory_(path_factory),
+ extra_format_options_(extra_format_options) {}
+
+Status FileIndexReadExecutor::Prepare() {
+ 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(indexes_, FileIndexEvaluator::Create(
+ data_schema, index_options,
path_factory_, file_,
+
split_read_.context_->GetCoreOptions().GetFileSystem(),
+ /*only_use_embedding_index=*/false,
split_read_.pool_));
+
+ std::map<std::string, DataField> read_by_name;
+ for (const auto& field : read_fields) {
+ read_by_name.emplace(field.Name(), field);
+ }
+ 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]));
+ }
+ // 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));
+ return PrepareSearch(field_names);
+}
+
+Status FileIndexReadExecutor::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) {
+ 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));
+ }
+ 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(split_read_.context_->GetPartitionKeys(),
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 FileIndexReadExecutor::PrepareSearch(const std::map<std::string,
std::string>& field_names) {
+ if (split_read_.context_->GetVectorSearch()) {
+ vector_search_ =
std::make_shared<VectorSearch>(*split_read_.context_->GetVectorSearch());
+ PAIMON_ASSIGN_OR_RAISE(
+ vector_search_->field_name,
+ ResolveSearchFieldName(vector_search_->field_name, field_names,
file_->file_name));
+ PAIMON_ASSIGN_OR_RAISE(
+ [[maybe_unused]] std::shared_ptr<FileIndexReader> reader,
+ indexes_->GetSearchReader(vector_search_->field_name,
+
FileIndexReader::SearchType::VECTOR_SEARCH));
+ }
+ if (split_read_.context_->GetFullTextSearch()) {
+ full_text_search_ =
+
std::make_shared<FullTextSearch>(*split_read_.context_->GetFullTextSearch());
+ PAIMON_ASSIGN_OR_RAISE(
+ full_text_search_->field_name,
+ ResolveSearchFieldName(full_text_search_->field_name, field_names,
file_->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();
+}
+
+Result<RoaringBitmap32> FileIndexReadExecutor::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 (dynamic_cast<Bitmap64DeletionVector*>(dv.get())) {
+ return Status::NotImplemented(
+ "Precise file-index evaluation does not support bitmap64
deletion vectors");
+ }
+ 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<FileIndexReadExecutor::Evaluation> FileIndexReadExecutor::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<FileIndexReadExecutor::Evaluation>
FileIndexReadExecutor::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>> FileIndexReadExecutor::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>> FileIndexReadExecutor::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<FileIndexReadExecutor::Evaluation> FileIndexReadExecutor::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) {
+ PAIMON_ASSIGN_OR_RAISE(index_result,
indexes_->EvaluateLeaf(iter->second, needs_scores));
Review Comment:
Please add a TODO here: in the future, add `candidate` (may exclude dv) to
`EvaluateLeaf` to improve index skipping.
##########
src/paimon/common/file_index/empty/empty_file_index_reader.h:
##########
@@ -37,15 +38,33 @@ class Literal;
/// Vector and full-text searches return no matches.
class EmptyFileIndexReader : public FileIndexReader {
public:
+ explicit EmptyFileIndexReader(std::vector<SearchType> search_types = {})
+ : search_types_(std::move(search_types)) {}
Review Comment:
Please avoid default parameters in the production code.
##########
include/paimon/file_index/file_index_reader.h:
##########
@@ -42,8 +44,18 @@ struct FullTextMatch;
/// of the indexed data; otherwise, index filtering results may be incorrect.
class PAIMON_EXPORT FileIndexReader : public
FunctionVisitor<std::shared_ptr<FileIndexResult>> {
public:
+ enum class SearchType { FULL_TEXT_MATCH, FULL_TEXT_SEARCH, VECTOR_SEARCH };
+
~FileIndexReader() override = default;
+ /// Whether this reader can execute the requested search without probing
it with a query.
+ virtual bool SupportsSearch(SearchType search_type) const;
+
+ /// Estimated matching fraction in [0, 1], or unknown. Estimates only
affect execution order.
+ /// Existing scalar indexes use the default unknown estimate.
+ virtual std::optional<double> EstimateSelectivity(
+ const std::shared_ptr<LeafPredicate>& predicate) const;
Review Comment:
Existing indexes like bitmap could also roughly infer selectivity. Please
add a TODO indicating we can support this in the future.
##########
src/paimon/core/operation/abstract_split_read.h:
##########
@@ -158,7 +171,8 @@ class AbstractSplitRead : public SplitRead {
const BinaryRow& partition, std::unique_ptr<ReaderBuilder>
reader_builder,
const FieldMappingBuilder* field_mapping_builder,
DeletionVector::Factory dv_factory,
const std::optional<std::vector<Range>>& row_ranges,
- const std::shared_ptr<DataFilePathFactory>& data_file_path_factory)
const;
+ const std::shared_ptr<DataFilePathFactory>& data_file_path_factory,
+ const RoaringBitmap32* exact_selection = nullptr) const;
Review Comment:
Please avoid default parameters in the production code.
##########
src/paimon/core/operation/abstract_split_read.h:
##########
@@ -131,6 +142,8 @@ class AbstractSplitRead : public SplitRead {
const std::optional<std::vector<std::string>>& write_cols);
private:
+ friend class FileIndexReadExecutor;
+
Review Comment:
Why do we need `friend`? I can’t see any clear reason why these two classes
require friendship—please adjust it.
##########
src/paimon/core/operation/file_index_read_executor.cpp:
##########
@@ -0,0 +1,553 @@
+/*
+ * 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/file_index_read_executor.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/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;
+}
+
+} // namespace
+
+Result<std::unique_ptr<FileIndexReadExecutor>> FileIndexReadExecutor::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::unique_ptr<FileIndexReadExecutor> executor(new FileIndexReadExecutor(
+ split_read, partition, file, field_mapping_builder, path_factory,
extra_format_options));
+ PAIMON_RETURN_NOT_OK(executor->Prepare());
+ return executor;
+}
+
+FileIndexReadExecutor::FileIndexReadExecutor(
+ 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)
+ : split_read_(split_read),
+ partition_(partition),
+ file_(file),
+ field_mapping_builder_(field_mapping_builder),
+ path_factory_(path_factory),
+ extra_format_options_(extra_format_options) {}
+
+Status FileIndexReadExecutor::Prepare() {
+ 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(indexes_, FileIndexEvaluator::Create(
+ data_schema, index_options,
path_factory_, file_,
+
split_read_.context_->GetCoreOptions().GetFileSystem(),
+ /*only_use_embedding_index=*/false,
split_read_.pool_));
+
+ std::map<std::string, DataField> read_by_name;
+ for (const auto& field : read_fields) {
+ read_by_name.emplace(field.Name(), field);
+ }
+ std::map<int32_t, std::pair<int32_t, DataField>> data_by_id;
+ for (size_t i = 0; i < data_fields.size(); ++i) {
Review Comment:
Please add comments to clarify the meaning of the `data_by_id` key and value.
##########
src/paimon/core/operation/file_store_scan.cpp:
##########
@@ -122,9 +122,13 @@ Result<bool> FileStoreScan::TestFileIndex(const
std::shared_ptr<Predicate>& pred
std::map<std::string, std::string> file_index_options;
PAIMON_ASSIGN_OR_RAISE(file_index_options,
FileIndexOptions::RewriteFieldNames(
core_options_.ToMap(),
field_name_mapping));
+ PAIMON_ASSIGN_OR_RAISE(
+ std::unique_ptr<FileIndexEvaluator> evaluator,
+ FileIndexEvaluator::Create(data_arrow_schema, file_index_options,
/*path_factory=*/nullptr,
+ meta, /*file_system=*/nullptr,
/*only_use_embedding_index=*/true,
+ pool_));
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FileIndexResult> index_result,
- FileIndexEvaluator::Evaluate(data_arrow_schema,
file_index_options,
- data_predicate, meta,
pool_));
+ evaluator->Evaluate(data_predicate));
return index_result->IsRemain();
Review Comment:
Upstream already checks `meta->embedded_index != nullptr` before entering
this path, so the `only_use_embedding_index` parameter seems to have no effect.
If `embedding_index` exists, we should always use it; if so, just return early.
Please remove this parameter.
##########
src/paimon/core/operation/file_index_read_executor.cpp:
##########
@@ -0,0 +1,553 @@
+/*
+ * 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/file_index_read_executor.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/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;
+}
+
+} // namespace
+
+Result<std::unique_ptr<FileIndexReadExecutor>> FileIndexReadExecutor::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::unique_ptr<FileIndexReadExecutor> executor(new FileIndexReadExecutor(
+ split_read, partition, file, field_mapping_builder, path_factory,
extra_format_options));
+ PAIMON_RETURN_NOT_OK(executor->Prepare());
+ return executor;
+}
+
+FileIndexReadExecutor::FileIndexReadExecutor(
+ 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)
+ : split_read_(split_read),
+ partition_(partition),
+ file_(file),
+ field_mapping_builder_(field_mapping_builder),
+ path_factory_(path_factory),
+ extra_format_options_(extra_format_options) {}
+
+Status FileIndexReadExecutor::Prepare() {
+ 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(indexes_, FileIndexEvaluator::Create(
+ data_schema, index_options,
path_factory_, file_,
+
split_read_.context_->GetCoreOptions().GetFileSystem(),
+ /*only_use_embedding_index=*/false,
split_read_.pool_));
+
+ std::map<std::string, DataField> read_by_name;
+ for (const auto& field : read_fields) {
+ read_by_name.emplace(field.Name(), field);
+ }
+ 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]));
+ }
+ // 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));
+ return PrepareSearch(field_names);
+}
+
+Status FileIndexReadExecutor::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) {
+ 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));
+ }
+ 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(split_read_.context_->GetPartitionKeys(),
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 FileIndexReadExecutor::PrepareSearch(const std::map<std::string,
std::string>& field_names) {
+ if (split_read_.context_->GetVectorSearch()) {
+ vector_search_ =
std::make_shared<VectorSearch>(*split_read_.context_->GetVectorSearch());
+ PAIMON_ASSIGN_OR_RAISE(
+ vector_search_->field_name,
+ ResolveSearchFieldName(vector_search_->field_name, field_names,
file_->file_name));
+ PAIMON_ASSIGN_OR_RAISE(
+ [[maybe_unused]] std::shared_ptr<FileIndexReader> reader,
+ indexes_->GetSearchReader(vector_search_->field_name,
+
FileIndexReader::SearchType::VECTOR_SEARCH));
+ }
+ if (split_read_.context_->GetFullTextSearch()) {
+ full_text_search_ =
+
std::make_shared<FullTextSearch>(*split_read_.context_->GetFullTextSearch());
+ PAIMON_ASSIGN_OR_RAISE(
+ full_text_search_->field_name,
+ ResolveSearchFieldName(full_text_search_->field_name, field_names,
file_->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();
+}
+
+Result<RoaringBitmap32> FileIndexReadExecutor::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 (dynamic_cast<Bitmap64DeletionVector*>(dv.get())) {
+ return Status::NotImplemented(
+ "Precise file-index evaluation does not support bitmap64
deletion vectors");
+ }
Review Comment:
The `dynamic_cast<Bitmap64DeletionVector*>(dv.get())` seems
unnecessary—right now it should always route into the `else if (dv)` branch.
##########
test/inte/write_and_read_inte_with_index_test.cpp:
##########
@@ -1723,6 +1787,138 @@ TEST_P(WriteAndReadInteWithIndexTest,
TestBitmapIndexWithDv) {
}
}
+TEST_P(WriteAndReadInteWithIndexTest, TestPkDvPreciseScalarPredicates) {
+ std::shared_ptr<DataSplitImpl> split;
+ ASSERT_NO_FATAL_FAILURE(CreatePkDvSplit(&split));
+ const arrow::FieldVector fields = {arrow::field("key", arrow::int32()),
+ arrow::field("f0", arrow::utf8()),
+ arrow::field("f3", arrow::float64())};
+ auto name =
+ PredicateBuilder::Equal(1, "f0", FieldType::STRING,
Literal(FieldType::STRING, "Bob", 3));
+ auto value = PredicateBuilder::GreaterThan(2, "f3", FieldType::DOUBLE,
Literal(15.0));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Predicate> conjunction,
+ PredicateBuilder::And({name, value}));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Predicate> disjunction,
+ PredicateBuilder::Or({name, value}));
+ CheckPkDvPreciseResult(split, fields, R"([[0,5,"Bob",16.1]])",
conjunction);
+ CheckPkDvPreciseResult(
+ split, fields,
+
R"([[0,1,"Bob",12.1],[0,5,"Bob",16.1],[0,6,"Tony",17.1],[0,7,"Alice",18.1]])",
disjunction);
+ CheckPkDvPreciseResult(split, fields,
+
R"([[0,1,"Bob",12.1],[0,2,"Emily",13.1],[0,3,"Tony",14.1],
+ [0,5,"Bob",16.1],[0,6,"Tony",17.1],[0,7,"Alice",18.1]])",
+ /*predicate=*/nullptr);
+}
+
+TEST_P(WriteAndReadInteWithIndexTest, TestPkDvMixedFilePredicatesAndScores) {
+ const std::string query = R"({"match":{"query":"names"}})";
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<Bytes> index,
+ MockSearchFileIndex::MakeQueryIndex(
+ "f0", MockSearchFileIndex::FULL_TEXT_IDENTIFIER,
+ {{query, {{{0, 0.875f}, {4, 0.75f}, {7, 0.5f}, {1, 0.25f}, {6,
0.125f}}, 0.5}}},
+ pool_));
+ std::shared_ptr<DataSplitImpl> split;
+ ASSERT_NO_FATAL_FAILURE(CreatePkDvSplit(&split, index));
+ {
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<Predicate> text,
+ PredicateBuilder::FullTextMatch(2,
std::make_shared<FullTextMatch>("f0", query)));
+ auto scalar = PredicateBuilder::GreaterThan(3, "f3",
FieldType::DOUBLE, Literal(15.0));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Predicate> predicate,
+ PredicateBuilder::And({text, scalar}));
+ CheckPkDvPreciseResult(
+ split,
+ {arrow::field("_f0_INDEX_SCORE", arrow::float32()),
arrow::field("key", arrow::int32()),
+ arrow::field("f0", arrow::utf8()), arrow::field("f3",
arrow::float64())},
+ R"([[0,0.125,6,"Tony",17.1],[0,0.5,7,"Alice",18.1]])", predicate);
+ }
+ {
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<Predicate> text,
+ PredicateBuilder::FullTextMatch(1,
std::make_shared<FullTextMatch>("f0", query)));
+ auto scalar = PredicateBuilder::GreaterThan(2, "f3",
FieldType::DOUBLE, Literal(15.0));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Predicate> predicate,
+ PredicateBuilder::Or({text, scalar}));
+ CheckPkDvPreciseResult(
+ split,
+ {arrow::field("key", arrow::int32()), arrow::field("f0",
arrow::utf8()),
+ arrow::field("f3", arrow::float64())},
+
R"([[0,1,"Bob",12.1],[0,5,"Bob",16.1],[0,6,"Tony",17.1],[0,7,"Alice",18.1]])",
+ predicate);
+ }
+}
+
+TEST_P(WriteAndReadInteWithIndexTest, TestPkDvFullTextTopKFiltersBeforeLimit) {
+ const std::string query = R"({"match":{"query":"ranked"}})";
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Bytes> index,
+ MockSearchFileIndex::MakeQueryIndex("f0",
statistics_index_type_,
+ {{query,
+ {{{0, 1.0f},
+ {4, 0.875f},
+ {7, 0.75f},
+ {5, 0.625f},
+ {2, 0.5f},
+ {1, 0.375f},
+ {3, 0.25f},
+ {6, 0.125f}},
+
std::nullopt}}},
+ pool_));
+ auto search = std::make_shared<FullTextSearch>("f0", query, /*limit=*/2);
+ auto predicate = PredicateBuilder::LessThan(2, "f3", FieldType::DOUBLE,
Literal(17.0));
+ std::shared_ptr<DataSplitImpl> split;
+ ASSERT_NO_FATAL_FAILURE(CreatePkDvSplit(&split, index));
+ // Rows 0 and 4 are deleted, and row 7 fails the scalar filter. All three
rank ahead of
+ // the surviving TopK, whose scores must be reordered with the physical
output rows.
+ CheckPkDvPreciseResult(
+ split,
+ {arrow::field("key", arrow::int32()), arrow::field("_f0_INDEX_SCORE",
arrow::float32()),
+ arrow::field("f3", arrow::float64())},
+ R"([[0,2,0.5,13.1],[0,5,0.625,16.1]])", predicate,
/*vector_search=*/nullptr, search);
+ ASSERT_EQ(statistics_->executions.size(), 1);
+ ASSERT_TRUE(statistics_->executions.front().needs_scores);
+ ASSERT_FALSE(search->include_row_ids);
+}
+
+TEST_P(WriteAndReadInteWithIndexTest,
TestPkDvFullTextAndVectorTopKScoreAlignment) {
+ const std::string query = R"({"match":{"query":"names"}})";
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<Bytes> text_payload,
+ MockSearchFileIndex::MakeQueryPayload(
+ {{query, {{{0, 1.0f}, {4, 0.875f}, {2, 0.75f}, {6, 0.5f}, {7,
0.25f}}, 0.5}}}, pool_));
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Bytes> vector_payload,
+ MockSearchFileIndex::MakePayload({{0, 0.125f},
+ {4, 0.25f},
+ {5, 0.375f},
+ {7, 0.5f},
+ {2, 0.625f},
+ {6, 0.75f},
+ {1, 0.875f},
+ {3, 1.0f}},
+ pool_));
+ FileIndexFormat::ColumnIndexes indexes;
+ indexes["f0"][MockSearchFileIndex::FULL_TEXT_IDENTIFIER] = text_payload;
+ indexes["f3"][MockSearchFileIndex::VECTOR_IDENTIFIER] = vector_payload;
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Bytes> index,
+ MockSearchFileIndex::Serialize(indexes, pool_));
+ std::shared_ptr<DataSplitImpl> split;
+ ASSERT_NO_FATAL_FAILURE(CreatePkDvSplit(&split, index));
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<Predicate> text,
+ PredicateBuilder::FullTextMatch(2,
std::make_shared<FullTextMatch>("f0", query)));
+ auto search = std::make_shared<VectorSearch>(
+ "f3", /*limit=*/2, std::vector<float>{1.0f}, nullptr, nullptr,
+ VectorSearch::DistanceType::EUCLIDEAN, std::map<std::string,
std::string>{});
+ // The DV excludes rows 0 and 4 and FullTextMatch excludes row 5 before
vector TopK.
+ // The vector field itself is not projected, and both score columns follow
the final rows.
+ CheckPkDvPreciseResult(
+ split,
+ {arrow::field("_f3_INDEX_SCORE", arrow::float32()),
arrow::field("key", arrow::int32()),
+ arrow::field("f0", arrow::utf8()), arrow::field("_f0_INDEX_SCORE",
arrow::float32())},
+ R"([[0,0.625,2,"Emily",0.75],[0,0.5,7,"Alice",0.25]])", text, search);
+ ASSERT_FALSE(search->pre_filter);
+}
+
TEST_P(WriteAndReadInteWithIndexTest, TestWithAlterTable) {
Review Comment:
Please add an integrated test for schema evolution. Ideally, use a predicate
that cannot be applied after the evolution and must rely on scan data filtering.
##########
src/paimon/core/operation/file_index_read_executor.cpp:
##########
@@ -0,0 +1,553 @@
+/*
+ * 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/file_index_read_executor.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/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;
+}
+
+} // namespace
+
+Result<std::unique_ptr<FileIndexReadExecutor>> FileIndexReadExecutor::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::unique_ptr<FileIndexReadExecutor> executor(new FileIndexReadExecutor(
+ split_read, partition, file, field_mapping_builder, path_factory,
extra_format_options));
+ PAIMON_RETURN_NOT_OK(executor->Prepare());
+ return executor;
+}
+
+FileIndexReadExecutor::FileIndexReadExecutor(
+ 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)
+ : split_read_(split_read),
+ partition_(partition),
+ file_(file),
+ field_mapping_builder_(field_mapping_builder),
+ path_factory_(path_factory),
+ extra_format_options_(extra_format_options) {}
+
+Status FileIndexReadExecutor::Prepare() {
+ 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(indexes_, FileIndexEvaluator::Create(
+ data_schema, index_options,
path_factory_, file_,
+
split_read_.context_->GetCoreOptions().GetFileSystem(),
+ /*only_use_embedding_index=*/false,
split_read_.pool_));
+
+ std::map<std::string, DataField> read_by_name;
+ for (const auto& field : read_fields) {
+ read_by_name.emplace(field.Name(), field);
+ }
+ 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]));
+ }
+ // 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));
+ return PrepareSearch(field_names);
+}
+
+Status FileIndexReadExecutor::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) {
+ 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));
+ }
+ 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(split_read_.context_->GetPartitionKeys(),
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 FileIndexReadExecutor::PrepareSearch(const std::map<std::string,
std::string>& field_names) {
+ if (split_read_.context_->GetVectorSearch()) {
+ vector_search_ =
std::make_shared<VectorSearch>(*split_read_.context_->GetVectorSearch());
+ PAIMON_ASSIGN_OR_RAISE(
+ vector_search_->field_name,
+ ResolveSearchFieldName(vector_search_->field_name, field_names,
file_->file_name));
+ PAIMON_ASSIGN_OR_RAISE(
+ [[maybe_unused]] std::shared_ptr<FileIndexReader> reader,
+ indexes_->GetSearchReader(vector_search_->field_name,
+
FileIndexReader::SearchType::VECTOR_SEARCH));
+ }
+ if (split_read_.context_->GetFullTextSearch()) {
+ full_text_search_ =
+
std::make_shared<FullTextSearch>(*split_read_.context_->GetFullTextSearch());
+ PAIMON_ASSIGN_OR_RAISE(
+ full_text_search_->field_name,
+ ResolveSearchFieldName(full_text_search_->field_name, field_names,
file_->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();
+}
+
+Result<RoaringBitmap32> FileIndexReadExecutor::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 (dynamic_cast<Bitmap64DeletionVector*>(dv.get())) {
+ return Status::NotImplemented(
+ "Precise file-index evaluation does not support bitmap64
deletion vectors");
+ }
+ 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<FileIndexReadExecutor::Evaluation> FileIndexReadExecutor::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<FileIndexReadExecutor::Evaluation>
FileIndexReadExecutor::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>> FileIndexReadExecutor::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>> FileIndexReadExecutor::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<FileIndexReadExecutor::Evaluation> FileIndexReadExecutor::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) {
+ 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>>
FileIndexReadExecutor::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> FileIndexReadExecutor::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()));
+ }
+ 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(); });
+ RoaringBitmap32 matches;
+ while (true) {
+ PAIMON_ASSIGN_OR_RAISE(BatchReader::ReadBatchWithBitmap batch,
+ filtered_reader->NextBatchWithBitmap());
+ if (BatchReader::IsEofBatch(batch)) {
Review Comment:
It seems there may be a `batch` memory leak. At least make sure to free the
c schema/array.
--
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]
