diff --git a/CMakeLists.txt b/CMakeLists.txt
index a24a2c603..cf1b59837 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -306,8 +306,10 @@ include(san-config)
 
 # Code coverage
 if("${PAIMON_GENERATE_COVERAGE}")
-    set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} --coverage -DCOVERAGE_BUILD")
-    set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} --coverage -DCOVERAGE_BUILD")
+    set(CMAKE_C_FLAGS
+        "${CMAKE_C_FLAGS} --coverage -fprofile-update=atomic -DCOVERAGE_BUILD")
+    set(CMAKE_CXX_FLAGS
+        "${CMAKE_CXX_FLAGS} --coverage -fprofile-update=atomic -DCOVERAGE_BUILD")
 endif()
 
 # CMAKE_CXX_FLAGS now fully assembled
diff --git a/cmake_modules/ThirdpartyToolchain.cmake b/cmake_modules/ThirdpartyToolchain.cmake
index ac94fc2b4..8aa8f1183 100644
--- a/cmake_modules/ThirdpartyToolchain.cmake
+++ b/cmake_modules/ThirdpartyToolchain.cmake
@@ -430,6 +430,8 @@ string(REPLACE "-Wglobal-constructors" "" EP_C_FLAGS ${EP_C_FLAGS})
 # Remove coverage flags from third-party dependencies to avoid gcov dependency
 string(REPLACE "--coverage" "" EP_CXX_FLAGS ${EP_CXX_FLAGS})
 string(REPLACE "--coverage" "" EP_C_FLAGS ${EP_C_FLAGS})
+string(REPLACE " -fprofile-update=atomic" "" EP_CXX_FLAGS ${EP_CXX_FLAGS})
+string(REPLACE " -fprofile-update=atomic" "" EP_C_FLAGS ${EP_C_FLAGS})
 string(REPLACE "-DCOVERAGE_BUILD" "" EP_CXX_FLAGS ${EP_CXX_FLAGS})
 string(REPLACE "-DCOVERAGE_BUILD" "" EP_C_FLAGS ${EP_C_FLAGS})
 if(NOT MSVC_TOOLCHAIN)
diff --git a/src/paimon/common/global_index/bitmap/bitmap_global_index_test.cpp b/src/paimon/common/global_index/bitmap/bitmap_global_index_test.cpp
index d7ed988ef..22b259d70 100644
--- a/src/paimon/common/global_index/bitmap/bitmap_global_index_test.cpp
+++ b/src/paimon/common/global_index/bitmap/bitmap_global_index_test.cpp
@@ -34,6 +34,8 @@
 #include "paimon/core/global_index/global_index_file_manager.h"
 #include "paimon/data/timestamp.h"
 #include "paimon/global_index/bitmap_global_index_result.h"
+#include "paimon/predicate/predicate_builder.h"
+#include "paimon/predicate/predicate_utils.h"
 #include "paimon/testing/mock/mock_index_path_factory.h"
 #include "paimon/testing/utils/testharness.h"
 
