Merge 17f08af into 30ebe53

ydb-platform · Oct 16, 2024 · 7a20392 · 7a20392
2 parents 30ebe53 + 17f08af
commit 7a20392
Show file tree

Hide file tree

Showing 4 changed files with 76 additions and 14 deletions.
diff --git a/ydb/core/kqp/common/kqp_tx.cpp b/ydb/core/kqp/common/kqp_tx.cpp
@@ -343,5 +343,70 @@ bool HasOltpTableWriteInTx(const NKqpProto::TKqpPhyQuery& physicalQuery) {
     return false;
 }
 
+bool HasUncommittedChangesRead(const ui64 currentTx, const NKqpProto::TKqpPhyQuery& physicalQuery) {
+    auto getTable = [](const NKqpProto::TKqpPhyTableId& table) {
+        return NKikimr::TTableId(table.GetOwnerId(), table.GetTableId());
+    };
+
+    THashSet<NKikimr::TTableId> ModifiedTables;
+    for (size_t index = 0; index < physicalQuery.TransactionsSize(); ++index) {
+        const bool canBeUncommittedChangesRead = (index >= currentTx);
+        const auto &tx = physicalQuery.GetTransactions()[index];
+        for (const auto &stage : tx.GetStages()) {
+            for (const auto &tableOp : stage.GetTableOps()) {
+                switch (tableOp.GetTypeCase()) {
+                    case NKqpProto::TKqpPhyTableOperation::kReadRange:
+                    case NKqpProto::TKqpPhyTableOperation::kLookup:
+                    case NKqpProto::TKqpPhyTableOperation::kReadRanges: {
+                        if (canBeUncommittedChangesRead && ModifiedTables.contains(getTable(tableOp.GetTable()))) {
+                            return true;
+                        }
+                        break;
+                    }
+                    case NKqpProto::TKqpPhyTableOperation::kReadOlapRange:
+                    case NKqpProto::TKqpPhyTableOperation::kUpsertRows:
+                    case NKqpProto::TKqpPhyTableOperation::kDeleteRows:
+                        ModifiedTables.insert(getTable(tableOp.GetTable()));
+                        break;
+                    default:
+                        YQL_ENSURE(false, "unexpected type");
+                }
+            }
+
+            for (const auto& input : stage.GetInputs()) {
+                if (input.GetTypeCase() == NKqpProto::TKqpPhyConnection::kStreamLookup) {
+                    if (canBeUncommittedChangesRead && ModifiedTables.contains(getTable(input.GetStreamLookup().GetTable()))) {
+                        return true;
+                    }
+                } else {
+                    return true;
+                }
+            }
+
+            for (const auto& source : stage.GetSources()) {
+                if (source.GetTypeCase() == NKqpProto::TKqpSource::kReadRangesSource) {
+                    if (canBeUncommittedChangesRead && ModifiedTables.contains(getTable(source.GetReadRangesSource().GetTable()))) {
+                        return true;
+                    }
+                } else {
+                    return true;
+                }
+            }
+
+            for (const auto& sink : stage.GetSinks()) {
+                if (sink.GetTypeCase() == NKqpProto::TKqpSink::kInternalSink) {
+                    YQL_ENSURE(sink.GetInternalSink().GetSettings().Is<NKikimrKqp::TKqpTableSinkSettings>());
+                    NKikimrKqp::TKqpTableSinkSettings settings;
+                    YQL_ENSURE(sink.GetInternalSink().GetSettings().UnpackTo(&settings), "Failed to unpack settings");
+                    ModifiedTables.insert(getTable(settings.GetTable()));
+                } else {
+                    return true;
+                }
+            }
+        }
+    }
+    return false;
+}
+
 } // namespace NKqp
 } // namespace NKikimr
diff --git a/ydb/core/kqp/common/kqp_tx.h b/ydb/core/kqp/common/kqp_tx.h
@@ -232,6 +232,11 @@ class TKqpTransactionContext : public NYql::TKikimrTransactionContextBase  {
         ParamsState = MakeIntrusive<TParamsState>();
         SnapshotHandle.Snapshot = IKqpGateway::TKqpSnapshot::InvalidSnapshot;
         HasImmediateEffects = false;
+
+        HasOlapTable = false;
+        HasOltpTable = false;
+        HasTableWrite = false;
+        HasUncommittedChangesRead = false;
     }
 
     TKqpTransactionInfo GetInfo() const;
