diff --git a/docs/source/user_guide/global_index.rst b/docs/source/user_guide/global_index.rst
index afee33914..f3dca7445 100644
--- a/docs/source/user_guide/global_index.rst
+++ b/docs/source/user_guide/global_index.rst
@@ -181,6 +181,31 @@ ranges is ranked only by the range that starts first, or ends first if both star
 A shard ignores the ``include_row_ids`` outside the rows it ranks and is not searched if none
 remain. The result iterates rows by ascending row id, not by score.
 
+When ``deletion-vectors.enabled`` and ``row-tracking.enabled`` are set, readers created by
+``GlobalIndexScan`` rank only the rows that are live in the snapshot the scan reads, as Java Paimon
+does, so rows deleted by deletion vectors do not take ``limit`` slots. BM25 statistics still come
+from the whole index. A later snapshot can delete some of the matched rows, so pass one snapshot id
+to both the index scan and the data scan:
+
+.. code-block:: cpp
+
+   PAIMON_ASSIGN_OR_RAISE(
+       std::unique_ptr<paimon::GlobalIndexScan> index_scan,
+       paimon::GlobalIndexScan::Create(table_path, snapshot_id, /*partitions=*/std::nullopt,
+                                       /*options=*/{}, /*file_system=*/nullptr,
+                                       /*executor=*/nullptr, pool));
+   PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<paimon::GlobalIndexReader> reader,
+                          index_scan->CreateReader("f0", "full-text",
+                                                   /*row_range_index=*/std::nullopt));
+   PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<paimon::ScoredGlobalIndexResult> result,
+                          reader->VisitFullTextSearch(std::make_shared<paimon::FullTextSearch>(
+                              "f0", R"({"match":{"query":"paimon lake"}})", /*limit=*/10)));
+
+   paimon::ScanContextBuilder scan_builder(table_path);
+   scan_builder.AddOption(paimon::Options::SCAN_MODE, "from-snapshot")
+       .AddOption(paimon::Options::SCAN_SNAPSHOT_ID, std::to_string(snapshot_id))
+       .SetGlobalIndexResult(result);
+
 **Configuration:**
 
 Table options with the ``full-text.`` prefix are passed to the engine with the prefix removed.
diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt
index da5213dc0..b3d1a1e24 100644
--- a/src/paimon/CMakeLists.txt
+++ b/src/paimon/CMakeLists.txt
@@ -269,9 +269,11 @@ set(PAIMON_CORE_SRCS
     core/deletionvectors/deletion_vectors_index_file.cpp
     core/deletionvectors/deletion_vector_index_file_writer.cpp
     core/global_index/global_index_evaluator_impl.cpp
+    core/global_index/global_index_live_row_filter.cpp
     core/global_index/global_index_scan.cpp
     core/global_index/global_index_scan_impl.cpp
     core/global_index/global_index_write_task.cpp
+    core/global_index/live_row_global_index_reader.cpp
     core/index/index_file_handler.cpp
     core/index/global_index_meta.cpp
     core/index/index_file_meta_serializer.cpp
@@ -865,6 +867,7 @@ if(PAIMON_BUILD_TESTS)
                     core/global_index/global_index_file_manager_test.cpp
                     core/global_index/global_index_evaluator_impl_test.cpp
                     core/global_index/indexed_split_test.cpp
+                    core/global_index/live_row_global_index_reader_test.cpp
                     core/manifest/file_source_test.cpp
                     core/manifest/file_kind_test.cpp
                     core/manifest/manifest_entry_writer_test.cpp
diff --git a/src/paimon/core/global_index/global_index_live_row_filter.cpp b/src/paimon/core/global_index/global_index_live_row_filter.cpp
new file mode 100644
index 000000000..faf9e0408
--- /dev/null
+++ b/src/paimon/core/global_index/global_index_live_row_filter.cpp
@@ -0,0 +1,213 @@
+/*
+ * 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/global_index/global_index_live_row_filter.h"
+
+#include <cstdint>
+#include <map>
+#include <memory>
+#include <optional>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "fmt/format.h"
+#include "paimon/common/types/data_field.h"
+#include "paimon/core/deletionvectors/deletion_vector.h"
+#include "paimon/core/index/index_file_handler.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/manifest/index_manifest_file.h"
+#include "paimon/core/manifest/manifest_file.h"
+#include "paimon/core/manifest/manifest_list.h"
+#include "paimon/core/operation/append_only_file_store_scan.h"
+#include "paimon/core/operation/data_evolution_file_store_scan.h"
+#include "paimon/core/operation/file_store_scan.h"
+#include "paimon/core/schema/schema_manager.h"
+#include "paimon/core/table/bucket_mode.h"
+#include "paimon/core/table/source/append_only_split_generator.h"
+#include "paimon/core/table/source/data_evolution_split_generator.h"
+#include "paimon/core/table/source/data_split_impl.h"
+#include "paimon/core/table/source/scan_mode.h"
+#include "paimon/core/table/source/snapshot/snapshot_reader.h"
+#include "paimon/core/table/source/split_generator.h"
+#include "paimon/core/utils/branch_manager.h"
+#include "paimon/core/utils/field_mapping.h"
+#include "paimon/core/utils/file_store_path_factory.h"
+#include "paimon/core/utils/index_file_path_factories.h"
+#include "paimon/core/utils/snapshot_manager.h"
+#include "paimon/scan_context.h"
+#include "paimon/table/source/plan.h"
+#include "paimon/utils/range.h"
+
+namespace paimon {
+namespace {
+/// Creates a reader over the data files of an append table in the partitions accepted by
+/// `partition_filter`, as Java `FileStoreTable#newSnapshotReader` with `withPartitionFilter` does.
+Result<std::unique_ptr<SnapshotReader>> CreateSnapshotReader(
+    const std::string& root_path, const std::shared_ptr<TableSchema>& table_schema,
+    const std::shared_ptr<PredicateFilter>& partition_filter, const CoreOptions& options,
+    const std::shared_ptr<Executor>& executor, const std::shared_ptr<MemoryPool>& pool) {
+    std::shared_ptr<FileSystem> fs = options.GetFileSystem();
+    std::string branch = BranchManager::NormalizeBranch(options.GetBranch());
+    auto arrow_schema = DataField::ConvertDataFieldsToArrowSchema(table_schema->Fields());
+    PAIMON_ASSIGN_OR_RAISE(std::vector<std::string> external_paths, options.CreateExternalPaths());
+    PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> global_index_external_path,
+                           options.CreateGlobalIndexExternalPath());
+    PAIMON_ASSIGN_OR_RAISE(
+        std::shared_ptr<FileStorePathFactory> path_factory,
+        FileStorePathFactory::Create(
+            root_path, arrow_schema, table_schema->PartitionKeys(),
+            options.GetPartitionDefaultName(), options.GetFileFormat()->Identifier(),
+            options.DataFilePrefix(), options.LegacyPartitionNameEnabled(), external_paths,
+            global_index_external_path, options.IndexFileInDataFileDir(), pool));
+    PAIMON_ASSIGN_OR_RAISE(
+        std::shared_ptr<ManifestList> manifest_list,
+        ManifestList::Create(fs, options.GetManifestFormat(), options.GetManifestCompression(),
+                             path_factory, options.GetCache(), pool));
+    PAIMON_ASSIGN_OR_RAISE(
+        std::shared_ptr<arrow::Schema> partition_schema,
+        FieldMapping::GetPartitionSchema(arrow_schema, table_schema->PartitionKeys()));
+    PAIMON_ASSIGN_OR_RAISE(
+        std::shared_ptr<ManifestFile> manifest_file,
+        ManifestFile::Create(fs, options.GetManifestFormat(), options.GetManifestCompression(),
+                             path_factory, options.GetManifestTargetFileSize(), pool, options,
+                             partition_schema));
+    auto snapshot_manager = std::make_shared<SnapshotManager>(fs, root_path, branch);
+    auto schema_manager = std::make_shared<SchemaManager>(fs, root_path, branch);
+    auto scan_filter = std::make_shared<ScanFilter>(
+        /*predicate=*/nullptr,
+        /*partition_filters=*/std::vector<std::map<std::string, std::string>>(),
+        /*bucket_filter=*/std::nullopt);
+
+    std::unique_ptr<FileStoreScan> scan;
+    std::unique_ptr<SplitGenerator> split_generator;
+    if (options.DataEvolutionEnabled()) {
+        PAIMON_ASSIGN_OR_RAISE(
+            scan, DataEvolutionFileStoreScan::Create(
+                      snapshot_manager, schema_manager, manifest_list, manifest_file, table_schema,
+                      arrow_schema, scan_filter, options, executor, pool));
+        split_generator = std::make_unique<DataEvolutionSplitGenerator>(
+            options.GetSourceSplitTargetSize(), options.GetSourceSplitOpenFileCost(),
+            options.BlobSplitByFileSize());
+    } else {
+        PAIMON_ASSIGN_OR_RAISE(
+            scan, AppendOnlyFileStoreScan::Create(snapshot_manager, schema_manager, manifest_list,
+                                                  manifest_file, table_schema, arrow_schema,
+                                                  scan_filter, options, executor, pool));
+        split_generator = std::make_unique<AppendOnlySplitGenerator>(
+            options.GetSourceSplitTargetSize(), options.GetSourceSplitOpenFileCost(),
+            BucketMode::BUCKET_UNAWARE);
+    }
+    scan->WithTablePath(root_path)->WithPartitionFilter(partition_filter, partition_schema);
+
+    PAIMON_ASSIGN_OR_RAISE(
+        std::unique_ptr<IndexManifestFile> index_manifest_file,
+        IndexManifestFile::Create(fs, options.GetManifestFormat(), options.GetManifestCompression(),
+                                  path_factory, options.GetBucket(), pool, options));
+    auto index_file_handler = std::make_unique<IndexFileHandler>(
+        fs, std::move(index_manifest_file), std::make_shared<IndexFilePathFactories>(path_factory),
+        options.DeletionVectorsBitmap64(), pool);
+    return std::make_unique<SnapshotReader>(
+        std::move(scan), path_factory, std::move(split_generator), std::move(index_file_handler));
+}
+
+/// Removes the rows deleted by the deletion vectors of `split` from `live_rows`. Each deletion
+/// vector is keyed by the data file whose first row id its positions are relative to.
+Status SubtractDeletedRows(const DataSplitImpl& split, const std::string& root_path,
+                           const Snapshot& snapshot, const std::shared_ptr<FileSystem>& fs,
+                           const std::shared_ptr<MemoryPool>& pool, RoaringBitmap64* live_rows) {
+    const std::vector<std::shared_ptr<DataFileMeta>>& files = split.DataFiles();
+    DeletionVector::Factory deletion_vector_factory = DeletionVector::CreateFactory(
+        fs, DeletionVector::CreateDeletionFileMap(files, split.DeletionFiles()), pool);
+    for (const auto& file : files) {
+        if (!file->first_row_id) {
+            continue;
+        }
+        int64_t first_row_id = file->first_row_id.value();
+        Result<std::shared_ptr<DeletionVector>> deletion_vector_result =
+            deletion_vector_factory(file->file_name);
+        if (!deletion_vector_result.ok()) {
+            const Status& status = deletion_vector_result.status();
+            return status.WithMessage(fmt::format(
+                "failed to read deletion vector of data file {} in snapshot {} of table {}: {}",
+                file->file_name, snapshot.Id(), root_path, status.message()));
+        }
+        std::shared_ptr<DeletionVector> deletion_vector = std::move(deletion_vector_result).value();
+        if (!deletion_vector || deletion_vector->IsEmpty()) {
+            continue;
+        }
+        RoaringBitmap64 deleted_rows;
+        PAIMON_RETURN_NOT_OK(deletion_vector->ForEachDeletedPosition(
+            [&deleted_rows, first_row_id](int64_t position) {
+                deleted_rows.Add(first_row_id + position);
+            }));
+        *live_rows -= deleted_rows;
+    }
+    return Status::OK();
+}
+}  // namespace
+
+Result<std::optional<RoaringBitmap64>> GlobalIndexLiveRowFilter::LiveRows(
+    const std::string& root_path, const std::shared_ptr<TableSchema>& table_schema,
+    const Snapshot& snapshot, const std::shared_ptr<PredicateFilter>& partition_filter,
+    const CoreOptions& options, const std::shared_ptr<Executor>& executor,
+    const std::shared_ptr<MemoryPool>& pool) {
+    // Without row tracking, as in primary-key tables, data files carry no row ids to filter.
+    if (!options.DeletionVectorsEnabled() || !options.RowTrackingEnabled()) {
+        return std::optional<RoaringBitmap64>();
+    }
+    PAIMON_ASSIGN_OR_RAISE(
+        std::unique_ptr<SnapshotReader> snapshot_reader,
+        CreateSnapshotReader(root_path, table_schema, partition_filter, options, executor, pool));
+    snapshot_reader->WithMode(ScanMode::ALL)->WithSnapshot(snapshot);
+    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Plan> plan, snapshot_reader->Read());
+
+    std::vector<std::shared_ptr<DataSplitImpl>> splits;
+    splits.reserve(plan->Splits().size());
+    for (const auto& split : plan->Splits()) {
+        auto data_split = std::dynamic_pointer_cast<DataSplitImpl>(split);
+        if (!data_split) {
+            return Status::Invalid("cannot cast split to DataSplitImpl when reading live rows");
+        }
+        splits.push_back(std::move(data_split));
+    }
+
+    RoaringBitmap64 live_rows;
+    // Phase 1: union the row id range of every data file.
+    for (const auto& split : splits) {
+        for (const auto& file : split->DataFiles()) {
+            if (file->first_row_id && file->row_count > 0) {
+                PAIMON_ASSIGN_OR_RAISE(Range row_id_range, file->NonNullRowIdRange());
+                // AddRange is half-open and `row_id_range.to` may be the largest int64 value.
+                live_rows.AddRange(row_id_range.from, row_id_range.to);
+                live_rows.Add(row_id_range.to);
+            }
+        }
+    }
+    // Phase 2: subtract each deletion vector. Every range is added first, so a row deleted by one
+    // file's deletion vector is not added back by another file covering it, and only one deletion
+    // vector is held at a time.
+    for (const auto& split : splits) {
+        PAIMON_RETURN_NOT_OK(SubtractDeletedRows(*split, root_path, snapshot,
+                                                 options.GetFileSystem(), pool, &live_rows));
+    }
+    return std::optional<RoaringBitmap64>(std::move(live_rows));
+}
+
+}  // namespace paimon
diff --git a/src/paimon/core/global_index/global_index_live_row_filter.h b/src/paimon/core/global_index/global_index_live_row_filter.h
new file mode 100644
index 000000000..4f607e699
--- /dev/null
+++ b/src/paimon/core/global_index/global_index_live_row_filter.h
@@ -0,0 +1,63 @@
+/*
+ * 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 <optional>
+#include <string>
+
+#include "paimon/common/predicate/predicate_filter.h"
+#include "paimon/core/core_options.h"
+#include "paimon/core/schema/table_schema.h"
+#include "paimon/core/snapshot.h"
+#include "paimon/executor.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/result.h"
+#include "paimon/utils/roaring_bitmap64.h"
+
+namespace paimon {
+/// Live-row filtering shared by global index search readers, aligned with Java
+/// `GlobalIndexLiveRowFilter`. Deletion vectors are applied only when rows are read, so a search
+/// that ranks deleted rows can fill its top-k slots with rows the read then drops.
+class GlobalIndexLiveRowFilter {
+ public:
+    GlobalIndexLiveRowFilter() = delete;
+    ~GlobalIndexLiveRowFilter() = delete;
+
+    /// Returns the global row ids of the rows that are live in `snapshot`: the row id ranges of
+    /// the data files in the partitions accepted by `partition_filter`, minus the rows deleted by
+    /// their deletion vectors. Returns nullopt when deletion vectors or row tracking are disabled,
+    /// so every row counts as live.
+    /// @param root_path        Root directory of the table.
+    /// @param table_schema     Schema of the table.
+    /// @param snapshot         The snapshot to read; pin it to the one the global index is
+    ///                         read from, so row ids and live rows agree.
+    /// @param partition_filter Predicate over the partition fields, or null for all partitions.
+    /// @param options          Options of the table.
+    /// @param executor         Executor for reading manifests.
+    /// @param pool             Memory pool for reading manifests and deletion vectors.
+    static Result<std::optional<RoaringBitmap64>> LiveRows(
+        const std::string& root_path, const std::shared_ptr<TableSchema>& table_schema,
+        const Snapshot& snapshot, const std::shared_ptr<PredicateFilter>& partition_filter,
+        const CoreOptions& options, const std::shared_ptr<Executor>& executor,
+        const std::shared_ptr<MemoryPool>& pool);
+};
+
+}  // namespace paimon
diff --git a/src/paimon/core/global_index/global_index_scan_impl.cpp b/src/paimon/core/global_index/global_index_scan_impl.cpp
index 511be25c3..b20437d4a 100644
--- a/src/paimon/core/global_index/global_index_scan_impl.cpp
+++ b/src/paimon/core/global_index/global_index_scan_impl.cpp
@@ -31,24 +31,25 @@
 #include "paimon/common/global_index/union_global_index_reader.h"
 #include "paimon/common/utils/scope_guard.h"
 #include "paimon/core/global_index/global_index_evaluator_impl.h"
+#include "paimon/core/global_index/global_index_live_row_filter.h"
 #include "paimon/core/index/index_file_handler.h"
 #include "paimon/global_index/bitmap_global_index_result.h"
 #include "paimon/global_index/global_indexer.h"
 #include "paimon/global_index/global_indexer_factory.h"
 
 namespace paimon {
-GlobalIndexScanImpl::GlobalIndexScanImpl(const std::shared_ptr<TableSchema>& table_schema,
-                                         const CoreOptions& options,
-                                         const std::shared_ptr<IndexPathFactory>& path_factory,
-                                         IndexMetaMap&& index_metas,
-                                         const std::shared_ptr<Executor>& executor,
-                                         const std::shared_ptr<MemoryPool>& pool)
+GlobalIndexScanImpl::GlobalIndexScanImpl(
+    const std::shared_ptr<TableSchema>& table_schema, const CoreOptions& options,
+    const std::shared_ptr<IndexPathFactory>& path_factory, IndexMetaMap&& index_metas,
+    LiveRowGlobalIndexReader::LiveRowsSupplier live_rows_supplier,
+    const std::shared_ptr<Executor>& executor, const std::shared_ptr<MemoryPool>& pool)
     : pool_(pool),
       table_schema_(table_schema),
       options_(options),
       index_file_manager_(std::make_shared<GlobalIndexFileManager>(
           options.GetFileSystem(), path_factory, /*checkpoint_path_factory=*/nullptr)),
       index_metas_(std::move(index_metas)),
+      live_rows_supplier_(std::move(live_rows_supplier)),
       executor_(executor) {}
 
 Result<std::unique_ptr<GlobalIndexScanImpl>> GlobalIndexScanImpl::Create(
@@ -122,8 +123,18 @@ Result<std::unique_ptr<GlobalIndexScanImpl>> GlobalIndexScanImpl::Create(
         index_metas[index_meta->index_field_id][index_file_meta->IndexType()][range].push_back(
             index_file_meta);
     }
-    return std::unique_ptr<GlobalIndexScanImpl>(new GlobalIndexScanImpl(
-        table_schema, options, path_factory, std::move(index_metas), final_executor, pool));
+    // Full-text searches rank only the rows live in the snapshot the index is read from.
+    LiveRowGlobalIndexReader::LiveRowsSupplier live_rows_supplier;
+    if (options.DeletionVectorsEnabled()) {
+        live_rows_supplier = LiveRowGlobalIndexReader::CacheLiveRows(
+            [root_path, table_schema, snapshot, partitions, options, final_executor, pool]() {
+                return GlobalIndexLiveRowFilter::LiveRows(
+                    root_path, table_schema, snapshot, partitions, options, final_executor, pool);
+            });
+    }
+    return std::unique_ptr<GlobalIndexScanImpl>(
+        new GlobalIndexScanImpl(table_schema, options, path_factory, std::move(index_metas),
+                                std::move(live_rows_supplier), final_executor, pool));
 }
 
 Result<std::shared_ptr<GlobalIndexEvaluator>> GlobalIndexScanImpl::GetOrCreateIndexEvaluator() {
@@ -235,8 +246,15 @@ Result<std::vector<std::shared_ptr<GlobalIndexReader>>> GlobalIndexScanImpl::Cre
         if (union_readers.empty()) {
             continue;
         }
-        readers.push_back(std::make_shared<UnionGlobalIndexReader>(
-            std::move(union_readers), std::move(row_ranges), std::move(search_ranges), executor_));
+        if (live_rows_supplier_) {
+            readers.push_back(LiveRowGlobalIndexReader::CreateUnion(
+                std::move(union_readers), std::move(row_ranges), std::move(search_ranges),
+                executor_, live_rows_supplier_));
+        } else {
+            readers.push_back(std::make_shared<UnionGlobalIndexReader>(
+                std::move(union_readers), std::move(row_ranges), std::move(search_ranges),
+                executor_));
+        }
     }
     return readers;
 }
