Make transaction not to see aborted changes (#27293)

Co-authored-by: Kirill Vasilenko <kir-vasilenko@ybd.tech>

Author: kirillvasilenko

ydb/core/tx/columnshard/columnshard_impl.cpp

@@ -176,15 +176,14 @@ NOlap::TSnapshot TColumnShard::GetMaxReadVersion() const {
     auto plannedTx = ProgressTxController->GetPlannedTx();
     if (plannedTx) {
         // We may only read just before the first transaction in the queue
-        auto maxReadVersion = TRowVersion(plannedTx->Step, plannedTx->TxId).Prev();
-        return NOlap::TSnapshot(maxReadVersion.Step, maxReadVersion.TxId);
-    }
-    ui64 step = LastPlannedStep;
-    if (MediatorTimeCastEntry) {
-        ui64 mediatorStep = MediatorTimeCastEntry->Get(TabletID());
-        step = Max(step, mediatorStep);
+        auto firstPlannedSnapshot = NOlap::TSnapshot(plannedTx->Step, plannedTx->TxId);
+        return firstPlannedSnapshot.GetPreviousSnapshot();
+    } else {
+        // the same snapshot is used by bulk upsert and aborts
+        // aborts are fine, but be careful with bulk upsert,
+        // it must correctly break conflicting serializable txs
+        return GetCurrentSnapshotForInternalModification();
     }
-    return NOlap::TSnapshot(step, Max<ui64>());
 }
 
 ui64 TColumnShard::GetOutdatedStep() const {

ydb/core/tx/columnshard/common/snapshot.h

@@ -50,7 +50,7 @@ class TSnapshot {
     }
 
     constexpr bool Valid() const noexcept {
-        return PlanStep && TxId;
+        return PlanStep != 0 && TxId != 0;
     }
 
     static constexpr TSnapshot Zero() noexcept {
@@ -106,6 +106,15 @@ class TSnapshot {
     explicit operator size_t() const {
         return CombineHashes(PlanStep, TxId);
     }
+
+    TSnapshot GetPreviousSnapshot() const {
+        AFL_VERIFY(Valid());
+        if (TxId == 0) {
+            return TSnapshot(PlanStep - 1, ::Max<ui64>());
+        } else {
+            return TSnapshot(PlanStep, TxId - 1);
+        }
+    }
 };
 
 } // namespace NKikimr::NOlap

ydb/core/tx/columnshard/engines/reader/common_reader/iterator/source.h

@@ -447,6 +447,7 @@ class IDataSource: public ICursorEntity, public NArrow::NSSA::IDataSource {
             return false;
         }
         if (DoAddTxConflict()) {
+            StageData->Clear();
             StageData->Abort();
             return true;
         }

ydb/core/tx/columnshard/engines/reader/simple_reader/iterator/context.cpp

@@ -40,7 +40,7 @@ std::shared_ptr<TFetchingScript> TSpecialReadContext::DoGetColumnsFetchingPlan(
         portionState = GetPortionStateAtScanStart(portion->GetPortionInfo());
     }
 
-    const bool needSnapshots = GetReadMetadata()->GetRequestSnapshot() < portionState.MaxRecordSnapshot || portionState.Conflicting;
+    const bool needConflictDetector = portionState.Conflicting;
 
     const bool useIndexes = false;
     const bool hasDeletions = source->GetHasDeletions();
@@ -54,21 +54,21 @@ std::shared_ptr<TFetchingScript> TSpecialReadContext::DoGetColumnsFetchingPlan(
 
     const bool preventDuplicates = NeedDuplicateFiltering() && !portionState.Conflicting;
     {
-        auto& result = CacheFetchingScripts[needSnapshots ? 1 : 0][partialUsageByPK ? 1 : 0][useIndexes ? 1 : 0][needShardingFilter ? 1 : 0]
+        auto& result = CacheFetchingScripts[needConflictDetector ? 1 : 0][partialUsageByPK ? 1 : 0][useIndexes ? 1 : 0][needShardingFilter ? 1 : 0]
                                            [hasDeletions ? 1 : 0][preventDuplicates ? 1 : 0];
         if (result.NeedInitialization()) {
             TGuard<TMutex> g(Mutex);
             if (auto gInit = result.StartInitialization()) {
                 gInit->InitializationFinished(BuildColumnsFetchingPlan(
-                    needSnapshots, partialUsageByPK, useIndexes, needShardingFilter, hasDeletions, preventDuplicates, isFinalSyncPoint));
+                    needConflictDetector, partialUsageByPK, useIndexes, needShardingFilter, hasDeletions, preventDuplicates, isFinalSyncPoint));
             }
             AFL_VERIFY(!result.NeedInitialization());
         }
         return result.GetScriptVerified();
     }
 }
 
