Skip to content

Commit 1c3ad36

Browse files
nikvas0nikvas0
committed
HTAP: ProposeTx + EvWrite (ydb-platform#9236)
1 parent 5ef387d commit 1c3ad36

File tree

9 files changed

+92
-113
lines changed

9 files changed

+92
-113
lines changed

‎ydb/core/kqp/common/kqp_tx.h‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -268,7 +268,7 @@ class TKqpTransactionContext : public NYql::TKikimrTransactionContextBase {
268268
}
269269

270270
bool ShouldExecuteDeferredEffects() const {
271-
if (HasUncommittedChangesRead) {
271+
if (HasUncommittedChangesRead || HasOlapTable) {
272272
YQL_ENSURE(EnableImmediateEffects);
273273
return !DeferredEffects.Empty();
274274
}
@@ -298,7 +298,7 @@ class TKqpTransactionContext : public NYql::TKikimrTransactionContextBase {
298298
}
299299

300300
bool CanDeferEffects() const {
301-
if (HasUncommittedChangesRead || AppData()->FeatureFlags.GetEnableForceImmediateEffectsExecution()) {
301+
if (HasUncommittedChangesRead || AppData()->FeatureFlags.GetEnableForceImmediateEffectsExecution() || HasOlapTable) {
302302
YQL_ENSURE(EnableImmediateEffects);
303303
return false;
304304
}

‎ydb/core/kqp/executer_actor/kqp_data_executer.cpp‎

Lines changed: 40 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -124,21 +124,19 @@ class TKqpDataExecuter : public TKqpExecuterBase<TKqpDataExecuter, EExecType::Da
124124
TKqpDataExecuter(IKqpGateway::TExecPhysicalRequest&& request, const TString& database,
125125
const TIntrusiveConstPtr<NACLib::TUserToken>& userToken,
126126
TKqpRequestCounters::TPtr counters, bool streamResult,
127-
const NKikimrConfig::TTableServiceConfig::TExecuterRetriesConfig& executerRetriesConfig,
127+
const NKikimrConfig::TTableServiceConfig& tableServiceConfig,
128128
NYql::NDq::IDqAsyncIoFactory::TPtr asyncIoFactory,
129-
const NKikimrConfig::TTableServiceConfig::EChannelTransportVersion chanTransportVersion,
130-
const NKikimrConfig::TTableServiceConfig::TAggregationConfig& aggregation,
131129
const TActorId& creator, const TIntrusivePtr<TUserRequestContext>& userRequestContext,
132-
const bool useEvWrite, ui32 statementResultIndex, const std::optional<TKqpFederatedQuerySetup>& federatedQuerySetup,
133-
const TGUCSettings::TPtr& GUCSettings, const TShardIdToTableInfoPtr& shardIdToTableInfo, const bool htapTx)
134-
: TBase(std::move(request), database, userToken, counters, executerRetriesConfig, chanTransportVersion, aggregation,
130+
ui32 statementResultIndex, const std::optional<TKqpFederatedQuerySetup>& federatedQuerySetup,
131+
const TGUCSettings::TPtr& GUCSettings, const TShardIdToTableInfoPtr& shardIdToTableInfo)
132+
: TBase(std::move(request), database, userToken, counters, tableServiceConfig,
135133
userRequestContext, statementResultIndex, TWilsonKqp::DataExecuter, "DataExecuter", streamResult)
136134
, AsyncIoFactory(std::move(asyncIoFactory))
137-
, UseEvWrite(useEvWrite)
135+
, UseEvWriteForOltp(tableServiceConfig.GetEnableOltpSink())
136+
, HtapTx(tableServiceConfig.GetEnableHtapTx())
138137
, FederatedQuerySetup(federatedQuerySetup)
139138
, GUCSettings(GUCSettings)
140139
, ShardIdToTableInfo(shardIdToTableInfo)
141-
, HtapTx(htapTx)
142140
{
143141
Target = creator;
144142

@@ -1487,7 +1485,7 @@ class TKqpDataExecuter : public TKqpExecuterBase<TKqpDataExecuter, EExecType::Da
14871485
auto& stage = stageInfo.Meta.GetStage(stageInfo.Id);
14881486

14891487
auto getShardTask = [&](ui64 shardId) -> TTask& {
1490-
YQL_ENSURE(!UseEvWrite);
1488+
YQL_ENSURE(!UseEvWriteForOltp);
14911489
auto it = shardTasks.find(shardId);
14921490
if (it != shardTasks.end()) {
14931491
return TasksGraph.GetTask(it->second);
@@ -1627,7 +1625,7 @@ class TKqpDataExecuter : public TKqpExecuterBase<TKqpDataExecuter, EExecType::Da
16271625

16281626
void ExecuteDatashardTransaction(ui64 shardId, NKikimrTxDataShard::TKqpTransaction& kqpTx, const bool isOlap)
16291627
{
1630-
YQL_ENSURE(!UseEvWrite);
1628+
YQL_ENSURE(!UseEvWriteForOltp);
16311629
TShardState shardState;
16321630
shardState.State = ImmediateTx ? TShardState::EState::Executing : TShardState::EState::Preparing;
16331631
shardState.DatashardState.ConstructInPlace();
@@ -2025,7 +2023,6 @@ class TKqpDataExecuter : public TKqpExecuterBase<TKqpDataExecuter, EExecType::Da
20252023
TDatashardTxs datashardTxs;
20262024
TEvWriteTxs evWriteTxs;
20272025
BuildDatashardTxs(datashardTasks, datashardTxs, evWriteTxs, topicTxs);
2028-
YQL_ENSURE(evWriteTxs.empty() || datashardTxs.empty());
20292026

20302027
// Single-shard datashard transactions are always immediate
20312028
ImmediateTx = (datashardTxs.size() + evWriteTxs.size() + Request.TopicOperations.GetSize() + sourceScanPartitionsCount) <= 1
@@ -2256,7 +2253,7 @@ class TKqpDataExecuter : public TKqpExecuterBase<TKqpDataExecuter, EExecType::Da
22562253
YQL_ENSURE(!locksList.empty(), "unexpected empty locks list in DataShardLocks");
22572254
NKikimrDataEvents::TKqpLocks* locks = nullptr;
22582255

2259-
if (UseEvWrite) {
2256+
if (UseEvWriteForOltp || ShardIdToTableInfo->Get(shardId).IsOlap) {
22602257
if (auto it = evWriteTxs.find(shardId); it != evWriteTxs.end()) {
22612258
locks = it->second->MutableLocks();
22622259
} else {
@@ -2328,15 +2325,17 @@ class TKqpDataExecuter : public TKqpExecuterBase<TKqpDataExecuter, EExecType::Da
23282325
// Note: currently persistent channels are never used
23292326
!HasPersistentChannels &&
23302327
// Can't use volatile transactions for EvWrite at current time
2331-
!UseEvWrite);
2328+
evWriteTxs.empty());
23322329

23332330
const bool useGenericReadSets = (
23342331
// Use generic readsets when feature is explicitly enabled
23352332
AppData()->FeatureFlags.GetEnableDataShardGenericReadSets() ||
23362333
// Volatile transactions must always use generic readsets
23372334
VolatileTx ||
23382335
// Transactions with topics must always use generic readsets
2339-
!topicTxs.empty());
2336+
!topicTxs.empty() ||
2337+
// HTAP transactions always use generic readsets
2338+
!evWriteTxs.empty());
23402339

23412340
if (!locksMap.empty() || VolatileTx ||
23422341
Request.TopicOperations.HasReadOperations() || Request.TopicOperations.HasWriteOperations())
@@ -2479,10 +2478,22 @@ class TKqpDataExecuter : public TKqpExecuterBase<TKqpDataExecuter, EExecType::Da
24792478

24802479
for (auto& [shardId, shardTx] : datashardTxs) {
24812480
shardTx->MutableLocks()->SetOp(NKikimrDataEvents::TKqpLocks::Commit);
2482-
*shardTx->MutableLocks()->MutableSendingShards() = sendingShards;
2483-
*shardTx->MutableLocks()->MutableReceivingShards() = receivingShards;
2484-
if (arbiter) {
2485-
shardTx->MutableLocks()->SetArbiterShard(arbiter);
2481+
if (columnShardArbiter) {
2482+
shardTx->MutableLocks()->AddSendingShards(*columnShardArbiter);
2483+
shardTx->MutableLocks()->AddReceivingShards(*columnShardArbiter);
2484+
if (sendingShardsSet.contains(shardId)) {
2485+
shardTx->MutableLocks()->AddSendingShards(shardId);
2486+
}
2487+
if (receivingShardsSet.contains(shardId)) {
2488+
shardTx->MutableLocks()->AddReceivingShards(shardId);
2489+
}
2490+
AFL_ENSURE(!arbiter);
2491+
} else {
2492+
*shardTx->MutableLocks()->MutableSendingShards() = sendingShards;
2493+
*shardTx->MutableLocks()->MutableReceivingShards() = receivingShards;
2494+
if (arbiter) {
2495+
shardTx->MutableLocks()->SetArbiterShard(arbiter);
2496+
}
24862497
}
24872498
}
24882499

@@ -2836,11 +2847,11 @@ class TKqpDataExecuter : public TKqpExecuterBase<TKqpDataExecuter, EExecType::Da
28362847

28372848
private:
28382849
NYql::NDq::IDqAsyncIoFactory::TPtr AsyncIoFactory;
2839-
bool UseEvWrite = false;
2850+
const bool UseEvWriteForOltp = false;
2851+
const bool HtapTx = false;
28402852
const std::optional<TKqpFederatedQuerySetup> FederatedQuerySetup;
28412853
const TGUCSettings::TPtr GUCSettings;
28422854
TShardIdToTableInfoPtr ShardIdToTableInfo;
2843-
const bool HtapTx = false;
28442855

28452856
bool HasExternalSources = false;
28462857
bool SecretSnapshotRequired = false;
@@ -2879,18 +2890,17 @@ class TKqpDataExecuter : public TKqpExecuterBase<TKqpDataExecuter, EExecType::Da
28792890

28802891
} // namespace
28812892

2882-
IActor* CreateKqpDataExecuter(IKqpGateway::TExecPhysicalRequest&& request, const TString& database, const TIntrusiveConstPtr<NACLib::TUserToken>& userToken,
2883-
TKqpRequestCounters::TPtr counters, bool streamResult, const NKikimrConfig::TTableServiceConfig::TAggregationConfig& aggregation,
2884-
const NKikimrConfig::TTableServiceConfig::TExecuterRetriesConfig& executerRetriesConfig,
2885-
NYql::NDq::IDqAsyncIoFactory::TPtr asyncIoFactory, const NKikimrConfig::TTableServiceConfig::EChannelTransportVersion chanTransportVersion, const TActorId& creator,
2886-
const TIntrusivePtr<TUserRequestContext>& userRequestContext,
2887-
const bool useEvWrite, ui32 statementResultIndex,
2893+
IActor* CreateKqpDataExecuter(IKqpGateway::TExecPhysicalRequest&& request, const TString& database,
2894+
const TIntrusiveConstPtr<NACLib::TUserToken>& userToken, TKqpRequestCounters::TPtr counters, bool streamResult,
2895+
const NKikimrConfig::TTableServiceConfig& tableServiceConfig,
2896+
NYql::NDq::IDqAsyncIoFactory::TPtr asyncIoFactory,
2897+
const TActorId& creator, const TIntrusivePtr<TUserRequestContext>& userRequestContext, ui32 statementResultIndex,
28882898
const std::optional<TKqpFederatedQuerySetup>& federatedQuerySetup, const TGUCSettings::TPtr& GUCSettings,
2889-
const TShardIdToTableInfoPtr& shardIdToTableInfo, const bool htapTx)
2899+
const TShardIdToTableInfoPtr& shardIdToTableInfo)
28902900
{
2891-
return new TKqpDataExecuter(std::move(request), database, userToken, counters, streamResult, executerRetriesConfig,
2892-
std::move(asyncIoFactory), chanTransportVersion, aggregation, creator, userRequestContext,
2893-
useEvWrite, statementResultIndex, federatedQuerySetup, GUCSettings, shardIdToTableInfo, htapTx);
2901+
return new TKqpDataExecuter(std::move(request), database, userToken, counters, streamResult,
2902+
tableServiceConfig, std::move(asyncIoFactory), creator, userRequestContext,
2903+
statementResultIndex, federatedQuerySetup, GUCSettings, shardIdToTableInfo);
28942904
}
28952905

28962906
} // namespace NKqp

‎ydb/core/kqp/executer_actor/kqp_executer.h‎

Lines changed: 3 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -102,14 +102,11 @@ struct TKqpFederatedQuerySetup;
102102

103103
IActor* CreateKqpExecuter(IKqpGateway::TExecPhysicalRequest&& request, const TString& database,
104104
const TIntrusiveConstPtr<NACLib::TUserToken>& userToken, TKqpRequestCounters::TPtr counters,
105-
const NKikimrConfig::TTableServiceConfig::TAggregationConfig& aggregation,
106-
const NKikimrConfig::TTableServiceConfig::TExecuterRetriesConfig& executerRetriesConfig,
105+
const NKikimrConfig::TTableServiceConfig& tableServiceConfig,
107106
NYql::NDq::IDqAsyncIoFactory::TPtr asyncIoFactory, TPreparedQueryHolder::TConstPtr preparedQuery,
108-
const NKikimrConfig::TTableServiceConfig::EChannelTransportVersion chanTransportVersion, const TActorId& creator,
109-
const TIntrusivePtr<TUserRequestContext>& userRequestContext,
110-
const bool useEvWrite, ui32 statementResultIndex,
107+
const TActorId& creator, const TIntrusivePtr<TUserRequestContext>& userRequestContext, ui32 statementResultIndex,
111108
const std::optional<TKqpFederatedQuerySetup>& federatedQuerySetup, const TGUCSettings::TPtr& GUCSettings,
112-
const TShardIdToTableInfoPtr& shardIdToTableInfo, const bool htapTx);
109+
const TShardIdToTableInfoPtr& shardIdToTableInfo);
113110

114111
IActor* CreateKqpSchemeExecuter(
115112
TKqpPhyTxHolder::TConstPtr phyTx, NKikimrKqp::EQueryType queryType, const TActorId& target,

‎ydb/core/kqp/executer_actor/kqp_executer_impl.cpp‎

Lines changed: 14 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -78,23 +78,19 @@ TActorId ReportToRl(ui64 ru, const TString& database, const TString& userToken,
7878

7979
IActor* CreateKqpExecuter(IKqpGateway::TExecPhysicalRequest&& request, const TString& database,
8080
const TIntrusiveConstPtr<NACLib::TUserToken>& userToken, TKqpRequestCounters::TPtr counters,
81-
const NKikimrConfig::TTableServiceConfig::TAggregationConfig& aggregation,
82-
const NKikimrConfig::TTableServiceConfig::TExecuterRetriesConfig& executerRetriesConfig,
81+
const NKikimrConfig::TTableServiceConfig& tableServiceConfig,
8382
NYql::NDq::IDqAsyncIoFactory::TPtr asyncIoFactory, TPreparedQueryHolder::TConstPtr preparedQuery,
84-
const NKikimrConfig::TTableServiceConfig::EChannelTransportVersion chanTransportVersion, const TActorId& creator,
85-
const TIntrusivePtr<TUserRequestContext>& userRequestContext,
86-
const bool useEvWrite, ui32 statementResultIndex,
83+
const TActorId& creator, const TIntrusivePtr<TUserRequestContext>& userRequestContext, ui32 statementResultIndex,
8784
const std::optional<TKqpFederatedQuerySetup>& federatedQuerySetup, const TGUCSettings::TPtr& GUCSettings,
88-
const TShardIdToTableInfoPtr& shardIdToTableInfo, const bool htapTx)
85+
const TShardIdToTableInfoPtr& shardIdToTableInfo)
8986
{
9087
if (request.Transactions.empty()) {
9188
// commit-only or rollback-only data transaction
9289
return CreateKqpDataExecuter(
9390
std::move(request), database, userToken, counters, false,
94-
aggregation, executerRetriesConfig, std::move(asyncIoFactory), chanTransportVersion, creator,
95-
userRequestContext, useEvWrite, statementResultIndex,
96-
federatedQuerySetup, /*GUCSettings*/nullptr, shardIdToTableInfo, htapTx
97-
);
91+
tableServiceConfig, std::move(asyncIoFactory), creator,
92+
userRequestContext, statementResultIndex,
93+
federatedQuerySetup, /*GUCSettings*/nullptr, shardIdToTableInfo);
9894
}
9995

10096
TMaybe<NKqpProto::TKqpPhyTx::EType> txsType;
@@ -114,24 +110,24 @@ IActor* CreateKqpExecuter(IKqpGateway::TExecPhysicalRequest&& request, const TSt
114110
case NKqpProto::TKqpPhyTx::TYPE_DATA:
115111
return CreateKqpDataExecuter(
116112
std::move(request), database, userToken, counters, false,
117-
aggregation, executerRetriesConfig, std::move(asyncIoFactory), chanTransportVersion, creator,
118-
userRequestContext, useEvWrite, statementResultIndex,
119-
federatedQuerySetup, /*GUCSettings*/nullptr, shardIdToTableInfo, htapTx
113+
tableServiceConfig, std::move(asyncIoFactory), creator,
114+
userRequestContext, statementResultIndex,
115+
federatedQuerySetup, /*GUCSettings*/nullptr, shardIdToTableInfo
120116
);
121117

122118
case NKqpProto::TKqpPhyTx::TYPE_SCAN:
123119
return CreateKqpScanExecuter(
124-
std::move(request), database, userToken, counters, aggregation,
125-
executerRetriesConfig, preparedQuery, chanTransportVersion, userRequestContext,
120+
std::move(request), database, userToken, counters,
121+
tableServiceConfig, preparedQuery, userRequestContext,
126122
statementResultIndex
127123
);
128124

129125
case NKqpProto::TKqpPhyTx::TYPE_GENERIC:
130126
return CreateKqpDataExecuter(
131127
std::move(request), database, userToken, counters, true,
132-
aggregation, executerRetriesConfig, std::move(asyncIoFactory), chanTransportVersion, creator,
133-
userRequestContext, useEvWrite, statementResultIndex,
134-
federatedQuerySetup, GUCSettings, shardIdToTableInfo, htapTx
128+
tableServiceConfig, std::move(asyncIoFactory), creator,
129+
userRequestContext, statementResultIndex,
130+
federatedQuerySetup, GUCSettings, shardIdToTableInfo
135131
);
136132

137133
default:

‎ydb/core/kqp/executer_actor/kqp_executer_impl.h‎

Lines changed: 8 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -123,9 +123,7 @@ class TKqpExecuterBase : public TActorBootstrapped<TDerived> {
123123
TKqpExecuterBase(IKqpGateway::TExecPhysicalRequest&& request, const TString& database,
124124
const TIntrusiveConstPtr<NACLib::TUserToken>& userToken,
125125
TKqpRequestCounters::TPtr counters,
126-
const NKikimrConfig::TTableServiceConfig::TExecuterRetriesConfig& executerRetriesConfig,
127-
const NKikimrConfig::TTableServiceConfig::EChannelTransportVersion chanTransportVersion,
128-
const NKikimrConfig::TTableServiceConfig::TAggregationConfig& aggregation,
126+
const NKikimrConfig::TTableServiceConfig tableServiceConfig,
129127
const TIntrusivePtr<TUserRequestContext>& userRequestContext,
130128
ui32 statementResultIndex, ui64 spanVerbosity = 0, TString spanName = "KqpExecuterBase", bool streamResult = false)
131129
: Request(std::move(request))
@@ -134,16 +132,16 @@ class TKqpExecuterBase : public TActorBootstrapped<TDerived> {
134132
, Counters(counters)
135133
, ExecuterSpan(spanVerbosity, std::move(Request.TraceId), spanName)
136134
, Planner(nullptr)
137-
, ExecuterRetriesConfig(executerRetriesConfig)
138-
, AggregationSettings(aggregation)
135+
, ExecuterRetriesConfig(tableServiceConfig.GetExecuterRetriesConfig())
136+
, AggregationSettings(tableServiceConfig.GetAggregationConfig())
139137
, HasOlapTable(false)
140138
, StreamResult(streamResult)
141139
, StatementResultIndex(statementResultIndex)
142140
{
143141
TasksGraph.GetMeta().Snapshot = IKqpGateway::TKqpSnapshot(Request.Snapshot.Step, Request.Snapshot.TxId);
144142
TasksGraph.GetMeta().Arena = MakeIntrusive<NActors::TProtoArenaHolder>();
145143
TasksGraph.GetMeta().Database = Database;
146-
TasksGraph.GetMeta().ChannelTransportVersion = chanTransportVersion;
144+
TasksGraph.GetMeta().ChannelTransportVersion = tableServiceConfig.GetChannelTransportVersion();
147145
TasksGraph.GetMeta().UserRequestContext = userRequestContext;
148146
ResponseEv = std::make_unique<TEvKqpExecuter::TEvTxResponse>(Request.TxAlloc, ExecType);
149147
ResponseEv->Orbit = std::move(Request.Orbit);
@@ -1993,20 +1991,15 @@ class TKqpExecuterBase : public TActorBootstrapped<TDerived> {
19931991

19941992
IActor* CreateKqpDataExecuter(IKqpGateway::TExecPhysicalRequest&& request, const TString& database,
19951993
const TIntrusiveConstPtr<NACLib::TUserToken>& userToken, TKqpRequestCounters::TPtr counters, bool streamResult,
1996-
const NKikimrConfig::TTableServiceConfig::TAggregationConfig& aggregation,
1997-
const NKikimrConfig::TTableServiceConfig::TExecuterRetriesConfig& executerRetriesConfig,
1994+
const NKikimrConfig::TTableServiceConfig& tableServiceConfig,
19981995
NYql::NDq::IDqAsyncIoFactory::TPtr asyncIoFactory,
1999-
const NKikimrConfig::TTableServiceConfig::EChannelTransportVersion chanTransportVersion, const TActorId& creator,
2000-
const TIntrusivePtr<TUserRequestContext>& userRequestContext,
2001-
const bool useEvWrite, ui32 statementResultIndex,
1996+
const TActorId& creator, const TIntrusivePtr<TUserRequestContext>& userRequestContext, ui32 statementResultIndex,
20021997
const std::optional<TKqpFederatedQuerySetup>& federatedQuerySetup, const TGUCSettings::TPtr& GUCSettings,
2003-
const TShardIdToTableInfoPtr& shardIdToTableInfo, const bool htapTx);
1998+
const TShardIdToTableInfoPtr& shardIdToTableInfo);
20041999

20052000
IActor* CreateKqpScanExecuter(IKqpGateway::TExecPhysicalRequest&& request, const TString& database,
20062001
const TIntrusiveConstPtr<NACLib::TUserToken>& userToken, TKqpRequestCounters::TPtr counters,
2007-
const NKikimrConfig::TTableServiceConfig::TAggregationConfig& aggregation,
2008-
const NKikimrConfig::TTableServiceConfig::TExecuterRetriesConfig& executerRetriesConfig,
2009-
TPreparedQueryHolder::TConstPtr preparedQuery, const NKikimrConfig::TTableServiceConfig::EChannelTransportVersion chanTransportVersion,
2002+
const NKikimrConfig::TTableServiceConfig& tableServiceConfig, TPreparedQueryHolder::TConstPtr preparedQuery,
20102003
const TIntrusivePtr<TUserRequestContext>& userRequestContext, ui32 statementResultIndex);
20112004

20122005
} // namespace NKqp

0 commit comments

Comments
 (0)