diff --git a/src/paimon/core/global_index/global_index_scan_impl.h b/src/paimon/core/global_index/global_index_scan_impl.h
index a7921d77c..bb26ae8ee 100644
--- a/src/paimon/core/global_index/global_index_scan_impl.h
+++ b/src/paimon/core/global_index/global_index_scan_impl.h
@@ -29,6 +29,7 @@
 #include "paimon/core/core_options.h"
 #include "paimon/core/global_index/global_index_evaluator.h"
 #include "paimon/core/global_index/global_index_file_manager.h"
+#include "paimon/core/global_index/live_row_global_index_reader.h"
 #include "paimon/core/index/index_file_meta.h"
 #include "paimon/core/index/index_path_factory.h"
 #include "paimon/core/schema/table_schema.h"
@@ -67,7 +68,9 @@ class GlobalIndexScanImpl : public GlobalIndexScan {
     GlobalIndexScanImpl(const std::shared_ptr<TableSchema>& table_schema,
                         const CoreOptions& options,
                         const std::shared_ptr<IndexPathFactory>& path_factory,
-                        IndexMetaMap&& index_metas, const std::shared_ptr<Executor>& executor,
+                        IndexMetaMap&& index_metas,
+                        LiveRowGlobalIndexReader::LiveRowsSupplier live_rows_supplier,
+                        const std::shared_ptr<Executor>& executor,
                         const std::shared_ptr<MemoryPool>& pool);
 
     Result<std::shared_ptr<GlobalIndexEvaluator>> GetOrCreateIndexEvaluator();
@@ -88,6 +91,8 @@ class GlobalIndexScanImpl : public GlobalIndexScan {
     CoreOptions options_;
     std::shared_ptr<GlobalIndexFileManager> index_file_manager_;
     IndexMetaMap index_metas_;
+    /// Supplies the rows live in the scanned snapshot; empty when deletion vectors are disabled.
+    LiveRowGlobalIndexReader::LiveRowsSupplier live_rows_supplier_;
     std::shared_ptr<Executor> executor_;
     std::shared_ptr<GlobalIndexEvaluator> evaluator_;
 };
diff --git a/src/paimon/core/global_index/live_row_global_index_reader.cpp b/src/paimon/core/global_index/live_row_global_index_reader.cpp
new file mode 100644
index 000000000..9bdf1449b
--- /dev/null
+++ b/src/paimon/core/global_index/live_row_global_index_reader.cpp
@@ -0,0 +1,232 @@
+/*
+ * 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/global_index/live_row_global_index_reader.h"
+
+#include <algorithm>
+#include <cassert>
+#include <mutex>
+#include <utility>
+
+#include "paimon/common/global_index/union_global_index_reader.h"
+
+namespace paimon {
+namespace {
+/// Returns whether `full_text_search` may rank a row of `range`: it has no include row ids, or one
+/// of them is in `range`.
+bool MayRankRowsOf(const FullTextSearch& full_text_search, const Range& range) {
+    if (!full_text_search.include_row_ids) {
+        return true;
+    }
+    const RoaringBitmap64& include_row_ids = full_text_search.include_row_ids.value();
+    auto first = include_row_ids.EqualOrLarger(range.from);
+    return first != include_row_ids.End() && *first <= range.to;
+}
+
+/// A union of LiveRowGlobalIndexReaders that loads the live rows on the visiting thread before a
+/// full-text search fans out to them.
+class LiveRowUnionGlobalIndexReader : public UnionGlobalIndexReader {
+ public:
+    /// @param ranked_ranges The global row ids the shards rank in a full-text search, leaving out
+    ///                      the shards that rank none.
+    LiveRowUnionGlobalIndexReader(std::vector<std::shared_ptr<GlobalIndexReader>>&& readers,
+                                  std::vector<Range>&& row_ranges,
+                                  std::vector<std::optional<Range>>&& search_ranges,
+                                  std::vector<Range>&& ranked_ranges,
+                                  const std::shared_ptr<Executor>& executor,
+                                  LiveRowGlobalIndexReader::LiveRowsSupplier live_rows_supplier)
+        : UnionGlobalIndexReader(std::move(readers), std::move(row_ranges),
+                                 std::move(search_ranges), executor),
+          ranked_ranges_(std::move(ranked_ranges)),
+          live_rows_supplier_(std::move(live_rows_supplier)) {}
+
+    Result<std::shared_ptr<ScoredGlobalIndexResult>> VisitFullTextSearch(
+        const std::shared_ptr<FullTextSearch>& full_text_search) override {
+        // Only a shard search needs live rows: the union rejects a search without a positive limit
+        // before searching any shard, and skips the shards whose rows the include row ids miss.
+        if (full_text_search && full_text_search->limit > 0 &&
+            std::any_of(ranked_ranges_.begin(), ranked_ranges_.end(), [&](const Range& range) {
+                return MayRankRowsOf(*full_text_search, range);
+            })) {
+            PAIMON_RETURN_NOT_OK(live_rows_supplier_());
+        }
+        return UnionGlobalIndexReader::VisitFullTextSearch(full_text_search);
+    }
+
+ private:
+    std::vector<Range> ranked_ranges_;
+    LiveRowGlobalIndexReader::LiveRowsSupplier live_rows_supplier_;
+};
+}  // namespace
+
+LiveRowGlobalIndexReader::LiveRowsSupplier LiveRowGlobalIndexReader::CacheLiveRows(
+    std::function<Result<std::optional<RoaringBitmap64>>()> load) {
+    struct State {
+        std::mutex mutex;
+        bool loaded = false;
+        std::shared_ptr<const RoaringBitmap64> live_rows;
+    };
+    auto state = std::make_shared<State>();
+    return [state, load = std::move(load)]() -> Result<std::shared_ptr<const RoaringBitmap64>> {
+        std::lock_guard<std::mutex> lock(state->mutex);
+        if (!state->loaded) {
+            PAIMON_ASSIGN_OR_RAISE(std::optional<RoaringBitmap64> live_rows, load());
+            if (live_rows) {
+                state->live_rows =
+                    std::make_shared<const RoaringBitmap64>(std::move(live_rows).value());
+            }
+            state->loaded = true;
+        }
+        return state->live_rows;
+    };
+}
+
+std::shared_ptr<GlobalIndexReader> LiveRowGlobalIndexReader::CreateUnion(
+    std::vector<std::shared_ptr<GlobalIndexReader>>&& shard_readers,
+    std::vector<Range>&& shard_ranges, std::vector<std::optional<Range>>&& search_ranges,
+    const std::shared_ptr<Executor>& executor, const LiveRowsSupplier& live_rows_supplier) {
+    assert(shard_readers.size() == shard_ranges.size());
+    std::vector<std::shared_ptr<GlobalIndexReader>> readers;
+    std::vector<Range> ranked_ranges;
+    readers.reserve(shard_readers.size());
+    ranked_ranges.reserve(shard_readers.size());
+    for (size_t i = 0; i < shard_readers.size(); i++) {
+        readers.push_back(std::make_shared<LiveRowGlobalIndexReader>(
+            std::move(shard_readers[i]), shard_ranges[i], live_rows_supplier));
+        if (search_ranges.empty()) {
+            ranked_ranges.push_back(shard_ranges[i]);
+        } else if (search_ranges[i]) {
+            ranked_ranges.push_back(search_ranges[i].value());
+        }
+    }
+    return std::make_shared<LiveRowUnionGlobalIndexReader>(
+        std::move(readers), std::move(shard_ranges), std::move(search_ranges),
+        std::move(ranked_ranges), executor, live_rows_supplier);
+}
+
+LiveRowGlobalIndexReader::LiveRowGlobalIndexReader(std::shared_ptr<GlobalIndexReader>&& wrapped,
+                                                   const Range& shard_range,
+                                                   LiveRowsSupplier live_rows_supplier)
+    : wrapped_(std::move(wrapped)),
+      shard_range_(shard_range),
+      live_rows_supplier_(std::move(live_rows_supplier)) {}
+
+Result<std::shared_ptr<GlobalIndexResult>> LiveRowGlobalIndexReader::VisitIsNotNull() {
+    return wrapped_->VisitIsNotNull();
+}
+
+Result<std::shared_ptr<GlobalIndexResult>> LiveRowGlobalIndexReader::VisitIsNull() {
+    return wrapped_->VisitIsNull();
+}
+
+Result<std::shared_ptr<GlobalIndexResult>> LiveRowGlobalIndexReader::VisitEqual(
+    const Literal& literal) {
+    return wrapped_->VisitEqual(literal);
+}
+
+Result<std::shared_ptr<GlobalIndexResult>> LiveRowGlobalIndexReader::VisitNotEqual(
+    const Literal& literal) {
+    return wrapped_->VisitNotEqual(literal);
+}
+
+Result<std::shared_ptr<GlobalIndexResult>> LiveRowGlobalIndexReader::VisitLessThan(
+    const Literal& literal) {
+    return wrapped_->VisitLessThan(literal);
+}
+
+Result<std::shared_ptr<GlobalIndexResult>> LiveRowGlobalIndexReader::VisitLessOrEqual(
+    const Literal& literal) {
+    return wrapped_->VisitLessOrEqual(literal);
+}
+
+Result<std::shared_ptr<GlobalIndexResult>> LiveRowGlobalIndexReader::VisitGreaterThan(
+    const Literal& literal) {
+    return wrapped_->VisitGreaterThan(literal);
+}
+
+Result<std::shared_ptr<GlobalIndexResult>> LiveRowGlobalIndexReader::VisitGreaterOrEqual(
+    const Literal& literal) {
+    return wrapped_->VisitGreaterOrEqual(literal);
+}
+
+Result<std::shared_ptr<GlobalIndexResult>> LiveRowGlobalIndexReader::VisitIn(
+    const std::vector<Literal>& literals) {
+    return wrapped_->VisitIn(literals);
+}
+
+Result<std::shared_ptr<GlobalIndexResult>> LiveRowGlobalIndexReader::VisitNotIn(
+    const std::vector<Literal>& literals) {
+    return wrapped_->VisitNotIn(literals);
+}
+
+Result<std::shared_ptr<GlobalIndexResult>> LiveRowGlobalIndexReader::VisitStartsWith(
+    const Literal& prefix) {
+    return wrapped_->VisitStartsWith(prefix);
+}
+
+Result<std::shared_ptr<GlobalIndexResult>> LiveRowGlobalIndexReader::VisitEndsWith(
+    const Literal& suffix) {
+    return wrapped_->VisitEndsWith(suffix);
+}
+
+Result<std::shared_ptr<GlobalIndexResult>> LiveRowGlobalIndexReader::VisitContains(
+    const Literal& literal) {
+    return wrapped_->VisitContains(literal);
+}
+
+Result<std::shared_ptr<GlobalIndexResult>> LiveRowGlobalIndexReader::VisitLike(
+    const Literal& literal) {
+    return wrapped_->VisitLike(literal);
+}
+
+Result<std::shared_ptr<ScoredGlobalIndexResult>> LiveRowGlobalIndexReader::VisitVectorSearch(
+    const std::shared_ptr<VectorSearch>& vector_search) {
+    return wrapped_->VisitVectorSearch(vector_search);
+}
+
+Result<std::shared_ptr<ScoredGlobalIndexResult>> LiveRowGlobalIndexReader::VisitFullTextSearch(
+    const std::shared_ptr<FullTextSearch>& full_text_search) {
+    if (!full_text_search || !MayRankRowsOf(*full_text_search, shard_range_)) {
+        return wrapped_->VisitFullTextSearch(full_text_search);
+    }
+    PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<const RoaringBitmap64> live_rows, live_rows_supplier_());
+    if (!live_rows) {
+        return wrapped_->VisitFullTextSearch(full_text_search);
+    }
+    RoaringBitmap64 include_row_ids;
+    // `to` may be the largest int64 value, so add it apart from the half-open range.
+    include_row_ids.AddRange(shard_range_.from, shard_range_.to);
+    include_row_ids.Add(shard_range_.to);
+    include_row_ids &= *live_rows;
+    if (full_text_search->include_row_ids) {
+        include_row_ids &= full_text_search->include_row_ids.value();
+    }
+    if (include_row_ids.Cardinality() == shard_range_.Count()) {
+        if (!full_text_search->include_row_ids) {
+            return wrapped_->VisitFullTextSearch(full_text_search);
+        }
+        return wrapped_->VisitFullTextSearch(std::make_shared<FullTextSearch>(
+            full_text_search->field_name, full_text_search->query, full_text_search->limit));
+    }
+    return wrapped_->VisitFullTextSearch(
+        std::make_shared<FullTextSearch>(full_text_search->field_name, full_text_search->query,
+                                         full_text_search->limit, std::move(include_row_ids)));
+}
+
+}  // namespace paimon
diff --git a/src/paimon/core/global_index/live_row_global_index_reader.h b/src/paimon/core/global_index/live_row_global_index_reader.h
new file mode 100644
index 000000000..4e2589c7e
--- /dev/null
+++ b/src/paimon/core/global_index/live_row_global_index_reader.h
@@ -0,0 +1,115 @@
+/*
+ * 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 <functional>
+#include <memory>
+#include <optional>
+#include <string>
+#include <vector>
+
+#include "paimon/executor.h"
+#include "paimon/global_index/global_index_reader.h"
+#include "paimon/utils/range.h"
+#include "paimon/utils/roaring_bitmap64.h"
+
+namespace paimon {
+
+/// A GlobalIndexReader over one index shard that ranks only live rows in full-text searches, so
+/// rows deleted by deletion vectors do not take top-k slots. As Java
+/// `DataEvolutionFullTextRead#includeRowIds` does, the shard is searched with the live rows of its
+/// range, intersected with the include row ids if given, while BM25 statistics still come from the
+/// whole index. Predicate visits are delegated unchanged, as the read drops deleted rows from their
+/// untruncated results, and so are vector searches.
+class LiveRowGlobalIndexReader : public GlobalIndexReader {
+ public:
+    /// Supplies the global row ids of live rows, or null when every row is live.
+    using LiveRowsSupplier = std::function<Result<std::shared_ptr<const RoaringBitmap64>>()>;
+
+    /// Returns a supplier that calls `load` on its first call and shares the live rows it returns
+    /// with later calls, so they are read at most once. A failed load is retried on the next call.
+    static LiveRowsSupplier CacheLiveRows(
+        std::function<Result<std::optional<RoaringBitmap64>>()> load);
+
+    /// Creates a union of the shard readers of one index whose full-text searches rank only live
+    /// rows. Each shard reader is wrapped in a LiveRowGlobalIndexReader. When a search may rank
+    /// rows of a shard, live rows are loaded on the visiting thread before it fans out to the
+    /// shards, so a load that reads manifests on `executor` never waits for it from one of its own
+    /// threads.
+    /// @param shard_readers      Readers of the shards. They take and return global row ids.
+    /// @param shard_ranges       The global row id range (both ends inclusive) of each shard.
+    /// @param search_ranges      The global row ids each shard ranks in a full-text search, as
+    ///                           passed to UnionGlobalIndexReader.
+    /// @param executor           Executor to search the shards in parallel, or null to search
+    ///                           them one after another.
+    /// @param live_rows_supplier Supplies the live rows to full-text searches.
+    static std::shared_ptr<GlobalIndexReader> CreateUnion(
+        std::vector<std::shared_ptr<GlobalIndexReader>>&& shard_readers,
+        std::vector<Range>&& shard_ranges, std::vector<std::optional<Range>>&& search_ranges,
+        const std::shared_ptr<Executor>& executor, const LiveRowsSupplier& live_rows_supplier);
+
+    /// Constructs a LiveRowGlobalIndexReader.
+    /// @param wrapped            The reader of the shard. It takes and returns global row ids.
+    /// @param shard_range        The global row id range (both ends inclusive) of the shard.
+    /// @param live_rows_supplier Supplies the live rows to full-text searches.
+    LiveRowGlobalIndexReader(std::shared_ptr<GlobalIndexReader>&& wrapped, const Range& shard_range,
+                             LiveRowsSupplier live_rows_supplier);
+
+    Result<std::shared_ptr<GlobalIndexResult>> VisitIsNotNull() override;
+    Result<std::shared_ptr<GlobalIndexResult>> VisitIsNull() override;
+    Result<std::shared_ptr<GlobalIndexResult>> VisitEqual(const Literal& literal) override;
+    Result<std::shared_ptr<GlobalIndexResult>> VisitNotEqual(const Literal& literal) override;
+    Result<std::shared_ptr<GlobalIndexResult>> VisitLessThan(const Literal& literal) override;
+    Result<std::shared_ptr<GlobalIndexResult>> VisitLessOrEqual(const Literal& literal) override;
+    Result<std::shared_ptr<GlobalIndexResult>> VisitGreaterThan(const Literal& literal) override;
+    Result<std::shared_ptr<GlobalIndexResult>> VisitGreaterOrEqual(const Literal& literal) override;
+    Result<std::shared_ptr<GlobalIndexResult>> VisitIn(
+        const std::vector<Literal>& literals) override;
+    Result<std::shared_ptr<GlobalIndexResult>> VisitNotIn(
+        const std::vector<Literal>& literals) override;
+    Result<std::shared_ptr<GlobalIndexResult>> VisitStartsWith(const Literal& prefix) override;
+    Result<std::shared_ptr<GlobalIndexResult>> VisitEndsWith(const Literal& suffix) override;
+    Result<std::shared_ptr<GlobalIndexResult>> VisitContains(const Literal& literal) override;
+    Result<std::shared_ptr<GlobalIndexResult>> VisitLike(const Literal& literal) override;
+
+    Result<std::shared_ptr<ScoredGlobalIndexResult>> VisitVectorSearch(
+        const std::shared_ptr<VectorSearch>& vector_search) override;
+
+    /// Searches only the live rows of the shard that are in the include row ids, if given. The
+    /// search is delegated unchanged when none of its include row ids is in the shard, or when it
+    /// has none and the whole shard is live.
+    Result<std::shared_ptr<ScoredGlobalIndexResult>> VisitFullTextSearch(
+        const std::shared_ptr<FullTextSearch>& full_text_search) override;
+
+    bool IsThreadSafe() const override {
+        return wrapped_->IsThreadSafe();
+    }
+
+    std::string GetIndexType() const override {
+        return wrapped_->GetIndexType();
+    }
+
+ private:
+    std::shared_ptr<GlobalIndexReader> wrapped_;
+    Range shard_range_;
+    LiveRowsSupplier live_rows_supplier_;
+};
+
+}  // namespace paimon
diff --git a/src/paimon/core/global_index/live_row_global_index_reader_test.cpp b/src/paimon/core/global_index/live_row_global_index_reader_test.cpp
new file mode 100644
index 000000000..bc727394c
--- /dev/null
+++ b/src/paimon/core/global_index/live_row_global_index_reader_test.cpp
@@ -0,0 +1,262 @@
+/*
+ * 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/global_index/live_row_global_index_reader.h"
+
+#include <cstdint>
+#include <memory>
+#include <mutex>
+#include <optional>
+#include <thread>
+#include <vector>
+
+#include "gtest/gtest.h"
+#include "paimon/executor.h"
+#include "paimon/predicate/full_text_search.h"
+#include "paimon/predicate/literal.h"
+#include "paimon/testing/mock/mock_global_index_reader.h"
+#include "paimon/testing/utils/testharness.h"
+#include "paimon/utils/range.h"
+#include "paimon/utils/roaring_bitmap64.h"
+
+namespace paimon::test {
+class LiveRowGlobalIndexReaderTest : public ::testing::Test {
+ public:
+    static LiveRowGlobalIndexReader::LiveRowsSupplier Supply(
+        const std::optional<RoaringBitmap64>& live_rows, int32_t* call_count) {
+        std::shared_ptr<const RoaringBitmap64> shared_live_rows;
+        if (live_rows) {
+            shared_live_rows = std::make_shared<const RoaringBitmap64>(live_rows.value());
+        }
+        return [shared_live_rows, call_count]() -> Result<std::shared_ptr<const RoaringBitmap64>> {
+            (*call_count)++;
+            return shared_live_rows;
+        };
+    }
+
+    static RoaringBitmap64 LiveRows(int64_t from, int64_t to, const std::vector<int64_t>& deleted) {
+        RoaringBitmap64 live_rows;
+        live_rows.AddRange(from, to + 1);
+        live_rows -= RoaringBitmap64::From(deleted);
+        return live_rows;
+    }
+
+    static std::shared_ptr<FullTextSearch> Search(
+        const std::optional<RoaringBitmap64>& include_row_ids = std::nullopt) {
+        return std::make_shared<FullTextSearch>("f0", R"({"match":{"query":"apple"}})",
+                                                /*limit=*/3, include_row_ids);
+    }
+};
+
+TEST_F(LiveRowGlobalIndexReaderTest, TestSearchUnchangedWhenEveryRowIsLive) {
+    auto mock_reader = std::make_shared<MockGlobalIndexReader>();
+    mock_reader->SetScoredResult({11, 12}, {0.5f, 0.9f});
+    int32_t call_count = 0;
+    LiveRowGlobalIndexReader reader(mock_reader, Range(10, 14), Supply(std::nullopt, &call_count));
+    std::shared_ptr<FullTextSearch> search = Search();
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<ScoredGlobalIndexResult> result,
+                         reader.VisitFullTextSearch(search));
+    ASSERT_EQ(mock_reader->CapturedFullTextSearch(), search);
+    ASSERT_EQ(result->ToString(), "row ids: {11,12}, scores: {0.50,0.90}");
+    ASSERT_EQ(call_count, 1);
+}
+
+TEST_F(LiveRowGlobalIndexReaderTest, TestSearchOnlyLiveRowsOfShard) {
+    auto mock_reader = std::make_shared<MockGlobalIndexReader>();
+    mock_reader->SetScoredResult({11, 13}, {0.9f, 0.5f});
+    int32_t call_count = 0;
+    LiveRowGlobalIndexReader reader(mock_reader, Range(10, 14),
+                                    Supply(LiveRows(0, 20, {1, 12}), &call_count));
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<ScoredGlobalIndexResult> result,
+                         reader.VisitFullTextSearch(Search()));
+    ASSERT_EQ(result->ToString(), "row ids: {11,13}, scores: {0.90,0.50}");
+    std::shared_ptr<FullTextSearch> captured = mock_reader->CapturedFullTextSearch();
+    ASSERT_TRUE(captured);
+    ASSERT_EQ(captured->field_name, "f0");
+    ASSERT_EQ(captured->query, R"({"match":{"query":"apple"}})");
+    ASSERT_EQ(captured->limit, 3);
+    ASSERT_TRUE(captured->include_row_ids);
+    ASSERT_EQ(captured->include_row_ids.value(), RoaringBitmap64::From({10, 11, 13, 14}));
+
+    constexpr int64_t kMax = RoaringBitmap64::MAX_VALUE;
+    LiveRowGlobalIndexReader max_reader(mock_reader, Range(kMax - 1, kMax),
+                                        Supply(RoaringBitmap64::From({kMax}), &call_count));
+    ASSERT_OK(max_reader.VisitFullTextSearch(Search()));
+    captured = mock_reader->CapturedFullTextSearch();
+    ASSERT_TRUE(captured && captured->include_row_ids);
+    ASSERT_EQ(captured->include_row_ids.value(), RoaringBitmap64::From({kMax}));
+}
+
+TEST_F(LiveRowGlobalIndexReaderTest, TestIncludeRowIdsClippedToShard) {
+    auto mock_reader = std::make_shared<MockGlobalIndexReader>();
+    int32_t call_count = 0;
+    LiveRowGlobalIndexReader reader(mock_reader, Range(10, 14),
+                                    Supply(LiveRows(0, 20, {}), &call_count));
+    ASSERT_OK(
+        reader.VisitFullTextSearch(Search(RoaringBitmap64::From({1, 12, 20, int64_t{1} << 32}))));
+    std::shared_ptr<FullTextSearch> captured = mock_reader->CapturedFullTextSearch();
+    ASSERT_TRUE(captured && captured->include_row_ids);
+    ASSERT_EQ(captured->include_row_ids.value(), RoaringBitmap64::From({12}));
+
+    ASSERT_OK(reader.VisitFullTextSearch(Search(LiveRows(9, 15, {}))));
+    captured = mock_reader->CapturedFullTextSearch();
+    ASSERT_TRUE(captured);
+    ASSERT_FALSE(captured->include_row_ids);
+}
+
+TEST_F(LiveRowGlobalIndexReaderTest, TestCreateUnionLoadsLiveRowsBeforeSearchingShards) {
+    constexpr int64_t kSecondShardFrom = int64_t{1} << 32;
+    auto first_shard = std::make_shared<MockGlobalIndexReader>();
+    auto second_shard = std::make_shared<MockGlobalIndexReader>();
+    second_shard->SetScoredResult({kSecondShardFrom}, {0.5f});
+    auto live_rows = std::make_shared<const RoaringBitmap64>(
+        RoaringBitmap64::From({kSecondShardFrom, kSecondShardFrom + 1}));
+    std::mutex mutex;
+    std::optional<std::thread::id> loading_thread;
+    LiveRowGlobalIndexReader::LiveRowsSupplier supplier =
+        [&]() -> Result<std::shared_ptr<const RoaringBitmap64>> {
+        std::lock_guard<std::mutex> lock(mutex);
+        if (!loading_thread) {
+            loading_thread = std::this_thread::get_id();
+        }
+        return live_rows;
+    };
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Executor> executor,
+                         CreateDefaultExecutor(/*thread_count=*/1));
+    std::shared_ptr<GlobalIndexReader> reader = LiveRowGlobalIndexReader::CreateUnion(
+        std::vector<std::shared_ptr<GlobalIndexReader>>({first_shard, second_shard}),
+        {Range(0, 1), Range(kSecondShardFrom, kSecondShardFrom + 1)},
+        {Range(0, 1), Range(kSecondShardFrom, kSecondShardFrom + 1)}, executor, supplier);
+
+    std::shared_ptr<FullTextSearch> search = Search();
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<ScoredGlobalIndexResult> result,
+                         reader->VisitFullTextSearch(search));
+    ASSERT_TRUE(loading_thread == std::this_thread::get_id());
+    std::shared_ptr<FullTextSearch> first_search = first_shard->CapturedFullTextSearch();
+    ASSERT_TRUE(first_search && first_search->include_row_ids);
+    ASSERT_TRUE(first_search->include_row_ids->IsEmpty());
+    ASSERT_EQ(second_shard->CapturedFullTextSearch(), search);
+    ASSERT_EQ(result->ToString(), "row ids: {4294967296}, scores: {0.50}");
+}
+
+TEST_F(LiveRowGlobalIndexReaderTest, TestCreateUnionRanksLiveRowsOfSearchRanges) {
+    auto first_shard = std::make_shared<MockGlobalIndexReader>();
+    first_shard->SetScoredResult({0, 1}, {0.9f, 0.2f});
+    auto second_shard = std::make_shared<MockGlobalIndexReader>();
+    second_shard->SetScoredResult({10, 11}, {0.8f, 0.7f});
+    int32_t call_count = 0;
+    std::shared_ptr<GlobalIndexReader> reader = LiveRowGlobalIndexReader::CreateUnion(
+        std::vector<std::shared_ptr<GlobalIndexReader>>({first_shard, second_shard}),
+        {Range(0, 9), Range(5, 14)}, {Range(0, 9), Range(10, 14)}, /*executor=*/nullptr,
+        Supply(LiveRows(0, 14, {3, 12}), &call_count));
+
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<ScoredGlobalIndexResult> result,
+                         reader->VisitFullTextSearch(Search()));
+    ASSERT_EQ(result->ToString(), "row ids: {0,10,11}, scores: {0.90,0.80,0.70}");
+    std::shared_ptr<FullTextSearch> first_search = first_shard->CapturedFullTextSearch();
+    ASSERT_TRUE(first_search && first_search->include_row_ids);
+    ASSERT_EQ(first_search->include_row_ids.value(), LiveRows(0, 9, {3}));
+    std::shared_ptr<FullTextSearch> second_search = second_shard->CapturedFullTextSearch();
+    ASSERT_TRUE(second_search && second_search->include_row_ids);
+    ASSERT_EQ(second_search->include_row_ids.value(), RoaringBitmap64::From({10, 11, 13, 14}));
+}
+
+TEST_F(LiveRowGlobalIndexReaderTest, TestCacheLiveRows) {
+    int32_t load_count = 0;
+    bool fail = true;
+    LiveRowGlobalIndexReader::LiveRowsSupplier supplier =
+        LiveRowGlobalIndexReader::CacheLiveRows([&]() -> Result<std::optional<RoaringBitmap64>> {
+            load_count++;
+            if (fail) {
+                return Status::IOError("cannot read manifest");
+            }
+            return std::optional<RoaringBitmap64>(RoaringBitmap64::From({1, 2}));
+        });
+    ASSERT_NOK_WITH_MSG(supplier(), "cannot read manifest");
+    fail = false;
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<const RoaringBitmap64> live_rows, supplier());
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<const RoaringBitmap64> cached_live_rows, supplier());
+    ASSERT_EQ(load_count, 2);
+    ASSERT_TRUE(live_rows);
+    ASSERT_EQ(*live_rows, RoaringBitmap64::From({1, 2}));
+    ASSERT_EQ(cached_live_rows, live_rows);
+
+    LiveRowGlobalIndexReader::LiveRowsSupplier every_row_live =
+        LiveRowGlobalIndexReader::CacheLiveRows([&]() -> Result<std::optional<RoaringBitmap64>> {
+            load_count++;
+            return std::optional<RoaringBitmap64>();
+        });
+    ASSERT_OK_AND_ASSIGN(live_rows, every_row_live());
+    ASSERT_OK_AND_ASSIGN(cached_live_rows, every_row_live());
+    ASSERT_FALSE(live_rows);
+    ASSERT_FALSE(cached_live_rows);
+    ASSERT_EQ(load_count, 3);
+}
+
+TEST_F(LiveRowGlobalIndexReaderTest, TestLiveRowsErrorOnlyFailsSearchesThatRankRows) {
+    auto mock_reader = std::make_shared<MockGlobalIndexReader>();
+    LiveRowGlobalIndexReader::LiveRowsSupplier failing_supplier =
+        []() -> Result<std::shared_ptr<const RoaringBitmap64>> {
+        return Status::IOError("cannot read deletion vector");
+    };
+    LiveRowGlobalIndexReader reader(mock_reader, Range(0, 9), failing_supplier);
+    ASSERT_NOK_WITH_MSG(reader.VisitFullTextSearch(Search()), "cannot read deletion vector");
+    ASSERT_EQ(mock_reader->InvocationCount(), 0);
+
+    std::shared_ptr<GlobalIndexReader> union_reader = LiveRowGlobalIndexReader::CreateUnion(
+        std::vector<std::shared_ptr<GlobalIndexReader>>({mock_reader}), {Range(0, 9)},
+        {Range(5, 9)}, /*executor=*/nullptr, failing_supplier);
+    ASSERT_NOK_WITH_MSG(union_reader->VisitFullTextSearch(Search()), "cannot read deletion vector");
+    ASSERT_NOK_WITH_MSG(union_reader->VisitFullTextSearch(std::make_shared<FullTextSearch>(
+                            "f0", R"({"match":{"query":"apple"}})", /*limit=*/0)),
+                        "full-text search requires a positive limit");
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<ScoredGlobalIndexResult> result,
+                         union_reader->VisitFullTextSearch(Search(RoaringBitmap64())));
+    ASSERT_EQ(result->ToString(), "row ids: {}, scores: {}");
+    ASSERT_OK_AND_ASSIGN(
+        result, union_reader->VisitFullTextSearch(Search(RoaringBitmap64::From({2, 100}))));
+    ASSERT_EQ(result->ToString(), "row ids: {}, scores: {}");
+    ASSERT_EQ(mock_reader->InvocationCount(), 0);
+
+    ASSERT_OK(reader.VisitFullTextSearch(Search(RoaringBitmap64::From({100}))));
+    std::shared_ptr<FullTextSearch> captured = mock_reader->CapturedFullTextSearch();
+    ASSERT_TRUE(captured && captured->include_row_ids);
+    ASSERT_EQ(captured->include_row_ids.value(), RoaringBitmap64::From({100}));
+}
+
+TEST_F(LiveRowGlobalIndexReaderTest, TestOtherVisitsDoNotReadLiveRows) {
+    auto mock_reader = std::make_shared<MockGlobalIndexReader>();
+    mock_reader->SetDefaultResult({1, 2});
+    mock_reader->SetScoredResult({1}, {0.5f});
+    int32_t call_count = 0;
+    LiveRowGlobalIndexReader reader(mock_reader, Range(0, 9),
+                                    Supply(LiveRows(0, 9, {1}), &call_count));
+
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexResult> result, reader.VisitEqual(Literal(1)));
+    ASSERT_EQ(result->ToString(), "{1,2}");
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<ScoredGlobalIndexResult> scored_result,
+                         reader.VisitVectorSearch(nullptr));
+    ASSERT_EQ(scored_result->ToString(), "row ids: {1}, scores: {0.50}");
+    ASSERT_OK_AND_ASSIGN(scored_result, reader.VisitFullTextSearch(nullptr));
+    ASSERT_EQ(call_count, 0);
+    ASSERT_EQ(reader.GetIndexType(), "fake");
+    ASSERT_TRUE(reader.IsThreadSafe());
+}
+
+}  // namespace paimon::test
diff --git a/src/paimon/core/io/data_file_meta.cpp b/src/paimon/core/io/data_file_meta.cpp
index cd8666098..6ac7f5436 100644
--- a/src/paimon/core/io/data_file_meta.cpp
+++ b/src/paimon/core/io/data_file_meta.cpp
@@ -20,6 +20,7 @@
 
 #include <algorithm>
 #include <cstddef>
