Skip to content

Commit 3b364c1

Browse files
nikvas0nikvas0
authored
[24-3-analytics] CTAS & Sinks fixes (#10326)
1 parent b2fc4b8 commit 3b364c1

File tree

7 files changed

+819
-20
lines changed

7 files changed

+819
-20
lines changed

‎.github/config/muted_ya.txt

+2-2
Original file line numberDiff line numberDiff line change
@@ -17,10 +17,9 @@ ydb/core/kqp/ut/tx KqpLocksTricky.TestNoLocksIssue+withSink
1717
ydb/core/kqp/ut/tx KqpSnapshotRead.ReadOnlyTxWithIndexCommitsOnConcurrentWrite+withSink
1818
ydb/core/kqp/ut/tx KqpSinkTx.InvalidateOnError
1919
ydb/core/kqp/ut/tx KqpSinkMvcc.ReadWriteTxFailsOnConcurrentWrite3
20-
ydb/core/kqp/ut/tx KqpSinkMvcc.OltpNamedStatement
2120
ydb/core/kqp/ut/tx KqpSinkMvcc.OlapNamedStatement
2221
ydb/core/kqp/ut/tx KqpSinkMvcc.OlapMultiSinks
23-
ydb/core/kqp/ut/tx KqpSinkMvcc.OltpMultiSinks
22+
ydb/core/kqp/ut/tx KqpLocks.MixedTxFail
2423
ydb/core/kqp/ut/query KqpLimits.QueryReplySize
2524
ydb/core/kqp/ut/query KqpQuery.QueryTimeout
2625
ydb/core/kqp/ut/service KqpQueryService.TableSink_OlapRWQueries
@@ -37,6 +36,7 @@ ydb/core/kqp/ut/scheme [44/50]*
3736
ydb/core/kqp/ut/service KqpQueryService.ExecuteQueryPgTableSelect
3837
ydb/core/kqp/ut/service KqpQueryService.QueryOnClosedSession
3938
ydb/core/kqp/ut/service KqpQueryService.TableSink_OltpUpdate
39+
ydb/core/kqp/ut/service KqpQueryService.TableSink_Htap*
4040
ydb/core/kqp/ut/service KqpService.CloseSessionsWithLoad
4141
ydb/core/kqp/ut/service [38/50]*
4242
ydb/core/persqueue/ut [37/40] chunk chunk

‎ydb/core/kqp/opt/kqp_opt_build_txs.cpp

+94-12
Original file line numberDiff line numberDiff line change
@@ -560,16 +560,20 @@ class TKqpBuildTxsTransformer : public TSyncTransformerBase {
560560
}
561561

562562
if (!query.Effects().Empty()) {
563-
auto tx = BuildTx(query.Effects().Ptr(), ctx, /* isPrecompute */ false);
564-
if (!tx) {
565-
return TStatus::Error;
566-
}
563+
auto collectedEffects = CollectEffects(query.Effects(), ctx);
567564

568-
if (!CheckEffectsTx(tx.Cast(), query, ctx)) {
569-
return TStatus::Error;
570-
}
565+
for (auto& effects : collectedEffects) {
566+
auto tx = BuildTx(effects.Ptr(), ctx, /* isPrecompute */ false);
567+
if (!tx) {
568+
return TStatus::Error;
569+
}
571570

572-
BuildCtx->PhysicalTxs.emplace_back(tx.Cast());
571+
if (!CheckEffectsTx(tx.Cast(), effects, ctx)) {
572+
return TStatus::Error;
573+
}
574+
575+
BuildCtx->PhysicalTxs.emplace_back(tx.Cast());
576+
}
573577
}
574578

575579
return TStatus::Ok;
@@ -581,8 +585,86 @@ class TKqpBuildTxsTransformer : public TSyncTransformerBase {
581585
}
582586

583587
private:
584-
bool HasTableEffects(const TKqlQuery& query) const {
585-
for (const TExprBase& effect : query.Effects()) {
588+
TVector<TExprList> CollectEffects(const TExprList& list, TExprContext& ctx) {
589+
struct TEffectsInfo {
590+
enum class EType {
591+
KQP_EFFECT,
592+
KQP_SINK,
593+
EXTERNAL_SINK,
594+
};
595+
596+
EType Type;
597+
THashSet<TStringBuf> TablesPathIds;
598+
TVector<TExprNode::TPtr> Exprs;
599+
};
600+
TVector<TEffectsInfo> effectsInfos;
601+
602+
for (const auto& expr : list) {
603+
if (auto sinkEffect = expr.Maybe<TKqpSinkEffect>()) {
604+
const size_t sinkIndex = FromString(TStringBuf(sinkEffect.Cast().SinkIndex()));
605+
const auto stage = sinkEffect.Cast().Stage().Maybe<TDqStageBase>();
606+
YQL_ENSURE(stage);
607+
YQL_ENSURE(stage.Cast().Outputs());
608+
const auto outputs = stage.Cast().Outputs().Cast();
609+
YQL_ENSURE(sinkIndex < outputs.Size());
610+
const auto sink = outputs.Item(sinkIndex).Maybe<TDqSink>();
611+
YQL_ENSURE(sink);
612+
613+
const auto sinkSettings = sink.Cast().Settings().Maybe<TKqpTableSinkSettings>();
614+
if (!sinkSettings) {
615+
// External writes always use their own physical transaction.
616+
effectsInfos.emplace_back();
617+
effectsInfos.back().Type = TEffectsInfo::EType::EXTERNAL_SINK;
618+
effectsInfos.back().Exprs.push_back(expr.Ptr());
619+
} else {
620+
// Two table sinks can't be executed in one physical transaction if they write into one table.
621+
const TStringBuf tablePathId = sinkSettings.Cast().Table().PathId().Value();
622+
623+
auto it = std::find_if(
624+
std::begin(effectsInfos),
625+
std::end(effectsInfos),
626+
[&tablePathId](const auto& effectsInfo) {
627+
return effectsInfo.Type == TEffectsInfo::EType::KQP_SINK
628+
&& !effectsInfo.TablesPathIds.contains(tablePathId);
629+
});
630+
if (it == std::end(effectsInfos)) {
631+
effectsInfos.emplace_back();
632+
it = std::prev(std::end(effectsInfos));
633+
it->Type = TEffectsInfo::EType::KQP_SINK;
634+
}
635+
it->TablesPathIds.insert(tablePathId);
636+
it->Exprs.push_back(expr.Ptr());
637+
}
638+
} else {
639+
// Table effects are executed all in one physical transaction.
640+
auto it = std::find_if(
641+
std::begin(effectsInfos),
642+
std::end(effectsInfos),
643+
[](const auto& effectsInfo) { return effectsInfo.Type == TEffectsInfo::EType::KQP_EFFECT; });
644+
if (it == std::end(effectsInfos)) {
645+
effectsInfos.emplace_back();
646+
it = std::prev(std::end(effectsInfos));
647+
it->Type = TEffectsInfo::EType::KQP_EFFECT;
648+
}
649+
it->Exprs.push_back(expr.Ptr());
650+
}
651+
}
652+
653+
TVector<TExprList> results;
654+
655+
for (const auto& effects : effectsInfos) {
656+
auto builder = Build<TExprList>(ctx, list.Pos());
657+
for (const auto& expr : effects.Exprs) {
658+
builder.Add(expr);
659+
}
660+
results.push_back(builder.Done());
661+
}
662+
663+
return results;
664+
}
665+
666+
bool HasTableEffects(const TExprList& effectsList) const {
667+
for (const TExprBase& effect : effectsList) {
586668
if (auto maybeSinkEffect = effect.Maybe<TKqpSinkEffect>()) {
587669
// (KqpSinkEffect (DqStage (... ((DqSink '0 (DataSink '"kikimr") ...)))) '0)
588670
auto sinkEffect = maybeSinkEffect.Cast();
@@ -608,7 +690,7 @@ class TKqpBuildTxsTransformer : public TSyncTransformerBase {
608690
return false;
609691
}
610692

611-
bool CheckEffectsTx(TKqpPhysicalTx tx, const TKqlQuery& query, TExprContext& ctx) const {
693+
bool CheckEffectsTx(TKqpPhysicalTx tx, const TExprList& effectsList, TExprContext& ctx) const {
612694
TMaybeNode<TExprBase> blackistedNode;
613695
VisitExpr(tx.Ptr(), [&blackistedNode](const TExprNode::TPtr& exprNode) {
614696
if (blackistedNode) {
@@ -630,7 +712,7 @@ class TKqpBuildTxsTransformer : public TSyncTransformerBase {
630712
return true;
631713
});
632714

633-
if (blackistedNode && HasTableEffects(query)) {
715+
if (blackistedNode && HasTableEffects(effectsList)) {
634716
ctx.AddError(TIssue(ctx.GetPosition(blackistedNode.Cast().Pos()), TStringBuilder()
635717
<< "Callable not expected in effects tx: " << blackistedNode.Cast<TCallable>().CallableName()));
636718
return false;

‎ydb/core/kqp/session_actor/kqp_session_actor.cpp

+2-1
Original file line numberDiff line numberDiff line change
@@ -856,7 +856,8 @@ class TKqpSessionActor : public TActorBootstrapped<TKqpSessionActor> {
856856
QueryState->TxCtx->HasOlapTable |= ::NKikimr::NKqp::HasOlapTableReadInTx(phyQuery) || ::NKikimr::NKqp::HasOlapTableWriteInTx(phyQuery);
857857
QueryState->TxCtx->HasOltpTable |= ::NKikimr::NKqp::HasOltpTableReadInTx(phyQuery) || ::NKikimr::NKqp::HasOltpTableWriteInTx(phyQuery);
858858
QueryState->TxCtx->HasTableWrite |= ::NKikimr::NKqp::HasOlapTableWriteInTx(phyQuery) || ::NKikimr::NKqp::HasOltpTableWriteInTx(phyQuery);
859-
if (QueryState->TxCtx->HasOlapTable && QueryState->TxCtx->HasOltpTable && QueryState->TxCtx->HasTableWrite && !Settings.TableService.GetEnableHtapTx()) {
859+
if (QueryState->TxCtx->HasOlapTable && QueryState->TxCtx->HasOltpTable && QueryState->TxCtx->HasTableWrite
860+
&& !Settings.TableService.GetEnableHtapTx() && !QueryState->IsSplitted()) {
860861
ReplyQueryError(Ydb::StatusIds::PRECONDITION_FAILED,
861862
"Write transactions between column and row tables are disabled at current time.");
862863
return false;

‎ydb/core/kqp/ut/federated_query/s3/kqp_federated_query_ut.cpp

+1-2
Original file line numberDiff line numberDiff line change
@@ -1213,8 +1213,7 @@ Y_UNIT_TEST_SUITE(KqpFederatedQuery) {
12131213
auto db = kikimr->GetQueryClient();
12141214
auto resultFuture = db.ExecuteQuery(sql, NYdb::NQuery::TTxControl::BeginTx().CommitTx());
12151215
resultFuture.Wait();
1216-
UNIT_ASSERT_C(!resultFuture.GetValueSync().IsSuccess(), resultFuture.GetValueSync().GetIssues().ToString());
1217-
UNIT_ASSERT_STRING_CONTAINS(resultFuture.GetValueSync().GetIssues().ToString(), "Callable not expected in effects tx: Unwrap");
1216+
UNIT_ASSERT_C(resultFuture.GetValueSync().IsSuccess(), resultFuture.GetValueSync().GetIssues().ToString());
12181217
}
12191218
}
12201219

0 commit comments

Comments
 (0)