feat(file-index): support file-local search with PK+DV filtering

include/paimon/file_index/file_index_format.h

@@ -102,10 +102,12 @@ class PAIMON_EXPORT FileIndexFormat {
     /// stream.
     /// @param input_stream Input stream containing serialized index data.
     /// @param pool Memory pool for temporary allocations during reading.
+    /// @param options Table options used to configure File Index backends.
     /// @return A unique pointer to a `Reader` on success, or an error if the stream is invalid
     ///         (e.g., wrong magic, unsupported version, or corrupted data).
     static Result<std::unique_ptr<Reader>> CreateReader(
-        const std::shared_ptr<InputStream>& input_stream, const std::shared_ptr<MemoryPool>& pool);
+        const std::shared_ptr<InputStream>& input_stream, const std::shared_ptr<MemoryPool>& pool,
+        const std::map<std::string, std::string>& options);
 
     /// Creates a `Writer` which serializes a complete V1 file index container.
     ///

include/paimon/file_index/file_index_reader.h

@@ -23,7 +23,10 @@
 #include <vector>
 
 #include "paimon/file_index/file_index_result.h"
+#include "paimon/file_index/scored_file_index_result.h"
+#include "paimon/predicate/full_text_search.h"
 #include "paimon/predicate/function_visitor.h"
+#include "paimon/predicate/vector_search.h"
 #include "paimon/result.h"
 #include "paimon/visibility.h"
 
@@ -38,6 +41,8 @@ namespace paimon {
 /// of the indexed data; otherwise, index filtering results may be incorrect.
 class PAIMON_EXPORT FileIndexReader : public FunctionVisitor<std::shared_ptr<FileIndexResult>> {
  public:
+    ~FileIndexReader() override = default;
+
     Result<std::shared_ptr<FileIndexResult>> VisitIsNotNull() override;
 
     Result<std::shared_ptr<FileIndexResult>> VisitIsNull() override;
@@ -66,6 +71,14 @@ class PAIMON_EXPORT FileIndexReader : public FunctionVisitor<std::shared_ptr<Fil
     Result<std::shared_ptr<FileIndexResult>> VisitContains(const Literal& literal) override;
 
     Result<std::shared_ptr<FileIndexResult>> VisitLike(const Literal& literal) override;
+
+    /// Execute vector search and return scored file-local physical row positions.
+    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(
+        const std::shared_ptr<FullTextSearch>& full_text_search);
 };
 
 }  // namespace paimon

include/paimon/file_index/file_indexer.h

@@ -19,6 +19,9 @@
 #pragma once
 
 #include <memory>
+#include <optional>
+#include <string>
+#include <vector>
 
 #include "paimon/file_index/file_index_reader.h"
 #include "paimon/file_index/file_index_result.h"
@@ -35,6 +38,11 @@ class PAIMON_EXPORT FileIndexer {
  public:
     virtual ~FileIndexer() = default;
 
+    /// 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);
+    }
+
     /// Create `FileIndexReader` with input stream.
     ///
     /// @param arrow_schema ArrowSchema derived from arrow schema or struct type with