@@ -333,6 +338,7 @@ class TKqpTransactionContext : public NYql::TKikimrTransactionContextBase  {
     bool HasOlapTable = false;
     bool HasOltpTable = false;
     bool HasTableWrite = false;
+    bool HasUncommittedChangesRead = false;
 
     TShardIdToTableInfoPtr ShardIdToTableInfo = std::make_shared<TShardIdToTableInfo>();
 };
@@ -498,4 +504,6 @@ bool HasOlapTableWriteInTx(const NKqpProto::TKqpPhyQuery& physicalQuery);
 bool HasOltpTableReadInTx(const NKqpProto::TKqpPhyQuery& physicalQuery);
 bool HasOltpTableWriteInTx(const NKqpProto::TKqpPhyQuery& physicalQuery);
 
+bool HasUncommittedChangesRead(const ui64 currentTx, const NKqpProto::TKqpPhyQuery& physicalQuery);
+
 }  // namespace NKikimr::NKqp
diff --git a/ydb/core/kqp/provider/yql_kikimr_provider.h b/ydb/core/kqp/provider/yql_kikimr_provider.h
@@ -292,7 +292,6 @@ class TKikimrTransactionContextBase : public TThrRefBase {
         Invalidated = false;
         Readonly = false;
         Closed = false;
-        HasUncommittedChangesRead = false;
     }
 
     void SetTempTables(NKikimr::NKqp::TKqpTempTablesState::TConstPtr tempTablesState) {
@@ -409,17 +408,6 @@ class TKikimrTransactionContextBase : public TThrRefBase {
             }
 
             auto& currentOps = TableOperations[table];
-            const bool currentModify = currentOps & KikimrModifyOps();
-            if (currentModify) {
-                if (KikimrReadOps() & newOp) {
-                    HasUncommittedChangesRead = true;
-                }
-
-                if ((*info)->GetHasIndexTables()) {
-                    HasUncommittedChangesRead = true;
-                }
-            }
-
             currentOps |= newOp;
         }
 
@@ -429,7 +417,6 @@ class TKikimrTransactionContextBase : public TThrRefBase {
     virtual ~TKikimrTransactionContextBase() = default;
 
 public:
-    bool HasUncommittedChangesRead = false;
     THashMap<TString, TYdbOperations> TableOperations;
     THashMap<TKikimrPathId, TString> TableByIdMap;
     TMaybe<NKikimrKqp::EIsolationLevel> EffectiveIsolationLevel;

diff --git a/ydb/core/kqp/session_actor/kqp_session_actor.cpp b/ydb/core/kqp/session_actor/kqp_session_actor.cpp
@@ -865,6 +865,8 @@ class TKqpSessionActor : public TActorBootstrapped<TKqpSessionActor> {
                             "Write transactions between column and row tables are disabled at current time.");
             return false;
         }
+        QueryState->TxCtx->HasUncommittedChangesRead = ::NKikimr::NKqp::HasUncommittedChangesRead(QueryState->CurrentTx, phyQuery);
+
         QueryState->TxCtx->SetTempTables(QueryState->TempTablesState);
         auto [success, issues] = QueryState->TxCtx->ApplyTableOperations(phyQuery.GetTableOps(), phyQuery.GetTableInfos(),
             EKikimrQueryType::Dml);
@@ -1236,7 +1238,7 @@ class TKqpSessionActor : public TActorBootstrapped<TKqpSessionActor> {
         } else if (QueryState->ShouldAcquireLocks(tx) && (!txCtx.HasOlapTable || Settings.TableService.GetEnableOlapSink())) {
             request.AcquireLocksTxId = txCtx.Locks.GetLockTxId();
 
-            if (txCtx.HasUncommittedChangesRead || Config->FeatureFlags.GetEnableForceImmediateEffectsExecution() || txCtx.HasOlapTable) {
+            if (!txCtx.CanDeferEffects()) {
                 request.UseImmediateEffects = true;
             }
         }