+#include <limits>
 #include <utility>
 
 #include "arrow/type_fwd.h"
@@ -167,6 +168,17 @@ Result<int64_t> DataFileMeta::CreationTimeEpochMillis() const {
     return utc_ts.GetMillisecond();
 }
 
+Result<Range> DataFileMeta::NonNullRowIdRange() const {
+    PAIMON_ASSIGN_OR_RAISE(int64_t start, NonNullFirstRowId());
+    if (start < 0 || row_count <= 0 ||
+        row_count - 1 > std::numeric_limits<int64_t>::max() - start) {
+        return Status::Invalid(
+            fmt::format("Invalid row id range of {}: first row id {}, row count {}.", file_name,
+                        start, row_count));
+    }
+    return Range(start, start + (row_count - 1));
+}
+
 Result<std::optional<RoaringBitmap32>> DataFileMeta::ToFileSelection(
     const std::optional<std::vector<Range>>& row_ranges) const {
     if (!row_ranges) {
diff --git a/src/paimon/core/io/data_file_meta.h b/src/paimon/core/io/data_file_meta.h
index ee3d4a9f5..0c9409c50 100644
--- a/src/paimon/core/io/data_file_meta.h
+++ b/src/paimon/core/io/data_file_meta.h
@@ -127,6 +127,11 @@ struct DataFileMeta {
         return Status::Invalid(fmt::format("First row id of {} should not be null.", file_name));
     }
 
+    /// Returns the global row ids of this file, as Java `nonNullRowIdRange` does. Fails if the
+    /// first row id is null or negative, the file has no rows, or its last row id would pass the
+    /// largest int64 value.
+    Result<Range> NonNullRowIdRange() const;
+
     // empty row_ranges indicates all rows in the file are needed, return null bitmap
     Result<std::optional<RoaringBitmap32>> ToFileSelection(
         const std::optional<std::vector<Range>>& row_ranges) const;
diff --git a/src/paimon/core/io/data_file_meta_test.cpp b/src/paimon/core/io/data_file_meta_test.cpp
index f76bc0ae8..b35631e42 100644
--- a/src/paimon/core/io/data_file_meta_test.cpp
+++ b/src/paimon/core/io/data_file_meta_test.cpp
@@ -18,6 +18,8 @@
 
 #include "paimon/core/io/data_file_meta.h"
 
+#include <limits>
+
 #include "gtest/gtest.h"
 #include "paimon/status.h"
 #include "paimon/testing/utils/binary_row_generator.h"
@@ -180,6 +182,34 @@ TEST(DataFileMetaTest, TestNonNullFirstRowId) {
     }
 }
 
+TEST(DataFileMetaTest, TestNonNullRowIdRange) {
+    auto create_file_meta = [](int64_t row_count, const std::optional<int64_t>& first_row_id) {
+        return std::make_shared<DataFileMeta>(
+            "data-0.orc", /*file_size=*/645, row_count, BinaryRow::EmptyRow(),
+            BinaryRow::EmptyRow(), SimpleStats::EmptyStats(), SimpleStats::EmptyStats(),
+            /*min_sequence_number=*/0, /*max_sequence_number=*/0, /*schema_id=*/0,
+            /*level=*/0, /*extra_files=*/std::vector<std::optional<std::string>>(),
+            /*creation_time=*/Timestamp(1737111915429ll, 0),
+            /*delete_row_count=*/0, /*embedded_index=*/nullptr, FileSource::Append(),
+            /*value_stats_cols=*/std::nullopt,
+            /*external_path=*/std::nullopt, first_row_id, /*write_cols=*/std::nullopt,
+            /*column_max_sequence_numbers=*/std::nullopt);
+    };
+    constexpr int64_t kMax = std::numeric_limits<int64_t>::max();
+    ASSERT_OK_AND_ASSIGN(Range range, create_file_meta(10, 100)->NonNullRowIdRange());
+    ASSERT_EQ(range, Range(100, 109));
+    ASSERT_OK_AND_ASSIGN(range, create_file_meta(2, kMax - 1)->NonNullRowIdRange());
+    ASSERT_EQ(range, Range(kMax - 1, kMax));
+    ASSERT_NOK_WITH_MSG(create_file_meta(3, kMax - 1)->NonNullRowIdRange(),
+                        "Invalid row id range of data-0.orc");
+    ASSERT_NOK_WITH_MSG(create_file_meta(0, 100)->NonNullRowIdRange(),
+                        "Invalid row id range of data-0.orc");
+    ASSERT_NOK_WITH_MSG(create_file_meta(10, -1)->NonNullRowIdRange(),
+                        "Invalid row id range of data-0.orc");
+    ASSERT_NOK_WITH_MSG(create_file_meta(10, std::nullopt)->NonNullRowIdRange(),
+                        "First row id of data-0.orc should not be null.");
+}
+
 TEST(DataFileMetaTest, TestToFileSelection) {
     // row id range [100, 110)
     auto file_meta = std::make_shared<DataFileMeta>(
diff --git a/src/paimon/core/operation/file_store_scan.h b/src/paimon/core/operation/file_store_scan.h
index b73ad095d..25911d0e5 100644
--- a/src/paimon/core/operation/file_store_scan.h
+++ b/src/paimon/core/operation/file_store_scan.h
@@ -130,6 +130,15 @@ class FileStoreScan {
         return this;
     }
 
+    /// Replaces the partition filter taken from the scan filter; a null filter accepts every
+    /// partition. `partition_schema` holds the partition fields the filter is indexed by.
+    FileStoreScan* WithPartitionFilter(const std::shared_ptr<PredicateFilter>& partition_filter,
+                                       const std::shared_ptr<arrow::Schema>& partition_schema) {
+        partition_filter_ = partition_filter;
+        partition_schema_ = partition_schema;
+        return this;
+    }
+
     FileStoreScan* WithTablePath(const std::string& table_path) {
         table_path_ = table_path;
         return this;
diff --git a/src/paimon/core/table/source/data_evolution_split_generator.cpp b/src/paimon/core/table/source/data_evolution_split_generator.cpp
index 0a76e75a0..62f334db3 100644
--- a/src/paimon/core/table/source/data_evolution_split_generator.cpp
+++ b/src/paimon/core/table/source/data_evolution_split_generator.cpp
@@ -28,6 +28,7 @@
 #include "paimon/common/utils/bin_packing.h"
 #include "paimon/common/utils/range_helper.h"
 #include "paimon/core/io/data_file_meta.h"
+#include "paimon/utils/range.h"
 
 namespace paimon {
 Result<std::vector<SplitGenerator::SplitGroup>> DataEvolutionSplitGenerator::SplitForBatch(
@@ -37,8 +38,8 @@ Result<std::vector<SplitGenerator::SplitGroup>> DataEvolutionSplitGenerator::Spl
             return meta->NonNullFirstRowId();
         },
         [](const std::shared_ptr<DataFileMeta>& meta) -> Result<int64_t> {
-            PAIMON_ASSIGN_OR_RAISE(int64_t first_row_id, meta->NonNullFirstRowId());
-            return first_row_id + meta->row_count - 1;
+            PAIMON_ASSIGN_OR_RAISE(Range row_id_range, meta->NonNullRowIdRange());
+            return row_id_range.to;
         });
 
     PAIMON_ASSIGN_OR_RAISE(std::vector<std::vector<std::shared_ptr<DataFileMeta>>> ranges,
diff --git a/src/paimon/core/table/source/split_generator_test.cpp b/src/paimon/core/table/source/split_generator_test.cpp
index 9f1368700..4ba05c3d5 100644
--- a/src/paimon/core/table/source/split_generator_test.cpp
+++ b/src/paimon/core/table/source/split_generator_test.cpp
@@ -21,6 +21,7 @@
 
 #include <algorithm>
 #include <cstdint>
+#include <limits>
 #include <memory>
 #include <optional>
 #include <string>
@@ -260,6 +261,22 @@ TEST_F(SplitGeneratorTest, TestDataEvolutionBlobSplitByFileSize) {
     }
 }
 
+TEST_F(SplitGeneratorTest, TestDataEvolutionRowIdRangeAtInt64Max) {
+    constexpr int64_t kMax = std::numeric_limits<int64_t>::max();
+    DataEvolutionSplitGenerator split_generator(/*target_split_size=*/1000, /*open_file_cost=*/10,
+                                                /*count_blob_size=*/true);
+    ASSERT_OK_AND_ASSIGN(std::vector<SplitGenerator::SplitGroup> split_groups,
+                         split_generator.SplitForBatch(
+                             {CreateDataFileMetaWithRowId("f1", /*file_size=*/100, /*row_count=*/2,
+                                                          /*first_row_id=*/kMax - 1),
+                              CreateDataFileMetaWithRowId("f2", /*file_size=*/100, /*row_count=*/1,
+                                                          /*first_row_id=*/kMax)}));
+    CheckResult(split_groups, {{"f1", "f2"}}, {false});
+    ASSERT_NOK_WITH_MSG(split_generator.SplitForBatch({CreateDataFileMetaWithRowId(
+                            "f3", /*file_size=*/100, /*row_count=*/3, /*first_row_id=*/kMax - 1)}),
+                        "Invalid row id range of f3");
+}
+
 TEST_F(SplitGeneratorTest, TestMergeTree) {
     std::vector<std::shared_ptr<DataFileMeta>> files = {
         CreateDataFileMeta("1", 0, 10),  CreateDataFileMeta("2", 0, 12),
diff --git a/test/inte/data_evolution_table_test.cpp b/test/inte/data_evolution_table_test.cpp
index c508fd199..2924e5413 100644
--- a/test/inte/data_evolution_table_test.cpp
+++ b/test/inte/data_evolution_table_test.cpp
@@ -32,9 +32,16 @@
 #include "paimon/common/utils/path_util.h"
 #include "paimon/common/utils/scope_guard.h"
 #include "paimon/common/utils/vector_store_utils.h"
+#include "paimon/core/core_options.h"
+#include "paimon/core/global_index/global_index_live_row_filter.h"
 #include "paimon/core/global_index/indexed_split_impl.h"
+#include "paimon/core/operation/file_store_scan.h"
+#include "paimon/core/schema/schema_manager.h"
+#include "paimon/core/snapshot.h"
 #include "paimon/core/table/source/data_split_impl.h"
+#include "paimon/core/utils/snapshot_manager.h"
 #include "paimon/defs.h"
+#include "paimon/executor.h"
 #include "paimon/format/mosaic/mosaic_format_defs.h"
 #include "paimon/fs/file_system.h"
 #include "paimon/global_index/bitmap_global_index_result.h"
@@ -163,8 +170,8 @@ class DataEvolutionTableTest : public ::testing::Test,
     /// created with, which a test that later evolves the schema has to pass along.
     /// `extra_options` is merged in for tests that also pin a read batch or split target size.
     std::map<std::string, std::string> CreateDataEvolutionTable(
-        bool deletion_vectors_enabled,
-        const std::map<std::string, std::string>& extra_options = {}) const {
+        bool deletion_vectors_enabled, const std::map<std::string, std::string>& extra_options = {},
+        const std::vector<std::string>& partition_keys = {}) const {
         std::map<std::string, std::string> options = {{Options::FILE_FORMAT, FileFormat()},
                                                       {Options::FILE_SYSTEM, "local"},
                                                       {Options::ROW_TRACKING_ENABLED, "true"},
@@ -175,7 +182,7 @@ class DataEvolutionTableTest : public ::testing::Test,
                             std::get<2>(GetParam()) ? "true" : "false");
         }
         options.insert(extra_options.begin(), extra_options.end());
-        CreateTable(/*partition_keys=*/{}, options);
+        CreateTable(partition_keys, options);
         return options;
     }
 
@@ -3386,6 +3393,72 @@ TEST_P(DataEvolutionDvTableTest, TestGlobalIndexScoresWithMergedFileFiltering) {
     CheckGlobalIndexScoresWithFiltering(/*merge_files=*/true);
 }
 
+TEST_P(DataEvolutionDvTableTest, TestGlobalIndexLiveRows) {
+    CreateDataEvolutionTable(/*deletion_vectors_enabled=*/true, /*extra_options=*/{},
+                             /*partition_keys=*/{"f1"});
+    std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
+    ASSERT_OK_AND_ASSIGN(
+        std::vector<std::shared_ptr<CommitMessage>> group_msgs0,
+        WriteAndCommitGroup(table_path, /*first_row_id=*/0, /*f0_values=*/{0, 1, 2, 3},
+                            /*partition=*/{{"f1", "p0"}}));
+    ASSERT_OK_AND_ASSIGN(
+        std::vector<std::shared_ptr<CommitMessage>> group_msgs1,
+        WriteAndCommitGroup(table_path, /*first_row_id=*/4, /*f0_values=*/{4, 5, 6, 7},
+                            /*partition=*/{{"f1", "p1"}}));
+    ASSERT_OK_AND_ASSIGN(std::vector<std::string> anchor_file_names,
+                         PlannedAnchorFileNames(table_path));
+    ASSERT_EQ(anchor_file_names.size(), 2);
+
+    std::shared_ptr<FileSystem> fs = dir_->GetFileSystem();
+    SchemaManager schema_manager(fs, table_path);
+    ASSERT_OK_AND_ASSIGN(std::optional<std::shared_ptr<TableSchema>> table_schema,
+                         schema_manager.Latest());
+    ASSERT_TRUE(table_schema);
+    ASSERT_OK_AND_ASSIGN(CoreOptions options,
+                         CoreOptions::FromMap(table_schema.value()->Options(), fs));
+    SnapshotManager snapshot_manager(fs, table_path);
+    ASSERT_OK_AND_ASSIGN(std::optional<Snapshot> snapshot_before_deletion,
+                         snapshot_manager.LatestSnapshot());
+    ASSERT_TRUE(snapshot_before_deletion);
+
+    ASSERT_OK(CommitDeletionVectors(table_path, group_msgs0[0],
+                                    {{anchor_file_names[0], /*deleted_positions=*/{1, 3}}}));
+    ASSERT_OK(CommitDeletionVectors(table_path, group_msgs1[0],
+                                    {{anchor_file_names[1], /*deleted_positions=*/{0}}}));
+    ASSERT_OK_AND_ASSIGN(std::optional<Snapshot> snapshot, snapshot_manager.LatestSnapshot());
+    ASSERT_TRUE(snapshot);
+
+    std::shared_ptr<Executor> executor = CreateDefaultExecutor();
+    auto read_live_rows = [&](const Snapshot& read_snapshot,
+                              const std::shared_ptr<PredicateFilter>& partition_filter) {
+        return GlobalIndexLiveRowFilter::LiveRows(table_path, table_schema.value(), read_snapshot,
+                                                  partition_filter, options, executor,
+                                                  GetDefaultPool());
+    };
+    ASSERT_OK_AND_ASSIGN(std::optional<RoaringBitmap64> live_rows,
+                         read_live_rows(snapshot.value(), /*partition_filter=*/nullptr));
+    ASSERT_TRUE(live_rows);
+    ASSERT_EQ(live_rows.value(), RoaringBitmap64::From({0, 2, 5, 6, 7}));
+
+    auto arrow_schema = DataField::ConvertDataFieldsToArrowSchema(table_schema.value()->Fields());
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<PredicateFilter> p0_filter,
+                         FileStoreScan::CreatePartitionPredicate(
+                             table_schema.value()->PartitionKeys(),
+                             options.GetPartitionDefaultName(), arrow_schema, {{{"f1", "p0"}}}));
+    ASSERT_OK_AND_ASSIGN(live_rows, read_live_rows(snapshot.value(), p0_filter));
+    ASSERT_TRUE(live_rows);
+    ASSERT_EQ(live_rows.value(), RoaringBitmap64::From({0, 2}));
+
+    ASSERT_OK_AND_ASSIGN(live_rows, read_live_rows(snapshot_before_deletion.value(),
+                                                   /*partition_filter=*/nullptr));
+    ASSERT_TRUE(live_rows);
+    ASSERT_EQ(live_rows.value(), RoaringBitmap64::From({0, 1, 2, 3, 4, 5, 6, 7}));
+
+    ASSERT_OK(fs->Delete(PathUtil::JoinPath(table_path, "index")));
+    ASSERT_NOK_WITH_MSG(read_live_rows(snapshot.value(), /*partition_filter=*/nullptr),
+                        "failed to read deletion vector of data file");
+}
+
 TEST_P(DataEvolutionDvTableTest, TestReadWithDeletionVectors) {
     CreateDataEvolutionTable(/*deletion_vectors_enabled=*/true);
     std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
diff --git a/test/inte/global_index_test.cpp b/test/inte/global_index_test.cpp
index 9c36766b2..97b169823 100644
--- a/test/inte/global_index_test.cpp
+++ b/test/inte/global_index_test.cpp
@@ -47,6 +47,7 @@
 #include "paimon/status.h"
 #include "paimon/testing/utils/binary_row_generator.h"
 #include "paimon/testing/utils/counting_cache_test_utils.h"
+#include "paimon/testing/utils/deletion_vector_test_helper.h"
 #include "paimon/testing/utils/io_exception_helper.h"
 #include "paimon/testing/utils/test_helper.h"
 #include "paimon/testing/utils/testharness.h"
@@ -283,6 +284,113 @@ class GlobalIndexTest : public ::testing::Test, public ::testing::WithParamInter
         CheckFullTextTopK(readers[0], query, std::nullopt, {3});
     }
 
+    void CheckSearchSkipsDeletedRows(const std::string& index_type,
+                                     const std::map<std::string, std::string>& index_options) {
+        arrow::FieldVector fields = {arrow::field("f0", arrow::utf8()),
+                                     arrow::field("f1", arrow::int32())};
+        auto schema = arrow::schema(fields);
+        std::map<std::string, std::string> options = {{Options::FILE_FORMAT, file_format_},
+                                                      {Options::FILE_SYSTEM, "local"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::DELETION_VECTORS_ENABLED, "true"}};
+        CreateTable(/*partition_keys=*/{}, schema, options);
+        std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar");
+
+        auto src_array1 = arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
+["apple apple apple apple apple apple apple apple apple apple", 0],
+["apple apple apple apple apple apple apple apple apple", 1],
+["apple apple apple apple apple apple apple apple", 2],
+["apple apple apple apple apple apple apple", 3],
+["apple apple apple apple apple apple", 4]
+    ])")
+                              .ValueOrDie();
+        auto src_array2 = arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields), R"([
+["apple apple apple apple apple", 5],
+["apple apple apple apple", 6],
+["apple apple apple", 7],
+["apple apple", 8],
+["apple", 9]
+    ])")
+                              .ValueOrDie();
+        ASSERT_OK_AND_ASSIGN(auto commit_msgs1,
+                             WriteArray(table_path, schema->field_names(), src_array1));
+        ASSERT_OK(Commit(table_path, commit_msgs1));
+        ASSERT_OK_AND_ASSIGN(auto commit_msgs2,
+                             WriteArray(table_path, schema->field_names(), src_array2));
+        ASSERT_OK(Commit(table_path, commit_msgs2));
+        ASSERT_OK(WriteIndex(table_path, /*partition_filters=*/{}, "f0", index_type, index_options,
+                             Range(0, 4)));
+        ASSERT_OK(WriteIndex(table_path, /*partition_filters=*/{}, "f0", index_type, index_options,
+                             Range(5, 9)));
+
+        auto create_scan = [&]() {
+            return GlobalIndexScan::Create(table_path, /*snapshot_id=*/std::nullopt,
+                                           /*partitions=*/std::nullopt, /*options=*/{}, fs_,
+                                           /*executor=*/nullptr, pool_);
+        };
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexScan> scan_before_deletion, create_scan());
+
+        ASSERT_OK_AND_ASSIGN(auto split, ScanData(table_path, /*partition_filters=*/{}));
+        ASSERT_OK_AND_ASSIGN(std::vector<std::string> anchor_file_names,
+                             DeletionVectorTestHelper::RetrieveAnchorFileNames(split->DataFiles()));
+        ASSERT_EQ(anchor_file_names.size(), 2u);
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<CommitMessage> dv_commit_msg,
+                             DeletionVectorTestHelper::CreateDeletionVectorCommitMessage(
+                                 dir_->GetFileSystem(), table_path, file_format_, commit_msgs1[0],
+                                 {{anchor_file_names[0], /*deleted_positions=*/{0, 1, 2, 3}},
+                                  {anchor_file_names[1], /*deleted_positions=*/{0}}},
+                                 pool_));
+        ASSERT_OK(Commit(table_path, {dv_commit_msg}));
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<GlobalIndexScan> scan, create_scan());
+
+        std::string query = R"({"match":{"query":"apple"}})";
+        ASSERT_OK_AND_ASSIGN(
+            auto readers_before_deletion,
+            scan_before_deletion->CreateReaders("f0", /*row_range_index=*/std::nullopt));
+        ASSERT_EQ(readers_before_deletion.size(), 1u);
+        CheckFullTextTopK(readers_before_deletion[0], query, std::nullopt,
+                          {0, 1, 2, 3, 4, 5, 6, 7, 8, 9});
+        ASSERT_OK_AND_ASSIGN(
+            ScoredRows rows_before_deletion,
+            SearchFullText(readers_before_deletion[0], query, /*limit=*/100, std::nullopt));
+        RoaringBitmap64 deleted_rows = RoaringBitmap64::From({0, 1, 2, 3, 5});
+        ScoredRows expected_live_rows;
+        for (const auto& row : rows_before_deletion) {
+            if (!deleted_rows.Contains(row.first)) {
+                expected_live_rows.push_back(row);
+            }
+        }
+
+        ASSERT_OK_AND_ASSIGN(auto readers,
+                             scan->CreateReaders("f0", /*row_range_index=*/std::nullopt));
+        ASSERT_EQ(readers.size(), 1u);
+        ASSERT_OK_AND_ASSIGN(ScoredRows live_rows,
+                             SearchFullText(readers[0], query, /*limit=*/100, std::nullopt));
+        ASSERT_EQ(live_rows, expected_live_rows);
+        CheckFullTextTopK(readers[0], query, std::nullopt, {4, 6, 7, 8, 9});
+        CheckFullTextTopK(readers[0], query, RoaringBitmap64::From({0, 5, 8}), {8});
+        CheckFullTextTopK(readers[0], query, RoaringBitmap64::From({0, 1}), {});
+
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<ScoredGlobalIndexResult> index_result,
+                             readers[0]->VisitFullTextSearch(
+                                 std::make_shared<FullTextSearch>("f0", query, /*limit=*/5)));
+        ASSERT_OK_AND_ASSIGN(auto plan, ScanGlobalIndexAndData(table_path, /*predicate=*/nullptr,
+                                                               /*options=*/{}, index_result));
+        auto expected_array =
+            arrow::ipc::internal::json::ArrayFromJSON(
+                arrow::struct_({SpecialFields::ValueKind().ArrowField(), fields[1]}),
+                R"([
+[0, 4],
+[0, 6],
+[0, 7],
+[0, 8],
+[0, 9]
+    ])")
+                .ValueOrDie();
+        ASSERT_OK(ReadData(table_path, {"f1"}, expected_array, /*predicate=*/nullptr, plan));
+    }
+
     Result<std::shared_ptr<Plan>> ScanGlobalIndexAndData(
         const std::string& table_path, const std::shared_ptr<Predicate>& predicate,
         const std::map<std::string, std::string>& options = {},
@@ -2924,6 +3032,14 @@ TEST_P(GlobalIndexTest, TestLuceneSearchOverlappingIndexRanges) {
                                        {"lucene-fts.write.tmp.directory", tmp_dir->Str()}},
                                       R"({"match":{"query":"document","operator":"And"}})");
 }
+
+TEST_P(GlobalIndexTest, TestLuceneSearchSkipsRowsDeletedByDeletionVectors) {
+    auto tmp_dir = paimon::test::UniqueTestDirectory::Create();
+    ASSERT_TRUE(tmp_dir);
+    CheckSearchSkipsDeletedRows("lucene-fts",
+                                {{"lucene-fts.write.omit-term-freq-and-position", "false"},
+                                 {"lucene-fts.write.tmp.directory", tmp_dir->Str()}});
+}
 #endif
 
 #ifdef PAIMON_ENABLE_FULL_TEXT
@@ -3071,6 +3187,10 @@ TEST_P(GlobalIndexTest, TestFullTextSearchOverlappingIndexRanges) {
     CheckSearchOverlappingIndexRanges("full-text", {{"full-text.tokenizer", "default"}},
                                       R"({"match":{"query":"document"}})");
 }
+
+TEST_P(GlobalIndexTest, TestFullTextSearchSkipsRowsDeletedByDeletionVectors) {
+    CheckSearchSkipsDeletedRows("full-text", {{"full-text.tokenizer", "default"}});
+}
 #endif
 
 TEST_P(GlobalIndexTest, TestBTreeWriteCommitScanReadIndex) {