@@ -125,10 +127,31 @@ TEST_F(BitmapGlobalIndexTest, WriteAndReadStringPredicates) {
     CheckResult(reader->VisitGreaterThan(StringLiteral("banana")), {4, 5});
     CheckResult(reader->VisitGreaterOrEqual(StringLiteral("banana")), {3, 4, 5});
 
-    CheckResult(reader->VisitStartsWith(StringLiteral("ban")), {3, 4});
-    CheckResult(reader->VisitEndsWith(StringLiteral("le")), {0, 1});
-    CheckResult(reader->VisitContains(StringLiteral("an")), {3, 4});
-    CheckResult(reader->VisitLike(StringLiteral("ban_n_")), {3});
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<Predicate> starts,
+        PredicateBuilder::StartsWith(0, "f0", FieldType::STRING, StringLiteral("ban")));
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<Predicate> ends,
+        PredicateBuilder::EndsWith(0, "f0", FieldType::STRING, StringLiteral("le")));
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<Predicate> contains,
+        PredicateBuilder::Contains(0, "f0", FieldType::STRING, StringLiteral("an")));
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<Predicate> like,
+        PredicateBuilder::Like(0, "f0", FieldType::STRING, StringLiteral("ban_n_")));
+    using IndexResult = std::shared_ptr<GlobalIndexResult>;
+    CheckResult(PredicateUtils::VisitPredicate<IndexResult>(
+                    std::dynamic_pointer_cast<LeafPredicate>(starts), reader),
+                {3, 4});
+    CheckResult(PredicateUtils::VisitPredicate<IndexResult>(
+                    std::dynamic_pointer_cast<LeafPredicate>(ends), reader),
+                {0, 1});
+    CheckResult(PredicateUtils::VisitPredicate<IndexResult>(
+                    std::dynamic_pointer_cast<LeafPredicate>(contains), reader),
+                {3, 4});
+    CheckResult(PredicateUtils::VisitPredicate<IndexResult>(
+                    std::dynamic_pointer_cast<LeafPredicate>(like), reader),
+                {3});
     ASSERT_NOK_WITH_MSG(reader->VisitVectorSearch(nullptr),
                         "Vector search is not supported in BitmapIndexReader");
     ASSERT_NOK_WITH_MSG(reader->VisitFullTextSearch(nullptr),
diff --git a/src/paimon/common/reader/prefetch_file_batch_reader_impl_test.cpp b/src/paimon/common/reader/prefetch_file_batch_reader_impl_test.cpp
index 54ed531ec..a741077e9 100644
--- a/src/paimon/common/reader/prefetch_file_batch_reader_impl_test.cpp
+++ b/src/paimon/common/reader/prefetch_file_batch_reader_impl_test.cpp
@@ -47,6 +47,12 @@
 
 namespace paimon::test {
 
+TEST(PrefetchFileBatchReaderTest, DefaultPreBufferRangeIsEmpty) {
+    MockFileBatchReader reader(nullptr, arrow::struct_({}), 1);
+    ASSERT_OK_AND_ASSIGN(auto ranges, reader.PrefetchFileBatchReader::PreBufferRange());
+    ASSERT_TRUE(ranges.empty());
+}
+
 class ControlledMockFileBatchReader : public MockFileBatchReader {
  public:
     ControlledMockFileBatchReader(const std::shared_ptr<arrow::Array>& data,
@@ -150,6 +156,21 @@ class IoReadingMockFileBatchReader : public MockFileBatchReader {
     std::shared_ptr<InputStream> input_stream_;
 };
 
+class StreamCapturingMockReaderBuilder : public MockFormatReaderBuilder {
+ public:
+    using MockFormatReaderBuilder::MockFormatReaderBuilder;
+
+    Result<std::unique_ptr<FileBatchReader>> Build(
+        const std::shared_ptr<InputStream>& input_stream) const override {
+        std::lock_guard<std::mutex> lock(mutex_);
+        input_streams_.push_back(input_stream);
+        return MockFormatReaderBuilder::Build(input_stream);
+    }
+
+    mutable std::mutex mutex_;
+    mutable std::vector<std::shared_ptr<InputStream>> input_streams_;
+};
+
 class IoReadingMockFormatReaderBuilder : public ReaderBuilder {
  public:
     IoReadingMockFormatReaderBuilder(const std::shared_ptr<arrow::Array>& data,
@@ -493,6 +514,164 @@ TEST_F(PrefetchFileBatchReaderImplTest, TestFailedIoMetrics) {
     ASSERT_EQ(latency_count, read_requests);
 }
 
+TEST_F(PrefetchFileBatchReaderImplTest, TestSequentialIoMetricsAndPositionForwarding) {
+    const std::string path = PathUtil::JoinPath(dir_->Str(), "stream-data");
+    ASSERT_OK(local_fs_->WriteFile(path, "abcdef", /*overwrite=*/false));
+    StreamCapturingMockReaderBuilder builder(PrepareArray(1), data_type_, 1);
+    ASSERT_OK_AND_ASSIGN(
+        auto reader,
+        PrefetchFileBatchReaderImpl::Create(
+            path, 6, &builder, local_fs_, /*prefetch_max_parallel_num=*/1, /*batch_size=*/1,
+            /*prefetch_batch_count=*/1, /*enable_adaptive_prefetch_strategy=*/false, executor_,
+            /*initialize_read_ranges=*/false, /*read_ahead_cache_enabled=*/false, CacheConfig(),
+            /*enable_io_metrics=*/true, WarmupLevel::NONE, pool_, GetArrowPool(pool_)));
+    ASSERT_EQ(builder.input_streams_.size(), 1u);
+    const auto& stream = builder.input_streams_.front();
+    ASSERT_OK(stream->Seek(2, FS_SEEK_SET));
+    ASSERT_OK_AND_ASSIGN(int64_t position, stream->GetPos());
+    ASSERT_EQ(position, 2);
+    char buffer[8] = {};
+    ASSERT_OK_AND_ASSIGN(int64_t bytes_read, stream->Read(buffer, 4));
+    ASSERT_EQ(bytes_read, 4);
+    ASSERT_EQ(std::string(buffer, bytes_read), "cdef");
+    ASSERT_OK_AND_ASSIGN(position, stream->GetPos());
+    ASSERT_EQ(position, 6);
+    ASSERT_OK(stream->Seek(-2, FS_SEEK_END));
+    ASSERT_OK_AND_ASSIGN(bytes_read, stream->Read(buffer, 2));
+    ASSERT_EQ(std::string(buffer, bytes_read), "ef");
+    ASSERT_OK(stream->Seek(0, FS_SEEK_SET));
+    ASSERT_NOK(stream->Read(buffer, 8));
+    const auto metrics = reader->GetReaderMetrics();
+    ASSERT_OK_AND_ASSIGN(uint64_t requests, metrics->GetCounter(PrefetchIoMetrics::READ_REQUESTS));
+    ASSERT_EQ(requests, 3);
+    ASSERT_OK_AND_ASSIGN(uint64_t requested,
+                         metrics->GetCounter(PrefetchIoMetrics::READ_REQUESTED_BYTES));
+    ASSERT_EQ(requested, 14);
+    ASSERT_OK_AND_ASSIGN(uint64_t physical,
+                         metrics->GetCounter(PrefetchIoMetrics::READ_PHYSICAL_BYTES));
+    ASSERT_EQ(physical, 6);
+    ASSERT_OK_AND_ASSIGN(uint64_t failures, metrics->GetCounter(PrefetchIoMetrics::READ_FAILED));
+    ASSERT_EQ(failures, 1);
+}
+
+TEST_F(PrefetchFileBatchReaderImplTest, TestMinimumQueueCapacityAndFileSchema) {
+    auto data = PrepareArray(10);
+    MockFormatReaderBuilder builder(data, data_type_, 2);
+    ASSERT_OK_AND_ASSIGN(
+        auto reader,
+        PrefetchFileBatchReaderImpl::Create(
+            "", 0, &builder, mock_fs_, /*prefetch_max_parallel_num=*/3, /*batch_size=*/2,
+            /*prefetch_batch_count=*/1, /*enable_adaptive_prefetch_strategy=*/false, executor_,
+            /*initialize_read_ranges=*/true, /*read_ahead_cache_enabled=*/false, CacheConfig(),
+            /*enable_io_metrics=*/false, WarmupLevel::NONE, pool_, GetArrowPool(pool_)));
+    ASSERT_EQ(reader->prefetch_queue_capacity_, 1u);
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<ArrowSchema> c_schema, reader->GetFileSchema());
+    auto imported_schema = arrow::ImportSchema(c_schema.get());
+    ASSERT_TRUE(imported_schema.ok()) << imported_schema.status().ToString();
+    std::shared_ptr<arrow::Schema> schema = std::move(imported_schema).ValueOrDie();
+    ASSERT_TRUE(schema->Equals(*arrow::schema(fields_)));
+    ASSERT_OK_AND_ASSIGN(auto result, CollectResultAndRowIds(reader.get()));
+    ASSERT_TRUE(result.first);
+    ASSERT_TRUE(result.first->Equals(*std::make_shared<arrow::ChunkedArray>(data)));
+}
+
+TEST_F(PrefetchFileBatchReaderImplTest, TestDiscardedBatchesAreCountedAndReleased) {
+    for (int32_t scenario : {0, 1, 2}) {
+        SCOPED_TRACE(scenario);
+        MockFormatReaderBuilder builder(PrepareArray(20), data_type_, 5);
+        ASSERT_OK_AND_ASSIGN(
+            auto reader,
+            PrefetchFileBatchReaderImpl::Create(
+                "", 0, &builder, mock_fs_, /*prefetch_max_parallel_num=*/1, /*batch_size=*/5,
+                /*prefetch_batch_count=*/2, /*enable_adaptive_prefetch_strategy=*/false, executor_,
+                /*initialize_read_ranges=*/true, /*read_ahead_cache_enabled=*/false, CacheConfig(),
+                /*enable_io_metrics=*/false, WarmupLevel::NONE, pool_, GetArrowPool(pool_)));
+        auto* source = dynamic_cast<MockFileBatchReader*>(reader->readers_.front().get());
+        ASSERT_TRUE(source);
+        if (scenario == 2) {
+            ASSERT_OK(reader->SetReadRanges({{0, 5}}));
+            ASSERT_OK(source->SetReadRanges({{10, 15}}));
+        }
+        ASSERT_OK_AND_ASSIGN(auto batch, source->NextBatchWithBitmap());
+        ASSERT_FALSE(BatchReader::IsEofBatch(batch));
+        auto imported_array = arrow::ImportArray(batch.first.first.get(), batch.first.second.get());
+        ASSERT_TRUE(imported_array.ok()) << imported_array.status().ToString();
+        std::shared_ptr<arrow::Array> array = std::move(imported_array).ValueOrDie();
+        if (scenario == 0) {
+            array = PrepareArray(0);
+        }
+        std::weak_ptr<arrow::ArrayData> lifetime = array->data();
+        ASSERT_TRUE(
+            arrow::ExportArray(*array, batch.first.first.get(), batch.first.second.get()).ok());
+        array.reset();
+        if (scenario == 1) {
+            batch.second = RoaringBitmap32();
+        }
+        ASSERT_FALSE(lifetime.expired());
+        ASSERT_OK(reader->HandleReadResult(/*reader_idx=*/0, {0, 5}, std::move(batch)));
+        ASSERT_TRUE(lifetime.expired());
+        ASSERT_TRUE(reader->prefetch_queues_.front()->empty());
+        ASSERT_OK_AND_ASSIGN(uint64_t discarded, reader->GetReaderMetrics()->GetCounter(
+                                                     PrefetchMetrics::DISCARDED_BATCHES));
+        ASSERT_EQ(discarded, 1);
+        ASSERT_OK_AND_ASSIGN(uint64_t produced, reader->GetReaderMetrics()->GetCounter(
+                                                    PrefetchMetrics::PRODUCED_BATCHES));
+        ASSERT_EQ(produced, 0);
+    }
+}
+
+TEST_F(PrefetchFileBatchReaderImplTest, TestBatchFilteredIntoLaterReadRange) {
+    auto data = PrepareArray(20);
+    RoaringBitmap32 bitmap;
+    bitmap.Add(12);
+    bitmap.Add(13);
+    MockFormatReaderBuilder builder(data, data_type_, bitmap, 5);
+    ASSERT_OK_AND_ASSIGN(
+        auto reader,
+        PrefetchFileBatchReaderImpl::Create(
+            "", 0, &builder, mock_fs_, /*prefetch_max_parallel_num=*/1, /*batch_size=*/5,
+            /*prefetch_batch_count=*/2, /*enable_adaptive_prefetch_strategy=*/false, executor_,
+            /*initialize_read_ranges=*/true, /*read_ahead_cache_enabled=*/false, CacheConfig(),
+            /*enable_io_metrics=*/false, WarmupLevel::NONE, pool_, GetArrowPool(pool_)));
+    auto* mock_reader = dynamic_cast<MockFileBatchReader*>(reader->readers_[0].get());
+    ASSERT_TRUE(mock_reader);
+    mock_reader->EnableRandomizeBatchSize(false);
+    ASSERT_OK(reader->SetReadRanges({{0, 5}, {10, 15}}));
+    ASSERT_OK_AND_ASSIGN(auto batch_with_bitmap, reader->NextBatchWithBitmap());
+    ASSERT_FALSE(BatchReader::IsEofBatch(batch_with_bitmap));
+    ASSERT_EQ(batch_with_bitmap.second, RoaringBitmap32::From({2, 3}));
+    // Row IDs address the original batch, before the selection bitmap is applied.
+    ASSERT_OK_AND_ASSIGN(uint64_t first_row_id, reader->GetPreviousBatchFileRowId(2));
+    ASSERT_OK_AND_ASSIGN(uint64_t last_row_id, reader->GetPreviousBatchFileRowId(3));
+    ASSERT_EQ(first_row_id, 12);
+    ASSERT_EQ(last_row_id, 13);
+    ASSERT_OK_AND_ASSIGN(auto batch, ReaderUtils::ApplyBitmapToReadBatch(
+                                         std::move(batch_with_bitmap), GetArrowPool(pool_)));
+    ASSERT_OK_AND_ASSIGN(auto result, ReadResultCollector::GetArray(std::move(batch)));
+    ASSERT_TRUE(result);
+    ASSERT_TRUE(result->Equals(data->Slice(12, 2)));
+    ASSERT_OK_AND_ASSIGN(auto eof_batch, reader->NextBatchWithBitmap());
+    ASSERT_TRUE(BatchReader::IsEofBatch(eof_batch));
+    ASSERT_OK_AND_ASSIGN(uint64_t produced,
+                         reader->GetReaderMetrics()->GetCounter(PrefetchMetrics::PRODUCED_BATCHES));
+    ASSERT_EQ(produced, 1);
+}
+
+TEST_F(PrefetchFileBatchReaderImplTest, TestWorkloopStopsAfterReadError) {
+    MockFormatReaderBuilder builder(PrepareArray(10), data_type_, 5);
+    ASSERT_OK_AND_ASSIGN(
+        auto reader,
+        PrefetchFileBatchReaderImpl::Create(
+            "", 0, &builder, mock_fs_, /*prefetch_max_parallel_num=*/1, /*batch_size=*/5,
+            /*prefetch_batch_count=*/2, /*enable_adaptive_prefetch_strategy=*/false, executor_,
+            /*initialize_read_ranges=*/true, /*read_ahead_cache_enabled=*/false, CacheConfig(),
+            /*enable_io_metrics=*/false, WarmupLevel::NONE, pool_, GetArrowPool(pool_)));
+    reader->SetReadStatus(Status::IOError("reader stopped"));
+    reader->Workloop();
+    ASSERT_NOK_WITH_MSG(reader->GetReadStatus(), "reader stopped");
+    ASSERT_TRUE(reader->prefetch_queues_.front()->empty());
+}
+
 TEST_F(PrefetchFileBatchReaderImplTest, FilterReadRangesWithoutBitmap) {
     std::vector<std::pair<uint64_t, uint64_t>> read_ranges = {
         {0, 1000},    {1000, 2000}, {2000, 3000}, {3000, 4000}, {4000, 5000},
diff --git a/src/paimon/core/catalog/file_system_catalog_test.cpp b/src/paimon/core/catalog/file_system_catalog_test.cpp
index 51e7c642d..f02af7be3 100644
--- a/src/paimon/core/catalog/file_system_catalog_test.cpp
+++ b/src/paimon/core/catalog/file_system_catalog_test.cpp
@@ -87,6 +87,70 @@ TEST(FileSystemCatalogTest, TestDatabaseExists) {
     ASSERT_EQ(db_location, PathUtil::JoinPath(dir->Str(), "db1.db"));
 }
 
+TEST(FileSystemCatalogTest, TestGlobalSystemTableDiscoveryAndSchema) {
+    std::map<std::string, std::string> options = {{Options::FILE_SYSTEM, "local"}};
+    ASSERT_OK_AND_ASSIGN(CoreOptions core_options, CoreOptions::FromMap(options));
+    auto dir = UniqueTestDirectory::Create();
+    ASSERT_TRUE(dir);
+    FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str(), options);
+    ASSERT_EQ(options, catalog.GetOptions());
+    ASSERT_OK_AND_ASSIGN(bool exists, catalog.DatabaseExists("sys"));
+    ASSERT_TRUE(exists);
+    ASSERT_OK_AND_ASSIGN(bool system_table, catalog.IsSystemTable(Identifier("sys", "tables")));
+    ASSERT_TRUE(system_table);
+    ASSERT_OK_AND_ASSIGN(exists, catalog.TableExists(Identifier("sys", "tables")));
+    ASSERT_TRUE(exists);
+    ASSERT_OK_AND_ASSIGN(exists, catalog.TableExists(Identifier("sys", "missing")));
+    ASSERT_FALSE(exists);
+    ASSERT_OK_AND_ASSIGN(exists, catalog.TableExists(Identifier("sys", "catalog_options")));
+    ASSERT_FALSE(exists);
+
+    options[CatalogOptionsSystemTable::kEnabledOption] = "true";
+    FileSystemCatalog enabled_catalog(core_options.GetFileSystem(), dir->Str(), options);
+    ASSERT_OK_AND_ASSIGN(exists, enabled_catalog.TableExists(Identifier("sys", "catalog_options")));
+    ASSERT_TRUE(exists);
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Schema> schema,
+                         enabled_catalog.LoadTableSchema(Identifier("sys", "catalog_options")));
+    ASSERT_NE(nullptr, std::dynamic_pointer_cast<SystemTableSchema>(schema));
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<ArrowSchema> c_schema, schema->GetArrowSchema());
+    auto arrow_schema = arrow::ImportSchema(c_schema.get());
+    ASSERT_TRUE(arrow_schema.ok()) << arrow_schema.status().ToString();
+    ASSERT_TRUE(arrow_schema.ValueOrDie()->Equals(*arrow::schema(
+        {arrow::field("key", arrow::utf8(), false), arrow::field("value", arrow::utf8(), false)})));
+}
+
+TEST(FileSystemCatalogTest, TestTableDiscoveryWithoutSchemaZero) {
+    std::map<std::string, std::string> options = {{Options::FILE_SYSTEM, "local"}};
+    ASSERT_OK_AND_ASSIGN(CoreOptions core_options, CoreOptions::FromMap(options));
+    auto dir = UniqueTestDirectory::Create();
+    ASSERT_TRUE(dir);
+    std::shared_ptr<FileSystem> fs = core_options.GetFileSystem();
+    FileSystemCatalog catalog(fs, dir->Str(), options);
+    ASSERT_OK(catalog.CreateDatabase("db1", {}, false));
+    Identifier identifier("db1", "evolved");
+    ASSERT_OK_AND_ASSIGN(std::string table_path, catalog.GetTableLocation(identifier));
+    SchemaManager schema_manager(fs, table_path);
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableSchema> schema,
+                         TableSchema::Create(3, arrow::schema({arrow::field("id", arrow::int32())}),
+                                             {}, {}, options));
+    ASSERT_OK_AND_ASSIGN(std::string json, schema->ToJsonString());
+    ASSERT_OK(fs->AtomicStore(schema_manager.ToSchemaPath(3), json));
+    // An incomplete table directory must not appear in listings alongside the evolved table.
+    std::string incomplete_path = PathUtil::JoinPath(dir->Str(), "db1.db/incomplete");
+    ASSERT_OK(fs->Mkdirs(incomplete_path));
+    ASSERT_OK_AND_ASSIGN(std::vector<std::string> tables, catalog.ListTables("db1"));
+    ASSERT_EQ((std::vector<std::string>{"evolved"}), tables);
+    ASSERT_OK_AND_ASSIGN(bool exists, catalog.TableExists(identifier));
+    ASSERT_TRUE(exists);
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Schema> loaded, catalog.LoadTableSchema(identifier));
+    std::shared_ptr<DataSchema> data_schema = std::dynamic_pointer_cast<DataSchema>(loaded);
+    ASSERT_NE(nullptr, data_schema);
+    ASSERT_EQ(3, data_schema->Id());
+    ASSERT_OK(catalog.DropTable(Identifier("db1", "incomplete"), true));
+    ASSERT_OK_AND_ASSIGN(exists, fs->Exists(incomplete_path));
+    ASSERT_TRUE(exists);
+}
+
 TEST(FileSystemCatalogTest, TestInvalidCreateDatabase) {
     {
         std::map<std::string, std::string> options;
diff --git a/src/paimon/core/operation/commit/conflict_detection_test.cpp b/src/paimon/core/operation/commit/conflict_detection_test.cpp
index 3fb751acf..c0122e776 100644
--- a/src/paimon/core/operation/commit/conflict_detection_test.cpp
+++ b/src/paimon/core/operation/commit/conflict_detection_test.cpp
@@ -238,6 +238,44 @@ class ConflictDetectionTest : public testing::Test {
     arrow::FieldVector fields_;
 };
 
+TEST_F(ConflictDetectionTest, TestEmptyAndUnrelatedEntriesSkipRowIdChecks) {
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> schema,
+                         TableSchema::Create(0, arrow::schema(fields_), {"f1"}, {}, {}));
+    ASSERT_OK_AND_ASSIGN(CoreOptions options,
+                         CoreOptions::FromMap({{Options::DATA_EVOLUTION_ENABLED, "true"}}));
+    ConflictDetection detection(schema, options, nullptr, nullptr, nullptr, nullptr, "test-user",
+                                "test-table", nullptr);
+    const BinaryRow partition = CreateIntRow(10);
+    const ManifestEntry untracked = CreateManifestEntry("untracked", FileKind::Add());
+    const ManifestEntry deleted =
+        CreateManifestEntryWithFirstRowId("deleted", partition, FileKind::Delete(), 0, 0, 10);
+    const ManifestEntry new_rows =
+        CreateManifestEntryWithFirstRowId("new-rows", partition, FileKind::Add(), 0, 100, 10);
+    ASSERT_OK(CheckConflicts(detection, {}, {}, Snapshot::CommitKind::Append()));
+    ASSERT_OK(detection.CheckRowIdExistence({}, {untracked, deleted, new_rows}, 100));
+    ASSERT_OK(detection.CheckRowIdExistence({}, {new_rows}, std::nullopt));
+    const auto deleted_index =
+        CreateGlobalIndexEntry("deleted-index", partition, 0, FileKind::Delete(), 0, 9);
+    const auto dv_index = CreateDvIndexEntry("deletion-vector", partition, 0);
+    ASSERT_OK(detection.CheckGlobalIndexRowIdExistence({}, {deleted_index, dv_index}));
+}
+
+TEST_F(ConflictDetectionTest, TestRowIdChecksAreDisabledWithoutDataEvolution) {
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> schema,
+                         TableSchema::Create(0, arrow::schema(fields_), {"f1"}, {}, {}));
+    ASSERT_OK_AND_ASSIGN(CoreOptions options, CoreOptions::FromMap({}));
+    ConflictDetection detection(schema, options, nullptr, nullptr, nullptr, nullptr, "test-user",
+                                "test-table", nullptr);
+    const BinaryRow partition = CreateIntRow(10);
+    const ManifestEntry missing =
+        CreateManifestEntryWithFirstRowId("missing", partition, FileKind::Add(), 0, 0, 10);
+    ASSERT_OK(detection.CheckRowIdExistence({}, {missing}, 100));
+    ASSERT_OK(detection.CheckGlobalIndexRowIdExistence(
+        {}, {CreateGlobalIndexEntry("missing-index", partition, 0, 0, 9)}));
+    ASSERT_OK(detection.CheckForRowIdFromSnapshot(MakeSnapshot(Snapshot::CommitKind::Append()),
+                                                  {missing}, {}, std::nullopt));
+}
+
 TEST_F(ConflictDetectionTest, TestFileDeletionConflicts) {
     ASSERT_OK_AND_ASSIGN(
         std::shared_ptr<TableSchema> table_schema,
diff --git a/src/paimon/core/operation/expire_snapshots_test.cpp b/src/paimon/core/operation/expire_snapshots_test.cpp
index f64c0ff03..43008f3f8 100644
--- a/src/paimon/core/operation/expire_snapshots_test.cpp
+++ b/src/paimon/core/operation/expire_snapshots_test.cpp
@@ -19,6 +19,7 @@
 #include "paimon/core/operation/expire_snapshots.h"
 
 #include <cstdint>
+#include <limits>
 #include <optional>
 #include <set>
 #include <utility>
@@ -75,6 +76,25 @@ class DeleteFailingFileSystem : public LocalFileSystem {
     std::string failed_path_;
 };
 
+// Simulates a concurrent expiration or tag deletion at a specific existence check.
+class DisappearingFileSystem : public LocalFileSystem {
+ public:
+    DisappearingFileSystem(std::string path, int32_t remove_on_check)
+        : path_(std::move(path)), remove_on_check_(remove_on_check) {}
+
+    Result<bool> Exists(const std::string& path) const override {
+        if (path == path_ && ++checks_ == remove_on_check_) {
+            PAIMON_RETURN_NOT_OK(LocalFileSystem::Delete(path));
+        }
+        return LocalFileSystem::Exists(path);
+    }
+
+ private:
+    std::string path_;
+    int32_t remove_on_check_;
+    mutable int32_t checks_ = 0;
+};
+
 }  // namespace
 
 class ExpireSnapshotsTest : public testing::Test {
@@ -225,6 +245,33 @@ class ExpireSnapshotsTest : public testing::Test {
         return index_path_factory->ToPath(entry.index_file);
     }
 
+    Result<Snapshot> WriteSnapshot(int64_t id, const std::vector<ManifestEntry>& base = {},
+                                   const std::vector<ManifestEntry>& delta = {},
+                                   int64_t time_millis = 0,
+                                   const std::optional<std::string>& statistics = std::nullopt) {
+        PAIMON_ASSIGN_OR_RAISE(std::vector<ManifestFileMeta> base_files,
+                               manifest_file_->Write(base));
+        PAIMON_ASSIGN_OR_RAISE(std::vector<ManifestFileMeta> delta_files,
+                               manifest_file_->Write(delta));
+        using ListMeta = std::pair<std::string, int64_t>;
+        PAIMON_ASSIGN_OR_RAISE(ListMeta base_list, manifest_list_->Write(base_files));
+        PAIMON_ASSIGN_OR_RAISE(ListMeta delta_list, manifest_list_->Write(delta_files));
+        Snapshot snapshot(id, 0, base_list.first, base_list.second, delta_list.first,
+                          delta_list.second, std::nullopt, std::nullopt, std::nullopt, "test", id,
+                          Snapshot::CommitKind::Append(), time_millis, 0, 0, std::nullopt,
+                          std::nullopt, statistics, std::nullopt, std::nullopt);
+        SnapshotManager manager(fs_, test_data_path_);
+        PAIMON_ASSIGN_OR_RAISE(std::string json, snapshot.ToJsonString());
+        PAIMON_RETURN_NOT_OK(fs_->AtomicStore(manager.SnapshotPath(id), json));
+        return snapshot;
+    }
+
+    Status WriteTag(const Snapshot& snapshot, const std::string& name) {
+        TagManager manager(fs_, test_data_path_);
+        PAIMON_ASSIGN_OR_RAISE(std::string json, snapshot.ToJsonString());
+        return fs_->AtomicStore(manager.TagPath(name), json);
+    }
+
  private:
     std::unique_ptr<UniqueTestDirectory> dir_;
     std::string test_data_path_;
@@ -355,6 +402,155 @@ TEST_F(ExpireSnapshotsTest, TestExpireSourceBackedBTreeIndexFiles) {
                                                     &missing_skipping_set));
 }
 
+TEST_F(ExpireSnapshotsTest, TestTimeRetentionStopsAtRecentSnapshot) {
+    auto manager = std::make_shared<SnapshotManager>(fs_, test_data_path_);
+    ASSERT_OK_AND_ASSIGN(CoreOptions options,
+                         CoreOptions::FromMap({{Options::SNAPSHOT_NUM_RETAINED_MIN, "1"},
+                                               {Options::SNAPSHOT_TIME_RETAINED, "1h"}}));
+    ExpireSnapshots expire(manager, path_factory_, manifest_list_, manifest_file_,
+                           index_manifest_file_, fs_, options.GetExpireConfig(), false, executor_);
+    ASSERT_OK(WriteSnapshot(1).status());
+    ASSERT_OK(WriteSnapshot(2, {}, {}, std::numeric_limits<int64_t>::max() / 2).status());
+    ASSERT_OK(WriteSnapshot(3, {}, {}, std::numeric_limits<int64_t>::max() / 2).status());
+    ASSERT_OK_AND_ASSIGN(int32_t count, expire.Expire());
+    ASSERT_EQ(1, count);
+    ASSERT_OK_AND_ASSIGN(bool exists, manager->SnapshotExists(1));
+    ASSERT_FALSE(exists);
+    ASSERT_OK_AND_ASSIGN(exists, manager->SnapshotExists(2));
+    ASSERT_TRUE(exists);
+    ASSERT_OK_AND_ASSIGN(exists, manager->SnapshotExists(3));
+    ASSERT_TRUE(exists);
+    ASSERT_OK_AND_ASSIGN(std::optional<int64_t> earliest, manager->EarliestSnapshotId());
+    ASSERT_EQ(std::optional<int64_t>(2), earliest);
+}
+
+TEST_F(ExpireSnapshotsTest, TestSnapshotsDisappearBeforeEarliestLookup) {
+    SnapshotManager paths(fs_, test_data_path_);
+    ASSERT_OK(WriteSnapshot(1).status());
+    auto disappearing_fs = std::make_shared<DisappearingFileSystem>(paths.SnapshotDirectory(), 1);
+    auto manager = std::make_shared<SnapshotManager>(disappearing_fs, test_data_path_);
+    ASSERT_OK_AND_ASSIGN(CoreOptions options, CoreOptions::FromMap({}));
+    ExpireSnapshots expire(manager, path_factory_, manifest_list_, manifest_file_,
+                           index_manifest_file_, disappearing_fs, options.GetExpireConfig(), false,
+                           executor_);
+    ASSERT_OK_AND_ASSIGN(int32_t count, expire.Expire());
+    ASSERT_EQ(0, count);
+    ASSERT_OK_AND_ASSIGN(bool exists, fs_->Exists(paths.SnapshotDirectory()));
+    ASSERT_FALSE(exists);
+}
+
+TEST_F(ExpireSnapshotsTest, TestConcurrentExpirationSkipsDisappearingSnapshot) {
+    SnapshotManager paths(fs_, test_data_path_);
+    for (int64_t id : {1, 2, 3}) {
+        ASSERT_OK(WriteSnapshot(id).status());
+    }
+    auto disappearing_fs = std::make_shared<DisappearingFileSystem>(paths.SnapshotPath(2), 2);
+    auto manager = std::make_shared<SnapshotManager>(disappearing_fs, test_data_path_);
+    ASSERT_OK_AND_ASSIGN(CoreOptions options, CoreOptions::FromMap({}));
+    ExpireSnapshots expire(manager, path_factory_, manifest_list_, manifest_file_,
+                           index_manifest_file_, disappearing_fs, options.GetExpireConfig(), false,
+                           executor_);
+    ASSERT_OK_AND_ASSIGN(int32_t count, expire.ExpireUntil(1, 3, 3));
+    ASSERT_EQ(0, count);
+    ASSERT_OK_AND_ASSIGN(bool exists, fs_->Exists(paths.SnapshotPath(1)));
+    ASSERT_TRUE(exists);
+    ASSERT_OK_AND_ASSIGN(exists, fs_->Exists(paths.SnapshotPath(2)));
+    ASSERT_FALSE(exists);
+    ASSERT_OK_AND_ASSIGN(exists, fs_->Exists(paths.SnapshotPath(3)));
+    ASSERT_TRUE(exists);
+    ASSERT_OK_AND_ASSIGN(std::optional<int64_t> earliest, manager->EarliestSnapshotId());
+    ASSERT_EQ(std::optional<int64_t>(3), earliest);
+}
+
+TEST_F(ExpireSnapshotsTest, TestIndexManifestDisappearsBeforeRead) {
+    auto manager = std::make_shared<SnapshotManager>(fs_, test_data_path_);
+    ASSERT_OK_AND_ASSIGN(CoreOptions options, CoreOptions::FromMap({}));
+    ASSERT_OK_AND_ASSIGN(IndexManifestEntry entry, CreateSourceBackedBTreeEntry("kept.index", 0));
+    ASSERT_OK_AND_ASSIGN(std::string payload_path, IndexFilePath(entry));
+    ASSERT_OK(fs_->WriteFile(payload_path, "index", false));
+    ASSERT_OK_AND_ASSIGN(std::optional<std::string> manifest,
+                         index_manifest_file_->WriteIndexFiles(std::nullopt, {entry}));
+    ASSERT_TRUE(manifest.has_value());
+    std::string manifest_path = path_factory_->ToManifestFilePath(manifest.value());
+    auto disappearing_fs = std::make_shared<DisappearingFileSystem>(manifest_path, 2);
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<IndexManifestFile> index_manifest,
+                         IndexManifestFile::Create(disappearing_fs, options.GetManifestFormat(),
+                                                   options.GetManifestCompression(), path_factory_,
+                                                   BucketMode::BUCKET_UNAWARE, mem_pool_, options));
+    ExpireSnapshots expire(manager, path_factory_, manifest_list_, manifest_file_, index_manifest,
+                           disappearing_fs, options.GetExpireConfig(), false, executor_);
+    std::set<std::string> retained = {"retained.index"};
+    ASSERT_OK(expire.CleanUnusedIndexManifest(manifest, &retained));
+    ASSERT_EQ((std::set<std::string>{"retained.index"}), retained);
+    ASSERT_OK_AND_ASSIGN(bool exists, fs_->Exists(manifest_path));
+    ASSERT_FALSE(exists);
+    ASSERT_OK_AND_ASSIGN(exists, fs_->Exists(payload_path));
+    ASSERT_TRUE(exists);
+}
+
+TEST_F(ExpireSnapshotsTest, TestTagListingSkipsDeletedTagsAndSortsBySnapshotId) {
+    ASSERT_OK_AND_ASSIGN(Snapshot first, WriteSnapshot(1));
+    ASSERT_OK_AND_ASSIGN(Snapshot second, WriteSnapshot(2));
+    ASSERT_OK_AND_ASSIGN(Snapshot third, WriteSnapshot(3));
+    ASSERT_OK(WriteTag(third, "a-third"));
+    ASSERT_OK(WriteTag(first, "z-first"));
+    ASSERT_OK(WriteTag(second, "deleted"));
+    TagManager tags(fs_, test_data_path_);
+    auto disappearing_fs = std::make_shared<DisappearingFileSystem>(tags.TagPath("deleted"), 1);
+    auto manager = std::make_shared<SnapshotManager>(disappearing_fs, test_data_path_);
+    ASSERT_OK_AND_ASSIGN(CoreOptions options, CoreOptions::FromMap({}));
+    ExpireSnapshots expire(manager, path_factory_, manifest_list_, manifest_file_,
+                           index_manifest_file_, disappearing_fs, options.GetExpireConfig(), false,
+                           executor_);
+    ASSERT_OK_AND_ASSIGN(std::vector<Snapshot> snapshots, expire.GetTaggedSnapshots());
+    ASSERT_EQ(2, snapshots.size());
+    ASSERT_EQ(1, snapshots[0].Id());
+    ASSERT_EQ(3, snapshots[1].Id());
+}
+
+TEST_F(ExpireSnapshotsTest, TestUnreadableTaggedManifestKeepsDataWhileExpiringMetadata) {
+    ManifestEntry added = CreateManifestEntry("protected.data", 0, FileKind::Add());
+    ManifestEntry deleted = CreateManifestEntry("protected.data", 0, FileKind::Delete());
+    ASSERT_OK_AND_ASSIGN(Snapshot tagged, WriteSnapshot(1, {added}));
+    ASSERT_OK(WriteTag(tagged, "protected"));
+    ASSERT_OK(WriteSnapshot(2, {}, {deleted}).status());
+    ASSERT_OK_AND_ASSIGN(std::string bucket, path_factory_->BucketPath(added.Partition(), 0));
+    std::string data_path = PathUtil::JoinPath(bucket, added.FileName());
+    ASSERT_OK(fs_->Mkdirs(bucket));
+    ASSERT_OK(fs_->WriteFile(data_path, "protected", false));
+    std::vector<ManifestFileMeta> manifests;
+    ASSERT_OK(manifest_list_->Read(tagged.BaseManifestList(), nullptr, std::nullopt, &manifests));
+    ASSERT_EQ(1, manifests.size());
+    manifest_file_->DeleteQuietly(manifests[0].FileName());
+    auto manager = std::make_shared<SnapshotManager>(fs_, test_data_path_);
+    ASSERT_OK_AND_ASSIGN(CoreOptions options, CoreOptions::FromMap({}));
+    ExpireSnapshots expire(manager, path_factory_, manifest_list_, manifest_file_,
+                           index_manifest_file_, fs_, options.GetExpireConfig(), false, executor_);
+    ASSERT_OK_AND_ASSIGN(int32_t count, expire.ExpireUntil(1, 2, 2));
+    ASSERT_EQ(1, count);
+    ASSERT_OK_AND_ASSIGN(bool exists, manager->SnapshotExists(1));
+    ASSERT_FALSE(exists);
+    ASSERT_OK_AND_ASSIGN(exists, manager->SnapshotExists(2));
+    ASSERT_TRUE(exists);
+    ASSERT_OK_AND_ASSIGN(exists, fs_->Exists(data_path));
+    ASSERT_TRUE(exists);
+    ASSERT_OK_AND_ASSIGN(std::optional<Tag> tag, TagManager(fs_, test_data_path_).Get("protected"));
+    ASSERT_TRUE(tag.has_value());
+}
+
+TEST_F(ExpireSnapshotsTest, TestRetainedStatisticsAreIncludedInManifestSkippingSet) {
+    ASSERT_OK_AND_ASSIGN(Snapshot snapshot, WriteSnapshot(1, {}, {}, 0, "statistics-retained"));
+    auto manager = std::make_shared<SnapshotManager>(fs_, test_data_path_);
+    ASSERT_OK_AND_ASSIGN(CoreOptions options, CoreOptions::FromMap({}));
+    ExpireSnapshots expire(manager, path_factory_, manifest_list_, manifest_file_,
+                           index_manifest_file_, fs_, options.GetExpireConfig(), false, executor_);
+    std::set<std::string> retained;
+    ASSERT_OK(expire.GetManifestSkippingSet({snapshot}, &retained));
+    ASSERT_EQ(1, retained.count("statistics-retained"));
+    ASSERT_EQ(1, retained.count(snapshot.BaseManifestList()));
+    ASSERT_EQ(1, retained.count(snapshot.DeltaManifestList()));
+}
+
 TEST_F(ExpireSnapshotsTest, TestExternalIndexDeleteFailureKeepsManifestForRetry) {
     auto mgr = std::make_shared<SnapshotManager>(fs_, test_data_path_);
     ASSERT_OK_AND_ASSIGN(CoreOptions options, CoreOptions::FromMap({}));
@@ -575,4 +771,65 @@ TEST_F(ExpireSnapshotsTest, TestCleanUnusedDataFilePreservesRetainedExtraFiles)
     }
 }
 
