Skip to content

[Stable-24-3] Fixes for sinks #10655

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions ydb/core/kqp/counters/kqp_counters.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -814,6 +814,18 @@ TKqpCounters::TKqpCounters(const ::NMonitoring::TDynamicCounterPtr& counters, co
DataShardIteratorMessages = KqpGroup->GetCounter("IteratorReads/DatashardMessages", true);
IteratorDeliveryProblems = KqpGroup->GetCounter("IteratorReads/DeliveryProblems", true);

/* sink writes */
WriteActorsShardResolve = KqpGroup->GetCounter("SinkWrites/WriteActorShardResolve", true);
WriteActorsCount = KqpGroup->GetCounter("SinkWrites/WriteActorsCount", false);
WriteActorImmediateWrites = KqpGroup->GetCounter("SinkWrites/WriteActorImmediateWrites", true);
WriteActorImmediateWritesRetries = KqpGroup->GetCounter("SinkWrites/WriteActorImmediateWritesRetries", true);
WriteActorWritesSizeHistogram =
KqpGroup->GetHistogram("SinkWrites/WriteActorWritesSize", NMonitoring::ExponentialHistogram(28, 2, 1));
WriteActorWritesOperationsHistogram =
KqpGroup->GetHistogram("SinkWrites/WriteActorWritesOperations", NMonitoring::ExponentialHistogram(20, 2, 1));
WriteActorWritesLatencyHistogram =
KqpGroup->GetHistogram("SinkWrites/WriteActorWritesLatencyMs", NMonitoring::ExponentialHistogram(20, 2, 1));

/* sequencers */

SequencerActorsCount = KqpGroup->GetCounter("Sequencer/ActorCount", false);
Expand Down
9 changes: 9 additions & 0 deletions ydb/core/kqp/counters/kqp_counters.h
Original file line number Diff line number Diff line change
Expand Up @@ -409,6 +409,15 @@ class TKqpCounters : public TKqpCountersBase, public NYql::NDq::TSpillingCounter
::NMonitoring::TDynamicCounters::TCounterPtr DataShardIteratorMessages;
::NMonitoring::TDynamicCounters::TCounterPtr IteratorDeliveryProblems;

// Sink write counters
::NMonitoring::TDynamicCounters::TCounterPtr WriteActorsShardResolve;
::NMonitoring::TDynamicCounters::TCounterPtr WriteActorsCount;
::NMonitoring::TDynamicCounters::TCounterPtr WriteActorImmediateWrites;
::NMonitoring::TDynamicCounters::TCounterPtr WriteActorImmediateWritesRetries;
NMonitoring::THistogramPtr WriteActorWritesSizeHistogram;
NMonitoring::THistogramPtr WriteActorWritesOperationsHistogram;
NMonitoring::THistogramPtr WriteActorWritesLatencyHistogram;

// Scheduler signals
::NMonitoring::TDynamicCounters::TCounterPtr SchedulerThrottled;
::NMonitoring::TDynamicCounters::TCounterPtr SchedulerCapacity;
Expand Down
7 changes: 3 additions & 4 deletions ydb/core/kqp/expr_nodes/kqp_expr_nodes.json
Original file line number Diff line number Diff line change
Expand Up @@ -546,10 +546,9 @@
"Match": {"Type": "Callable", "Name": "KqpTableSinkSettings"},
"Children": [
{"Index": 0, "Name": "Table", "Type": "TKqpTable"},
{"Index": 1, "Name": "Columns", "Type": "TCoAtomList"},
{"Index": 2, "Name": "InconsistentWrite", "Type": "TCoAtom"},
{"Index": 3, "Name": "Mode", "Type": "TCoAtom"},
{"Index": 4, "Name": "Settings", "Type": "TCoNameValueTupleList", "Optional": true}
{"Index": 1, "Name": "InconsistentWrite", "Type": "TCoAtom"},
{"Index": 2, "Name": "Mode", "Type": "TCoAtom"},
{"Index": 3, "Name": "Settings", "Type": "TCoNameValueTupleList", "Optional": true}
]
},
{
Expand Down
10 changes: 3 additions & 7 deletions ydb/core/kqp/opt/kqp_opt_effects.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -231,7 +231,7 @@ TCoAtomList BuildKeyColumnsList(const TKikimrTableDescription& table, TPositionH
.Done();
}

TDqStage RebuildPureStageWithSink(TExprBase expr, const TKqpTable& table, const TCoAtomList& columns,
TDqStage RebuildPureStageWithSink(TExprBase expr, const TKqpTable& table,
const bool allowInconsistentWrites, const TStringBuf mode, TExprContext& ctx) {
Y_DEBUG_ABORT_UNLESS(IsDqPureExpr(expr));

Expand All @@ -253,7 +253,6 @@ TDqStage RebuildPureStageWithSink(TExprBase expr, const TKqpTable& table, const
.Index().Value("0").Build()
.Settings<TKqpTableSinkSettings>()
.Table(table)
.Columns(columns)
.InconsistentWrite(allowInconsistentWrites
? ctx.NewAtom(expr.Pos(), "true")
: ctx.NewAtom(expr.Pos(), "false"))
Expand Down Expand Up @@ -311,7 +310,7 @@ bool BuildUpsertRowsEffect(const TKqlUpsertRows& node, TExprContext& ctx, const
if (IsDqPureExpr(node.Input())) {
if (sinkEffect) {
stageInput = RebuildPureStageWithSink(
node.Input(), node.Table(), node.Columns(),
node.Input(), node.Table(),
settings.AllowInconsistentWrites, settings.Mode, ctx);
effect = Build<TKqpSinkEffect>(ctx, node.Pos())
.Stage(stageInput.Cast().Ptr())
Expand Down Expand Up @@ -349,7 +348,6 @@ bool BuildUpsertRowsEffect(const TKqlUpsertRows& node, TExprContext& ctx, const
.Index().Value("0").Build()
.Settings<TKqpTableSinkSettings>()
.Table(node.Table())
.Columns(node.Columns())
.InconsistentWrite(settings.AllowInconsistentWrites
? ctx.NewAtom(node.Pos(), "true")
: ctx.NewAtom(node.Pos(), "false"))
Expand Down Expand Up @@ -459,7 +457,7 @@ bool BuildDeleteRowsEffect(const TKqlDeleteRows& node, TExprContext& ctx, const
if (IsDqPureExpr(node.Input())) {
if (sinkEffect) {
const auto keyColumns = BuildKeyColumnsList(table, node.Pos(), ctx);
stageInput = RebuildPureStageWithSink(node.Input(), node.Table(), keyColumns, false, "delete", ctx);
stageInput = RebuildPureStageWithSink(node.Input(), node.Table(), false, "delete", ctx);
effect = Build<TKqpSinkEffect>(ctx, node.Pos())
.Stage(stageInput.Cast().Ptr())
.SinkIndex().Build("0")
Expand All @@ -486,7 +484,6 @@ bool BuildDeleteRowsEffect(const TKqlDeleteRows& node, TExprContext& ctx, const
auto input = dqUnion.Output().Stage().Program().Body();

if (sinkEffect) {
const auto keyColumns = BuildKeyColumnsList(table, node.Pos(), ctx);
auto sink = Build<TDqSink>(ctx, node.Pos())
.DataSink<TKqpTableSink>()
.Category(ctx.NewAtom(node.Pos(), NYql::KqpTableSinkName))
Expand All @@ -495,7 +492,6 @@ bool BuildDeleteRowsEffect(const TKqlDeleteRows& node, TExprContext& ctx, const
.Index().Value("0").Build()
.Settings<TKqpTableSinkSettings>()
.Table(node.Table())
.Columns(keyColumns)
.InconsistentWrite(ctx.NewAtom(node.Pos(), "false"))
.Mode(ctx.NewAtom(node.Pos(), "delete"))
.Settings()
Expand Down
45 changes: 34 additions & 11 deletions ydb/core/kqp/query_compiler/kqp_query_compiler.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -141,6 +141,15 @@ void FillTablesMap(const TKqpTable& table, const TCoAtomList& columns,
THashMap<TStringBuf, THashSet<TStringBuf>>& tablesMap)
{
FillTablesMap(table, tablesMap);
for (const auto& column : columns) {
tablesMap[table.Path()].emplace(column);
}
}

void FillTablesMap(const TKqpTable& table, const TVector<TStringBuf>& columns,
THashMap<TStringBuf, THashSet<TStringBuf>>& tablesMap)
{
FillTablesMap(table, tablesMap);

for (const auto& column : columns) {
tablesMap[table.Path()].emplace(column);
Expand Down Expand Up @@ -763,7 +772,7 @@ class TKqpQueryCompiler : public IKqpQueryCompiler {
YQL_ENSURE(maybeSinkNode);
auto sinkNode = maybeSinkNode.Cast();
auto* sinkProto = stageProto.AddSinks();
FillSink(sinkNode, sinkProto, tablesMap, ctx);
FillSink(sinkNode, sinkProto, tablesMap, stage, ctx);
sinkProto->SetOutputIndex(FromString(TStringBuf(sinkNode.Index())));

if (IsTableSink(sinkNode.DataSink().Cast<TCoDataSink>().Category())) {
Expand Down Expand Up @@ -1029,23 +1038,38 @@ class TKqpQueryCompiler : public IKqpQueryCompiler {
}
}

void FillKqpSink(const TDqSink& sink, NKqpProto::TKqpSink* protoSink, THashMap<TStringBuf, THashSet<TStringBuf>>& tablesMap) {
void FillKqpSink(const TDqSink& sink, NKqpProto::TKqpSink* protoSink, THashMap<TStringBuf, THashSet<TStringBuf>>& tablesMap, const TDqPhyStage& stage) {
if (auto settings = sink.Settings().Maybe<TKqpTableSinkSettings>()) {
NKqpProto::TKqpInternalSink& internalSinkProto = *protoSink->MutableInternalSink();
internalSinkProto.SetType(TString(NYql::KqpTableSinkName));
NKikimrKqp::TKqpTableSinkSettings settingsProto;
FillTablesMap(settings.Table().Cast(), settings.Columns().Cast(), tablesMap);

const auto& tupleType = stage.Ref().GetTypeAnn()->Cast<TTupleExprType>();
YQL_ENSURE(tupleType);
YQL_ENSURE(tupleType->GetSize() == 1);
const auto& listType = tupleType->GetItems()[0]->Cast<TListExprType>();
YQL_ENSURE(listType);
const auto& structType = listType->GetItemType()->Cast<TStructExprType>();
YQL_ENSURE(structType);

TVector<TStringBuf> columns;
columns.reserve(structType->GetSize());
for (const auto& item : structType->GetItems()) {
columns.emplace_back(item->GetName());
}

FillTablesMap(settings.Table().Cast(), columns, tablesMap);
FillTableId(settings.Table().Cast(), *settingsProto.MutableTable());

const auto tableMeta = TablesData->ExistingTable(Cluster, settings.Table().Cast().Path()).Metadata;

for (const auto& columnName : tableMeta->KeyColumnNames) {
const auto columnMeta = tableMeta->Columns.FindPtr(columnName);
YQL_ENSURE(columnMeta != nullptr, "Unknown column in sink: \"" + columnName + "\"");
YQL_ENSURE(columnMeta != nullptr, "Unknown column in sink: \"" + TString(columnName) + "\"");

auto keyColumnProto = settingsProto.AddKeyColumns();
keyColumnProto->SetId(columnMeta->Id);
keyColumnProto->SetName(columnName);
keyColumnProto->SetName(TString(columnName));
keyColumnProto->SetTypeId(columnMeta->TypeInfo.GetTypeId());

if (columnMeta->TypeInfo.GetTypeId() == NScheme::NTypeIds::Pg) {
Expand All @@ -1055,14 +1079,13 @@ class TKqpQueryCompiler : public IKqpQueryCompiler {
}
}

for (const auto& column : settings.Columns().Cast()) {
const auto columnName = column.StringValue();
for (const auto& columnName : columns) {
const auto columnMeta = tableMeta->Columns.FindPtr(columnName);
YQL_ENSURE(columnMeta != nullptr, "Unknown column in sink: \"" + columnName + "\"");
YQL_ENSURE(columnMeta != nullptr, "Unknown column in sink: \"" + TString(columnName) + "\"");

auto columnProto = settingsProto.AddColumns();
columnProto->SetId(columnMeta->Id);
columnProto->SetName(columnName);
columnProto->SetName(TString(columnName));
columnProto->SetTypeId(columnMeta->TypeInfo.GetTypeId());

if (columnMeta->TypeInfo.GetTypeId() == NScheme::NTypeIds::Pg) {
Expand Down Expand Up @@ -1102,11 +1125,11 @@ class TKqpQueryCompiler : public IKqpQueryCompiler {
|| dataSinkCategory == NYql::KqpTableSinkName;
}

void FillSink(const TDqSink& sink, NKqpProto::TKqpSink* protoSink, THashMap<TStringBuf, THashSet<TStringBuf>>& tablesMap, TExprContext& ctx) {
void FillSink(const TDqSink& sink, NKqpProto::TKqpSink* protoSink, THashMap<TStringBuf, THashSet<TStringBuf>>& tablesMap, const TDqPhyStage& stage, TExprContext& ctx) {
Y_UNUSED(ctx);
const TStringBuf dataSinkCategory = sink.DataSink().Cast<TCoDataSink>().Category();
if (IsTableSink(dataSinkCategory)) {
FillKqpSink(sink, protoSink, tablesMap);
FillKqpSink(sink, protoSink, tablesMap, stage);
} else {
// Delegate sink filling to dq integration of specific provider
const auto provider = TypesCtx.DataSinkMap.find(dataSinkCategory);
Expand Down
45 changes: 41 additions & 4 deletions ydb/core/kqp/runtime/kqp_write_actor.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
#include <ydb/core/tx/tx.h>
#include <ydb/library/actors/core/actorsystem.h>
#include <ydb/library/actors/core/interconnect.h>
#include <ydb/library/wilson_ids/wilson.h>
#include <ydb/library/yql/dq/actors/compute/dq_compute_actor_impl.h>
#include <ydb/library/yql/public/issue/yql_issue_message.h>

Expand Down Expand Up @@ -134,10 +135,13 @@ class TKqpDirectWriteActor : public TActorBootstrapped<TKqpDirectWriteActor>, pu
, InconsistentTx(
Settings.GetInconsistentTx())
, MemoryLimit(MessageSettings.InFlightMemoryLimitPerActorBytes)
, WriteActorSpan(TWilsonKqp::WriteActor, NWilson::TTraceId(args.TraceId), "WriteActor")
{
YQL_ENSURE(std::holds_alternative<ui64>(TxId));
YQL_ENSURE(!ImmediateTx);
EgressStats.Level = args.StatsLevel;

Counters->WriteActorsCount->Inc();
}

void Bootstrap() {
Expand Down Expand Up @@ -244,6 +248,7 @@ class TKqpDirectWriteActor : public TActorBootstrapped<TKqpDirectWriteActor>, pu
}

void ResolveTable() {
Counters->WriteActorsShardResolve->Inc();
SchemeEntry.reset();
SchemeRequest.reset();

Expand All @@ -267,8 +272,11 @@ class TKqpDirectWriteActor : public TActorBootstrapped<TKqpDirectWriteActor>, pu
entry.ShowPrivatePath = true;
request->ResultSet.emplace_back(entry);

Send(MakeSchemeCacheID(), new TEvTxProxySchemeCache::TEvInvalidateTable(TableId, {}));
Send(MakeSchemeCacheID(), new TEvTxProxySchemeCache::TEvNavigateKeySet(request));
WriteActorStateSpan = NWilson::TSpan(TWilsonKqp::WriteActorTableNavigate, WriteActorSpan.GetTraceId(),
"WaitForShardsResolve", NWilson::EFlags::AUTO_END);

Send(MakeSchemeCacheID(), new TEvTxProxySchemeCache::TEvInvalidateTable(TableId, {}), 0, 0, WriteActorSpan.GetTraceId());
Send(MakeSchemeCacheID(), new TEvTxProxySchemeCache::TEvNavigateKeySet(request), 0, 0, WriteActorSpan.GetTraceId());
}

void Handle(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr& ev) {
Expand Down Expand Up @@ -327,7 +335,7 @@ class TKqpDirectWriteActor : public TActorBootstrapped<TKqpDirectWriteActor>, pu
request->ResultSet.emplace_back(std::move(keyRange));

TAutoPtr<TEvTxProxySchemeCache::TEvResolveKeySet> resolveReq(new TEvTxProxySchemeCache::TEvResolveKeySet(request));
Send(MakeSchemeCacheID(), resolveReq.Release(), 0, 0);
Send(MakeSchemeCacheID(), resolveReq.Release(), 0, 0, WriteActorSpan.GetTraceId());
}

void Handle(TEvTxProxySchemeCache::TEvResolveKeySetResult::TPtr& ev) {
Expand Down Expand Up @@ -368,6 +376,8 @@ class TKqpDirectWriteActor : public TActorBootstrapped<TKqpDirectWriteActor>, pu
}()
<< ", Cookie=" << ev->Cookie);



switch (ev->Get()->GetStatus()) {
case NKikimrDataEvents::TEvWriteResult::STATUS_UNSPECIFIED: {
CA_LOG_E("Got UNSPECIFIED for table `"
Expand Down Expand Up @@ -542,6 +552,11 @@ class TKqpDirectWriteActor : public TActorBootstrapped<TKqpDirectWriteActor>, pu
EgressStats.Chunks++;
EgressStats.Splits++;
EgressStats.Resume();

if (auto it = SendTime.find(shardId); it != std::end(SendTime)) {
Counters->WriteActorWritesLatencyHistogram->Collect((TInstant::Now() - it->second).MilliSeconds());
SendTime.erase(it);
}
}
resumeNotificator.CheckMemory();
}
Expand Down Expand Up @@ -579,7 +594,6 @@ class TKqpDirectWriteActor : public TActorBootstrapped<TKqpDirectWriteActor>, pu
NYql::NDqProto::StatusIds::UNAVAILABLE);
return;
}

auto evWrite = std::make_unique<NKikimr::NEvents::TDataEvents::TEvWrite>(
NKikimrDataEvents::TEvWrite::MODE_IMMEDIATE);

Expand Down Expand Up @@ -613,6 +627,16 @@ class TKqpDirectWriteActor : public TActorBootstrapped<TKqpDirectWriteActor>, pu
ShardedWriteController->GetDataFormat());
}

if (metadata->SendAttempts == 0) {
Counters->WriteActorImmediateWrites->Inc();
Counters->WriteActorWritesSizeHistogram->Collect(serializationResult.TotalDataSize);
Counters->WriteActorWritesOperationsHistogram->Collect(metadata->OperationsCount);

SendTime[shardId] = TInstant::Now();
} else {
Counters->WriteActorImmediateWritesRetries->Inc();
}

CA_LOG_D("Send EvWrite to ShardID=" << shardId << ", TxId=" << evWrite->Record.GetTxId()
<< ", TxMode=" << evWrite->Record.GetTxMode()
<< ", LockTxId=" << evWrite->Record.GetLockTxId() << ", LockNodeId=" << evWrite->Record.GetLockNodeId()
Expand Down Expand Up @@ -708,6 +732,13 @@ class TKqpDirectWriteActor : public TActorBootstrapped<TKqpDirectWriteActor>, pu
NYql::TIssues issues;
issues.AddIssue(std::move(issue));

if (WriteActorStateSpan) {
WriteActorStateSpan.EndError(issues.ToOneLineString());
}
if (WriteActorSpan) {
WriteActorSpan.EndError(issues.ToOneLineString());
}

Callbacks->OnAsyncOutputError(OutputIndex, std::move(issues), statusCode);
}

Expand All @@ -717,6 +748,8 @@ class TKqpDirectWriteActor : public TActorBootstrapped<TKqpDirectWriteActor>, pu
}

void Prepare() {
WriteActorStateSpan.EndOk();

YQL_ENSURE(SchemeEntry);
ResolveAttempts = 0;

Expand Down Expand Up @@ -788,12 +821,16 @@ class TKqpDirectWriteActor : public TActorBootstrapped<TKqpDirectWriteActor>, pu
std::optional<NSchemeCache::TSchemeCacheRequest::TEntry> SchemeRequest;
ui64 ResolveAttempts = 0;

THashMap<ui64, TInstant> SendTime;
THashMap<ui64, TLockInfo> LocksInfo;
bool Finished = false;

const i64 MemoryLimit;

IShardedWriteControllerPtr ShardedWriteController = nullptr;

NWilson::TSpan WriteActorSpan;
NWilson::TSpan WriteActorStateSpan;
};

void RegisterKqpWriteActor(NYql::NDq::TDqAsyncIoFactory& factory, TIntrusivePtr<TKqpCounters> counters) {
Expand Down
11 changes: 8 additions & 3 deletions ydb/core/kqp/runtime/kqp_write_table.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -893,14 +893,17 @@ class TShardsInfo {

void MakeNextBatches(i64 maxDataSize, ui64 maxCount) {
YQL_ENSURE(BatchesInFlight == 0);
YQL_ENSURE(!IsEmpty());
i64 dataSize = 0;
// For columnshard batch can be slightly larger than the limit.
while (BatchesInFlight < maxCount
&& BatchesInFlight < Batches.size()
&& dataSize + GetBatch(BatchesInFlight)->GetMemory() <= maxDataSize) {
&& (dataSize + GetBatch(BatchesInFlight)->GetMemory() <= maxDataSize || BatchesInFlight == 0)) {
dataSize += GetBatch(BatchesInFlight)->GetMemory();
++BatchesInFlight;
}
YQL_ENSURE(BatchesInFlight == Batches.size() || GetBatch(BatchesInFlight)->GetMemory() <= maxDataSize);
YQL_ENSURE(BatchesInFlight != 0);
YQL_ENSURE(BatchesInFlight == maxCount || BatchesInFlight == Batches.size() || dataSize + GetBatch(BatchesInFlight)->GetMemory() >= maxDataSize);
}

const IPayloadSerializer::IBatchPtr& GetBatch(size_t index) const {
Expand Down Expand Up @@ -1204,7 +1207,9 @@ class TShardedWriteController : public IShardedWriteController {
if (force) {
for (auto& [shardId, batches] : Serializer->FlushBatchesForce()) {
for (auto& batch : batches) {
ShardsInfo.GetShard(shardId).PushBatch(std::move(batch));
if (batch && !batch->IsEmpty()) {
ShardsInfo.GetShard(shardId).PushBatch(std::move(batch));
}
}
}
} else {
Expand Down
Loading
Loading