@@ -51,8 +59,8 @@ class PAIMON_EXPORT FileIndexer {
 
     /// Create `FileIndexWriter` for arrow schema.
     ///
-    /// @param arrow_schema ArrowSchema derived from arrow schema or struct type with
-    /// specified indexed field.
+    /// @param arrow_schema ArrowSchema containing the indexed field first, followed by any
+    /// fields requested through `GetExtraFieldNames()`.
     /// @param pool Memory pool for memory allocation.
     /// @return A `FileIndexWriter` to write index.
     virtual Result<std::shared_ptr<FileIndexWriter>> CreateWriter(

include/paimon/file_index/scored_file_index_result.h

@@ -0,0 +1,61 @@
+/*
+ * 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 <memory>
+#include <string>
+#include <vector>
+
+#include "paimon/file_index/file_index_result.h"
+#include "paimon/result.h"
+#include "paimon/utils/roaring_bitmap32.h"
+#include "paimon/visibility.h"
+
+namespace paimon {
+
+/// File-local vector 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,
+                                                                 std::vector<float>&& scores);
+
+    bool IsEmpty() const {
+        return row_positions_.IsEmpty();
+    }
+
+    Result<bool> IsRemain() const override;
+
+    const RoaringBitmap32& GetRowPositions() const {
+        return row_positions_;
+    }
+
+    const std::vector<float>& GetScores() const {
+        return scores_;
+    }
+
+    std::string ToString() const override;
+
+ private:
+    ScoredFileIndexResult(RoaringBitmap32&& row_positions, std::vector<float>&& scores);
+
+    RoaringBitmap32 row_positions_;
+    std::vector<float> scores_;
+};
+
+}  // namespace paimon

include/paimon/predicate/full_text_search.h

@@ -92,8 +92,9 @@ struct PAIMON_EXPORT FullTextSearch {
     std::string query;
     /// Type of search to perform.
     SearchType search_type;
-    /// A pre-filter based on **global row IDs**, implemented by leveraging another global index.
-    /// Only rows whose global row ID is present in `pre_filter` will be included during search.
+    /// A pre-filter whose row ID domain is determined by the reader: global row IDs for
+    /// GlobalIndexReader and physical file-local row IDs for FileIndexReader.
+    /// Only rows whose row ID is present in `pre_filter` will be included during search.
     /// If not set, all rows will be included.
     std::optional<RoaringBitmap64> pre_filter;
     /// Whether to compute and return relevance scores (e.g. BM25). The 4-path matrix:

include/paimon/predicate/vector_search.h

@@ -30,11 +30,12 @@
 namespace paimon {
 /// `VectorSearch` to perform vector similarity search.
 struct PAIMON_EXPORT VectorSearch {
-    /// `PreFilter`: A lightweight pre-filtering function applied **before** similarity
-    /// scoring. It operates solely on **global row ids** and is typically driven by other global
-    /// index, such as bitmap, or range index. This filter enables early pruning of irrelevant
-    /// candidates (e.g., "only consider rows with label X"), significantly reducing the search
-    /// space. Returns true to include the row in vector search process; false to exclude it.
+    /// `PreFilter`: A lightweight pre-filtering function applied **before** similarity scoring.
+    /// The row id domain is determined by the reader executing the search: `GlobalIndexReader`
+    /// supplies global row ids, while `FileIndexReader` supplies physical row ids local to the
+    /// current data file. This filter enables early pruning of irrelevant candidates (e.g., "only
+    /// consider rows with label X"), significantly reducing the search space. Returns true to
+    /// include the row in vector search process; false to exclude it.
     ///
     /// @note Must be thread-safe.
     using PreFilter = std::function<bool(int64_t)>;
@@ -65,7 +66,8 @@ struct PAIMON_EXPORT VectorSearch {
     int32_t limit;
     /// The query vector (must match the dimensionality of the indexed vectors).
     std::vector<float> query;
-    /// A pre-filter based on **global row ids**, implemented by leveraging other global index
+    /// A pre-filter whose row id domain is determined by the reader: global row ids for Global
+    /// Index and file-local physical row ids for File Index.
     std::function<bool(int64_t)> pre_filter;
     /// A runtime filtering condition that may involve graph traversal of
     /// structured attributes. **Using this parameter often yields better

include/paimon/read_context.h

@@ -37,11 +37,13 @@
 namespace paimon {
 class Catalog;
 class Executor;
+struct FullTextSearch;
 class FormatTable;
 class MemoryPool;
 class Predicate;
 class FileSystem;
 class RealtimeContext;
+struct VectorSearch;
 
 /// `ReadContext` is some configuration for read operations.
 ///
@@ -53,10 +55,13 @@ class PAIMON_EXPORT ReadContext {
     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, 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<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,
@@ -96,6 +101,18 @@ class PAIMON_EXPORT ReadContext {
         return predicate_;
     }
 
+    const std::shared_ptr<VectorSearch>& GetVectorSearch() const {
+        return vector_search_;
+    }
+
+    const std::shared_ptr<FullTextSearch>& GetFullTextSearch() const {
+        return full_text_search_;
+    }
+
+    bool HasFileIndexSearch() const {
+        return vector_search_ != nullptr || full_text_search_ != nullptr;
+    }
+
     bool EnablePredicateFilter() const {
         return enable_predicate_filter_;
     }
@@ -179,6 +196,8 @@ class PAIMON_EXPORT ReadContext {
     std::vector<std::string> read_field_names_;
     std::vector<int32_t> read_field_ids_;
     std::shared_ptr<Predicate> predicate_;
+    std::shared_ptr<VectorSearch> vector_search_;
+    std::shared_ptr<FullTextSearch> full_text_search_;
     bool enable_predicate_filter_;
     bool enable_prefetch_;
     bool enable_late_materializing_;
@@ -320,6 +339,18 @@ class PAIMON_EXPORT ReadContextBuilder {
     /// @return Reference to this builder for method chaining.
     ReadContextBuilder& SetPredicate(const std::shared_ptr<Predicate>& predicate);
 
+    /// Configure file-local vector search. The limit is applied independently to every data file.
+    /// Vector search and full-text search are mutually exclusive.
+    /// @param vector_search Vector search request, or nullptr to clear it.
+    /// @return Reference to this builder for method chaining.
+    ReadContextBuilder& SetVectorSearch(const std::shared_ptr<VectorSearch>& vector_search);
+
+    /// Configure file-local full-text search. The limit is applied independently to every data
+    /// file. Full-text search and vector search are mutually exclusive.
+    /// @param full_text_search Full-text search request, or nullptr to clear it.
+    /// @return Reference to this builder for method chaining.
+    ReadContextBuilder& SetFullTextSearch(const std::shared_ptr<FullTextSearch>& full_text_search);
+
     /// 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

@@ -73,6 +73,7 @@ set(PAIMON_COMMON_SRCS
     common/file_index/file_index_format.cpp
     common/file_index/file_index_reader.cpp
     common/file_index/file_index_result.cpp
+    common/file_index/scored_file_index_result.cpp
     common/format/column_stats.cpp
     common/format/file_format_factory.cpp
     common/fs/file_system.cpp
@@ -81,7 +82,6 @@ set(PAIMON_COMMON_SRCS
     common/fs/file_system_factory.cpp
     common/global_index/union_global_index_reader.cpp
     common/global_index/offset_global_index_reader.cpp
-    common/global_index/complete_index_score_batch_reader.cpp
     common/global_index/bitmap_scored_global_index_result.cpp
     common/global_index/bitmap_global_index_result.cpp
     common/global_index/global_index_result.cpp
@@ -138,6 +138,8 @@ set(PAIMON_COMMON_SRCS
     common/predicate/predicate_utils.cpp
     common/predicate/starts_with.cpp
     common/reader/batch_reader.cpp
+    common/reader/complete_index_score_batch_reader.cpp
+    common/reader/complete_index_score_file_batch_reader.cpp
     common/reader/concat_batch_reader.cpp
     common/reader/late_materializing_file_batch_reader.cpp
     common/reader/data_file_reader_factory.cpp
@@ -593,6 +595,7 @@ if(PAIMON_BUILD_TESTS)
                     common/file_index/file_indexer_factory_test.cpp
                     common/file_index/file_index_result_test.cpp
                     common/file_index/file_index_reader_test.cpp
+                    common/file_index/scored_file_index_result_test.cpp
                     common/file_index/file_index_format_test.cpp
                     common/file_index/empty/empty_file_index_reader_test.cpp
                     common/file_index/bitmap/bitmap_index_result_test.cpp
@@ -608,7 +611,6 @@ if(PAIMON_BUILD_TESTS)
                     common/file_index/rangebitmap/range_bitmap_type_adapter_test.cpp
                     common/file_index/bloomfilter/bloom_filter_file_index_test.cpp
                     common/file_index/bloomfilter/fast_hash_test.cpp
-                    common/global_index/complete_index_score_batch_reader_test.cpp
                     common/global_index/global_index_result_test.cpp
                     common/global_index/global_index_utils_test.cpp
                     common/global_index/offset_global_index_reader_test.cpp
@@ -649,6 +651,7 @@ if(PAIMON_BUILD_TESTS)
                     common/predicate/predicate_test.cpp
                     common/predicate/predicate_utils_test.cpp
                     common/predicate/predicate_validator_test.cpp
+                    common/reader/complete_index_score_batch_reader_test.cpp
                     common/reader/concat_batch_reader_test.cpp
                     common/reader/data_file_reader_factory_test.cpp
                     common/reader/late_materializing_file_batch_reader_test.cpp

src/paimon/common/file_index/empty/empty_file_index_reader.h

@@ -22,17 +22,32 @@
 #include <utility>
 #include <vector>
 
+#include "paimon/file_index/bitmap_index_result.h"
 #include "paimon/file_index/file_index_reader.h"
 #include "paimon/file_index/file_index_result.h"
+#include "paimon/file_index/scored_file_index_result.h"
 #include "paimon/result.h"
+#include "paimon/utils/roaring_bitmap32.h"
 
 namespace paimon {
 class Literal;
 
-/// Empty file index which has no writer and no serialized bytes.
+/// Reader for a file index entry with no serialized index data.
 /// No data in the file index, which mean this file has no related records.
+/// Vector and full-text searches return no matches.
 class EmptyFileIndexReader : public FileIndexReader {
  public:
+    Result<std::shared_ptr<ScoredFileIndexResult>> VisitVectorSearch(
+        const std::shared_ptr<VectorSearch>& vector_search) override {
+        return ScoredFileIndexResult::Create(RoaringBitmap32(), {});
+    }
+
+    Result<std::shared_ptr<FileIndexResult>> VisitFullTextSearch(
+        const std::shared_ptr<FullTextSearch>& full_text_search) override {
+        return std::make_shared<BitmapIndexResult>(
+            []() -> Result<RoaringBitmap32> { return RoaringBitmap32(); });
+    }
+
     Result<std::shared_ptr<FileIndexResult>> VisitEqual(const Literal& literal) override {
         return FileIndexResult::Skip();
     }

src/paimon/common/file_index/empty/empty_file_index_reader_test.cpp

@@ -19,9 +19,17 @@
 #include "paimon/common/file_index/empty/empty_file_index_reader.h"
 
 #include <cstdint>
+#include <map>
+#include <memory>
+#include <optional>
+#include <vector>
 
 #include "gtest/gtest.h"
+#include "paimon/file_index/bitmap_index_result.h"
+#include "paimon/predicate/full_text_search.h"
 #include "paimon/predicate/literal.h"
+#include "paimon/predicate/vector_search.h"
+#include "paimon/testing/utils/testharness.h"
 
 namespace paimon::test {
 TEST(EmptyFileIndexReaderTest, TestSimple) {
@@ -44,4 +52,29 @@ TEST(EmptyFileIndexReaderTest, TestSimple) {
     ASSERT_TRUE(reader.VisitNotEqual(lit0).value()->IsRemain().value());
     ASSERT_TRUE(reader.VisitNotIn({lit0}).value()->IsRemain().value());
 }
+
+TEST(EmptyFileIndexReaderTest, TestVectorSearchReturnsNoMatches) {
+    EmptyFileIndexReader reader;
+    auto search = std::make_shared<VectorSearch>(
+        "embedding", /*limit=*/1, std::vector<float>{1.0f}, nullptr, nullptr,
+        VectorSearch::DistanceType::EUCLIDEAN, std::map<std::string, std::string>{});
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<ScoredFileIndexResult> result,
+                         reader.VisitVectorSearch(search));
+    EXPECT_TRUE(result->GetRowPositions().IsEmpty());
+    EXPECT_TRUE(result->GetScores().empty());
+}
+
+TEST(EmptyFileIndexReaderTest, TestFullTextSearchReturnsNoMatches) {
+    EmptyFileIndexReader reader;
+    auto search = std::make_shared<FullTextSearch>(
+        "body", /*limit=*/1, "word", FullTextSearch::SearchType::MATCH_ANY, std::nullopt);
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<FileIndexResult> result,
+                         reader.VisitFullTextSearch(search));
+    ASSERT_OK_AND_ASSIGN(bool is_remain, result->IsRemain());
+    EXPECT_FALSE(is_remain);
+    auto bitmap_result = std::dynamic_pointer_cast<BitmapIndexResult>(result);
+    ASSERT_NE(nullptr, bitmap_result);
+    ASSERT_OK_AND_ASSIGN(const RoaringBitmap32* bitmap, bitmap_result->GetBitmap());
+    EXPECT_TRUE(bitmap->IsEmpty());
+}
 }  // namespace paimon::test