+TEST_F(ExpireSnapshotsTest, TestCleansEmptyPartitionHierarchyAndKeepsNonemptyPartition) {
+    auto manager = std::make_shared<SnapshotManager>(fs_, test_data_path_);
+    ASSERT_OK_AND_ASSIGN(
+        CoreOptions options,
+        CoreOptions::FromMap({{Options::SNAPSHOT_CLEAN_EMPTY_DIRECTORIES, "true"}}));
+    ExpireSnapshots expire(manager, path_factory_, manifest_list_, manifest_file_,
+                           index_manifest_file_, fs_, options.GetExpireConfig(), false, executor_);
+    std::vector<ManifestEntry> deletions;
+    std::vector<std::string> partition_paths;
+    for (int16_t value : {3, 4, 5}) {
+        BinaryRow partition(2);
+        BinaryRowWriter writer(&partition, 20, mem_pool_.get());
+        writer.WriteBoolean(0, value != 5);
+        writer.WriteShort(1, value);
+        writer.Complete();
+        ManifestEntry entry = CreateManifestEntry("expired.data", 0, FileKind::Delete());
+        deletions.emplace_back(FileKind::Delete(), partition, 0, 3, entry.File());
+        ASSERT_OK_AND_ASSIGN(std::string bucket, path_factory_->BucketPath(partition, 0));
+        ASSERT_OK(fs_->Mkdirs(bucket));
+        ASSERT_OK(fs_->WriteFile(PathUtil::JoinPath(bucket, "expired.data"), "data", false));
+        if (value == 5) {
+            ASSERT_OK(fs_->WriteFile(PathUtil::JoinPath(bucket, "live.data"), "live", false));
+        }
+        ASSERT_OK_AND_ASSIGN(std::vector<std::string> hierarchy,
+                             path_factory_->GetHierarchicalPartitionPath(partition));
+        ASSERT_EQ(2, hierarchy.size());
+        partition_paths.push_back(hierarchy[0]);
+    }
+    ASSERT_OK_AND_ASSIGN(std::vector<ManifestFileMeta> manifests, manifest_file_->Write(deletions));
+    using ListMeta = std::pair<std::string, int64_t>;
+    ASSERT_OK_AND_ASSIGN(ListMeta list, manifest_list_->Write(manifests));
+    ExpireSnapshots::DataFilePathFactoryCache cache;
+    ASSERT_OK(expire.CleanUnusedDataFiles(list.first, {}, &cache));
+    ASSERT_OK(expire.CleanEmptyDirectories());
+    for (size_t i = 0; i < partition_paths.size(); ++i) {
+        ASSERT_OK_AND_ASSIGN(bool exists, fs_->Exists(partition_paths[i]));
+        ASSERT_EQ(i == 2, exists) << partition_paths[i];
+    }
+    ASSERT_OK_AND_ASSIGN(bool exists, fs_->Exists(test_data_path_));
+    ASSERT_TRUE(exists);
+}
+
+TEST_F(ExpireSnapshotsTest, TestEmptyUnpartitionedBucketDoesNotDeleteTableDirectory) {
+    partition_keys_.clear();
+    path_factory_ = CreateFactory(test_data_path_);
+    auto manager = std::make_shared<SnapshotManager>(fs_, test_data_path_);
+    ASSERT_OK_AND_ASSIGN(
+        CoreOptions options,
+        CoreOptions::FromMap({{Options::SNAPSHOT_CLEAN_EMPTY_DIRECTORIES, "true"}}));
+    ExpireSnapshots expire(manager, path_factory_, manifest_list_, manifest_file_,
+                           index_manifest_file_, fs_, options.GetExpireConfig(), false, executor_);
+    ASSERT_OK_AND_ASSIGN(std::string bucket, path_factory_->BucketPath(BinaryRow::EmptyRow(), 0));
+    ASSERT_OK(fs_->Mkdirs(bucket));
+    expire.deletion_buckets_[BinaryRow::EmptyRow()].insert(0);
+    ASSERT_OK(expire.CleanEmptyDirectories());
+    ASSERT_OK_AND_ASSIGN(bool exists, fs_->Exists(bucket));
+    ASSERT_FALSE(exists);
+    ASSERT_OK_AND_ASSIGN(exists, fs_->Exists(test_data_path_));
+    ASSERT_TRUE(exists);
+}
+
 }  // namespace paimon::test
