feat(file-index): support precise file index predicate evaluation

docs/source/user_guide/file_index.rst

@@ -21,6 +21,20 @@ File Index
 Full-text predicates and root TopK
 ----------------------------------
 
+Enable ``ReadContextBuilder::EnablePreciseFileIndexEvaluate(true)`` to precisely
+filter rows at the file-index-query stage, before reading all columns requested by
+the query from each data file. It evaluates the complete predicate to compute
+exact matching row positions within that file.
+
+For a root full-text or vector TopK request, predicate filtering, row-range
+restrictions and deletion vectors are applied before the TopK limit. Filtering after
+TopK truncation could discard selected rows. The final row positions are then passed
+to the data-file reader to restrict the read of all requested columns.
+
+The option defaults to ``false`` and requires ``file-index.read.enabled=true``.
+Full-text predicates and root TopK requests require this option; configuring
+them while it is disabled is rejected by ``ReadContextBuilder::Finish()``.
+
 ``PredicateBuilder::FullTextMatch`` creates a ``LeafPredicate`` for a full-text
 condition without TopK truncation. It holds a shared ``FullTextMatch`` request,
 whose query uses the same JSON DSL as ``FullTextSearch``. Do not modify a request
@@ -45,7 +59,9 @@ positions in that schema. For example, the following constructs
        std::shared_ptr<paimon::Predicate> filter,
        paimon::PredicateBuilder::Or({text, price}));
    paimon::ReadContextBuilder builder(table_path);
-   builder.SetReadFieldNames({"body", "price"}).SetPredicate(filter);
+   builder.EnablePreciseFileIndexEvaluate(true)
+       .SetReadFieldNames({"body", "price"})
+       .SetPredicate(filter);
 
 ``SetVectorSearch`` and ``SetFullTextSearch`` configure a separate root TopK
 request. They are mutually exclusive and require a positive limit. The intended
@@ -54,7 +70,9 @@ field does not need to be projected. A full-text match inside a predicate never
 has a TopK limit.
 
 Score output is requested through an explicit ``_<field>_INDEX_SCORE``
-projection. For example, request the score of a full-text
+projection. Precise evaluation rejects the unnamed ``_INDEX_SCORE`` column,
+whether selected by field name, field ID or an explicit read schema.
+For example, request the score of a full-text
 predicate by projecting ``_body_INDEX_SCORE``:
 
 .. code-block:: cpp
@@ -65,7 +83,9 @@ predicate by projecting ``_body_INDEX_SCORE``:
        std::shared_ptr<paimon::Predicate> text,
        paimon::PredicateBuilder::FullTextMatch(0, request));
    paimon::ReadContextBuilder builder(table_path);
-   builder.SetReadFieldNames({"body", "_body_INDEX_SCORE"}).SetPredicate(text);
+   builder.EnablePreciseFileIndexEvaluate(true)
+       .SetReadFieldNames({"body", "_body_INDEX_SCORE"})
+       .SetPredicate(text);
 
 Score fields have type ``float32`` and preserve the source field's exact name,
 including case. Select them with ``SetReadFieldNames`` or ``SetReadSchema``;

include/paimon/file_index/file_index_reader.h

@@ -20,6 +20,7 @@
 
 #include <cstdint>
 #include <memory>
+#include <optional>
 #include <vector>
 
 #include "paimon/file_index/file_index_result.h"
@@ -31,6 +32,7 @@
 #include "paimon/visibility.h"
 
 namespace paimon {
+class LeafPredicate;
 struct FullTextMatch;
 /// Evaluates filter predicates against a file-level index to determine file eligibility.
 ///
@@ -42,8 +44,21 @@ 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;
+
+    /// Estimates the fraction of rows in the data file expected to match the predicate.
+    /// Returns a value in [0, 1], or std::nullopt if no estimate is available.
+    /// Values outside [0, 1] are also treated as unknown.
+    /// The estimate only guides predicate execution order; it is not used to filter rows.
+    /// TODO(lisizhuo.lsz): Add selectivity estimates for scalar indexes such as bitmap indexes.
+    virtual std::optional<double> EstimateSelectivity(
+        const std::shared_ptr<LeafPredicate>& predicate) const;
+
     Result<std::shared_ptr<FileIndexResult>> VisitIsNotNull() override;
 
     Result<std::shared_ptr<FileIndexResult>> VisitIsNull() override;
@@ -77,13 +92,16 @@ class PAIMON_EXPORT FileIndexReader : public FunctionVisitor<std::shared_ptr<Fil
     virtual Result<std::shared_ptr<ScoredFileIndexResult>> VisitVectorSearch(
         const std::shared_ptr<VectorSearch>& vector_search);
 
-    /// Execute full-text search and return matching file-local physical row positions.
-    virtual Result<std::shared_ptr<FileIndexResult>> VisitFullTextSearch(
+    /// Execute full-text TopK search and return scored file-local physical row positions.
+    /// Scores are required for merging results across files, regardless of score projection.
+    virtual Result<std::shared_ptr<ScoredFileIndexResult>> VisitFullTextSearch(
         const std::shared_ptr<FullTextSearch>& full_text_search);
 
     /// Return every matching file-local physical row position, without TopK truncation.
-    Result<std::shared_ptr<FileIndexResult>> VisitFullTextMatch(
-        const std::shared_ptr<FullTextMatch>& full_text_match) override;
+    /// When needs_scores is true, return a ScoredFileIndexResult, including for empty results.
+    /// When needs_scores is false, return a BitmapIndexResult, not a ScoredFileIndexResult.
+    virtual Result<std::shared_ptr<FileIndexResult>> VisitFullTextMatch(
+        const std::shared_ptr<FullTextMatch>& full_text_match, bool needs_scores = false);
 };
 
 }  // namespace paimon

