Skip to content

Commit 0d22c0d

Browse files
nikvas0nikvas0
committed
fix
1 parent 192b9be commit 0d22c0d

File tree

8 files changed

+106
-121
lines changed

8 files changed

+106
-121
lines changed

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

Lines changed: 0 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -283,7 +283,6 @@ bool BuildUpsertRowsEffect(const TKqlUpsertRows& node, TExprContext& ctx, const
283283
const auto& table = kqpCtx.Tables->ExistingTable(kqpCtx.Cluster, node.Table().Path());
284284

285285
sinkEffect = kqpCtx.IsGenericQuery() && table.Metadata->Kind == EKikimrTableKind::Olap;
286-
YQL_CLOG(INFO, ProviderKqp) << "#### SINK EFFECT: " << static_cast<int>(sinkEffect);
287286

288287
TKqpUpsertRowsSettings settings;
289288
if (node.Settings()) {
@@ -319,12 +318,6 @@ bool BuildUpsertRowsEffect(const TKqlUpsertRows& node, TExprContext& ctx, const
319318
auto program = stage.Program();
320319
auto input = program.Body();
321320

322-
YQL_CLOG(INFO, ProviderKqp) << "#### DQUNION: " << KqpExprToPrettyString(dqUnion, ctx);
323-
YQL_CLOG(INFO, ProviderKqp) << "#### dqUnion.Output().Stage(): " << KqpExprToPrettyString(dqUnion.Output().Stage(), ctx);
324-
YQL_CLOG(INFO, ProviderKqp) << "#### program: " << KqpExprToPrettyString(program, ctx);
325-
YQL_CLOG(INFO, ProviderKqp) << "#### input: " << KqpExprToPrettyString(input, ctx);
326-
YQL_CLOG(INFO, ProviderKqp) << "#### TABLE: " << KqpExprToPrettyString(node.Table(), ctx);
327-
328321
if (sinkEffect) {
329322
stageInput = Build<TDqStage>(ctx, node.Pos())
330323
.Inputs(stage.Inputs())

‎ydb/core/kqp/provider/yql_kikimr_datasink.cpp‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -614,8 +614,8 @@ class TKikimrDataSink : public TDataProviderBase
614614
return true;
615615
}
616616

617-
if (mode != "replace") {
618-
ctx.AddError(TIssue(ctx.GetPosition(node->Pos()), TStringBuilder() << "Write mode '" << static_cast<TStringBuf>(mode) << "' is not supported for olap tables"));
617+
if (mode != "replace" && mode != "drop" && mode != "drop_if_exists") {
618+
ctx.AddError(TIssue(ctx.GetPosition(node->Pos()), TStringBuilder() << "Write mode '" << static_cast<TStringBuf>(mode) << "' is not supported for olap tables."));
619619
return false;
620620
}
621621

‎ydb/core/kqp/runtime/kqp_write_actor.cpp‎

Lines changed: 97 additions & 102 deletions
Original file line numberDiff line numberDiff line change
@@ -80,7 +80,6 @@ class TKqpWriteActor : public TActorBootstrapped<TKqpWriteActor>, public NYql::N
8080
, Callbacks(args.Callback)
8181
, Counters(counters)
8282
, TypeEnv(args.TypeEnv)
83-
, Alloc(args.Alloc)
8483
, TxId(args.TxId)
8584
, TableId(Settings.GetTable().GetOwnerId(), Settings.GetTable().GetTableId())
8685
{
@@ -103,10 +102,6 @@ class TKqpWriteActor : public TActorBootstrapped<TKqpWriteActor>, public NYql::N
103102
virtual ~TKqpWriteActor() {
104103
}
105104

106-
TGuard<NKikimr::NMiniKQL::TScopedAlloc> BindAllocator() {
107-
return TypeEnv.BindAllocator();
108-
}
109-
110105
void CommitState(const NYql::NDqProto::TCheckpoint&) final {};
111106
void LoadState(const NYql::NDqProto::TSinkState&) final {};
112107

@@ -140,7 +135,7 @@ class TKqpWriteActor : public TActorBootstrapped<TKqpWriteActor>, public NYql::N
140135
Columns[index].PType,
141136
row.GetElement(index),
142137
TypeEnv,
143-
/* copy */ true);
138+
/* copy */ false);
144139
}
145140
BatchBuilder->AddRow(
146141
TConstArrayRef<TCell>{cells.begin(), cells.begin() + KeyColumns.size()},
@@ -250,80 +245,6 @@ class TKqpWriteActor : public TActorBootstrapped<TKqpWriteActor>, public NYql::N
250245
ProcessRows();
251246
}
252247

253-
void RuntimeError(const TString& message, NYql::NDqProto::StatusIds::StatusCode statusCode, const NYql::TIssues& subIssues = {}) {
254-
NYql::TIssue issue(message);
255-
for (const auto& i : subIssues) {
256-
issue.AddSubIssue(MakeIntrusive<NYql::TIssue>(i));
257-
}
258-
259-
NYql::TIssues issues;
260-
issues.AddIssue(std::move(issue));
261-
262-
Callbacks->OnAsyncOutputError(OutputIndex, std::move(issues), statusCode);
263-
}
264-
265-
void PassAway() override {
266-
TActorBootstrapped<TKqpWriteActor>::PassAway();
267-
}
268-
269-
void BuildColumns() {
270-
KeyColumns.reserve(Settings.KeyColumnsSize());
271-
i32 number = 0;
272-
for (const auto& column : Settings.GetKeyColumns()) {
273-
KeyColumns.emplace_back(
274-
column.GetName(),
275-
column.GetId(),
276-
NScheme::TTypeInfo {
277-
static_cast<NScheme::TTypeId>(column.GetTypeId()),
278-
column.GetTypeId() == NScheme::NTypeIds::Pg
279-
? NPg::TypeDescFromPgTypeId(column.GetTypeInfo().GetPgTypeId())
280-
: nullptr
281-
},
282-
column.GetTypeInfo().GetPgTypeMod(),
283-
number++
284-
);
285-
}
286-
287-
ColumnIds.reserve(Settings.ColumnsSize());
288-
Columns.reserve(Settings.ColumnsSize());
289-
number = 0;
290-
for (const auto& column : Settings.GetColumns()) {
291-
ColumnIds.push_back(column.GetId());
292-
Columns.emplace_back(
293-
column.GetName(),
294-
column.GetId(),
295-
NScheme::TTypeInfo {
296-
static_cast<NScheme::TTypeId>(column.GetTypeId()),
297-
column.GetTypeId() == NScheme::NTypeIds::Pg
298-
? NPg::TypeDescFromPgTypeId(column.GetTypeInfo().GetPgTypeId())
299-
: nullptr
300-
},
301-
column.GetTypeInfo().GetPgTypeMod(),
302-
number++
303-
);
304-
}
305-
}
306-
307-
void PrepareBatchBuilder() {
308-
std::vector<std::pair<TString, NScheme::TTypeInfo>> columns;
309-
for (const auto& column : Columns) {
310-
columns.emplace_back(column.Name, column.PType);
311-
}
312-
std::set<std::string> notNullColumns;
313-
for (const auto& column : Settings.GetColumns()) {
314-
if (column.GetNotNull()) {
315-
notNullColumns.insert(column.GetName());
316-
}
317-
}
318-
319-
BatchBuilder = std::make_unique<NArrow::TArrowBatchBuilder>(arrow::Compression::UNCOMPRESSED, notNullColumns);
320-
321-
TString err;
322-
if (!BatchBuilder->Start(columns, 0, 0, err)) {
323-
RuntimeError("Failed to start batch builder: " + err, NYql::NDqProto::StatusIds::PRECONDITION_FAILED);
324-
}
325-
}
326-
327248
void ProcessRows() {
328249
SplitBatchByShards();
329250
SendNewBatchesToShards();
@@ -333,27 +254,6 @@ class TKqpWriteActor : public TActorBootstrapped<TKqpWriteActor>, public NYql::N
333254
}
334255
}
335256

336-
NKikimr::NEvWrite::IShardsSplitter::IEvWriteDataAccessor::TPtr GetDataAccessor(
337-
const std::shared_ptr<arrow::RecordBatch>& batch) const {
338-
struct TDataAccessor : public NKikimr::NEvWrite::IShardsSplitter::IEvWriteDataAccessor {
339-
std::shared_ptr<arrow::RecordBatch> Batch;
340-
341-
TDataAccessor(const std::shared_ptr<arrow::RecordBatch>& batch)
342-
: Batch(batch) {
343-
}
344-
345-
std::shared_ptr<arrow::RecordBatch> GetDeserializedBatch() const override {
346-
return Batch;
347-
}
348-
349-
TString GetSerializedData() const override {
350-
return NArrow::SerializeBatchNoCompression(Batch);
351-
}
352-
};
353-
354-
return std::make_shared<TDataAccessor>(batch);
355-
}
356-
357257
void SplitBatchByShards() {
358258
if (!SchemeEntry || BatchBuilder->Bytes() == 0) {
359259
return;
@@ -485,6 +385,102 @@ class TKqpWriteActor : public TActorBootstrapped<TKqpWriteActor>, public NYql::N
485385
return result;
486386
}
487387

388+
void RuntimeError(const TString& message, NYql::NDqProto::StatusIds::StatusCode statusCode, const NYql::TIssues& subIssues = {}) {
389+
NYql::TIssue issue(message);
390+
for (const auto& i : subIssues) {
391+
issue.AddSubIssue(MakeIntrusive<NYql::TIssue>(i));
392+
}
393+
394+
NYql::TIssues issues;
395+
issues.AddIssue(std::move(issue));
396+
397+
Callbacks->OnAsyncOutputError(OutputIndex, std::move(issues), statusCode);
398+
}
399+
400+
void PassAway() override {
401+
TActorBootstrapped<TKqpWriteActor>::PassAway();
402+
}
403+
404+
void BuildColumns() {
405+
KeyColumns.reserve(Settings.KeyColumnsSize());
406+
i32 number = 0;
407+
for (const auto& column : Settings.GetKeyColumns()) {
408+
KeyColumns.emplace_back(
409+
column.GetName(),
410+
column.GetId(),
411+
NScheme::TTypeInfo {
412+
static_cast<NScheme::TTypeId>(column.GetTypeId()),
413+
column.GetTypeId() == NScheme::NTypeIds::Pg
414+
? NPg::TypeDescFromPgTypeId(column.GetTypeInfo().GetPgTypeId())
415+
: nullptr
416+
},
417+
column.GetTypeInfo().GetPgTypeMod(),
418+
number++
419+
);
420+
}
421+
422+
ColumnIds.reserve(Settings.ColumnsSize());
423+
Columns.reserve(Settings.ColumnsSize());
424+
number = 0;
425+
for (const auto& column : Settings.GetColumns()) {
426+
ColumnIds.push_back(column.GetId());
427+
Columns.emplace_back(
428+
column.GetName(),
429+
column.GetId(),
430+
NScheme::TTypeInfo {
431+
static_cast<NScheme::TTypeId>(column.GetTypeId()),
432+
column.GetTypeId() == NScheme::NTypeIds::Pg
433+
? NPg::TypeDescFromPgTypeId(column.GetTypeInfo().GetPgTypeId())
434+
: nullptr
435+
},
436+
column.GetTypeInfo().GetPgTypeMod(),
437+
number++
438+
);
439+
}
440+
}
441+
442+
void PrepareBatchBuilder() {
443+
std::vector<std::pair<TString, NScheme::TTypeInfo>> columns;
444+
for (const auto& column : Columns) {
445+
columns.emplace_back(column.Name, column.PType);
446+
}
447+
std::set<std::string> notNullColumns;
448+
for (const auto& column : Settings.GetColumns()) {
449+
if (column.GetNotNull()) {
450+
notNullColumns.insert(column.GetName());
451+
}
452+
}
453+
454+
BatchBuilder = std::make_unique<NArrow::TArrowBatchBuilder>(arrow::Compression::UNCOMPRESSED, notNullColumns);
455+
456+
TString err;
457+
if (!BatchBuilder->Start(columns, 0, 0, err)) {
458+
RuntimeError("Failed to start batch builder: " + err, NYql::NDqProto::StatusIds::PRECONDITION_FAILED);
459+
}
460+
}
461+
462+
NKikimr::NEvWrite::IShardsSplitter::IEvWriteDataAccessor::TPtr GetDataAccessor(
463+
const std::shared_ptr<arrow::RecordBatch>& batch) const {
464+
struct TDataAccessor : public NKikimr::NEvWrite::IShardsSplitter::IEvWriteDataAccessor {
465+
std::shared_ptr<arrow::RecordBatch> Batch;
466+
467+
TDataAccessor(const std::shared_ptr<arrow::RecordBatch>& batch)
468+
: Batch(batch) {
469+
}
470+
471+
std::shared_ptr<arrow::RecordBatch> GetDeserializedBatch() const override {
472+
return Batch;
473+
}
474+
475+
TString GetSerializedData() const override {
476+
return NArrow::SerializeBatchNoCompression(Batch);
477+
}
478+
};
479+
480+
return std::make_shared<TDataAccessor>(batch);
481+
}
482+
483+
488484
NActors::TActorId TxProxyId = MakeTxProxyID();
489485
NActors::TActorId PipeCacheId = NKikimr::MakePipePeNodeCacheID(false);
490486

@@ -495,7 +491,6 @@ class TKqpWriteActor : public TActorBootstrapped<TKqpWriteActor>, public NYql::N
495491
NYql::NDq::IDqComputeActorAsyncOutput::ICallbacks * Callbacks = nullptr;
496492
TIntrusivePtr<TKqpCounters> Counters;
497493
const NMiniKQL::TTypeEnvironment& TypeEnv;
498-
std::shared_ptr<NKikimr::NMiniKQL::TScopedAlloc> Alloc;
499494

500495
const NYql::NDq::TTxId TxId;
501496
const TTableId TableId;

‎ydb/core/kqp/ut/common/kqp_ut_common.cpp‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -442,13 +442,13 @@ void TKikimrRunner::CreateSampleTables() {
442442

443443
void TKikimrRunner::Initialize(const TKikimrSettings& settings) {
444444
// Server->GetRuntime()->SetLogPriority(NKikimrServices::FLAT_TX_SCHEMESHARD, NActors::NLog::PRI_DEBUG);
445-
Server->GetRuntime()->SetLogPriority(NKikimrServices::KQP_YQL, NActors::NLog::PRI_DEBUG);
445+
// Server->GetRuntime()->SetLogPriority(NKikimrServices::KQP_YQL, NActors::NLog::PRI_DEBUG);
446446
// Server->GetRuntime()->SetLogPriority(NKikimrServices::KQP_YQL, NActors::NLog::PRI_INFO);
447447
// Server->GetRuntime()->SetLogPriority(NKikimrServices::TX_DATASHARD, NActors::NLog::PRI_TRACE);
448-
Server->GetRuntime()->SetLogPriority(NKikimrServices::TX_COORDINATOR, NActors::NLog::PRI_DEBUG);
448+
// Server->GetRuntime()->SetLogPriority(NKikimrServices::TX_COORDINATOR, NActors::NLog::PRI_DEBUG);
449449
// Server->GetRuntime()->SetLogPriority(NKikimrServices::KQP_COMPUTE, NActors::NLog::PRI_DEBUG);
450450
// Server->GetRuntime()->SetLogPriority(NKikimrServices::KQP_TASKS_RUNNER, NActors::NLog::PRI_DEBUG);
451-
Server->GetRuntime()->SetLogPriority(NKikimrServices::KQP_EXECUTER, NActors::NLog::PRI_DEBUG);
451+
// Server->GetRuntime()->SetLogPriority(NKikimrServices::KQP_EXECUTER, NActors::NLog::PRI_DEBUG);
452452
// Server->GetRuntime()->SetLogPriority(NKikimrServices::TX_PROXY_SCHEME_CACHE, NActors::NLog::PRI_DEBUG);
453453
// Server->GetRuntime()->SetLogPriority(NKikimrServices::SCHEME_BOARD_REPLICA, NActors::NLog::PRI_DEBUG);
454454
// Server->GetRuntime()->SetLogPriority(NKikimrServices::KQP_WORKER, NActors::NLog::PRI_DEBUG);
@@ -457,14 +457,14 @@ void TKikimrRunner::Initialize(const TKikimrSettings& settings) {
457457
// Server->GetRuntime()->SetLogPriority(NKikimrServices::KQP_SLOW_LOG, NActors::NLog::PRI_TRACE);
458458
// Server->GetRuntime()->SetLogPriority(NKikimrServices::KQP_PROXY, NActors::NLog::PRI_DEBUG);
459459
// Server->GetRuntime()->SetLogPriority(NKikimrServices::KQP_COMPILE_SERVICE, NActors::NLog::PRI_DEBUG);
460-
Server->GetRuntime()->SetLogPriority(NKikimrServices::KQP_COMPILE_ACTOR, NActors::NLog::PRI_TRACE);
460+
// Server->GetRuntime()->SetLogPriority(NKikimrServices::KQP_COMPILE_ACTOR, NActors::NLog::PRI_TRACE);
461461
// Server->GetRuntime()->SetLogPriority(NKikimrServices::KQP_COMPILE_REQUEST, NActors::NLog::PRI_DEBUG);
462462
// Server->GetRuntime()->SetLogPriority(NKikimrServices::KQP_GATEWAY, NActors::NLog::PRI_DEBUG);
463463
// Server->GetRuntime()->SetLogPriority(NKikimrServices::RPC_REQUEST, NActors::NLog::PRI_DEBUG);
464464
// Server->GetRuntime()->SetLogPriority(NKikimrServices::KQP_RESOURCE_MANAGER, NActors::NLog::PRI_DEBUG);
465465
// Server->GetRuntime()->SetLogPriority(NKikimrServices::KQP_NODE, NActors::NLog::PRI_DEBUG);
466466
// Server->GetRuntime()->SetLogPriority(NKikimrServices::KQP_BLOBS_STORAGE, NActors::NLog::PRI_DEBUG);
467-
Server->GetRuntime()->SetLogPriority(NKikimrServices::TX_COLUMNSHARD, NActors::NLog::PRI_INFO);
467+
// Server->GetRuntime()->SetLogPriority(NKikimrServices::TX_COLUMNSHARD, NActors::NLog::PRI_INFO);
468468

469469
RunCall([this, domain = settings.DomainRoot]{
470470
this->Client->InitRootScheme(domain);

‎ydb/core/kqp/ut/olap/kqp_olap_ut.cpp‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5098,7 +5098,7 @@ Y_UNIT_TEST_SUITE(KqpOlap) {
50985098

50995099
UNIT_ASSERT(!result.IsSuccess());
51005100
std::string errorMsg = result.GetIssues().ToString();
5101-
UNIT_ASSERT_C(errorMsg.find("Data manipulation queries do not support column shard tables.") != std::string::npos, errorMsg);
5101+
UNIT_ASSERT_C(errorMsg.find("Write mode 'insert_abort' is not supported for olap tables.") != std::string::npos, errorMsg);
51025102
}
51035103

51045104
Y_UNIT_TEST(Olap_InsertFailsOnGenericQuery) {
@@ -5118,7 +5118,7 @@ Y_UNIT_TEST_SUITE(KqpOlap) {
51185118

51195119
UNIT_ASSERT(!result.IsSuccess());
51205120
std::string errorMsg = result.GetIssues().ToString();
5121-
UNIT_ASSERT_C(errorMsg.find("Data manipulation queries do not support column shard tables.") != std::string::npos, errorMsg);
5121+
UNIT_ASSERT_C(errorMsg.find("Write mode 'insert_abort' is not supported for olap tables.") != std::string::npos, errorMsg);
51225122
}
51235123

51245124
Y_UNIT_TEST(OlapRead_FailsOnDataQuery) {

‎ydb/library/yql/core/yql_expr_type_annotation.cpp‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3922,7 +3922,6 @@ template bool EnsureNewSeqType<false, false, true>(const TExprNode& node, TExprC
39223922
template bool EnsureNewSeqType<true, true, true>(const TExprNode& node, TExprContext& ctx, const TTypeAnnotationNode** itemType);
39233923
template bool EnsureNewSeqType<false, true, true>(const TExprNode& node, TExprContext& ctx, const TTypeAnnotationNode** itemType);
39243924
template bool EnsureNewSeqType<false, true, false>(const TExprNode& node, TExprContext& ctx, const TTypeAnnotationNode** itemType);
3925-
template bool EnsureNewSeqType<false, false, false>(const TExprNode& node, TExprContext& ctx, const TTypeAnnotationNode** itemType);
39263925

39273926
bool EnsureAnySeqType(const TExprNode& node, TExprContext& ctx) {
39283927
if (HasError(node.GetTypeAnn(), ctx)) {

‎ydb/library/yql/dq/actors/compute/dq_compute_actor_async_io.h‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -231,7 +231,6 @@ struct IDqAsyncIoFactory : public TThrRefBase {
231231
const THashMap<TString, TString>& TaskParams;
232232
const NKikimr::NMiniKQL::TTypeEnvironment& TypeEnv;
233233
const NKikimr::NMiniKQL::THolderFactory& HolderFactory;
234-
std::shared_ptr<NKikimr::NMiniKQL::TScopedAlloc> Alloc;
235234
IRandomProvider *const RandomProvider;
236235
};
237236

‎ydb/library/yql/dq/actors/compute/dq_compute_actor_impl.h‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1683,7 +1683,6 @@ class TDqComputeActorBase : public NActors::TActorBootstrapped<TDerived>
16831683
.TaskParams = taskParams,
16841684
.TypeEnv = typeEnv,
16851685
.HolderFactory = holderFactory,
1686-
.Alloc = TaskRunner->GetAllocatorPtr(),
16871686
.RandomProvider = TaskRunner ? TaskRunner->GetRandomProvider() : nullptr
16881687
});
16891688
} catch (const std::exception& ex) {

0 commit comments

Comments
 (0)