diff --git a/src/paimon/core/operation/file_store_commit_impl_test.cpp b/src/paimon/core/operation/file_store_commit_impl_test.cpp
index 96c6de34d..be1482052 100644
--- a/src/paimon/core/operation/file_store_commit_impl_test.cpp
+++ b/src/paimon/core/operation/file_store_commit_impl_test.cpp
@@ -67,6 +67,7 @@
 #include "paimon/core/manifest/manifest_file_meta.h"
 #include "paimon/core/manifest/manifest_list.h"
 #include "paimon/core/operation/commit/realtime_commit_properties.h"
+#include "paimon/core/operation/commit/row_id_column_conflict_checker.h"
 #include "paimon/core/operation/metrics/commit_metrics.h"
 #include "paimon/core/partition/partition_statistics.h"
 #include "paimon/core/schema/schema_manager.h"
@@ -1137,6 +1138,39 @@ TEST_F(FileStoreCommitImplTest, TestCatalogSchemaIsRereadOnEveryCommit) {
     }
 }
 
+TEST_F(FileStoreCommitImplTest, TestStatisticsAreInheritedOnlyForTheSameSchema) {
+    for (int64_t next_schema_id : {7, 8}) {
+        SCOPED_TRACE(next_schema_id);
+        auto catalog = CreateCatalogServingSchemaId(7);
+        catalog->CheckBaseSnapshotUuid();
+        ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileStoreCommit> commit, CreateCatalogCommit(catalog));
+        ASSERT_OK(commit->Commit(CommitMessagesOfRound(1), 1));
+        ASSERT_EQ(catalog->AcceptedSnapshots().size(), 1u);
+        const Snapshot& first = catalog->AcceptedSnapshots().front();
+        Snapshot with_statistics(
+            first.Version(), first.Id(), first.SchemaId(), first.BaseManifestList(),
+            first.BaseManifestListSize(), first.DeltaManifestList(), first.DeltaManifestListSize(),
+            first.ChangelogManifestList(), first.ChangelogManifestListSize(), first.IndexManifest(),
+            first.CommitUser(), first.CommitIdentifier(), first.GetCommitKind(), first.TimeMillis(),
+            first.TotalRecordCount(), first.DeltaRecordCount(), first.ChangelogRecordCount(),
+            first.Watermark(), std::string("statistics-1"), first.Properties(), first.NextRowId(),
+            first.Uuid());
+        catalog->SetHeldSnapshot(with_statistics);
+        catalog->SetTableSchema(TableSchemaWithId(next_schema_id));
+
+        ASSERT_OK(commit->Commit(CommitMessagesOfRound(2), 2));
+        ASSERT_EQ(catalog->CommitCalls().size(), 2u);
+        const Snapshot& second = catalog->AcceptedSnapshots().back();
+        ASSERT_EQ(second.Id(), 2);
+        ASSERT_EQ(second.SchemaId(), next_schema_id);
+        if (next_schema_id == with_statistics.SchemaId()) {
+            ASSERT_EQ(second.Statistics(), std::optional<std::string>("statistics-1"));
+        } else {
+            ASSERT_FALSE(second.Statistics().has_value());
+        }
+    }
+}
+
 TEST_F(FileStoreCommitImplTest, TestCatalogSchemaWhichIsNotADataTableSchemaIsRefused) {
     std::shared_ptr<MockVersionManagedCatalog> catalog = CreateCatalogServingSchemaId(7);
     ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileStoreCommit> commit, CreateCatalogCommit(catalog));