include/paimon/file_index/file_indexer.h

@@ -47,6 +47,10 @@ class PAIMON_EXPORT FileIndexer {
         return false;
     }
 
+    /// Create a reader for an empty index entry without serialized bytes.
+    /// Scalar indexers use the default reader. Search indexers must preserve their capabilities.
+    virtual std::shared_ptr<FileIndexReader> CreateEmptyReader() const;
+
     /// Extra columns required while building this index, in addition to the indexed column.
     virtual Result<std::optional<std::vector<std::string>>> GetExtraFieldNames() const {
         return std::optional<std::vector<std::string>>(std::nullopt);

include/paimon/file_index/scored_file_index_result.h

@@ -29,7 +29,7 @@
 
 namespace paimon {
 
-/// File-local vector search result. Scores correspond to row positions in ascending order.
+/// File-local search result. Scores correspond to row positions in ascending order.
 class PAIMON_EXPORT ScoredFileIndexResult : public FileIndexResult {
  public:
     static Result<std::shared_ptr<ScoredFileIndexResult>> Create(RoaringBitmap32&& row_positions,

include/paimon/predicate/function_visitor.h

@@ -18,20 +18,16 @@
 
 #pragma once
 
-#include <memory>
 #include <string>
 #include <vector>
 
 #include "paimon/predicate/leaf_predicate.h"
 #include "paimon/predicate/literal.h"
 #include "paimon/result.h"
-#include "paimon/status.h"
 #include "paimon/visibility.h"
 
 namespace paimon {
-struct FullTextMatch;
-
-/// A visitor interface for evaluating filter predicates on indexed columns.
+/// A visitor interface for evaluating scalar filter predicates on indexed columns.
 /// @tparam T The result type produced by each visit method (e.g., a file index result or global
 /// index result).
 template <typename T>
@@ -80,11 +76,5 @@ class PAIMON_EXPORT FunctionVisitor {
 
     /// Evaluates whether string values like the given string.
     virtual Result<T> VisitLike(const Literal& literal) = 0;
-
-    /// Evaluates a full-text condition without TopK truncation.
-    /// Visitors without full-text support return NotImplemented.
-    virtual Result<T> VisitFullTextMatch(const std::shared_ptr<FullTextMatch>& full_text_match) {
-        return Status::NotImplemented("This visitor does not support complete full-text matching");
-    }
 };
 }  // namespace paimon

include/paimon/predicate/predicate_utils.h

@@ -64,6 +64,7 @@ class PAIMON_EXPORT PredicateUtils {
     /// Whether the predicate contains a full-text matching condition.
     static bool ContainsFullTextMatch(const std::shared_ptr<Predicate>& predicate);
 
+    /// Dispatches scalar leaf predicates. Search predicates use dedicated index reader methods.
     template <typename T>
     static Result<T> VisitPredicate(const std::shared_ptr<LeafPredicate>& predicate,
                                     const std::shared_ptr<FunctionVisitor<T>>& visitor) {
@@ -117,21 +118,13 @@ class PAIMON_EXPORT PredicateUtils {
                 assert(predicate->Literals().size() == 1);
                 return visitor->VisitLike(predicate->Literals()[0]);
             }
-            case Function::Type::FULL_TEXT_MATCH: {
-                PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<FullTextMatch> full_text_match,
-                                       GetFullTextMatch(predicate));
-                return visitor->VisitFullTextMatch(full_text_match);
-            }
             default:
                 return Status::Invalid("invalid " + predicate->GetFunction().ToString() +
                                        " function in leaf predicate");
         }
     }
 
  private:
-    static Result<std::shared_ptr<FullTextMatch>> GetFullTextMatch(
-        const std::shared_ptr<LeafPredicate>& predicate);
-
     static Result<std::optional<std::shared_ptr<Predicate>>> ReconstructPredicateWithPickedFields(
         const std::shared_ptr<Predicate>& predicate,
         const std::map<std::string, int32_t>& picked_field_name_to_idx);

include/paimon/read_context.h

@@ -52,23 +52,26 @@ struct VectorSearch;
 /// @see ReadContextBuilder
 class PAIMON_EXPORT ReadContext {
  public:
-    ReadContext(
-        const std::string& path, const std::string& branch,
-        const std::vector<std::string>& read_field_names,
-        const std::vector<int32_t>& read_field_ids, const std::shared_ptr<Predicate>& predicate,
-        const std::shared_ptr<VectorSearch>& vector_search,
-        const std::shared_ptr<FullTextSearch>& full_text_search, bool enable_predicate_filter,
-        bool enable_prefetch, bool enable_late_materializing, uint32_t prefetch_batch_count,
-        uint32_t prefetch_max_parallel_num, bool enable_multi_thread_row_to_batch,
-        uint32_t row_to_batch_thread_number, const std::optional<std::string>& table_schema,
-        const std::shared_ptr<MemoryPool>& memory_pool, const std::shared_ptr<Executor>& executor,
-        const std::shared_ptr<FileSystem>& specific_file_system,
-        const std::map<std::string, std::string>& fs_scheme_to_identifier_map,
-        const std::shared_ptr<RealtimeContext>& realtime_context,
-        const std::map<std::string, std::string>& options, bool read_ahead_cache_enabled,
-        const CacheConfig& cache_config, const std::shared_ptr<Cache>& cache,
-        const std::shared_ptr<FormatTable>& format_table, WarmupLevel warmup_level,
-        const std::shared_ptr<Catalog>& catalog, const std::optional<Identifier>& identifier);
+    ReadContext(const std::string& path, const std::string& branch,
+                const std::vector<std::string>& read_field_names,
+                const std::vector<int32_t>& read_field_ids,
+                const std::shared_ptr<Predicate>& predicate,
+                const std::shared_ptr<VectorSearch>& vector_search,
+                const std::shared_ptr<FullTextSearch>& full_text_search,
+                bool enable_precise_file_index_evaluate, bool enable_predicate_filter,
+                bool enable_prefetch, bool enable_late_materializing, uint32_t prefetch_batch_count,
+                uint32_t prefetch_max_parallel_num, bool enable_multi_thread_row_to_batch,
+                uint32_t row_to_batch_thread_number, const std::optional<std::string>& table_schema,
+                const std::shared_ptr<MemoryPool>& memory_pool,
+                const std::shared_ptr<Executor>& executor,
+                const std::shared_ptr<FileSystem>& specific_file_system,
+                const std::map<std::string, std::string>& fs_scheme_to_identifier_map,
+                const std::shared_ptr<RealtimeContext>& realtime_context,
+                const std::map<std::string, std::string>& options, bool read_ahead_cache_enabled,
+                const CacheConfig& cache_config, const std::shared_ptr<Cache>& cache,
+                const std::shared_ptr<FormatTable>& format_table, WarmupLevel warmup_level,
+                const std::shared_ptr<Catalog>& catalog,
+                const std::optional<Identifier>& identifier);
 
     ~ReadContext();
 
@@ -113,6 +116,10 @@ class PAIMON_EXPORT ReadContext {
         return vector_search_ != nullptr || full_text_search_ != nullptr;
     }
 
+    bool EnablePreciseFileIndexEvaluate() const {
+        return enable_precise_file_index_evaluate_;
+    }
+
     bool EnablePredicateFilter() const {
         return enable_predicate_filter_;
     }
@@ -208,6 +215,7 @@ class PAIMON_EXPORT ReadContext {
     std::shared_ptr<Predicate> predicate_;
     std::shared_ptr<VectorSearch> vector_search_;
     std::shared_ptr<FullTextSearch> full_text_search_;
+    bool enable_precise_file_index_evaluate_;
     bool enable_predicate_filter_;
     bool enable_prefetch_;
     bool enable_late_materializing_;
@@ -352,39 +360,52 @@ 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): Align with Java Paimon so fields referenced by predicates no longer need
+    /// to appear in the projection.
+    /// Full-text matches can be combined with scalar predicates using PredicateBuilder::And/Or.
+    /// Predicates on virtual index score columns (_INDEX_SCORE or _<field>_INDEX_SCORE) are not
+    /// supported.
     ///
     /// @param predicate Shared pointer to the predicate for data filtering.
     /// @return Reference to this builder for method chaining.
+    /// @note Predicates containing FullTextMatch require EnablePreciseFileIndexEvaluate(true).
     ReadContextBuilder& SetPredicate(const std::shared_ptr<Predicate>& predicate);
 
     /// Configure the root vector TopK request, logically applied after the predicate filter.
     /// The search field need not be projected. SetVectorSearch is mutually exclusive with
     /// SetFullTextSearch and must have a positive limit.
     /// Scores are required internally for TopK. Project `_<field>_INDEX_SCORE` to include them
-    /// in the result. Per-field score output is not implemented yet; reading such a projection
-    /// returns NotImplemented.
-    /// TODO(lisizhuo.lsz): Execute combined read predicates and merge TopK across files.
+    /// in the result.
+    /// TODO(lisizhuo.lsz): Merge TopK across files.
     /// @param vector_search Vector search request, or nullptr to clear it.
     /// @return Reference to this builder for method chaining.
+    /// @note Requires EnablePreciseFileIndexEvaluate(true).
     ReadContextBuilder& SetVectorSearch(const std::shared_ptr<VectorSearch>& vector_search);
 
     /// Configure the root full-text TopK request, logically applied after the predicate filter.
     /// The search field need not be projected. Use PredicateBuilder::FullTextMatch for a
     /// full-text filter without TopK truncation. SetFullTextSearch is mutually exclusive with
     /// SetVectorSearch and must have a positive limit.
     /// Scores are required internally for TopK. Project `_<field>_INDEX_SCORE` to include them
-    /// in the result. Per-field score output is not implemented yet; reading such a projection
-    /// returns NotImplemented.
-    /// TODO(lisizhuo.lsz): Execute combined read predicates and merge TopK across files.
+    /// in the result.
+    /// TODO(lisizhuo.lsz): Merge TopK across files.
     /// @param full_text_search Full-text search request, or nullptr to clear it.
     /// @return Reference to this builder for method chaining.
+    /// @note Requires EnablePreciseFileIndexEvaluate(true).
     ReadContextBuilder& SetFullTextSearch(const std::shared_ptr<FullTextSearch>& full_text_search);
 
+    /// Compute exact file-local result row positions during file-index evaluation, using the
+    /// configured Predicate and optional VectorSearch or FullTextSearch request.
+    /// Requires file-index.read.enabled=true. Reads available file indexes and rechecks scalar
+    /// index candidates against the data. If a scalar predicate's field has no index, reads the
+    /// data to compute exact matching rows. Only scalar predicates may be evaluated without an
+    /// index. Score output must use named projections such as _<field>_INDEX_SCORE;
+    /// projecting _INDEX_SCORE by name, field ID or read schema is not supported in this mode.
+    /// The default is false.
+    /// @param enabled Whether to compute exact result row positions during file-index evaluation.
+    /// @return Reference to this builder for method chaining.
+    ReadContextBuilder& EnablePreciseFileIndexEvaluate(bool enabled);
+
     /// Whether to perform precise filtering according to predicates for data read from format
     /// reader.
     /// @param enabled Whether to enable precise filtering (default: false)

src/paimon/CMakeLists.txt

@@ -69,6 +69,7 @@ set(PAIMON_COMMON_SRCS
     common/factories/io_hook.cpp
     common/factories/factory_creator.cpp
     common/file_index/bitmap/bitmap_index_result.cpp
+    common/file_index/file_indexer.cpp
     common/file_index/file_indexer_factory.cpp
     common/file_index/file_index_format.cpp
     common/file_index/file_index_format_utils.cpp
@@ -412,6 +413,7 @@ set(PAIMON_CORE_SRCS
     core/operation/orphan_files_cleaner.cpp
     core/operation/orphan_files_cleaner_impl.cpp
     core/operation/raw_file_split_read.cpp
+    core/operation/precise_file_index_read.cpp
     core/operation/read_context.cpp
     core/operation/scan_context.cpp
     core/manifest/snapshot_live_manifest_entries.cpp
@@ -992,6 +994,7 @@ if(PAIMON_BUILD_TESTS)
                     core/operation/commit/realtime_deduplicate_dv_materializer_test.cpp
                     core/operation/commit/sequence_snapshot_properties_test.cpp
                     core/operation/key_value_file_store_write_test.cpp
+                    core/operation/precise_file_index_read_test.cpp
                     core/operation/internal_read_context_test.cpp
                     core/operation/abstract_split_read_test.cpp
                     core/operation/append_only_file_store_write_test.cpp