-std::shared_ptr<TFetchingScript> TSpecialReadContext::BuildColumnsFetchingPlan(const bool needSnapshots, const bool partialUsageByPredicateExt,
+std::shared_ptr<TFetchingScript> TSpecialReadContext::BuildColumnsFetchingPlan(const bool needConflictDetector, const bool partialUsageByPredicateExt,
     const bool /*useIndexes*/, const bool needFilterSharding, const bool needFilterDeletion, const bool preventDuplicates,
     const bool isFinalSyncPoint) const {
     const bool partialUsageByPredicate = partialUsageByPredicateExt && GetPredicateColumns()->GetColumnsCount();
@@ -90,7 +90,7 @@ std::shared_ptr<TFetchingScript> TSpecialReadContext::BuildColumnsFetchingPlan(c
         if (partialUsageByPredicate) {
             acc.AddFetchingStep(*GetPredicateColumns(), NArrow::NSSA::IMemoryCalculationPolicy::EStage::Filter);
         }
-        if (needSnapshots || GetFFColumns()->Cross(*GetSpecColumns())) {
+        if (needConflictDetector || GetFFColumns()->Cross(*GetSpecColumns())) {
             acc.AddFetchingStep(*GetSpecColumns(), NArrow::NSSA::IMemoryCalculationPolicy::EStage::Filter);
         }
         if (needFilterDeletion) {
@@ -101,9 +101,9 @@ std::shared_ptr<TFetchingScript> TSpecialReadContext::BuildColumnsFetchingPlan(c
             acc.AddAssembleStep(*GetPredicateColumns(), "PREDICATE", NArrow::NSSA::IMemoryCalculationPolicy::EStage::Filter, false);
             acc.AddStep(std::make_shared<TPredicateFilter>());
         }
-        if (needSnapshots || GetFFColumns()->Cross(*GetSpecColumns())) {
+        if (needConflictDetector || GetFFColumns()->Cross(*GetSpecColumns())) {
             acc.AddAssembleStep(*GetSpecColumns(), "SPEC", NArrow::NSSA::IMemoryCalculationPolicy::EStage::Filter, false);
-            acc.AddStep(std::make_shared<TSnapshotFilter>());
+            acc.AddStep(std::make_shared<TConflictDetector>());
         }
         if (preventDuplicates) {
             acc.AddStep(std::make_shared<TDuplicateFilter>());

ydb/core/tx/columnshard/engines/reader/simple_reader/iterator/fetching.cpp

@@ -3,6 +3,7 @@
 #include "source.h"
 
 #include <ydb/core/tx/columnshard/engines/filter.h>
+#include <ydb/core/tx/columnshard/engines/portions/written.h>
 #include <ydb/core/tx/columnshard/engines/reader/simple_reader/duplicates/events.h>
 #include <ydb/core/tx/conveyor_composite/usage/service.h>
 #include <ydb/core/tx/limiter/grouped_memory/usage/service.h>
@@ -21,6 +22,48 @@ TConclusion<bool> TPredicateFilter::DoExecuteInplace(const std::shared_ptr<NComm
     return true;
 }
 
+void VerifyConflictingPortion(const std::shared_ptr<NCommon::IDataSource>& source) {
+    // the portion must be a simple portion
+    AFL_VERIFY(source->GetType() == IDataSource::EType::SimplePortion);
+    auto* portionSource = static_cast<TPortionDataSource*>(source.get());
+    auto& info = portionSource->GetPortionInfo();
+    auto status = portionSource->GetContext()->GetPortionStateAtScanStart(info);
+
+    // let's check that the portion state is ok
+    // we may have here only written portions (not compacted)
+    AFL_VERIFY(info.GetPortionType() == EPortionType::Written);
+    const auto& wPortionInfo = static_cast<const TWrittenPortionInfo&>(info);
+    // we may have here only conflicting portions
+    AFL_VERIFY(status.Conflicting);
+    const auto& requestSnapshot = source->GetContext()->GetReadMetadata()->GetRequestSnapshot();
+    // if portion was already committed at the scan start, it must have commit snapshot greater than the request snapshot
+    if (status.Committed) {
+        AFL_VERIFY(wPortionInfo.GetCommitSnapshotVerified() > requestSnapshot)("error", "portion was committed and conflicting at the scan start, but has commit snapshot less than the request snapshot")("portion_info", wPortionInfo.DebugString())("request_snapshot", requestSnapshot.DebugString());
+    } else {
+        // if the portion was not committed it means now it may be:
+        // 1. still not committed
+        if (!wPortionInfo.IsCommitted()) {
+            // do nothing, it is just fine
+        // 2. committed and removed, in this case its snapshot must be greater or equal to the request snapshot
+        } else if (wPortionInfo.HasRemoveSnapshot()) {
+            AFL_VERIFY(wPortionInfo.GetCommitSnapshotVerified() >= requestSnapshot)("error", "portion was committed and conflicting at the scan start, but now it is removed and committed and has commit snapshot less than the request snapshot")("portion_info", wPortionInfo.DebugString())("request_snapshot", requestSnapshot.DebugString());
+        // 3. committed and not removed, in this case its snapshot must be greater than the request snapshot
+        } else {
+            AFL_VERIFY(wPortionInfo.GetCommitSnapshotVerified() > requestSnapshot)("error", "portion was committed and conflicting at the scan start, but now it is committed and has commit snapshot less than the request snapshot")("portion_info", wPortionInfo.DebugString())("request_snapshot", requestSnapshot.DebugString());
+        }
+    }
+    // source must not be empty, we will mark it as conflicting
+    AFL_VERIFY(source->GetRecordsCount() > 0)("error", "source has no records");
+}
+
+TConclusion<bool> TConflictDetector::DoExecuteInplace(
+    const std::shared_ptr<NCommon::IDataSource>& source, const TFetchingScriptCursor& /*step*/) const {
+    VerifyConflictingPortion(source);
+    // it is not empty (not filtered everything out by other filters) and conflicting, so we must mark the conflict here
+    AFL_VERIFY(source->AddTxConflict());
+    return true;
+}
+
 TConclusion<bool> TSnapshotFilter::DoExecuteInplace(
     const std::shared_ptr<NCommon::IDataSource>& source, const TFetchingScriptCursor& /*step*/) const {
     auto filter =

ydb/core/tx/columnshard/engines/reader/simple_reader/iterator/fetching.h

@@ -212,6 +212,18 @@ class TPredicateFilter: public IFetchingStep {
     }
 };
 
+class TConflictDetector: public IFetchingStep {
+private:
+    using TBase = IFetchingStep;
+
+public:
+    virtual TConclusion<bool> DoExecuteInplace(
+        const std::shared_ptr<NCommon::IDataSource>& source, const TFetchingScriptCursor& step) const override;
+    TConflictDetector()
+        : TBase("CONFLICT_DETECTOR") {
+    }
+};
+
 class TSnapshotFilter: public IFetchingStep {
 private:
     using TBase = IFetchingStep;

ydb/core/tx/columnshard/engines/storage/granule/granule.h

@@ -225,8 +225,14 @@ class TGranuleMeta: TNonCopyable {
         auto it = InsertedPortions.find(insertWriteId);
         AFL_VERIFY(it != InsertedPortions.end());
         AFL_VERIFY(InsertedPortionsById.contains(it->second->GetPortionId()));
-        it->second->SetCommitSnapshot(ssRemove);
+        // it is better to set remove snapshot before the commit snapshot
+        // because otherwise concurrent readers may see the portion as just committed while
+        // the commit snapshot is already set, but the remove snapshot is not set yet.
+        // this problem should be addressed properly by a synchronized (or atomic) access
+        // to this part of the portion info state https://github.com/ydb-platform/ydb/issues/27205.
+        // until then, this workaround is better than nothing.
         it->second->SetRemoveSnapshot(ssRemove);
+        it->second->SetCommitSnapshot(ssRemove);
         TDbWrapper wrapper(txc.DB, nullptr);
         it->second->CommitToDatabase(wrapper);
     }

ydb/core/tx/columnshard/operations/write.cpp

@@ -126,9 +126,10 @@ void TWriteOperation::AbortOnExecute(TColumnShard& owner, NTabletFlatExecutor::T
     TBlobGroupSelector dsGroupSelector(owner.Info());
     NOlap::TDbWrapper dbTable(txc.DB, &dsGroupSelector);
 
+    auto abortSnapshot = owner.GetCurrentSnapshotForInternalModification();
     for (auto&& i : InsertWriteIds) {
         owner.MutableIndexAs<NOlap::TColumnEngineForLogs>().MutableGranuleVerified(PathId.InternalPathId).AbortPortionOnExecute(
-            txc, i, owner.GetCurrentSnapshotForInternalModification());
+            txc, i, abortSnapshot);
     }
 }