vector: Fix building vector index on DeltaVS may lead to delta compact failure (#10311)

ref #9600, close #10310

* In `DeltaMergeStore::segmentEnsureDeltaLocalIndexAsync` and `DeltaMergeStore::segmentWaitDeltaLocalIndexReady` if we detect that `segment->getDelta()->isUpdating()`, it means the segment is running an update task (SegmentMergeDelta/SegmentMerge/SegmentSplit), just skip the local index build job
* `ColumnFilePersistedSet::installCompactionResults` if we detect the column file set is changed during minor compaction, then log an error and rollback the minor compaction
* `WriteBatches::rollbackWrittenLogAndData` will also remove the pending `removed_log` and `removed_data`

Signed-off-by: JaySon-Huang <tshent@qq.com>

Author: JaySon-Huang

dbms/src/Storages/DeltaMerge/Delta/ColumnFilePersistedSet.cpp

@@ -75,7 +75,7 @@ void ColumnFilePersistedSet::checkColumnFiles(const ColumnFilePersisteds & new_c
 
     RUNTIME_CHECK_MSG(
         new_rows == rows && new_deletes == deletes,
-        "Rows and deletes check failed. Actual: rows[{}], deletes[{}]. Expected: rows[{}], deletes[{}]. Current column "
+        "Rows and deletes check failed. Actual: rows={} deletes={} Expected: rows={} deletes={}. Current column "
         "files: {}, new column files: {}.", //
         new_rows,
         new_deletes,
@@ -316,6 +316,7 @@ bool ColumnFilePersistedSet::installCompactionResults(const MinorCompactionPtr &
     minor_compaction_version += 1;
     LOG_DEBUG(log, "{}, before commit compaction, persisted column files: {}", info(), detailInfo());
     ColumnFilePersisteds new_persisted_files;
+    // Generate the new persisted files by the compaction results.
     for (const auto & task : compaction->getTasks())
     {
         if (task.is_trivial_move)
@@ -333,23 +334,28 @@ bool ColumnFilePersistedSet::installCompactionResults(const MinorCompactionPtr &
                     || (file->getId() != (*old_persisted_files_iter)->getId())
                     || (file->getRows() != (*old_persisted_files_iter)->getRows())))
             {
-                throw Exception(
-                    ErrorCodes::LOGICAL_ERROR,
-                    "Compaction algorithm broken, "
-                    "compaction={{{}}} persisted_files={} "
+                // The ColumnFile in `to_compact` is not in the latest persisted files, skip
+                // the minor compaction.
+                LOG_WARNING(
+                    log,
+                    "Minor Compaction is skipped, "
+                    "compaction={{{}}} persisted_files={} task={} "
                     "old_persisted_files_iter.is_end={} "
                     "file->getId={} old_persist_files->getId={} file->getRows={} old_persist_files->getRows={}",
                     compaction->info(),
                     detailInfo(),
+                    task.toString(),
                     old_persisted_files_iter == persisted_files.end(),
                     file->getId(),
                     old_persisted_files_iter == persisted_files.end() ? -1 : (*old_persisted_files_iter)->getId(),
                     file->getRows(),
                     old_persisted_files_iter == persisted_files.end() ? -1 : (*old_persisted_files_iter)->getRows());
+                return false;
             }
             old_persisted_files_iter++;
         }
     }
+    // The new persisted_files contains the files that appended during the minor compaction.
     while (old_persisted_files_iter != persisted_files.end())
     {
         new_persisted_files.emplace_back(*old_persisted_files_iter);
@@ -387,7 +393,7 @@ ColumnFileSetSnapshotPtr ColumnFilePersistedSet::createSnapshot(const IColumnFil
     {
         LOG_ERROR(
             log,
-            "Rows and deletes check failed. Actual: rows[{}], deletes[{}]. Expected: rows[{}], deletes[{}].",
+            "Rows and deletes check failed. Actual: rows={} deletes={}. Expected: rows={} deletes={}",
             total_rows,
             total_deletes,
             rows.load(),

dbms/src/Storages/DeltaMerge/Delta/DeltaValueSpace.cpp

@@ -224,7 +224,7 @@ std::pair<ColumnFiles, ColumnFilePersisteds> DeltaValueSpace::cloneNewlyAppended
         snapshot_persisted_files.begin(),
         snapshot_persisted_files.end());
     // If there were flush since the snapshot, the flushed files should be behind the files in the snapshot.
-    // So let's place these "flused files" after the persisted files in snapshot.
+    // So let's place these "flushed files" after the persisted files in snapshot.
     head_persisted_files.insert(head_persisted_files.end(), flushed_mem_files.begin(), flushed_mem_files.end());
 
     auto new_persisted_files = persisted_file_set->diffColumnFiles(head_persisted_files);
@@ -354,9 +354,7 @@ bool DeltaValueSpace::flush(DMContext & context)
     SCOPE_EXIT({
         bool v = true;
         if (!is_flushing.compare_exchange_strong(v, false))
-            throw Exception(
-                fmt::format("Delta is expected to be flushing, delta={}", simple_info),
-                ErrorCodes::LOGICAL_ERROR);
+            throw Exception(ErrorCodes::LOGICAL_ERROR, "Delta is expected to be flushing, delta={}", simple_info);
     });
 
     LOG_DEBUG(log, "Flush start, delta={}", info());
@@ -475,6 +473,8 @@ bool DeltaValueSpace::compact(DMContext & context)
             /*tracing_id*/ fmt::format("minor_compact_{}", simpleInfo()));
     }
 
+    SYNC_FOR("DeltaValueSpace::compact_after_compaction_task_build");
+
     WriteBatches wbs(*context.storage_pool, context.getWriteLimiter());
     {
         // do compaction task
@@ -500,9 +500,11 @@ bool DeltaValueSpace::compact(DMContext & context)
         }
         if (!compaction_task->commit(persisted_file_set, wbs))
         {
-            LOG_WARNING(log, "Structure has been updated during compact, delta={}", simpleInfo());
+            LOG_DEBUG(
+                log,
+                "Structure has been updated during compact, delta compaction is stopped, delta={}",
+                simpleInfo());
             wbs.rollbackWrittenLogAndData();
-            LOG_DEBUG(log, "Compact stop because structure got updated, delta={}", simpleInfo());
             return false;
         }
         // Reset to the index of first file that can be compacted if the minor compaction succeed,

dbms/src/Storages/DeltaMerge/Delta/MinorCompaction.h

@@ -50,6 +50,19 @@ class MinorCompaction : public std::enable_shared_from_this<MinorCompaction>
             total_rows += column_file->getRows();
             total_bytes += column_file->getBytes();
         }
+
+        String toString() const
+        {
+            return fmt::format(
+                "Task{{to_compact_size={} first_file_index={} total_rows={} total_bytes={} is_trivial_move={} "
+                "result_rows={}}}",
+                to_compact.size(),
+                first_file_index,
+                total_rows,
+                total_bytes,
+                is_trivial_move,
+                result ? result->getRows() : 0);
+        }
     };
     using Tasks = std::vector<Task>;
 

dbms/src/Storages/DeltaMerge/DeltaMergeStore_InternalSegment.cpp

@@ -20,6 +20,7 @@
 #include <Storages/DeltaMerge/ColumnFile/ColumnFileDataProvider.h>
 #include <Storages/DeltaMerge/ColumnFile/ColumnFileTinyLocalIndexWriter.h>
 #include <Storages/DeltaMerge/DMContext.h>
+#include <Storages/DeltaMerge/Delta/DeltaValueSpace.h>
 #include <Storages/DeltaMerge/DeltaMergeStore.h>
 #include <Storages/DeltaMerge/File/DMFileLocalIndexWriter.h>
 #include <Storages/DeltaMerge/LocalIndexerScheduler.h>
@@ -851,75 +852,110 @@ void DeltaMergeStore::segmentEnsureStableLocalIndexWithErrorReport(
     }
 }
 
-bool DeltaMergeStore::segmentEnsureDeltaLocalIndexAsync(const SegmentPtr & segment)
+namespace
+{
+struct LocalIndexOnDeltaVSBuildInfo
+{
+    ColumnFileTinyLocalIndexWriter::LocalIndexBuildInfo build_info;
+    std::weak_ptr<DeltaValueSpace> delta_weak_ptr;
+};
+std::optional<LocalIndexOnDeltaVSBuildInfo> //
+genBuildInfoFromDeltaVS(const SegmentPtr & segment, const LocalIndexInfosSnapshot & local_index_infos)
 {
-    RUNTIME_CHECK(segment != nullptr);
-
-    auto local_index_infos_snap = getLocalIndexInfosSnapshot();
-    if (!local_index_infos_snap)
-        return false;
-
     // Acquire a lock to make sure delta is not changed during the process.
     auto lock = segment->getUpdateLock();
     if (!lock)
-        return false;
+        return std::nullopt;
+    // The segment is running an update(SegmentMergeDelta/SegmentMerge/SegmentSplit) task, skip the index build.
+    if (segment->getDelta()->isUpdating())
+        return std::nullopt;
+
+    // In case nothing to be built
     auto column_file_persisted_set = segment->getDelta()->getPersistedFileSet();
     if (!column_file_persisted_set)
-        return false;
+        return std::nullopt;
     auto build_info
-        = ColumnFileTinyLocalIndexWriter::getLocalIndexBuildInfo(local_index_infos_snap, column_file_persisted_set);
+        = ColumnFileTinyLocalIndexWriter::getLocalIndexBuildInfo(local_index_infos, column_file_persisted_set);
     if (!build_info.indexes_to_build || build_info.indexes_to_build->empty())
-        return false;
+        return std::nullopt;
+
     // Use weak_ptr to avoid blocking gc.
     auto delta_weak_ptr = std::weak_ptr<DeltaValueSpace>(segment->getDelta());
     lock->unlock();
+    return LocalIndexOnDeltaVSBuildInfo{build_info, delta_weak_ptr};
+}
+} // namespace
+
+bool DeltaMergeStore::segmentEnsureDeltaLocalIndexAsync(const SegmentPtr & segment)
+{
+    RUNTIME_CHECK(segment != nullptr);
+
+    auto local_index_infos_snap = getLocalIndexInfosSnapshot();
+    if (!local_index_infos_snap)
+        return false;
+
+    auto delta_vs_build_info = genBuildInfoFromDeltaVS(segment, local_index_infos_snap);
+    if (!delta_vs_build_info)
+    {
+        LOG_DEBUG(
+            log,
+            "segmentEnsureDeltaLocalIndexAsync - Give up because no index to build or delta is being updated, "
+            "segment={}",
+            segment->simpleInfo());
+        return true;
+    }
 
     auto store_weak_ptr = weak_from_this();
-    auto tracing_id = fmt::format("segmentEnsureDeltaLocalIndexAsync source_segment={}", segment->simpleInfo());
-    auto workload = [store_weak_ptr, build_info, delta_weak_ptr, segment, tracing_id]() -> void {
+    const auto source_segment_info = segment->simpleInfo();
+    auto workload = [store_weak_ptr, delta_vs_build_info, source_segment_info]() -> void {
         auto store = store_weak_ptr.lock();
         if (!store) // Store is destroyed before the task is executed.
             return;
-        auto delta = delta_weak_ptr.lock();
+        auto delta = delta_vs_build_info->delta_weak_ptr.lock();
         if (!delta) // Delta is destroyed before the task is executed.
             return;
+        auto tracing_id = fmt::format("segmentEnsureDeltaLocalIndexAsync source_segment={}", source_segment_info);
         auto dm_context = store->newDMContext( //
             store->global_context,
             store->global_context.getSettingsRef(),
             tracing_id);
-        const auto source_segment_info = segment->simpleInfo();
-        store->segmentEnsureDeltaLocalIndex(*dm_context, build_info.indexes_to_build, delta, source_segment_info);
+        store->segmentEnsureDeltaLocalIndex(
+            *dm_context,
+            delta_vs_build_info->build_info.indexes_to_build,
+            delta,
+            source_segment_info);
     };
 
     auto indexer_scheduler = global_context.getGlobalLocalIndexerScheduler();
     RUNTIME_CHECK(indexer_scheduler != nullptr);
     try
     {
         // new task of these index are generated, clear existing error_message in segment
-        segment->clearIndexBuildError(build_info.index_ids);
+        segment->clearIndexBuildError(delta_vs_build_info->build_info.index_ids);
 
         auto [ok, reason] = indexer_scheduler->pushTask(LocalIndexerScheduler::Task{
             .keyspace_id = keyspace_id,
             .table_id = physical_table_id,
-            .file_ids = build_info.file_ids,
-            .request_memory = build_info.estimated_memory_bytes,
+            .file_ids = delta_vs_build_info->build_info.file_ids,
+            .request_memory = delta_vs_build_info->build_info.estimated_memory_bytes,
             .workload = workload,
         });
         if (ok)
             return true;
 
-        segment->setIndexBuildError(build_info.index_ids, reason);
+        segment->setIndexBuildError(delta_vs_build_info->build_info.index_ids, reason);
+        auto tracing_id = fmt::format("segmentEnsureDeltaLocalIndexAsync source_segment={}", source_segment_info);
         LOG_ERROR(
             log->getChild(tracing_id),
             "Failed to generate async segment stable index task, index_ids={} reason={}",
-            build_info.index_ids,
+            delta_vs_build_info->build_info.index_ids,
             reason);
         return false;
     }
     catch (...)
     {
         const auto message = getCurrentExceptionMessage(false, false);
-        segment->setIndexBuildError(build_info.index_ids, message);
+        segment->setIndexBuildError(delta_vs_build_info->build_info.index_ids, message);
 
         tryLogCurrentException(log);
 
@@ -937,18 +973,15 @@ bool DeltaMergeStore::segmentWaitDeltaLocalIndexReady(const SegmentPtr & segment
     if (!local_index_infos_snap)
         return true;
 
-    // Acquire a lock to make sure delta is not changed during the process.
-    auto lock = segment->mustGetUpdateLock();
-    auto column_file_persisted_set = segment->getDelta()->getPersistedFileSet();
-    if (!column_file_persisted_set)
-        return false;
-    auto build_info
-        = ColumnFileTinyLocalIndexWriter::getLocalIndexBuildInfo(local_index_infos, column_file_persisted_set);
-    // Use weak_ptr to avoid blocking gc.
-    auto delta_weak_ptr = std::weak_ptr<DeltaValueSpace>(segment->getDelta());
-    lock.unlock();
-    if (!build_info.indexes_to_build || build_info.indexes_to_build->empty())
-        return false;
+    auto delta_vs_build_info = genBuildInfoFromDeltaVS(segment, local_index_infos_snap);
+    if (!delta_vs_build_info)
+    {
+        LOG_INFO(
+            log,
+            "WaitDeltaLocalIndexReady - Give up because no index to build or delta is being updated, segment={}",
+            segment->simpleInfo());
+        return true;
+    }
 
     auto segment_id = segment->segmentId();
     static constexpr size_t MAX_CHECK_TIME_SECONDS = 60; // 60s
@@ -964,11 +997,11 @@ bool DeltaMergeStore::segmentWaitDeltaLocalIndexReady(const SegmentPtr & segment
         if (!column_file_persisted_set)
             return false; // ColumnFilePersistedSet is not exist, return false
         bool all_indexes_built = true;
-        auto delta_ptr = delta_weak_ptr.lock();
+        auto delta_ptr = delta_vs_build_info->delta_weak_ptr.lock();
         if (auto lock = delta_ptr ? delta_ptr->getLock() : std::nullopt; lock)
         {
             const auto & column_files = column_file_persisted_set->getFiles();
-            for (const auto & index : *build_info.indexes_to_build)
+            for (const auto & index : *delta_vs_build_info->build_info.indexes_to_build)
             {
                 for (const auto & column_file : column_files)
                 {