Merge 642a42a into ffe45cd

ydb-platform · Sep 16, 2024 · 2a78401 · 2a78401
2 parents ffe45cd + 642a42a
commit 2a78401
Show file tree

Hide file tree

Showing 31 changed files with 648 additions and 186 deletions.
diff --git a/ydb/core/kqp/common/kqp_tx.cpp b/ydb/core/kqp/common/kqp_tx.cpp
@@ -14,17 +14,17 @@ NYql::TIssue GetLocksInvalidatedIssue(const TKqpTransactionContext& txCtx, const
     if (pathId.OwnerId() != 0) {
         auto table = txCtx.TableByIdMap.FindPtr(pathId);
         if (!table) {
-            return YqlIssue(TPosition(), TIssuesIds::KIKIMR_LOCKS_INVALIDATED, message << " Unknown table.");    
+            return YqlIssue(TPosition(), TIssuesIds::KIKIMR_LOCKS_INVALIDATED, message << " Unknown table.");
         }
-        return YqlIssue(TPosition(), TIssuesIds::KIKIMR_LOCKS_INVALIDATED, message << " Table: " << *table);
+        return YqlIssue(TPosition(), TIssuesIds::KIKIMR_LOCKS_INVALIDATED, message << " Table: " << "`" << *table << "`");
     } else {
         // Olap tables don't return SchemeShard in locks, thus we use tableId here.
         for (const auto& [pathId, table] : txCtx.TableByIdMap) {
             if (pathId.TableId() == pathId.TableId()) {
-                return YqlIssue(TPosition(), TIssuesIds::KIKIMR_LOCKS_INVALIDATED, message << " Table: " << table);
+                return YqlIssue(TPosition(), TIssuesIds::KIKIMR_LOCKS_INVALIDATED, message << " Table: " << "`" << table << "`");
             }
         }
-        return YqlIssue(TPosition(), TIssuesIds::KIKIMR_LOCKS_INVALIDATED, message << " Unknown table.");    
+        return YqlIssue(TPosition(), TIssuesIds::KIKIMR_LOCKS_INVALIDATED, message << " Unknown table."); 
     }
 }
 
@@ -36,6 +36,27 @@ TIssue GetLocksInvalidatedIssue(const TKqpTransactionContext& txCtx, const TKqpT
             invalidatedLock.GetPathId()));
 }
 