@@ -1705,6 +1739,44 @@ TEST_F(FileStoreCommitImplTest, TestRealtimeCommitIsIdempotentThroughTheCatalog)
     ASSERT_EQ(catalog->AcceptedSnapshots().size(), 1u);
 }
 
+TEST_F(FileStoreCommitImplTest, TestCommitRestoresAlreadyCommittedRealtimeSnapshot) {
+    ASSERT_OK(PrepareFirstCommitMessageFiles());
+    auto catalog = CreateCatalog();
+    catalog->CheckBaseSnapshotUuid();
+    CatalogCommitSpec spec;
+    spec.catalog = catalog;
+    spec.commit_user = "realtime_commit_user";
+    spec.options = {{Options::REALTIME_ENABLED, "true"}, {Options::BUCKET, "2"}};
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileStoreCommit> commit, CreateCatalogCommit(spec));
+    auto* commit_impl = dynamic_cast<FileStoreCommitImpl*>(commit.get());
+    ASSERT_TRUE(commit_impl);
+    const auto messages = CommitMessagesOfRound(1);
+    std::vector<RealtimeCommitProgress> progress;
+    ASSERT_NO_FATAL_FAILURE(BuildRealtimeProgress(commit_impl, messages, &progress));
+    ASSERT_OK_AND_ASSIGN(int64_t snapshot_id,
+                         commit_impl->CommitWithProgress(progress, 1, std::nullopt));
+    ASSERT_EQ(snapshot_id, 1);
+
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileStoreCommit> restored, CreateCatalogCommit(spec));
+    auto* restored_impl = dynamic_cast<FileStoreCommitImpl*>(restored.get());
+    ASSERT_TRUE(restored_impl);
+    std::map<RealtimePartitionBucket, OffsetRange> ranges;
+    for (const auto& entry : progress) {
+        ranges.emplace(entry.partition_bucket, entry.offset_range);
+    }
+    auto committable = restored_impl->CreateManifestCommittable(1, messages, std::nullopt, {});
+    ASSERT_OK(restored_impl->Commit(committable, /*check_append_files=*/false,
+                                    /*retry_on_conflict=*/true, ranges));
+    ASSERT_EQ(catalog->AcceptedSnapshots().size(), 1u);
+    ASSERT_EQ(catalog->CommitCalls().size(), 1u);
+    ASSERT_OK_AND_ASSIGN(uint64_t generated, restored_impl->GetCommitMetrics()->GetCounter(
+                                                 CommitMetrics::LAST_GENERATED_SNAPSHOTS));
+    ASSERT_EQ(generated, 0);
+    ASSERT_OK_AND_ASSIGN(uint64_t restored_id, restored_impl->GetCommitMetrics()->GetCounter(
+                                                   CommitMetrics::LAST_COMMITTED_SNAPSHOT_ID));
+    ASSERT_EQ(restored_id, snapshot_id);
+}
+
 TEST_F(FileStoreCommitImplTest, TestRealtimeCommitReportsAnUnpublishedHistory) {
     ASSERT_OK(PrepareFirstCommitMessageFiles());
     auto catalog = CreateCatalog();
@@ -1828,6 +1900,80 @@ TEST_F(FileStoreCommitImplTest, TestRowIdChecksUseCatalogLatestAndPublishedHisto
     ASSERT_EQ(catalog->AcceptedSnapshots().size(), 2u);
 }
 
+TEST_F(FileStoreCommitImplTest, TestRowIdConflictChecksSnapshotBoundsAndHistory) {
+    CatalogCommitSpec spec;
+    spec.catalog = CreateCatalog();
+    spec.options = {{Options::DATA_EVOLUTION_ENABLED, "true"},
+                    {Options::ROW_TRACKING_ENABLED, "true"}};
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileStoreCommit> commit, CreateCatalogCommit(spec));
+    auto* impl = dynamic_cast<FileStoreCommitImpl*>(commit.get());
+    ASSERT_TRUE(impl);
+    auto& detection = impl->conflict_detection_;
+    const BinaryRow partition = CreateIntRow(10);
+    auto make_entry = [&](const std::string& name, const FileKind& kind,
+                          std::optional<int64_t> row_id, const std::string& column) {
+        auto file = CreateAppendDataFileMeta(name, 10);
+        file->schema_id = 0;
+        file->first_row_id = row_id;
+        file->write_cols = std::vector<std::string>{column};
+        return ManifestEntry(kind, partition, /*bucket=*/0, /*total_buckets=*/2, file);
+    };
+    const ManifestEntry update = make_entry("update", FileKind::Add(), 0, "f2");
+    ASSERT_OK_AND_ASSIGN(auto checker, RowIdColumnConflictChecker::FromDataFiles(
+                                           impl->schema_manager_, {update.File()}));
+    auto make_snapshot = [&](int64_t id, const Snapshot::CommitKind& kind,
+                             const std::vector<ManifestEntry>& entries) -> Result<Snapshot> {
+        PAIMON_ASSIGN_OR_RAISE(std::vector<ManifestFileMeta> manifests,
+                               impl->manifest_file_->Write(entries));
+        std::pair<std::string, int64_t> delta_list;
+        PAIMON_ASSIGN_OR_RAISE(delta_list, impl->manifest_list_->Write(manifests));
+        return Snapshot(id, /*schema_id=*/0, /*base_manifest_list=*/"", std::nullopt,
+                        delta_list.first, delta_list.second, std::nullopt, std::nullopt,
+                        std::nullopt, "history-writer", id, kind, /*time_millis=*/0,
+                        /*total_record_count=*/100,
+                        /*delta_record_count=*/0, std::nullopt, std::nullopt, std::nullopt,
+                        std::nullopt, /*next_row_id=*/100);
+    };
+    std::vector<ManifestEntry> history = {
+        make_entry("deleted", FileKind::Delete(), 0, "f2"),
+        make_entry("untracked", FileKind::Add(), std::nullopt, "f2"),
+        make_entry("new-rows", FileKind::Add(), 100, "f2"),
+        make_entry("other-column", FileKind::Add(), 0, "f3")};
+    for (int64_t id : {-1, 1, 2}) {
+        ASSERT_OK_AND_ASSIGN(
+            Snapshot snapshot,
+            make_snapshot(
+                id, id == 2 ? Snapshot::CommitKind::Compact() : Snapshot::CommitKind::Append(),
+                id == 1 ? history : std::vector<ManifestEntry>{}));
+        ASSERT_OK_AND_ASSIGN(std::string json, snapshot.ToJsonString());
+        ASSERT_OK(file_system_->AtomicStore(impl->snapshot_manager_->SnapshotPath(id), json));
+    }
+    for (bool conflicts : {false, true}) {
+        SCOPED_TRACE(conflicts);
+        ASSERT_OK_AND_ASSIGN(
+            Snapshot latest,
+            make_snapshot(3, Snapshot::CommitKind::Append(),
+                          {make_entry("latest", FileKind::Add(), 0, conflicts ? "f2" : "f3")}));
+        for (int64_t from_snapshot : {-1, 1}) {
+            SCOPED_TRACE(from_snapshot);
+            detection.SetRowIdCheckFromSnapshot(from_snapshot);
+            Status status = detection.CheckForRowIdFromSnapshot(latest, {update}, {}, checker);
+            if (conflicts) {
+                ASSERT_NOK_WITH_MSG(status, "updating the same file");
+            } else {
+                ASSERT_OK(status);
+            }
+        }
+        for (int64_t from_snapshot : {3, 4}) {
+            detection.SetRowIdCheckFromSnapshot(from_snapshot);
+            ASSERT_OK(detection.CheckForRowIdFromSnapshot(latest, {update}, {}, checker));
+        }
+        detection.SetRowIdCheckFromSnapshot(1);
+        ASSERT_OK(detection.CheckForRowIdFromSnapshot(latest, {}, {}, checker));
+        ASSERT_OK(detection.CheckForRowIdFromSnapshot(latest, {update}, {}, std::nullopt));
+    }
+}
+
 TEST_F(FileStoreCommitImplTest, TestCatalogSnapshotLoaderFailurePropagates) {
     auto catalog = CreateCatalog();
     catalog->SetLoadSnapshotStatus(
diff --git a/src/paimon/core/operation/orphan_files_cleaner_test.cpp b/src/paimon/core/operation/orphan_files_cleaner_test.cpp
index 6043e0d79..1471d794b 100644
--- a/src/paimon/core/operation/orphan_files_cleaner_test.cpp
+++ b/src/paimon/core/operation/orphan_files_cleaner_test.cpp
@@ -51,6 +51,7 @@
 #include "paimon/core/snapshot.h"
 #include "paimon/core/utils/file_store_path_factory.h"
 #include "paimon/defs.h"
+#include "paimon/executor.h"
 #include "paimon/fs/local/local_file_system.h"
 #include "paimon/memory/memory_pool.h"
 #include "paimon/status.h"
@@ -149,9 +150,19 @@ TEST(OrphanFilesCleanerTest, TestPkTable) {
     std::string table_path = dir->Str();
     ASSERT_TRUE(TestUtil::CopyDirectory(test_data_path, table_path));
     CleanContextBuilder clean_context_builder(table_path);
-    ASSERT_OK_AND_ASSIGN(
-        std::unique_ptr<CleanContext> clean_context,
-        clean_context_builder.AddOption(Options::FILE_SYSTEM, "local").WithOlderThanMs(0).Finish());
+    std::shared_ptr<MemoryPool> pool = GetMemoryPool();
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Executor> executor, CreateDefaultExecutor(2));
+    // Replacing options must discard an earlier invalid value, and cleaning must work with
+    // the caller's memory pool and executor.
+    clean_context_builder.AddOption(Options::BUCKET, "invalid-bucket");
+    ASSERT_OK_AND_ASSIGN(std::unique_ptr<CleanContext> clean_context,
+                         clean_context_builder.SetOptions({{Options::FILE_SYSTEM, "local"}})
+                             .WithMemoryPool(pool)
+                             .WithExecutor(executor)
+                             .WithOlderThanMs(0)
+                             .Finish());
+    ASSERT_EQ(pool, clean_context->GetMemoryPool());
+    ASSERT_EQ(executor, clean_context->GetExecutor());
     ASSERT_OK_AND_ASSIGN(auto cleaner, OrphanFilesCleaner::Create(std::move(clean_context)));
     ASSERT_OK_AND_ASSIGN(std::set<std::string> cleaned_paths, cleaner->Clean());
     ASSERT_TRUE(cleaned_paths.empty());
diff --git a/src/paimon/format/parquet/page_filtered_row_group_reader_test.cpp b/src/paimon/format/parquet/page_filtered_row_group_reader_test.cpp
index d17379348..b8af016b3 100644
--- a/src/paimon/format/parquet/page_filtered_row_group_reader_test.cpp
+++ b/src/paimon/format/parquet/page_filtered_row_group_reader_test.cpp
@@ -67,6 +67,38 @@ class Predicate;
 
 namespace paimon::parquet::test {
 
+namespace {
+class SyntheticOffsetIndex : public ::parquet::OffsetIndex {
+ public:
+    explicit SyntheticOffsetIndex(std::vector<::parquet::PageLocation> pages)
+        : pages_(std::move(pages)) {}
+
+    const std::vector<::parquet::PageLocation>& page_locations() const override {
+        return pages_;
+    }
+
+ private:
+    std::vector<::parquet::PageLocation> pages_;
+};
+
+class SyntheticPageIndexReader : public ::parquet::RowGroupPageIndexReader {
+ public:
+    explicit SyntheticPageIndexReader(std::shared_ptr<::parquet::OffsetIndex> index)
+        : index_(std::move(index)) {}
+
+    std::shared_ptr<::parquet::ColumnIndex> GetColumnIndex(int32_t) override {
+        return nullptr;
+    }
+
+    std::shared_ptr<::parquet::OffsetIndex> GetOffsetIndex(int32_t) override {
+        return index_;
+    }
+
+ private:
+    std::shared_ptr<::parquet::OffsetIndex> index_;
+};
+}  // namespace
+
 class ReadAtTrackingInputStream : public InputStream {
  public:
     explicit ReadAtTrackingInputStream(std::shared_ptr<InputStream> input)
@@ -1766,6 +1798,70 @@ TEST_F(PageFilteredRowGroupReaderTest, MissingOffsetIndexFallsBackToSequentialRe
     ASSERT_EQ(85, values->Value(1));
 }
 
+TEST_F(PageFilteredRowGroupReaderTest, InvalidOffsetIndexFallsBackWithoutChangingSelectedRows) {
+    std::string file_name = dir_->Str() + "/invalid_offset_index.parquet";
+    auto data = MakeSequentialIntData(100);
+    WriteTestFile(file_name, data, /*write_batch_size=*/10, /*max_row_group_length=*/100);
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<InputStream> in, fs_->Open(file_name));
+    ASSERT_OK_AND_ASSIGN(int64_t length, in->Length());
+    auto stream = std::make_shared<ArrowInputStreamAdapter>(in, length, arrow_pool_);
+    std::unique_ptr<::parquet::arrow::FileReader> reader;
+    ASSERT_TRUE(::parquet::arrow::OpenFile(stream, arrow_pool_.get(), &reader).ok());
+    auto column = reader->parquet_reader()->metadata()->RowGroup(0)->ColumnChunk(0);
+    int64_t offset = column->data_page_offset();
+    int64_t size = column->total_compressed_size();
+    std::vector<std::vector<::parquet::PageLocation>> invalid_pages = {
+        {},
+        {{offset, 1, 1}},
+        {{offset - 1, 1, 0}},
+        {{offset, 0, 0}},
+        {{offset, 2, 0}, {offset + 1, 2, 10}},
+        {{offset, 1, 0}, {offset + size, 1, 10}},
+    };
+    RowRanges rows;
+    rows.Add(RowRanges::Range(15, 19));
+    rows.Add(RowRanges::Range(85, 89));
+    TargetRowGroup target(0, true, rows);
+    // A missing index and each malformed layout must select the same rows using the
+    // sequential fallback, including through the streaming reader's schema API.
+    for (size_t i = 0; i <= invalid_pages.size(); ++i) {
+        SCOPED_TRACE(i);
+        std::shared_ptr<::parquet::OffsetIndex> index;
+        if (i < invalid_pages.size()) {
+            index = std::make_shared<SyntheticOffsetIndex>(invalid_pages[i]);
+            ASSERT_FALSE(
+                PageFilteredRowGroupReader::MakeDataPageReadPlan(rows, index, *column, 100));
+        }
+        auto page_index = std::make_shared<SyntheticPageIndexReader>(index);
+        std::vector<arrow::io::ReadRange> ranges = PageFilteredRowGroupReader::ComputePageRanges(
+            target, {0}, page_index, reader->parquet_reader());
+        ASSERT_EQ(1, ranges.size());
+        ASSERT_EQ(offset, ranges[0].offset);
+        ASSERT_EQ(size, ranges[0].length);
+        ASSERT_OK_AND_ASSIGN(std::unique_ptr<arrow::RecordBatchReader> batches,
+                             PageFilteredRowGroupReader::ReadFilteredRowGroup(
+                                 target, {0}, arrow::io::CacheOptions::Defaults(), false, ranges,
+                                 /*max_chunksize=*/3, page_index, arrow_pool_, reader.get()));
+        ASSERT_TRUE(
+            batches->schema()->Equals(*arrow::schema({arrow::field("val", arrow::int32())})));
+        std::vector<int32_t> selected;
+        while (true) {
+            std::shared_ptr<arrow::RecordBatch> batch;
+            ASSERT_TRUE(batches->ReadNext(&batch).ok());
+            if (!batch) {
+                break;
+            }
+            ASSERT_LE(batch->num_rows(), 3);
+            auto values = std::dynamic_pointer_cast<arrow::Int32Array>(batch->column(0));
+            ASSERT_NE(nullptr, values);
+            for (int64_t row = 0; row < values->length(); ++row) {
+                selected.push_back(values->Value(row));
+            }
+        }
+        ASSERT_EQ((std::vector<int32_t>{15, 16, 17, 18, 19, 85, 86, 87, 88, 89}), selected);
+    }
+}
+
 /// Test: bitmap hits partial pages of a row group (no predicate).
 ///
 /// 200 rows, 10 rows per page, 100 rows per row group → 2 row groups.
diff --git a/src/paimon/fs/local/local_file_test.cpp b/src/paimon/fs/local/local_file_test.cpp
index f172780b2..96b551845 100644
--- a/src/paimon/fs/local/local_file_test.cpp
+++ b/src/paimon/fs/local/local_file_test.cpp
@@ -32,6 +32,17 @@ TEST(LocalFileTest, TestSchemeCaseInsensitive) {
     ASSERT_NOK(LocalFile::Create("s3:/tmp"));
 }
 
+TEST(LocalFileTest, TestGetParentFileWithoutSlash) {
+    for (const char* path : {"file", ""}) {
+        SCOPED_TRACE(path);
+        LocalFile file(path);
+        std::unique_ptr<LocalFile> parent = file.GetParentFile();
+        ASSERT_TRUE(parent);
+        ASSERT_TRUE(parent->IsEmpty());
+        ASSERT_TRUE(parent->GetPath().empty());
+    }
+}
+
 TEST(LocalFileTest, TestGetFileStatusForMissingFile) {
     auto test_root_dir = UniqueTestDirectory::Create();
     ASSERT_TRUE(test_root_dir);
diff --git a/src/paimon/rest/rest_credential_provider_test.cpp b/src/paimon/rest/rest_credential_provider_test.cpp
index 67c98fa1f..ec9d6292e 100644
--- a/src/paimon/rest/rest_credential_provider_test.cpp
+++ b/src/paimon/rest/rest_credential_provider_test.cpp
@@ -20,11 +20,13 @@
 
 #include <atomic>
 #include <chrono>
+#include <condition_variable>
 #include <map>
 #include <memory>
 #include <mutex>
 #include <optional>
 #include <string>
+#include <thread>
 #include <utility>
 
 #include "gtest/gtest.h"
@@ -180,6 +182,63 @@ TEST_F(RestCredentialProviderTest, RefreshesWithinTheSafeTime) {
     ASSERT_EQ(2, state_->request_count.load());
 }
 
+TEST_F(RestCredentialProviderTest, ConcurrentRefreshReusesTheTokenLoadedByAnotherCaller) {
+    std::atomic<int32_t> load_count{0};
+    std::mutex mutex;
+    std::condition_variable readers_arrived;
+    int32_t clock_calls = 0;
+    bool readers_synchronized = true;
+    const int64_t refreshed_expiration =
+        kExpiresAtMillis + 10 * RestApi::kTokenExpirationSafeTimeMillis;
+    RestCredentialProvider provider(
+        [&]() -> Result<GetTableTokenResponse> {
+            if (load_count.fetch_add(1) == 0) {
+                return GetTableTokenResponse({{"fs.oss.accessKeyId", "ak-1"}}, kExpiresAtMillis);
+            }
+            return GetTableTokenResponse({{"fs.oss.accessKeyId", "ak-2"}}, refreshed_expiration);
+        },
+        Identifier("db1", "t1"),
+        [&]() {
+            std::unique_lock<std::mutex> lock(mutex);
+            // Both callers must observe the expired token under their shared locks before
+            // either can obtain the write lock and refresh it.
+            if (++clock_calls <= 2) {
+                readers_arrived.notify_all();
+                if (!readers_arrived.wait_for(lock, std::chrono::seconds(10),
+                                              [&]() { return clock_calls >= 2; })) {
+                    readers_synchronized = false;
+                }
+            }
+            return std::chrono::system_clock::time_point(
+                std::chrono::milliseconds(kExpiresAtMillis));
+        });
+
+    // An absent token loads without consulting the clock. The seeded token is already
+    // expired at the injected time, so both concurrent callers will need a refresh.
+    ASSERT_OK_AND_ASSIGN(RestToken initial, provider.ValidToken());
+    ASSERT_EQ("ak-1", initial.token.at("fs.oss.accessKeyId"));
+    ASSERT_EQ(1, load_count.load());
+
+    std::optional<Result<RestToken>> first;
+    std::optional<Result<RestToken>> second;
+    std::thread first_caller([&]() { first.emplace(provider.ValidToken()); });
+    std::thread second_caller([&]() { second.emplace(provider.ValidToken()); });
+    first_caller.join();
+    second_caller.join();
+
+    ASSERT_TRUE(readers_synchronized);
+    ASSERT_TRUE(first.has_value());
+    ASSERT_TRUE(second.has_value());
+    ASSERT_OK(first->status());
+    ASSERT_OK(second->status());
+    ASSERT_EQ("ak-2", first->value().token.at("fs.oss.accessKeyId"));
+    ASSERT_EQ("ak-2", second->value().token.at("fs.oss.accessKeyId"));
+    ASSERT_EQ(refreshed_expiration, first->value().expires_at_millis);
+    ASSERT_EQ(refreshed_expiration, second->value().expires_at_millis);
+    // One initial load and one refresh, even though both callers observed expiration.
+    ASSERT_EQ(2, load_count.load());
+}
+
 TEST_F(RestCredentialProviderTest, ExpiredTokenReloadsOnEveryCall) {
     // an expiration the server did not report makes the credentials expire immediately
     {
diff --git a/test/generate_coverage.sh b/test/generate_coverage.sh
index 9f5298c35..fbc1a8398 100755
--- a/test/generate_coverage.sh
+++ b/test/generate_coverage.sh
@@ -21,9 +21,24 @@ set -e
 cd "$(dirname "$0")/../"
 repo_root=$PWD
 mkdir -p build && cd build
-cmake ../ -DCMAKE_BUILD_TYPE=Debug -DPAIMON_BUILD_TESTS=ON -DPAIMON_USE_ASAN=ON -DPAIMON_GENERATE_COVERAGE=ON -DCMAKE_EXPORT_COMPILE_COMMANDS=ON \
-    -DPAIMON_ENABLE_AVRO=ON -DPAIMON_ENABLE_ORC=ON \
-    -DPAIMON_ENABLE_MOSAIC=ON -DPAIMON_ENABLE_LANCE=ON -DPAIMON_ENABLE_VORTEX=ON
+cmake ../ \
+    -DCMAKE_BUILD_TYPE=Debug \
+    -DPAIMON_BUILD_TESTS=ON \
+    -DPAIMON_USE_ASAN=ON \
+    -DPAIMON_GENERATE_COVERAGE=ON \
+    -DCMAKE_EXPORT_COMPILE_COMMANDS=ON \
+    -DPAIMON_ENABLE_AVRO=ON \
+    -DPAIMON_ENABLE_ORC=ON \
+    -DPAIMON_ENABLE_MOSAIC=ON \
+    -DPAIMON_ENABLE_LANCE=ON \
+    -DPAIMON_ENABLE_VORTEX=ON \
+    -DPAIMON_ENABLE_JINDO=ON \
+    -DPAIMON_ENABLE_OSS=ON \
+    -DPAIMON_ENABLE_S3=ON \
+    -DPAIMON_ENABLE_LUMINA=ON \
+    -DPAIMON_ENABLE_LUCENE=ON \
+    -DPAIMON_ENABLE_FULL_TEXT=ON \
+    -DPAIMON_ENABLE_REST=ON
 make -j
 make test
 lcov --capture --directory src/paimon --directory test --output-file coverage.info