Skip to content

Commit d2fb053

Browse files
GrigoriyPAGrigoriyPA
authored
YQ-4602 supported stream lookup join (#25897)
1 parent c8c8e41 commit d2fb053

File tree

20 files changed

+740
-324
lines changed

20 files changed

+740
-324
lines changed

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -392,6 +392,7 @@ bool HasUncommittedChangesRead(THashSet<NKikimr::TTableId>& modifiedTables, cons
392392
case NKqpProto::TKqpPhyConnection::kResult:
393393
case NKqpProto::TKqpPhyConnection::kValue:
394394
case NKqpProto::TKqpPhyConnection::kMerge:
395+
case NKqpProto::TKqpPhyConnection::kDqSourceStreamLookup:
395396
case NKqpProto::TKqpPhyConnection::TYPE_NOT_SET:
396397
break;
397398
}

‎ydb/core/kqp/compute_actor/kqp_compute_actor.cpp‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@
1212
#include <ydb/core/kqp/runtime/kqp_vector_actor.h>
1313
#include <ydb/core/kqp/runtime/kqp_write_actor.h>
1414
#include <ydb/library/formats/arrow/protos/ssa.pb.h>
15+
#include <ydb/library/yql/dq/actors/input_transforms/dq_input_transform_lookup_factory.h>
1516
#include <ydb/library/yql/dq/comp_nodes/dq_block_hash_join.h>
1617
#include <ydb/library/yql/dq/comp_nodes/dq_hash_combine.h>
1718
#include <ydb/library/yql/dq/proto/dq_tasks.pb.h>
@@ -91,6 +92,7 @@ NYql::NDq::IDqAsyncIoFactory::TPtr CreateKqpAsyncIoFactory(
9192
RegisterKqpWriteActor(*factory, counters);
9293
RegisterSequencerActorFactory(*factory, counters);
9394
RegisterKqpVectorResolveActor(*factory, counters);
95+
NYql::NDq::RegisterDqInputTransformLookupActorFactory(*factory);
9496

9597
if (federatedQuerySetup) {
9698
auto s3HttpRetryPolicy = NYql::GetHTTPDefaultRetryPolicy(NYql::THttpRetryPolicyOptions{.RetriedCurlCodes = NYql::FqRetriedCurlCodes()});

‎ydb/core/kqp/compute_actor/ya.make‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -25,12 +25,13 @@ PEERDIR(
2525
ydb/library/formats/arrow/protos
2626
ydb/library/formats/arrow/common
2727
ydb/library/yql/dq/actors/compute
28+
ydb/library/yql/dq/actors/input_transforms
29+
ydb/library/yql/dq/comp_nodes
2830
ydb/library/yql/providers/generic/actors
2931
ydb/library/yql/providers/pq/async_io
3032
ydb/library/yql/providers/s3/actors_factory
3133
ydb/library/yql/providers/solomon/actors
3234
yql/essentials/public/issue
33-
ydb/library/yql/dq/comp_nodes
3435
)
3536

3637
GENERATE_ENUM_SERIALIZATION(kqp_compute_state.h)

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

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -518,6 +518,47 @@ void TKqpTasksGraph::BuildVectorResolveChannels(const TStageInfo& stageInfo, ui3
518518
inputStageInfo, outputIndex, enableSpilling, logFunc);
519519
}
520520

521+
void TKqpTasksGraph::BuildDqSourceStreamLookupChannels(const TStageInfo& stageInfo, ui32 inputIndex, const TStageInfo& inputStageInfo,
522+
ui32 outputIndex, const NKqpProto::TKqpPhyCnDqSourceStreamLookup& dqSourceStreamLookup, const TChannelLogFunc& logFunc) {
523+
YQL_ENSURE(stageInfo.Tasks.size() == 1);
524+
525+
auto* settings = GetMeta().Allocate<NDqProto::TDqInputTransformLookupSettings>();
526+
settings->SetLeftLabel(dqSourceStreamLookup.GetLeftLabel());
527+
settings->SetRightLabel(dqSourceStreamLookup.GetRightLabel());
528+
settings->SetJoinType(dqSourceStreamLookup.GetJoinType());
529+
settings->SetNarrowInputRowType(dqSourceStreamLookup.GetConnectionInputRowType());
530+
settings->SetNarrowOutputRowType(dqSourceStreamLookup.GetConnectionOutputRowType());
531+
settings->SetCacheLimit(dqSourceStreamLookup.GetCacheLimit());
532+
settings->SetCacheTtlSeconds(dqSourceStreamLookup.GetCacheTtlSeconds());
533+
settings->SetMaxDelayedRows(dqSourceStreamLookup.GetMaxDelayedRows());
534+
settings->SetIsMultiget(dqSourceStreamLookup.GetIsMultiGet());
535+
536+
const auto& leftJointKeys = dqSourceStreamLookup.GetLeftJoinKeyNames();
537+
settings->MutableLeftJoinKeyNames()->Assign(leftJointKeys.begin(), leftJointKeys.end());
538+
539+
const auto& rightJointKeys = dqSourceStreamLookup.GetRightJoinKeyNames();
540+
settings->MutableRightJoinKeyNames()->Assign(rightJointKeys.begin(), rightJointKeys.end());
541+
542+
auto& streamLookupSource = *settings->MutableRightSource();
543+
streamLookupSource.SetSerializedRowType(dqSourceStreamLookup.GetLookupRowType());
544+
const auto& compiledSource = dqSourceStreamLookup.GetLookupSource();
545+
streamLookupSource.SetProviderName(compiledSource.GetType());
546+
*streamLookupSource.MutableLookupSource() = compiledSource.GetSettings();
547+
548+
TTransform dqSourceStreamLookupTransform = {
549+
.Type = "StreamLookupInputTransform",
550+
.InputType = dqSourceStreamLookup.GetInputStageRowType(),
551+
.OutputType = dqSourceStreamLookup.GetOutputStageRowType(),
552+
};
553+
YQL_ENSURE(dqSourceStreamLookupTransform.Settings.PackFrom(*settings));
554+
555+
for (const auto taskId : stageInfo.Tasks) {
556+
GetTask(taskId).Inputs[inputIndex].Transform = dqSourceStreamLookupTransform;
557+
}
558+
559+
BuildUnionAllChannels(*this, stageInfo, inputIndex, inputStageInfo, outputIndex, /* enableSpilling */ false, logFunc);
560+
}
561+
521562
void TKqpTasksGraph::BuildKqpStageChannels(TStageInfo& stageInfo, ui64 txId, bool enableSpilling, bool enableShuffleElimination) {
522563
auto& stage = stageInfo.Meta.GetStage(stageInfo.Id);
523564

@@ -710,6 +751,12 @@ void TKqpTasksGraph::BuildKqpStageChannels(TStageInfo& stageInfo, ui64 txId, boo
710751
break;
711752
}
712753

754+
case NKqpProto::TKqpPhyConnection::kDqSourceStreamLookup: {
755+
BuildDqSourceStreamLookupChannels(stageInfo, inputIdx, inputStageInfo, outputIdx,
756+
input.GetDqSourceStreamLookup(), log);
757+
break;
758+
}
759+
713760
default:
714761
YQL_ENSURE(false, "Unexpected stage input type: " << (ui32)input.GetTypeCase());
715762
}
@@ -1370,6 +1417,8 @@ void TKqpTasksGraph::FillInputDesc(NYql::NDqProto::TTaskInput& inputDesc, const
13701417
}
13711418

13721419
transformProto->MutableSettings()->PackFrom(*input.Meta.VectorResolveSettings);
1420+
} else {
1421+
*transformProto->MutableSettings() = input.Transform->Settings;
13731422
}
13741423
}
13751424
}
@@ -1725,6 +1774,7 @@ bool TKqpTasksGraph::BuildComputeTasks(TStageInfo& stageInfo, const ui32 nodesCo
17251774
case NKqpProto::TKqpPhyConnection::kMap:
17261775
case NKqpProto::TKqpPhyConnection::kParallelUnionAll:
17271776
case NKqpProto::TKqpPhyConnection::kVectorResolve:
1777+
case NKqpProto::TKqpPhyConnection::kDqSourceStreamLookup:
17281778
break;
17291779
default:
17301780
YQL_ENSURE(false, "Unexpected connection type: " << (ui32)input.GetTypeCase() << Endl);

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -422,6 +422,8 @@ class TKqpTasksGraph : public NYql::NDq::TDqTasksGraph<TGraphMeta, TStageInfoMet
422422
void BuildVectorResolveChannels(const TStageInfo& stageInfo, ui32 inputIndex,
423423
const TStageInfo& inputStageInfo, ui32 outputIndex,
424424
const NKqpProto::TKqpPhyCnVectorResolve& vectorResolve, bool enableSpilling, const NYql::NDq::TChannelLogFunc& logFunc);
425+
void BuildDqSourceStreamLookupChannels(const TStageInfo& stageInfo, ui32 inputIndex, const TStageInfo& inputStageInfo,
426+
ui32 outputIndex, const NKqpProto::TKqpPhyCnDqSourceStreamLookup& dqSourceStreamLookup, const NYql::NDq::TChannelLogFunc& logFunc);
425427

426428
void FillOutputDesc(NYql::NDqProto::TTaskOutput& outputDesc, const TTaskOutput& output, ui32 outputIdx,
427429
bool enableSpilling, const TStageInfo& stageInfo) const;

‎ydb/core/kqp/host/kqp_host.cpp‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1964,6 +1964,7 @@ class TKqpHost : public IKqpHost {
19641964
if (FederatedQuerySetup->PqGateway) {
19651965
InitPqProvider();
19661966
}
1967+
TypesCtx->StreamLookupJoin = true;
19671968
}
19681969

19691970
InitPgProvider();

‎ydb/core/kqp/opt/logical/kqp_opt_log.cpp‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,7 @@ class TKqpLogicalOptTransformer : public TOptimizeTransformerBase {
4141
AddHandler(0, &TCoTake::Match, HNDL(RewriteTakeSortToTopSort));
4242
AddHandler(0, &TCoFlatMap::Match, HNDL(RewriteSqlInToEquiJoin));
4343
AddHandler(0, &TCoFlatMap::Match, HNDL(RewriteSqlInCompactToJoin));
44+
AddHandler(0, &TCoEquiJoin::Match, HNDL(RewriteStreamEquiJoinWithLookup));
4445
AddHandler(0, &TCoEquiJoin::Match, HNDL(OptimizeEquiJoinWithCosts));
4546
AddHandler(0, &TCoEquiJoin::Match, HNDL(RewriteEquiJoin));
4647
AddHandler(0, &TDqJoin::Match, HNDL(JoinToIndexLookup));
@@ -167,6 +168,12 @@ class TKqpLogicalOptTransformer : public TOptimizeTransformerBase {
167168
return output;
168169
}
169170

171+
TMaybeNode<TExprBase> RewriteStreamEquiJoinWithLookup(TExprBase node, TExprContext& ctx) {
172+
TExprBase output = DqRewriteStreamEquiJoinWithLookup(node, ctx, TypesCtx);
173+
DumpAppliedRule("KqpRewriteStreamEquiJoinWithLookup", node.Ptr(), output.Ptr(), ctx);
174+
return output;
175+
}
176+
170177
TMaybeNode<TExprBase> OptimizeEquiJoinWithCosts(TExprBase node, TExprContext& ctx) {
171178
auto maxDPhypDPTableSize = Config->MaxDPHypDPTableSize.Get().GetOrElse(TDqSettings::TDefault::MaxDPHypDPTableSize);
172179
auto optLevel = Config->CostBasedOptimizationLevel.Get().GetOrElse(Config->DefaultCostBasedOptimizationLevel);

‎ydb/core/kqp/opt/physical/kqp_opt_phy.cpp‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -71,6 +71,7 @@ class TKqpPhysicalOptTransformer : public TOptimizeTransformerBase {
7171
AddHandler(0, &TCoExtendBase::Match, HNDL(BuildExtendStage));
7272
AddHandler(0, &TDqJoin::Match, HNDL(RewriteRightJoinToLeft));
7373
AddHandler(0, &TDqJoin::Match, HNDL(RewriteLeftPureJoin<false>));
74+
AddHandler(0, &TDqJoin::Match, HNDL(RewriteStreamLookupJoin));
7475
AddHandler(0, &TDqJoin::Match, HNDL(BuildJoin<false>));
7576
AddHandler(0, &TDqPrecompute::Match, HNDL(BuildPrecompute));
7677
AddHandler(0, &TCoLMap::Match, HNDL(PushLMapToStage<false>));
@@ -507,6 +508,14 @@ class TKqpPhysicalOptTransformer : public TOptimizeTransformerBase {
507508
return output;
508509
}
509510

511+
TMaybeNode<TExprBase> RewriteStreamLookupJoin(TExprBase node, TExprContext& ctx) {
512+
TMaybeNode<TExprBase> output = DqRewriteStreamLookupJoin(node, ctx);
513+
if (output) {
514+
DumpAppliedRule("RewriteStreamLookupJoin", node.Ptr(), output.Cast().Ptr(), ctx);
515+
}
516+
return output;
517+
}
518+
510519
template <bool IsGlobal>
511520
TMaybeNode<TExprBase> BuildJoin(TExprBase node, TExprContext& ctx,
512521
IOptimizationContext& optCtx, const TGetParents& getParents)

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

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -666,7 +666,8 @@ class TKikimrDataSource : public TDataProviderBase {
666666
node.IsCallable(TDqReadWrap::CallableName()) ||
667667
node.IsCallable(TDqReadWideWrap::CallableName()) ||
668668
node.IsCallable(TDqReadBlockWideWrap::CallableName()) ||
669-
node.IsCallable(TDqSource::CallableName())
669+
node.IsCallable(TDqSource::CallableName()) ||
670+
node.IsCallable(TDqLookupSourceWrap::CallableName())
670671
)
671672
)
672673
{

‎ydb/core/kqp/query_compiler/kqp_query_compiler.cpp‎

Lines changed: 75 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
#include "kqp_query_compiler.h"
22

3+
#include <ydb/core/base/table_index.h>
34
#include <ydb/core/kqp/common/kqp_yql.h>
45
#include <ydb/core/kqp/gateway/utils/scheme_helpers.h>
56
#include <ydb/core/kqp/opt/kqp_opt.h>
@@ -8,24 +9,23 @@
89
#include <ydb/core/kqp/query_compiler/kqp_olap_compiler.h>
910
#include <ydb/core/kqp/query_data/kqp_predictor.h>
1011
#include <ydb/core/kqp/query_data/kqp_request_predictor.h>
11-
#include <ydb/core/ydb_convert/ydb_convert.h>
12-
13-
#include <ydb/core/base/table_index.h>
1412
#include <ydb/core/scheme/scheme_tabledefs.h>
13+
#include <ydb/core/ydb_convert/ydb_convert.h>
1514
#include <ydb/library/mkql_proto/mkql_proto.h>
16-
17-
#include <yql/essentials/core/dq_integration/yql_dq_integration.h>
1815
#include <ydb/library/yql/dq/opt/dq_opt.h>
1916
#include <ydb/library/yql/dq/type_ann/dq_type_ann.h>
2017
#include <ydb/library/yql/dq/tasks/dq_task_program.h>
21-
#include <yql/essentials/minikql/mkql_node_serialization.h>
22-
#include <yql/essentials/providers/common/mkql/yql_type_mkql.h>
23-
#include <yql/essentials/providers/common/provider/yql_provider_names.h>
24-
#include <yql/essentials/providers/common/structured_token/yql_token_builder.h>
18+
#include <ydb/library/yql/providers/dq/common/yql_dq_common.h>
2519
#include <ydb/library/yql/providers/dq/common/yql_dq_settings.h>
2620
#include <ydb/library/yql/providers/s3/statistics/yql_s3_statistics.h>
21+
22+
#include <yql/essentials/core/dq_integration/yql_dq_integration.h>
2723
#include <yql/essentials/core/yql_opt_utils.h>
2824
#include <yql/essentials/core/yql_type_helpers.h>
25+
#include <yql/essentials/minikql/mkql_node_serialization.h>
26+
#include <yql/essentials/providers/common/mkql/yql_type_mkql.h>
27+
#include <yql/essentials/providers/common/provider/yql_provider_names.h>
28+
#include <yql/essentials/providers/common/structured_token/yql_token_builder.h>
2929

3030

3131
namespace NKikimr {
@@ -589,6 +589,14 @@ TIssues ApplyOverridePlannerSettings(const TString& overridePlannerJson, NKqpPro
589589
return issues;
590590
}
591591

592+
TStringBuf RemoveJoinAliases(TStringBuf keyName) {
593+
if (const auto idx = keyName.find_last_of('.'); idx != TString::npos) {
594+
return keyName.substr(idx + 1);
595+
}
596+
597+
return keyName;
598+
}
599+
592600
class TKqpQueryCompiler : public IKqpQueryCompiler {
593601
public:
594602
TKqpQueryCompiler(const TString& cluster, const TIntrusivePtr<TKikimrTablesData> tablesData,
@@ -795,7 +803,7 @@ class TKqpQueryCompiler : public IKqpQueryCompiler {
795803
auto connection = input.Cast<TDqConnection>();
796804

797805
auto& protoInput = *stageProto.AddInputs();
798-
FillConnection(connection, stagesMap, protoInput, ctx, tablesMap, physicalStageByID);
806+
FillConnection(connection, stagesMap, protoInput, ctx, tablesMap, physicalStageByID, &stage, inputIndex);
799807
protoInput.SetInputIndex(inputIndex);
800808
}
801809
}
@@ -1017,7 +1025,7 @@ class TKqpQueryCompiler : public IKqpQueryCompiler {
10171025

10181026
auto& resultProto = *txProto.AddResults();
10191027
auto& connectionProto = *resultProto.MutableConnection();
1020-
FillConnection(connection, stagesMap, connectionProto, ctx, tablesMap, physicalStageByID);
1028+
FillConnection(connection, stagesMap, connectionProto, ctx, tablesMap, physicalStageByID, nullptr, 0);
10211029

10221030
const TTypeAnnotationNode* itemType = nullptr;
10231031
switch (connectionProto.GetTypeCase()) {
@@ -1452,7 +1460,9 @@ class TKqpQueryCompiler : public IKqpQueryCompiler {
14521460
NKqpProto::TKqpPhyConnection& connectionProto,
14531461
TExprContext& ctx,
14541462
THashMap<TStringBuf, THashSet<TStringBuf>>& tablesMap,
1455-
THashMap<ui64, NKqpProto::TKqpPhyStage*>& physicalStageByID
1463+
THashMap<ui64, NKqpProto::TKqpPhyStage*>& physicalStageByID,
1464+
const TDqPhyStage* stage,
1465+
ui32 inputIndex
14561466
) {
14571467
auto inputStageIndex = stagesMap.FindPtr(connection.Output().Stage().Ref().UniqueId());
14581468
YQL_ENSURE(inputStageIndex, "stage #" << connection.Output().Stage().Ref().UniqueId() << " not found in stages map: "
@@ -1819,6 +1829,59 @@ class TKqpQueryCompiler : public IKqpQueryCompiler {
18191829
return;
18201830
}
18211831

1832+
if (auto maybeDqSourceStreamLookup = connection.Maybe<TDqCnStreamLookup>()) {
1833+
const auto streamLookup = maybeDqSourceStreamLookup.Cast();
1834+
const auto lookupSourceWrap = streamLookup.RightInput().Cast<TDqLookupSourceWrap>();
1835+
1836+
const TStringBuf dataSourceCategory = lookupSourceWrap.DataSource().Category();
1837+
const auto provider = TypesCtx.DataSourceMap.find(dataSourceCategory);
1838+
YQL_ENSURE(provider != TypesCtx.DataSourceMap.end(), "Unsupported data source category: \"" << dataSourceCategory << "\"");
1839+
NYql::IDqIntegration* dqIntegration = provider->second->GetDqIntegration();
1840+
YQL_ENSURE(dqIntegration, "Unsupported dq source for provider: \"" << dataSourceCategory << "\"");
1841+
1842+
auto& dqSourceLookupCn = *connectionProto.MutableDqSourceStreamLookup();
1843+
auto& lookupSource = *dqSourceLookupCn.MutableLookupSource();
1844+
auto& lookupSourceSettings = *lookupSource.MutableSettings();
1845+
auto& lookupSourceType = *lookupSource.MutableType();
1846+
dqIntegration->FillLookupSourceSettings(lookupSourceWrap.Ref(), lookupSourceSettings, lookupSourceType);
1847+
YQL_ENSURE(!lookupSourceSettings.type_url().empty(), "Data source provider \"" << dataSourceCategory << "\" did't fill dq source settings for its dq source node");
1848+
YQL_ENSURE(lookupSourceType, "Data source provider \"" << dataSourceCategory << "\" did't fill dq source settings type for its dq source node");
1849+
1850+
const auto& streamLookupOutput = streamLookup.Output();
1851+
const auto connectionInputRowType = GetSeqItemType(streamLookupOutput.Ref().GetTypeAnn());
1852+
YQL_ENSURE(connectionInputRowType->GetKind() == ETypeAnnotationKind::Struct);
1853+
const auto connectionOutputRowType = GetSeqItemType(streamLookup.Ref().GetTypeAnn());
1854+
YQL_ENSURE(connectionOutputRowType->GetKind() == ETypeAnnotationKind::Struct);
1855+
YQL_ENSURE(stage);
1856+
dqSourceLookupCn.SetConnectionInputRowType(NYql::NCommon::GetSerializedTypeAnnotation(connectionInputRowType));
1857+
dqSourceLookupCn.SetConnectionOutputRowType(NYql::NCommon::GetSerializedTypeAnnotation(connectionOutputRowType));
1858+
dqSourceLookupCn.SetLookupRowType(NYql::NCommon::GetSerializedTypeAnnotation(lookupSourceWrap.RowType().Ref().GetTypeAnn()));
1859+
dqSourceLookupCn.SetInputStageRowType(NYql::NCommon::GetSerializedTypeAnnotation(GetSeqItemType(streamLookupOutput.Stage().Program().Ref().GetTypeAnn())));
1860+
dqSourceLookupCn.SetOutputStageRowType(NYql::NCommon::GetSerializedTypeAnnotation(GetSeqItemType(stage->Program().Args().Arg(inputIndex).Ref().GetTypeAnn())));
1861+
1862+
const TString leftLabel(streamLookup.LeftLabel());
1863+
dqSourceLookupCn.SetLeftLabel(leftLabel);
1864+
dqSourceLookupCn.SetRightLabel(streamLookup.RightLabel().StringValue());
1865+
dqSourceLookupCn.SetJoinType(streamLookup.JoinType().StringValue());
1866+
dqSourceLookupCn.SetCacheLimit(FromString<ui64>(streamLookup.MaxCachedRows()));
1867+
dqSourceLookupCn.SetCacheTtlSeconds(FromString<ui64>(streamLookup.TTL()));
1868+
dqSourceLookupCn.SetMaxDelayedRows(FromString<ui64>(streamLookup.MaxDelayedRows()));
1869+
1870+
if (const auto maybeMultiget = streamLookup.IsMultiget()) {
1871+
dqSourceLookupCn.SetIsMultiGet(FromString<bool>(maybeMultiget.Cast()));
1872+
}
1873+
1874+
for (const auto& key : streamLookup.LeftJoinKeyNames()) {
1875+
*dqSourceLookupCn.AddLeftJoinKeyNames() = leftLabel ? RemoveJoinAliases(key) : key;
1876+
}
1877+
1878+
for (const auto& key : streamLookup.RightJoinKeyNames()) {
1879+
*dqSourceLookupCn.AddRightJoinKeyNames() = RemoveJoinAliases(key);
1880+
}
1881+
1882+
return;
1883+
}
1884+
18221885
YQL_ENSURE(false, "Unexpected connection type: " << connection.CallableName());
18231886
}
18241887

0 commit comments

Comments
 (0)