+NYql::TIssue GetLocksInvalidatedIssue(const TShardIdToTableInfo& shardIdToTableInfo, const ui64& shardId) {
+    TStringBuilder message;
+    message << "Transaction locks invalidated.";
+
+    if (auto tableInfoPtr = shardIdToTableInfo.GetPtr(shardId); tableInfoPtr) {
+        message << " Tables: ";
+        bool first = true;
+        for (const auto& path : tableInfoPtr->Pathes) {
+            if (!first) {
+                message << ", ";
+                first = false;
+            }
+            message << "`" << path << "`";
+        }
+        return YqlIssue(TPosition(), TIssuesIds::KIKIMR_LOCKS_INVALIDATED, message);
+    } else {
+        message << " Unknown table.";   
+    }
+    return YqlIssue(TPosition(), TIssuesIds::KIKIMR_LOCKS_INVALIDATED, message);
+}
+
 std::pair<bool, std::vector<TIssue>> MergeLocks(const NKikimrMiniKQL::TType& type, const NKikimrMiniKQL::TValue& value,
     TKqpTransactionContext& txCtx)
 {

diff --git a/ydb/core/kqp/common/kqp_tx.h b/ydb/core/kqp/common/kqp_tx.h
@@ -121,6 +121,50 @@ struct TDeferredEffects {
     friend class TKqpTransactionContext;
 };
 
+struct TTableInfo {
+    bool IsOlap = false;
+    THashSet<TStringBuf> Pathes;
+};
+
+
+class TShardIdToTableInfo {
+public:
+    const TTableInfo& Get(ui64 shardId) const {
+        const auto* result = GetPtr(shardId);
+        AFL_ENSURE(result);
+        return *result;
+    }
+
+    const TTableInfo* GetPtr(ui64 shardId) const {
+        auto it = ShardIdToInfo.find(shardId);
+        return it != std::end(ShardIdToInfo)
+            ? &it->second
+            : nullptr;
+    }
+
+    void Add(ui64 shardId, bool isOlap, const TString& path) {
+        const auto [stringsIter, _] = Strings.insert(path);
+        const TStringBuf pathBuf = *stringsIter;
+        auto infoIter = ShardIdToInfo.find(shardId);
+        if (infoIter != std::end(ShardIdToInfo)) {
+            AFL_ENSURE(infoIter->second.IsOlap == isOlap);
+            infoIter->second.Pathes.insert(pathBuf);
+        } else {
+            ShardIdToInfo.emplace(
+                shardId,
+                TTableInfo{
+                    .IsOlap = isOlap,
+                    .Pathes = {pathBuf},
+                });
+        }
+    }
+
+private:
+    THashMap<ui64, TTableInfo> ShardIdToInfo;
+    std::unordered_set<TString> Strings;// Pointers aren't invalidated.
+};
+using TShardIdToTableInfoPtr = std::shared_ptr<TShardIdToTableInfo>;
+
 class TKqpTransactionContext : public NYql::TKikimrTransactionContextBase  {
 public:
     explicit TKqpTransactionContext(bool implicit, const NMiniKQL::IFunctionRegistry* funcRegistry,
@@ -287,6 +331,12 @@ class TKqpTransactionContext : public NYql::TKikimrTransactionContextBase  {
     TTxAllocatorState::TPtr TxAlloc;
 
     IKqpGateway::TKqpSnapshotHandle SnapshotHandle;
+
+    bool HasOlapTable = false;
+    bool HasOltpTable = false;
+    bool HasTableWrite = false;
+
+    TShardIdToTableInfoPtr ShardIdToTableInfo = std::make_shared<TShardIdToTableInfo>();
 };
 
 struct TTxId {
@@ -435,6 +485,7 @@ class TTransactionsCache {
 };
 
 NYql::TIssue GetLocksInvalidatedIssue(const TKqpTransactionContext& txCtx, const NYql::TKikimrPathId& pathId);
+NYql::TIssue GetLocksInvalidatedIssue(const TShardIdToTableInfo& shardIdToTableInfo, const ui64& shardId);
 std::pair<bool, std::vector<NYql::TIssue>> MergeLocks(const NKikimrMiniKQL::TType& type,
     const NKikimrMiniKQL::TValue& value, TKqpTransactionContext& txCtx);
 

diff --git a/ydb/core/kqp/compile_service/kqp_compile_actor.cpp b/ydb/core/kqp/compile_service/kqp_compile_actor.cpp
@@ -603,6 +603,7 @@ void ApplyServiceConfig(TKikimrConfiguration& kqpConfig, const TTableServiceConf
     kqpConfig.EnableCreateTableAs = serviceConfig.GetEnableCreateTableAs();
     kqpConfig.EnableOlapSink = serviceConfig.GetEnableOlapSink();
     kqpConfig.EnableOltpSink = serviceConfig.GetEnableOltpSink();
+    kqpConfig.EnableHtapTx = serviceConfig.GetEnableHtapTx();
     kqpConfig.BlockChannelsMode = serviceConfig.GetBlockChannelsMode();
     kqpConfig.IdxLookupJoinsPrefixPointLimit = serviceConfig.GetIdxLookupJoinPointsLimit();
     kqpConfig.OldLookupJoinBehaviour = serviceConfig.GetOldLookupJoinBehaviour();

diff --git a/ydb/core/kqp/compile_service/kqp_compile_service.cpp b/ydb/core/kqp/compile_service/kqp_compile_service.cpp
@@ -525,6 +525,7 @@ class TKqpCompileService : public TActorBootstrapped<TKqpCompileService> {
         bool enableColumnsWithDefault = TableServiceConfig.GetEnableColumnsWithDefault();
         bool enableOlapSink = TableServiceConfig.GetEnableOlapSink();
         bool enableOltpSink = TableServiceConfig.GetEnableOltpSink();
+        bool enableHtapTx = TableServiceConfig.GetEnableHtapTx();
         bool enableCreateTableAs = TableServiceConfig.GetEnableCreateTableAs();
         auto blockChannelsMode = TableServiceConfig.GetBlockChannelsMode();
 
@@ -558,6 +559,7 @@ class TKqpCompileService : public TActorBootstrapped<TKqpCompileService> {
             TableServiceConfig.GetEnableColumnsWithDefault() != enableColumnsWithDefault ||
             TableServiceConfig.GetEnableOlapSink() != enableOlapSink ||
             TableServiceConfig.GetEnableOltpSink() != enableOltpSink ||
+            TableServiceConfig.GetEnableHtapTx() != enableHtapTx ||
             TableServiceConfig.GetEnableCreateTableAs() != enableCreateTableAs ||
             TableServiceConfig.GetBlockChannelsMode() != blockChannelsMode ||
             TableServiceConfig.GetOldLookupJoinBehaviour() != oldLookupJoinBehaviour ||