12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934 |
- #include "yql_join.h"
- #include "yql_expr_type_annotation.h"
- #include "yql_opt_utils.h"
- #include <util/string/cast.h>
- #include <util/string/join.h>
- #include <util/string/type.h>
- namespace NYql {
- using namespace NNodes;
- namespace {
- const TTypeAnnotationNode* AddOptionalType(const TTypeAnnotationNode* type, TExprContext& ctx) {
- if (type->IsOptionalOrNull()) {
- return type;
- }
- return ctx.MakeType<TOptionalExprType>(type);
- }
- struct TJoinState {
- bool Used = false;
- };
- IGraphTransformer::TStatus ParseJoinKeys(TExprNode& side, TVector<std::pair<TStringBuf, TStringBuf>>& keys,
- TVector<const TTypeAnnotationNode*>& keyTypes, const TJoinLabels& labels,
- TExprContext& ctx, bool isCross) {
- if (!EnsureTuple(side, ctx)) {
- return IGraphTransformer::TStatus::Error;
- }
- for (auto& child : side.Children()) {
- if (!EnsureAtom(*child, ctx)) {
- return IGraphTransformer::TStatus::Error;
- }
- }
- if (isCross) {
- if (side.ChildrenSize() != 0) {
- ctx.AddError(TIssue(ctx.GetPosition(side.Pos()),
- TStringBuilder() << "Expected empty list"));
- return IGraphTransformer::TStatus::Error;
- }
- } else {
- if (side.ChildrenSize() < 2 || (side.ChildrenSize() % 2) != 0) {
- ctx.AddError(TIssue(ctx.GetPosition(side.Pos()),
- TStringBuilder() << "Expected non-empty list of atoms with even length"));
- return IGraphTransformer::TStatus::Error;
- }
- }
- keys.clear();
- for (ui32 i = 0; i < side.ChildrenSize(); i += 2) {
- auto table = side.Child(i)->Content();
- auto column = side.Child(i + 1)->Content();
- auto key = std::make_pair(table, column);
- keys.push_back(key);
- }
- for (auto& key : keys) {
- auto keyType = labels.FindColumn(key.first, key.second);
- if (!keyType) {
- ctx.AddError(TIssue(ctx.GetPosition(side.Pos()),
- TStringBuilder() << "Unknown column: " << key.second << " in correlation name: " << key.first));
- return IGraphTransformer::TStatus::Error;
- }
- if (!(*keyType)->IsHashable() || !(*keyType)->IsEquatable()) {
- ctx.AddError(TIssue(ctx.GetPosition(side.Pos()),
- TStringBuilder() << "Unsupported type of column: " << key.second << " in correlation name: " << key.first
- << ", type: " << *(*keyType)));
- return IGraphTransformer::TStatus::Error;
- }
- keyTypes.push_back(*keyType);
- }
- return IGraphTransformer::TStatus::Ok;
- }
- struct TGLobalJoinState {
- ui32 NestedJoins = 0;
- };
- bool AddEquiJoinLinkOptionHint(const std::string_view& side, std::unordered_set<std::string_view>& hints, const TExprNode& hintNode, TExprContext& ctx) {
- if (!EnsureAtom(hintNode, ctx)) {
- return false;
- }
- const auto pos = ctx.GetPosition(hintNode.Pos());
- if (hintNode.IsAtom({"unique", "small"})) {
- if (hints.contains(hintNode.IsAtom("small") ? "unique" : "small")) {
- ctx.AddError(TIssue(pos, TStringBuilder() << "Hints 'unique' and 'small' are not compatible"));
- return false;
- }
- } else if (!hintNode.IsAtom("any")) {
- ctx.AddError(TIssue(pos, TStringBuilder() << "Unknown hint: '" << hintNode.Content() << "' for " << side << " side"));
- return false;
- }
- hints.insert(hintNode.Content());
- return true;
- }
- IGraphTransformer::TStatus ParseJoins(const TJoinLabels& labels,
- TExprNode& joins, TVector<TJoinState>& joinsStates, THashSet<TStringBuf>& scope,
- TGLobalJoinState& globalState, bool strictKeys, TExprContext& ctx, const TUniqueConstraintNode** unique = nullptr, const TDistinctConstraintNode** distinct = nullptr);
- IGraphTransformer::TStatus ParseJoinScope(const TJoinLabels& labels,
- TExprNode& side, TVector<TJoinState>& joinsStates, THashSet<TStringBuf>& scope,
- TGLobalJoinState& globalState, bool strictKeys, const TUniqueConstraintNode*& unique, const TDistinctConstraintNode*& distinct, TExprContext& ctx) {
- if (side.IsAtom()) {
- const auto label = side.Content();
- const auto input = labels.FindInput(label);
- if (!input) {
- ctx.AddError(TIssue(ctx.GetPosition(side.Pos()),
- TStringBuilder() << "Unknown correlation name: " << label));
- return IGraphTransformer::TStatus::Error;
- }
- for (const auto& x : (*input)->Tables) {
- scope.insert(x);
- }
- const auto rename = [&](const TPartOfConstraintBase::TPathType& path) -> std::vector<TPartOfConstraintBase::TPathType> {
- if (path.empty())
- return {};
- auto newPath = path;
- newPath.front() = ctx.AppendString((*input)->FullName(newPath.front()));
- return {std::move(newPath)};
- };
- if (const auto u = (*input)->Unique) {
- unique = u->RenameFields(ctx, rename);
- }
- if (const auto d = (*input)->Distinct) {
- distinct = d->RenameFields(ctx, rename);
- }
- return IGraphTransformer::TStatus::Ok;
- }
- if (globalState.NestedJoins + 2 == labels.Inputs.size()) {
- ctx.AddError(TIssue(ctx.GetPosition(side.Pos()),
- TStringBuilder() << "Too many nested joins, expected exactly: " << (labels.Inputs.size() - 2)));
- return IGraphTransformer::TStatus::Error;
- }
- ++globalState.NestedJoins;
- return ParseJoins(labels, side, joinsStates, scope, globalState, strictKeys, ctx, &unique, &distinct);
- }
- IGraphTransformer::TStatus ParseJoins(const TJoinLabels& labels,
- TExprNode& joins, TVector<TJoinState>& joinsStates, THashSet<TStringBuf>& scope,
- TGLobalJoinState& globalState, bool strictKeys, TExprContext& ctx, const TUniqueConstraintNode** unique, const TDistinctConstraintNode** distinct) {
- if (!EnsureTupleSize(joins, 6, ctx)) {
- return IGraphTransformer::TStatus::Error;
- }
- const auto& joinType = joins.Head();
- if (!EnsureAtom(joinType, ctx)) {
- return IGraphTransformer::TStatus::Error;
- }
- if (!joinType.IsAtom({"Inner", "Left", "Right", "Full", "LeftOnly", "RightOnly", "Exclusion", "LeftSemi" , "RightSemi", "Cross"})) {
- ctx.AddError(TIssue(ctx.GetPosition(joinType.Pos()), TStringBuilder() << "Unsupported join type: " << joinType.Content()));
- return IGraphTransformer::TStatus::Error;
- }
- THashSet<TStringBuf> myLeftScope;
- const TUniqueConstraintNode* lUnique = nullptr;
- const TDistinctConstraintNode* lDistinct = nullptr;
- if (const auto status = ParseJoinScope(labels, *joins.Child(1), joinsStates, myLeftScope, globalState, strictKeys, lUnique, lDistinct, ctx); status.Level != IGraphTransformer::TStatus::Ok) {
- return status;
- }
- THashSet<TStringBuf> myRightScope;
- const TUniqueConstraintNode* rUnique = nullptr;
- const TDistinctConstraintNode* rDistinct = nullptr;
- if (const auto status = ParseJoinScope(labels, *joins.Child(2), joinsStates, myRightScope, globalState, strictKeys, rUnique, rDistinct, ctx); status.Level != IGraphTransformer::TStatus::Ok) {
- return status;
- }
- TVector<std::pair<TStringBuf, TStringBuf>> leftKeys;
- TVector<const TTypeAnnotationNode*> leftKeyTypes;
- const bool cross = joinType.IsAtom("Cross");
- if (const auto status = ParseJoinKeys(*joins.Child(3), leftKeys, leftKeyTypes, labels, ctx, cross); status.Level != IGraphTransformer::TStatus::Ok) {
- return status;
- }
- std::vector<std::string_view> lCheck;
- lCheck.reserve(leftKeys.size());
- for (const auto& x : leftKeys) {
- for (const auto& name : (*labels.FindInput(x.first))->AllNames(x.second))
- lCheck.emplace_back(ctx.AppendString(name));
- if (!myLeftScope.contains(x.first)) {
- ctx.AddError(TIssue(ctx.GetPosition(joins.Pos()),
- TStringBuilder() << "Correlation name " << x.first << " is out of scope"));
- return IGraphTransformer::TStatus::Error;
- }
- joinsStates[*labels.FindInputIndex(x.first)].Used = true;
- }
- TVector<std::pair<TStringBuf, TStringBuf>> rightKeys;
- TVector<const TTypeAnnotationNode*> rightKeyTypes;
- if (const auto status = ParseJoinKeys(*joins.Child(4), rightKeys, rightKeyTypes, labels, ctx, cross); status.Level != IGraphTransformer::TStatus::Ok) {
- return status;
- }
- std::vector<std::string_view> rCheck;
- rCheck.reserve(rightKeys.size());
- for (const auto& x : rightKeys) {
- for (const auto& name : (*labels.FindInput(x.first))->AllNames(x.second))
- rCheck.emplace_back(ctx.AppendString(name));
- if (!myRightScope.contains(x.first)) {
- ctx.AddError(TIssue(ctx.GetPosition(joins.Pos()),
- TStringBuilder() << "Correlation name " << x.first << " is out of scope"));
- return IGraphTransformer::TStatus::Error;
- }
- joinsStates[*labels.FindInputIndex(x.first)].Used = true;
- }
- if (leftKeys.size() != rightKeys.size()) {
- ctx.AddError(TIssue(ctx.GetPosition(joins.Pos()),
- TStringBuilder() << "Mismatch of key column count in equality between " << leftKeys.front().first
- << " and " << rightKeys.front().first));
- return IGraphTransformer::TStatus::Error;
- }
- for (auto i = 0U; i < leftKeyTypes.size(); ++i) {
- if (strictKeys && leftKeyTypes[i] != rightKeyTypes[i]) {
- ctx.AddError(TIssue(ctx.GetPosition(joins.Pos()),
- TStringBuilder() << "Strict key type match requested, but keys have different types: ("
- << leftKeys[i].first << "." << leftKeys[i].second
- << " has type: " << *leftKeyTypes[i] << ", " << rightKeys[i].first << "." << rightKeys[i].second
- << " has type: " << *rightKeyTypes[i] << ")"));
- return IGraphTransformer::TStatus::Error;
- }
- if (ECompareOptions::Uncomparable == CanCompare<true>(leftKeyTypes[i], rightKeyTypes[i])) {
- ctx.AddError(TIssue(ctx.GetPosition(joins.Pos()),
- TStringBuilder() << "Cannot compare key columns (" << leftKeys[i].first << "." << leftKeys[i].second
- << " has type: " << *leftKeyTypes[i] << ", " << rightKeys[i].first << "." << rightKeys[i].second
- << " has type: " << *rightKeyTypes[i] << ")"));
- return IGraphTransformer::TStatus::Error;
- }
- }
- if (cross) {
- for (const auto& x : myLeftScope) {
- joinsStates[*labels.FindInputIndex(x)].Used = true;
- }
- for (const auto& x : myRightScope) {
- joinsStates[*labels.FindInputIndex(x)].Used = true;
- }
- }
- scope.clear();
- const bool singleSide = joinType.Content().ends_with("Only") || joinType.Content().ends_with("Semi");
- const bool rightSide = joinType.Content().starts_with("Right");
- const bool leftSide = joinType.Content().starts_with("Left");
- if (!singleSide || !rightSide) {
- scope.insert(myLeftScope.cbegin(), myLeftScope.cend());
- }
- if (!singleSide || !leftSide) {
- scope.insert(myRightScope.cbegin(), myRightScope.cend());
- }
- const auto linkOptions = joins.Child(5);
- if (!EnsureTuple(*linkOptions, ctx)) {
- return IGraphTransformer::TStatus::Error;
- }
- std::optional<std::unordered_set<std::string_view>> leftHints, rightHints;
- bool hasJoinStrategyHint = false;
- for (auto child : linkOptions->Children()) {
- if (!EnsureTupleMinSize(*child, 1, ctx)) {
- return IGraphTransformer::TStatus::Error;
- }
- if (!EnsureAtom(child->Head(), ctx)) {
- return IGraphTransformer::TStatus::Error;
- }
- const auto& option = child->Head();
- if (option.IsAtom({"left", "right"})) {
- if (!EnsureTupleSize(*child, 2, ctx)) {
- return IGraphTransformer::TStatus::Error;
- }
- auto& hints = option.IsAtom("left") ? leftHints : rightHints;
- if (hints) {
- ctx.AddError(TIssue(ctx.GetPosition(option.Pos()), TStringBuilder() <<
- "Duplication of hints for " << option.Content() << " side"));
- return IGraphTransformer::TStatus::Error;
- }
- hints.emplace();
- if (child->Child(1)->IsAtom()) {
- if (!AddEquiJoinLinkOptionHint(option.Content(), *hints, *child->Child(1), ctx)) {
- return IGraphTransformer::TStatus::Error;
- }
- } else {
- if (!EnsureTuple(*child->Child(1), ctx)) {
- return IGraphTransformer::TStatus::Error;
- }
- for (auto hint : child->Child(1)->Children()) {
- if (!AddEquiJoinLinkOptionHint(option.Content(), *hints, *hint, ctx)) {
- return IGraphTransformer::TStatus::Error;
- }
- }
- }
- }
- else if (option.IsAtom("forceSortedMerge") || option.IsAtom("forceStreamLookup")) {
- if (!EnsureTupleSize(*child, 1, ctx)) {
- return IGraphTransformer::TStatus::Error;
- }
- if (hasJoinStrategyHint) {
- ctx.AddError(TIssue(ctx.GetPosition(option.Pos()), TStringBuilder() <<
- "Duplicate " << option.Content() << " link option"));
- return IGraphTransformer::TStatus::Error;
- }
- hasJoinStrategyHint = true;
- }
- else if (option.IsAtom("join_algo")) {
- //do nothing
- }
- else if (option.IsAtom("compact")) {
- if (!EnsureTupleSize(*child, 1, ctx)) {
- return IGraphTransformer::TStatus::Error;
- }
- }
- else {
- ctx.AddError(TIssue(ctx.GetPosition(option.Pos()), TStringBuilder() <<
- "Unknown option name: " << option.Content()));
- return IGraphTransformer::TStatus::Error;
- }
- }
- const bool lAny = leftHints && (leftHints->contains("unique") || leftHints->contains("any"));
- const bool rAny = rightHints && (rightHints->contains("unique") || rightHints->contains("any"));
- const bool lOneRow = lAny || lUnique && lUnique->ContainsCompleteSet(lCheck);
- const bool rOneRow = rAny || rUnique && rUnique->ContainsCompleteSet(rCheck);
- if (unique) {
- if (singleSide) {
- if (leftSide)
- *unique = lUnique;
- else if (rightSide)
- *unique = rUnique;
- } else if (!joinType.IsAtom("Cross")) {
- const bool exclusion = joinType.IsAtom("Exclusion") ;
- const bool useLeft = lUnique && (rOneRow || exclusion);
- const bool useRight = rUnique && (lOneRow || exclusion);
- if (useLeft && !useRight)
- *unique = lUnique;
- else if (useRight && !useLeft)
- *unique = rUnique;
- else if (useLeft && useRight)
- *unique = TUniqueConstraintNode::Merge(lUnique, rUnique, ctx);
- }
- }
- if (distinct) {
- if (singleSide) {
- if (leftSide)
- *distinct = lDistinct;
- else if (rightSide)
- *distinct = rDistinct;
- } else if (!joinType.IsAtom("Cross")) {
- const bool inner = joinType.IsAtom("Inner");
- const bool useLeft = lDistinct && rOneRow && (inner || leftSide);
- const bool useRight = rDistinct && lOneRow && (inner || rightSide);
- if (useLeft && !useRight)
- *distinct = lDistinct;
- else if (useRight && !useLeft)
- *distinct = rDistinct;
- else if (useLeft && useRight)
- *distinct = TDistinctConstraintNode::Merge(lDistinct, rDistinct, ctx);
- }
- }
- return IGraphTransformer::TStatus::Ok;
- }
- struct TFlattenState {
- TString Table;
- TTypeAnnotationNode::TListType AllTypes;
- };
- void CollectEquiJoinKeyColumnsFromLeaf(const TExprNode& columns, THashMap<TStringBuf, THashSet<TStringBuf>>& tableKeysMap) {
- YQL_ENSURE(columns.ChildrenSize() % 2 == 0);
- for (ui32 i = 0; i < columns.ChildrenSize(); i += 2) {
- auto table = columns.Child(i)->Content();
- auto column = columns.Child(i + 1)->Content();
- tableKeysMap[table].insert(column);
- }
- }
- void CollectEquiJoinKeyColumns(const TExprNode& joinTree, THashMap<TStringBuf, THashSet<TStringBuf>>& tableKeysMap) {
- auto& left = *joinTree.Child(1);
- if (!left.IsAtom()) {
- CollectEquiJoinKeyColumns(left, tableKeysMap);
- }
- auto& right = *joinTree.Child(2);
- if (!right.IsAtom()) {
- CollectEquiJoinKeyColumns(right, tableKeysMap);
- }
- CollectEquiJoinKeyColumnsFromLeaf(*joinTree.Child(3), tableKeysMap);
- CollectEquiJoinKeyColumnsFromLeaf(*joinTree.Child(4), tableKeysMap);
- }
- bool CollectEquiJoinOnlyParents(const TExprNode& current, const TExprNode* prev, ui32 depth,
- TVector<TEquiJoinParent>& results, const TExprNode* extractMembersInScope,
- const TParentsMap& parents)
- {
- if (depth == 0) {
- if (!prev || !TCoEquiJoin::Match(¤t)) {
- return false;
- }
- TCoEquiJoin equiJoin(¤t);
- for (ui32 i = 0; i + 2 < equiJoin.ArgCount(); ++i) {
- auto joinInput = equiJoin.Arg(i).Cast<TCoEquiJoinInput>();
- auto list = joinInput.List();
- if (list.Raw() == prev) {
- results.emplace_back(equiJoin.Raw(), i, extractMembersInScope);
- return true;
- }
- }
- YQL_ENSURE(false, "Unable to locate FlatMap in EquiJoin");
- }
- auto it = parents.find(¤t);
- if (it == parents.end() || it->second.empty()) {
- return false;
- }
- const TExprNode* extractMembers = extractMembersInScope;
- bool currentIsExtractMembers = TCoExtractMembers::Match(¤t);
- if (currentIsExtractMembers) {
- if (extractMembers) {
- // repeatable extract members should not actually happen
- return false;
- }
- extractMembers = current.Child(1);
- }
- auto nextPrev = (TCoFlatMapBase::Match(¤t) || currentIsExtractMembers) ? ¤t : prev;
- for (auto parent : it->second) {
- if (!CollectEquiJoinOnlyParents(*parent, nextPrev, currentIsExtractMembers ? depth : (depth - 1), results,
- extractMembers, parents))
- {
- return false;
- }
- }
- return true;
- }
- }
- TMaybe<TIssue> TJoinLabel::Parse(TExprContext& ctx, TExprNode& node, const TStructExprType* structType, const TUniqueConstraintNode* unique, const TDistinctConstraintNode* distinct) {
- Tables.clear();
- InputType = structType;
- Unique = unique;
- Distinct = distinct;
- if (auto atom = TMaybeNode<TCoAtom>(&node)) {
- if (auto err = ValidateLabel(ctx, atom.Cast())) {
- return err;
- }
- AddLabel = true;
- Tables.push_back(atom.Cast().Value());
- return {};
- }
- else if (auto tuple = TMaybeNode<TCoAtomList>(&node)) {
- if (tuple.Cast().Size() == 0) {
- return TIssue(ctx.GetPosition(node.Pos()), "Empty list of correlation names are not allowed");
- }
- for (const auto& child : tuple.Cast()) {
- if (auto err = ValidateLabel(ctx, child)) {
- return err;
- }
- Tables.push_back(child.Value());
- }
- Sort(Tables);
- auto prevLabel = Tables[0];
- for (ui32 i = 1; i < Tables.size(); ++i) {
- if (Tables[i] == prevLabel) {
- return TIssue(ctx.GetPosition(node.Pos()), TStringBuilder() << "Duplication of correlation names: " << prevLabel);
- }
- prevLabel = Tables[i];
- }
- // all labels are unique, ensure that all columns are under one of label
- for (auto column : InputType->GetItems()) {
- auto name = column->GetName();
- auto pos = name.find('.');
- if (pos == TString::npos) {
- return TIssue(ctx.GetPosition(node.Pos()), TStringBuilder() << "Expected columns name as table.name, but got: " << name);
- }
- auto table = name.substr(0, pos);
- if (!BinarySearch(Tables.begin(), Tables.end(), table)) {
- return TIssue(ctx.GetPosition(node.Pos()), TStringBuilder() << "Unknown table name: " << table);
- }
- auto columnName = name.substr(pos + 1);
- if (columnName.empty()) {
- return TIssue(ctx.GetPosition(node.Pos()), "Empty correlation name is not allowed");
- }
- }
- return {};
- }
- else {
- return TIssue(ctx.GetPosition(node.Pos()), TStringBuilder() << "Expected either atom or list, but got" << node.Type());
- }
- }
- TMaybe<TIssue> TJoinLabel::ValidateLabel(TExprContext& ctx, const TCoAtom& label) {
- if (label.Value().empty()) {
- return TIssue(ctx.GetPosition(label.Pos()), "Empty correlation name is not allowed");
- }
- if (label.Value().Contains('.')) {
- return TIssue(ctx.GetPosition(label.Pos()), "Dot symbol is not allowed in the correlation name");
- }
- return {};
- }
- TString TJoinLabel::FullName(const TStringBuf& column) const {
- if (AddLabel) {
- return FullColumnName(Tables.front(), column);
- } else {
- return TString(column);
- }
- }
- TVector<TString> TJoinLabel::AllNames(const TStringBuf& column) const {
- TVector<TString> result(Tables.size());
- std::transform(Tables.cbegin(), Tables.cend(), result.begin(), std::bind(&FullColumnName, std::placeholders::_1, std::cref(column)));
- return result;
- }
- TStringBuf TJoinLabel::ColumnName(const TStringBuf& column) const {
- auto pos = column.find('.');
- if (pos == TString::npos) {
- return column;
- }
- return column.substr(pos + 1);
- }
- TStringBuf TJoinLabel::TableName(const TStringBuf& column) const {
- auto pos = column.find('.');
- if (pos == TString::npos) {
- YQL_ENSURE(AddLabel);
- return Tables[0];
- }
- return column.substr(0, pos);
- }
- bool TJoinLabel::HasTable(const TStringBuf& table) const {
- return BinarySearch(Tables.begin(), Tables.end(), table);
- }
- TMaybe<const TTypeAnnotationNode*> TJoinLabel::FindColumn(const TStringBuf& table, const TStringBuf& column) const {
- auto pos = InputType->FindItem(MemberName(table, column));
- if (!pos) {
- return TMaybe<const TTypeAnnotationNode*>();
- }
- return InputType->GetItems()[*pos]->GetItemType();
- }
- TString TJoinLabel::MemberName(const TStringBuf& table, const TStringBuf& column) const {
- return AddLabel ? TString(column) : FullColumnName(table, column);
- }
- TVector<TString> TJoinLabel::EnumerateAllColumns() const {
- TVector<TString> result;
- if (AddLabel) {
- // add label to all columns
- for (auto& x : InputType->GetItems()) {
- result.push_back(FullColumnName(Tables[0], x->GetName()));
- }
- } else {
- for (auto& x : InputType->GetItems()) {
- result.push_back(TString(x->GetName()));
- }
- }
- return result;
- }
- TVector<TString> TJoinLabel::EnumerateAllMembers() const {
- TVector<TString> result;
- for (auto& x : InputType->GetItems()) {
- result.push_back(TString(x->GetName()));
- }
- return result;
- }
- TMaybe<TIssue> TJoinLabels::Add(TExprContext& ctx, TExprNode& node, const TStructExprType* structType, const TUniqueConstraintNode* unique, const TDistinctConstraintNode* distinct) {
- ui32 index = Inputs.size();
- Inputs.emplace_back();
- TJoinLabel& label = Inputs.back();
- if (auto err = label.Parse(ctx, node, structType, unique, distinct)) {
- return err;
- }
- for (auto& table : label.Tables) {
- if (!InputByTable.insert({ table, index }).second) {
- return TIssue(
- ctx.GetPosition(node.Pos()),
- TStringBuilder() << "Duplication of table name " << table);
- }
- }
- return {};
- }
- TMaybe<const TJoinLabel*> TJoinLabels::FindInput(const TStringBuf& table) const {
- auto inputIndex = InputByTable.FindPtr(table);
- if (!inputIndex) {
- return {};
- }
- return &Inputs[*inputIndex];
- }
- TMaybe<ui32> TJoinLabels::FindInputIndex(const TStringBuf& table) const {
- auto inputIndex = InputByTable.FindPtr(table);
- if (!inputIndex) {
- return{};
- }
- return *inputIndex;
- }
- TMaybe<const TTypeAnnotationNode*> TJoinLabels::FindColumn(const TStringBuf& table, const TStringBuf& column) const {
- auto tableIndex = InputByTable.FindPtr(table);
- if (!tableIndex) {
- return TMaybe<const TTypeAnnotationNode*>();
- }
- return Inputs[*tableIndex].FindColumn(table, column);
- }
- TMaybe<const TTypeAnnotationNode*> TJoinLabels::FindColumn(const TStringBuf& fullName) const {
- TStringBuf part1;
- TStringBuf part2;
- SplitTableName(fullName, part1, part2);
- return FindColumn(part1, part2);
- }
- TVector<TString> TJoinLabels::EnumerateColumns(const TStringBuf& table) const {
- TVector<TString> result;
- auto tableIndex = InputByTable.FindPtr(table);
- Y_ENSURE(tableIndex, "Unknown table:" << table);
- auto& label = Inputs[*tableIndex];
- if (label.AddLabel) {
- // add label to all columns
- for (auto& x : label.InputType->GetItems()) {
- result.push_back(FullColumnName(table, x->GetName()));
- }
- }
- else {
- // filter out some columns
- for (auto& x : label.InputType->GetItems()) {
- TStringBuf part1;
- TStringBuf part2;
- SplitTableName(x->GetName(), part1, part2);
- if (part1 == table) {
- result.push_back(TString(x->GetName()));
- }
- }
- }
- return result;
- }
- IGraphTransformer::TStatus ValidateEquiJoinOptions(TPositionHandle positionHandle, TExprNode& optionsNode,
- TJoinOptions& options, TExprContext& ctx)
- {
- auto position = ctx.GetPosition(positionHandle);
- if (!EnsureTuple(optionsNode, ctx)) {
- return IGraphTransformer::TStatus::Error;
- }
- options = TJoinOptions{};
- THashSet<TStringBuf> renameTargetSet;
- bool hasRename = false;
- for (auto child : optionsNode.Children()) {
- if (!EnsureTupleMinSize(*child, 1, ctx)) {
- return IGraphTransformer::TStatus::Error;
- }
- if (!EnsureAtom(*child->Child(0), ctx)) {
- return IGraphTransformer::TStatus::Error;
- }
- auto optionName = child->Child(0)->Content();
- if (optionName == "rename") {
- hasRename = true;
- if (!EnsureTupleSize(*child, 3, ctx)) {
- return IGraphTransformer::TStatus::Error;
- }
- if (!EnsureAtom(*child->Child(1), ctx)) {
- return IGraphTransformer::TStatus::Error;
- }
- if (!EnsureAtom(*child->Child(2), ctx)) {
- return IGraphTransformer::TStatus::Error;
- }
- auto& v = options.RenameMap[child->Child(1)->Content()];
- if (!child->Child(2)->Content().empty()) {
- if (!renameTargetSet.insert(child->Child(2)->Content()).second) {
- ctx.AddError(TIssue(position, TStringBuilder() <<
- "Duplicated target column: " << child->Child(2)->Content()));
- return IGraphTransformer::TStatus::Error;
- }
- v.push_back(child->Child(2)->Content());
- }
- } else if (optionName == "flatten") {
- options.Flatten = true;
- } else if (optionName == "strict_keys") {
- options.StrictKeys = true;
- } else if (optionName == "preferred_sort") {
- THashSet<TStringBuf> sortBySet;
- TVector<TStringBuf> sortBy;
- if (!EnsureTupleSize(*child, 2, ctx)) {
- return IGraphTransformer::TStatus::Error;
- }
- if (!EnsureTupleMinSize(*child->Child(1), 1, ctx)) {
- return IGraphTransformer::TStatus::Error;
- }
- for (auto column : child->Child(1)->Children()) {
- if (!EnsureAtom(*column, ctx)) {
- return IGraphTransformer::TStatus::Error;
- }
- if (!sortBySet.insert(column->Content()).second) {
- ctx.AddError(TIssue(ctx.GetPosition(column->Pos()), TStringBuilder() <<
- "Duplicated preferred_sort column: " << column->Content()));
- return IGraphTransformer::TStatus::Error;
- }
- sortBy.push_back(column->Content());
- }
- if (!options.PreferredSortSets.insert(sortBy).second) {
- ctx.AddError(TIssue(ctx.GetPosition(child->Child(1)->Pos()), TStringBuilder() <<
- "Duplicated preferred_sort set: " << JoinSeq(", ", sortBy)));
- }
- } else if (optionName == "cbo_passed") {
- // do nothing
- } else if (optionName == "join_algo") {
- // do nothing
- } else if (optionName == "compact") {
- options.Compact = true;
- } else {
- ctx.AddError(TIssue(position, TStringBuilder() <<
- "Unknown option name: " << optionName));
- return IGraphTransformer::TStatus::Error;
- }
- if (hasRename && options.Flatten) {
- ctx.AddError(TIssue(position, TStringBuilder() <<
- "Options flatten and rename are incompatible with each other"));
- return IGraphTransformer::TStatus::Error;
- }
- }
- return IGraphTransformer::TStatus::Ok;
- }
- IGraphTransformer::TStatus EquiJoinAnnotation(
- TPositionHandle positionHandle,
- const TStructExprType*& resultType,
- const TJoinLabels& labels,
- TExprNode& joins,
- const TJoinOptions& options,
- TExprContext& ctx
- ) {
- auto position = ctx.GetPosition(positionHandle);
- if (labels.InputByTable.size() < 2) {
- ctx.AddError(TIssue(position, TStringBuilder() << "Expected at least 2 table"));
- return IGraphTransformer::TStatus::Error;
- }
- TVector<TJoinState> joinsStates(labels.Inputs.size());
- TGLobalJoinState globalState;
- THashSet<TStringBuf> scope;
- auto parseStatus = ParseJoins(labels, joins, joinsStates, scope, globalState, options.StrictKeys, ctx);
- if (parseStatus.Level != IGraphTransformer::TStatus::Ok) {
- return parseStatus;
- }
- if (globalState.NestedJoins + 2 != labels.Inputs.size()) {
- ctx.AddError(TIssue(position,
- TStringBuilder() << "Too few nested joins, expected exactly: " << (labels.Inputs.size() - 2)));
- return IGraphTransformer::TStatus::Error;
- }
- for (ui32 i = 0; i < joinsStates.size(); ++i) {
- if (!joinsStates[i].Used) {
- ctx.AddError(TIssue(position, TStringBuilder() <<
- "Input with correlation name(s) " << JoinSeq(", ", labels.Inputs[i].Tables) << " was not used"));
- return IGraphTransformer::TStatus::Error;
- }
- }
- auto columnTypes = GetJoinColumnTypes(joins, labels, ctx);
- TVector<const TItemExprType*> resultFields;
- TMap<TString, TFlattenState> flattenFields; // column -> table
- THashSet<TString> processedRenames;
- for (auto it: labels.Inputs) {
- for (auto item: it.InputType->GetItems()) {
- TString fullName = it.FullName(item->GetName());
- auto type = columnTypes.FindPtr(fullName);
- if (type) {
- TVector<TStringBuf> fullNames;
- fullNames.push_back(fullName);
- if (!processedRenames.contains(fullName)) {
- auto renameIt = options.RenameMap.find(fullName);
- if (renameIt != options.RenameMap.end()) {
- fullNames = renameIt->second;
- processedRenames.insert(fullName);
- }
- }
- for (auto& fullName: fullNames) {
- if (options.Flatten) {
- auto tableName = it.TableName(fullName);
- auto columnName = it.ColumnName(fullName);
- auto iter = flattenFields.find(columnName);
- if (iter != flattenFields.end()) {
- if (AreSameJoinKeys(joins, tableName, columnName, iter->second.Table, columnName)) {
- iter->second.AllTypes.push_back(*type);
- continue;
- }
- ctx.AddError(TIssue(position, TStringBuilder() <<
- "Conflict of flattening output on columns " << fullName << " and " << iter->second.Table
- << "." << columnName));
- return IGraphTransformer::TStatus::Error;
- }
- TFlattenState state;
- state.AllTypes.push_back(*type);
- state.Table = TString(tableName);
- flattenFields.emplace(TString(columnName), state);
- } else {
- resultFields.push_back(ctx.MakeType<TItemExprType>(fullName, *type));
- }
- }
- }
- }
- }
- if (options.Flatten) {
- for (auto& x : flattenFields) {
- if (const auto commonType = CommonType(positionHandle, x.second.AllTypes, ctx)) {
- const bool unwrap = ETypeAnnotationKind::Optional == commonType->GetKind() &&
- std::any_of(x.second.AllTypes.cbegin(), x.second.AllTypes.cend(), [](const TTypeAnnotationNode* type) { return ETypeAnnotationKind::Optional != type->GetKind(); });
- resultFields.emplace_back(ctx.MakeType<TItemExprType>(x.first, unwrap ? commonType->Cast<TOptionalExprType>()->GetItemType() : commonType));
- } else
- return IGraphTransformer::TStatus::Error;
- }
- }
- resultType = ctx.MakeType<TStructExprType>(resultFields);
- if (!resultType->Validate(position, ctx)) {
- return IGraphTransformer::TStatus::Error;
- }
- return IGraphTransformer::TStatus::Ok;
- }
- IGraphTransformer::TStatus EquiJoinConstraints(
- TPositionHandle positionHandle,
- const TUniqueConstraintNode*& unique,
- const TDistinctConstraintNode*& distinct,
- const TJoinLabels& labels,
- TExprNode& joins,
- TExprContext& ctx
- ) {
- const auto position = ctx.GetPosition(positionHandle);
- YQL_ENSURE(labels.InputByTable.size() >= 2U);
- TVector<TJoinState> joinsStates(labels.Inputs.size());
- TGLobalJoinState globalState;
- THashSet<TStringBuf> scope;
- if (const auto parseStatus = ParseJoins(labels, joins, joinsStates, scope, globalState, false, ctx, &unique, &distinct); parseStatus.Level != IGraphTransformer::TStatus::Ok) {
- return parseStatus;
- }
- return IGraphTransformer::TStatus::Ok;
- }
- THashMap<TStringBuf, THashSet<TStringBuf>> CollectEquiJoinKeyColumnsByLabel(const TExprNode& joinTree) {
- THashMap<TStringBuf, THashSet<TStringBuf>> result;
- CollectEquiJoinKeyColumns(joinTree, result);
- return result;
- };
- bool IsLeftJoinSideOptional(const TStringBuf& joinType) {
- if (joinType == "Right" || joinType == "Full" || joinType == "Exclusion") {
- return true;
- }
- return false;
- }
- bool IsRightJoinSideOptional(const TStringBuf& joinType) {
- if (joinType == "Left" || joinType == "Full" || joinType == "Exclusion") {
- return true;
- }
- return false;
- }
- TExprNode::TPtr FilterOutNullJoinColumns(TPositionHandle pos, const TExprNode::TPtr& input,
- const TJoinLabel& label, const TSet<TString>& optionalKeyColumns, TExprContext& ctx) {
- if (optionalKeyColumns.empty()) {
- return input;
- }
- TExprNode::TListType optColumns;
- for (auto fullColumnName : optionalKeyColumns) {
- TStringBuf table;
- TStringBuf column;
- SplitTableName(fullColumnName, table, column);
- auto memberName = label.MemberName(table, column);
- optColumns.push_back(ctx.NewAtom(pos, memberName));
- }
- return ctx.Builder(pos)
- .Callable("SkipNullMembers")
- .Add(0, input)
- .List(1)
- .Add(std::move(optColumns))
- .Seal()
- .Seal()
- .Build();
- }
- TMap<TStringBuf, TVector<TStringBuf>> LoadJoinRenameMap(const TExprNode& settings) {
- TMap<TStringBuf, TVector<TStringBuf>> res;
- for (const auto& child : settings.Children()) {
- if (child->Head().IsAtom("rename")) {
- auto& v = res[child->Child(1)->Content()];
- if (!child->Child(2)->Content().empty()) {
- v.push_back(child->Child(2)->Content());
- }
- }
- }
- return res;
- }
- TCoLambda BuildJoinRenameLambda(TPositionHandle pos, const TMap<TStringBuf, TVector<TStringBuf>>& renameMap,
- const TStructExprType& joinResultType, TExprContext& ctx)
- {
- THashMap<TStringBuf, TStringBuf> reverseRenameMap;
- for (const auto& [oldName , targets] : renameMap) {
- for (TStringBuf newName : targets) {
- reverseRenameMap[newName] = oldName;
- }
- }
- TCoArgument rowArg = Build<TCoArgument>(ctx, pos)
- .Name("row")
- .Done();
- TVector<TExprBase> renameTuples;
- for (auto& item : joinResultType.GetItems()) {
- TStringBuf newName = item->GetName();
- auto renamedFrom = reverseRenameMap.FindPtr(newName);
- TStringBuf oldName = renamedFrom ? *renamedFrom : newName;
- auto tuple = Build<TCoNameValueTuple>(ctx, pos)
- .Name().Build(newName)
- .Value<TCoMember>()
- .Struct(rowArg)
- .Name().Build(oldName)
- .Build()
- .Done();
- renameTuples.push_back(tuple);
- }
- return Build<TCoLambda>(ctx, pos)
- .Args({rowArg})
- .Body<TCoAsStruct>()
- .Add(renameTuples)
- .Build()
- .Done();
- }
- TSet<TVector<TStringBuf>> LoadJoinSortSets(const TExprNode& settings) {
- TSet<TVector<TStringBuf>> res;
- for (const auto& child : settings.Children()) {
- if (child->Child(0)->Content() == "preferred_sort") {
- TVector<TStringBuf> sortBy;
- for (auto column : child->Child(1)->Children()) {
- sortBy.push_back(column->Content());
- }
- res.insert(sortBy);
- }
- }
- return res;
- }
- THashMap<TString, const TTypeAnnotationNode*> GetJoinColumnTypes(const TExprNode& joins,
- const TJoinLabels& labels, TExprContext& ctx) {
- return GetJoinColumnTypes(joins, labels, joins.Child(0)->Content(), ctx);
- }
- THashMap<TString, const TTypeAnnotationNode*> GetJoinColumnTypes(const TExprNode& joins,
- const TJoinLabels& labels, const TStringBuf& joinType, TExprContext& ctx) {
- THashMap<TString, const TTypeAnnotationNode*> finalType;
- THashMap<TString, const TTypeAnnotationNode*> leftType;
- THashMap<TString, const TTypeAnnotationNode*> rightType;
- bool isLeftOptional = IsLeftJoinSideOptional(joinType);
- bool isRightOptional = IsRightJoinSideOptional(joinType);
- if (joins.Child(1)->IsAtom()) {
- auto name = joins.Child(1)->Content();
- auto input = *labels.FindInput(name);
- for (auto& x : input->InputType->GetItems()) {
- leftType[input->FullName(x->GetName())] = x->GetItemType();
- }
- }
- else {
- leftType = GetJoinColumnTypes(*joins.Child(1), labels, ctx);
- }
- if (joins.Child(2)->IsAtom()) {
- auto name = joins.Child(2)->Content();
- auto input = *labels.FindInput(name);
- for (auto& x : input->InputType->GetItems()) {
- rightType[input->FullName(x->GetName())] = x->GetItemType();
- }
- }
- else {
- rightType = GetJoinColumnTypes(*joins.Child(2), labels, ctx);
- }
- if (isLeftOptional) {
- for (auto& x : leftType) {
- x.second = AddOptionalType(x.second, ctx);
- }
- }
- if (isRightOptional) {
- for (auto& x : rightType) {
- x.second = AddOptionalType(x.second, ctx);
- }
- }
- if (joinType != "RightOnly" && joinType != "RightSemi") {
- for (auto& x : leftType) {
- finalType.insert({ x.first, x.second });
- }
- }
- if (joinType != "LeftOnly" && joinType != "LeftSemi") {
- for (auto& x : rightType) {
- finalType.insert({ x.first, x.second });
- }
- }
- return finalType;
- }
- bool AreSameJoinKeys(const TExprNode& joins, const TStringBuf& table1, const TStringBuf& column1, const TStringBuf& table2, const TStringBuf& column2) {
- if (!joins.Child(1)->IsAtom()) {
- if (AreSameJoinKeys(*joins.Child(1), table1, column1, table2, column2)) {
- return true;
- }
- }
- if (!joins.Child(2)->IsAtom()) {
- if (AreSameJoinKeys(*joins.Child(2), table1, column1, table2, column2)) {
- return true;
- }
- }
- for (ui32 i = 0; i < joins.Child(3)->ChildrenSize(); i += 2) {
- if (joins.Child(3)->Child(i)->Content() == table1) {
- if (joins.Child(4)->Child(i)->Content() == table2 &&
- joins.Child(3)->Child(i + 1)->Content() == column1 &&
- joins.Child(4)->Child(i + 1)->Content() == column2) {
- return true;
- }
- }
- else if (joins.Child(3)->Child(i)->Content() == table2) {
- if (joins.Child(4)->Child(i)->Content() == table1 &&
- joins.Child(3)->Child(i + 1)->Content() == column2 &&
- joins.Child(4)->Child(i + 1)->Content() == column1) {
- return true;
- }
- }
- }
- return false;
- }
- std::pair<bool, bool> IsRequiredSide(const TExprNode::TPtr& joinTree, const TJoinLabels& labels, ui32 inputIndex) {
- auto joinType = joinTree->Child(0)->Content();
- auto left = joinTree->ChildPtr(1);
- auto right = joinTree->ChildPtr(2);
- if (joinType == "Inner" || joinType == "Left" || joinType == "LeftOnly" || joinType == "LeftSemi" || joinType == "RightSemi" || joinType == "Cross") {
- if (!left->IsAtom()) {
- auto x = IsRequiredSide(left, labels, inputIndex);
- if (x.first) {
- return x;
- }
- }
- else {
- auto table = left->Content();
- if (*labels.FindInputIndex(table) == inputIndex) {
- return { true, joinType == "Inner" || joinType == "LeftSemi" };
- }
- }
- }
- if (joinType == "Inner" || joinType == "Right" || joinType == "RightOnly" || joinType == "RightSemi" || joinType == "LeftSemi" || joinType == "Cross") {
- if (!right->IsAtom()) {
- auto x = IsRequiredSide(right, labels, inputIndex);
- if (x.first) {
- return x;
- }
- }
- else {
- auto table = right->Content();
- if (*labels.FindInputIndex(table) == inputIndex) {
- return{ true, joinType == "Inner" || joinType == "RightSemi"};
- }
- }
- }
- return{ false, false };
- }
- TMaybe<bool> IsFilteredSide(const TExprNode::TPtr& joinTree, const TJoinLabels& labels, ui32 inputIndex) {
- auto joinType = joinTree->Child(0)->Content();
- auto left = joinTree->ChildPtr(1);
- auto right = joinTree->ChildPtr(2);
- TMaybe<bool> isLeftFiltered;
- if (!left->IsAtom()) {
- isLeftFiltered = IsFilteredSide(left, labels, inputIndex);
- } else {
- auto table = left->Content();
- if (*labels.FindInputIndex(table) == inputIndex) {
- if (joinType == "Inner" || joinType == "LeftOnly" || joinType == "LeftSemi") {
- isLeftFiltered = true;
- } else if (joinType != "RightOnly" && joinType != "RightSemi") {
- isLeftFiltered = false;
- }
- }
- }
- TMaybe<bool> isRightFiltered;
- if (!right->IsAtom()) {
- isRightFiltered = IsFilteredSide(right, labels, inputIndex);
- } else {
- auto table = right->Content();
- if (*labels.FindInputIndex(table) == inputIndex) {
- if (joinType == "Inner" || joinType == "RightOnly" || joinType == "RightSemi") {
- isRightFiltered = true;
- } else if (joinType != "LeftOnly" && joinType != "LeftSemi") {
- isRightFiltered = false;
- }
- }
- }
- YQL_ENSURE(!(isLeftFiltered.Defined() && isRightFiltered.Defined()));
- if (!isLeftFiltered.Defined() && !isRightFiltered.Defined()) {
- return {};
- }
- return isLeftFiltered.Defined() ? isLeftFiltered : isRightFiltered;
- }
- void AppendEquiJoinRenameMap(TPositionHandle pos, const TMap<TStringBuf, TVector<TStringBuf>>& newRenameMap,
- TExprNode::TListType& joinSettingNodes, TExprContext& ctx) {
- for (auto& x : newRenameMap) {
- if (x.second.empty()) {
- joinSettingNodes.push_back(ctx.Builder(pos)
- .List()
- .Atom(0, "rename")
- .Atom(1, x.first)
- .Atom(2, "")
- .Seal()
- .Build());
- continue;
- }
- for (auto& y : x.second) {
- if (x.first == y && x.second.size() == 1) {
- continue;
- }
- joinSettingNodes.push_back(ctx.Builder(pos)
- .List()
- .Atom(0, "rename")
- .Atom(1, x.first)
- .Atom(2, y)
- .Seal()
- .Build());
- }
- }
- }
- void AppendEquiJoinSortSets(TPositionHandle pos, const TSet<TVector<TStringBuf>>& newSortSets,
- TExprNode::TListType& joinSettingNodes, TExprContext& ctx)
- {
- for (auto& ss : newSortSets) {
- YQL_ENSURE(!ss.empty());
- joinSettingNodes.push_back(ctx.Builder(pos)
- .List()
- .Atom(0, "preferred_sort")
- .List(1)
- .Do([&](TExprNodeBuilder& parent) -> TExprNodeBuilder& {
- for (ui32 i = 0; i < ss.size(); ++i) {
- parent.Atom(i, ss[i]);
- }
- return parent;
- })
- .Seal()
- .Seal()
- .Build());
- }
- }
- TMap<TStringBuf, TVector<TStringBuf>> UpdateUsedFieldsInRenameMap(
- const TMap<TStringBuf, TVector<TStringBuf>>& renameMap,
- const TSet<TStringBuf>& usedFields,
- const TStructExprType* structType
- ) {
- TMap<TStringBuf, TStringBuf> reversedRenameMap;
- TMap<TStringBuf, TVector<TStringBuf>> newRenameMap;
- for (auto& x : renameMap) {
- if (!x.second.empty()) {
- for (auto& y : x.second) {
- reversedRenameMap[y] = x.first;
- }
- }
- else {
- // previous drops
- newRenameMap[x.first].clear();
- }
- }
- for (auto& item : structType->GetItems()) {
- bool needRemove = !usedFields.contains(item->GetName());
- if (auto renamed = reversedRenameMap.FindPtr(item->GetName())) {
- if (needRemove) {
- if (newRenameMap[*renamed].empty()) {
- newRenameMap[*renamed].push_back("");
- }
- }
- else {
- if (!newRenameMap[*renamed].empty() && newRenameMap[*renamed][0].empty()) {
- newRenameMap[*renamed].clear(); // Do not remove column because it will be renamed.
- }
- newRenameMap[*renamed].push_back(item->GetName());
- }
- }
- else {
- if (needRemove) {
- newRenameMap[item->GetName()].push_back("");
- }
- }
- }
- for (auto& x : newRenameMap) {
- if (AnyOf(x.second, [](const TStringBuf& value) { return !value.empty(); })) {
- continue;
- }
- x.second.clear();
- }
- return newRenameMap;
- }
- TVector<TEquiJoinParent> CollectEquiJoinOnlyParents(const TCoFlatMapBase& flatMap, const TParentsMap& parents)
- {
- TVector<TEquiJoinParent> result;
- if (!CollectEquiJoinOnlyParents(flatMap.Ref(), nullptr, 2, result, nullptr, parents)) {
- result.clear();
- }
- return result;
- }
- TEquiJoinLinkSettings GetEquiJoinLinkSettings(const TExprNode& linkSettings) {
- TEquiJoinLinkSettings result;
- result.Pos = linkSettings.Pos();
- auto collectHints = [](TSet<TString>& hints, const TExprNode& hintsNode) {
- if (hintsNode.IsAtom()) {
- hints.insert(ToString(hintsNode.Content()));
- } else {
- for (auto h : hintsNode.Children()) {
- YQL_ENSURE(h->IsAtom());
- hints.insert(ToString(h->Content()));
- }
- }
- };
- if (auto left = GetSetting(linkSettings, "left")) {
- collectHints(result.LeftHints, *left->Child(1));
- }
- if (auto right = GetSetting(linkSettings, "right")) {
- collectHints(result.RightHints, *right->Child(1));
- }
- if (auto algo = GetSetting(linkSettings, "join_algo")) {
- YQL_ENSURE(algo->Child(1)->IsAtom());
- result.JoinAlgo = FromString<EJoinAlgoType>(algo->Child(1)->Content());
- }
- result.ForceSortedMerge = HasSetting(linkSettings, "forceSortedMerge");
-
- if (HasSetting(linkSettings, "forceStreamLookup")) {
- result.JoinAlgo = EJoinAlgoType::StreamLookupJoin;
- }
- if (HasSetting(linkSettings, "compact")) {
- result.Compact = true;
- }
- return result;
- }
- TExprNode::TPtr BuildEquiJoinLinkSettings(const TEquiJoinLinkSettings& linkSettings, TExprContext& ctx) {
- auto builder = [&](const TStringBuf& side) -> TExprNode::TPtr {
- return ctx.Builder(linkSettings.Pos)
- .List()
- .Atom(0, side)
- .List(1)
- .Do([&](TExprNodeBuilder& parent) -> TExprNodeBuilder& {
- ui32 i = 0;
- for (auto h : (side == "left" ? linkSettings.LeftHints : linkSettings.RightHints)) {
- parent.Atom(i++, h);
- }
- return parent;
- })
- .Seal()
- .Seal()
- .Build();
- };
- TExprNode::TListType settings;
- if (linkSettings.ForceSortedMerge) {
- settings.push_back(ctx.NewList(linkSettings.Pos, { ctx.NewAtom(linkSettings.Pos, "forceSortedMerge", TNodeFlags::Default) }));
- }
- if (linkSettings.LeftHints) {
- settings.push_back(builder("left"));
- }
- if (linkSettings.RightHints) {
- settings.push_back(builder("right"));
- }
- if (linkSettings.Compact) {
- settings.push_back(ctx.NewList(linkSettings.Pos, { ctx.NewAtom(linkSettings.Pos, "compact", TNodeFlags::Default) }));
- }
- return ctx.NewList(linkSettings.Pos, std::move(settings));
- }
- TExprNode::TPtr RemapNonConvertibleMemberForJoin(TPositionHandle pos, const TExprNode::TPtr& memberValue,
- const TTypeAnnotationNode& memberType, const TTypeAnnotationNode& unifiedType, TExprContext& ctx)
- {
- TExprNode::TPtr result = memberValue;
- if (&memberType != &unifiedType) {
- result = ctx.Builder(pos)
- .Callable("StrictCast")
- .Add(0, std::move(result))
- .Add(1, ExpandType(pos, unifiedType, ctx))
- .Seal()
- .Build();
- }
- if (RemoveOptionalType(&unifiedType)->GetKind() != ETypeAnnotationKind::Data) {
- if (unifiedType.HasOptionalOrNull()) {
- result = ctx.Builder(pos)
- .Callable("If")
- .Callable(0, "HasNull")
- .Add(0, result)
- .Seal()
- .Callable(1, "Null")
- .Seal()
- .Callable(2, "StablePickle")
- .Add(0, result)
- .Seal()
- .Seal()
- .Build();
- } else {
- result = ctx.NewCallable(pos, "StablePickle", { result });
- }
- }
- return result;
- }
- TExprNode::TPtr PrepareListForJoin(TExprNode::TPtr list, const TTypeAnnotationNode::TListType& keyTypes, TExprNode::TListType& keys, TExprNode::TListType&& payloads, bool payload, bool optional, bool filter, TExprContext& ctx) {
- const auto pos = list->Pos();
- const auto filterPayloads = [&payloads](TExprNodeBuilder& parent) -> TExprNodeBuilder& {
- if (payloads.empty())
- parent.Arg(1, "row");
- else
- parent.Callable(1, "FilterMembers")
- .Arg(0, "row")
- .List(1)
- .Add(std::move(payloads))
- .Seal()
- .Seal();
- return parent;
- };
- if (keyTypes.empty() && 1U == keys.size()) {
- return payload ?
- ctx.Builder(pos)
- .Callable("Map")
- .Add(0, std::move(list))
- .Lambda(1)
- .Param("row")
- .List()
- .Add(0, std::move(keys.front()))
- .Do(filterPayloads)
- .Seal()
- .Seal()
- .Seal()
- .Build():
- ctx.Builder(pos)
- .Callable("List")
- .Callable(0, "ListType")
- .Callable(0, "DataType")
- .Atom(0, "Bool", TNodeFlags::Default)
- .Seal()
- .Seal()
- .Seal()
- .Build();
- }
- if (1U == keyTypes.size()) {
- const auto keyType = ctx.MakeType<TOptionalExprType>(keyTypes.front());
- list = payload ? optional ?
- ctx.Builder(pos)
- .Callable("Map")
- .Add(0, std::move(list))
- .Lambda(1)
- .Param("row")
- .List()
- .Callable(0, "StrictCast")
- .Callable(0, "Member")
- .Arg(0, "row")
- .Add(1, std::move(keys.front()))
- .Seal()
- .Add(1, ExpandType(pos, *keyType, ctx))
- .Seal()
- .Do(filterPayloads)
- .Seal()
- .Seal()
- .Seal()
- .Build():
- ctx.Builder(pos)
- .Callable("FlatMap")
- .Add(0, std::move(list))
- .Lambda(1)
- .Param("row")
- .Callable("FlatMap")
- .Callable(0, "StrictCast")
- .Callable(0, "Member")
- .Arg(0, "row")
- .Add(1, std::move(keys.front()))
- .Seal()
- .Add(1, ExpandType(pos, *keyType, ctx))
- .Seal()
- .Lambda(1)
- .Param("key")
- .Callable("Just")
- .List(0)
- .Arg(0, "key")
- .Do(filterPayloads)
- .Seal()
- .Seal()
- .Seal()
- .Seal()
- .Seal()
- .Seal()
- .Build():
- ctx.Builder(pos)
- .Callable(optional ? "Map" : "FlatMap")
- .Add(0, std::move(list))
- .Lambda(1)
- .Param("row")
- .Callable("StrictCast")
- .Callable(0, "Member")
- .Arg(0, "row")
- .Add(1, std::move(keys.front()))
- .Seal()
- .Add(1, ExpandType(pos, *keyType, ctx))
- .Seal()
- .Seal()
- .Seal()
- .Build();
- } else {
- const auto keyType = ctx.MakeType<TOptionalExprType>(ctx.MakeType<TTupleExprType>(keyTypes));
- list = payload ? optional ?
- ctx.Builder(pos)
- .Callable("Map")
- .Add(0, std::move(list))
- .Lambda(1)
- .Param("row")
- .List()
- .Callable(0, "StrictCast")
- .List(0)
- .Do([&](TExprNodeBuilder& parent) -> TExprNodeBuilder& {
- ui32 pos = 0;
- for (auto& key : keys) {
- parent.Callable(pos++, "Member")
- .Arg(0, "row")
- .Add(1, std::move(key))
- .Seal();
- }
- return parent;
- })
- .Seal()
- .Add(1, ExpandType(pos, *keyType, ctx))
- .Seal()
- .Do(filterPayloads)
- .Seal()
- .Seal()
- .Seal()
- .Build():
- ctx.Builder(pos)
- .Callable("FlatMap")
- .Add(0, std::move(list))
- .Lambda(1)
- .Param("row")
- .Callable("FlatMap")
- .Callable(0, "StrictCast")
- .List(0)
- .Do([&](TExprNodeBuilder& parent) -> TExprNodeBuilder& {
- ui32 pos = 0;
- for (auto& key : keys) {
- parent.Callable(pos++, "Member")
- .Arg(0, "row")
- .Add(1, std::move(key))
- .Seal();
- }
- return parent;
- })
- .Seal()
- .Add(1, ExpandType(pos, *keyType, ctx))
- .Seal()
- .Lambda(1)
- .Param("key")
- .Callable("Just")
- .List(0)
- .Arg(0, "key")
- .Do(filterPayloads)
- .Seal()
- .Seal()
- .Seal()
- .Seal()
- .Seal()
- .Seal()
- .Build():
- ctx.Builder(pos)
- .Callable(optional ? "Map" : "FlatMap")
- .Add(0, std::move(list))
- .Lambda(1)
- .Param("row")
- .Callable("StrictCast")
- .List(0)
- .Do([&](TExprNodeBuilder& parent) -> TExprNodeBuilder& {
- ui32 pos = 0;
- for (auto& key : keys) {
- parent.Callable(pos++, "Member")
- .Arg(0, "row")
- .Add(1, std::move(key))
- .Seal();
- }
- return parent;
- })
- .Seal()
- .Add(1, ExpandType(pos, *keyType, ctx))
- .Seal()
- .Seal()
- .Seal()
- .Build();
- }
- if (optional && filter) {
- list = payload ?
- ctx.Builder(pos)
- .Callable("Filter")
- .Add(0, std::move(list))
- .Lambda(1)
- .Param("row")
- .Callable(0, "Exists")
- .Callable(0, "Nth")
- .Arg(0, "row")
- .Atom(1, 0U)
- .Seal()
- .Seal()
- .Seal()
- .Seal()
- .Build():
- ctx.Builder(pos)
- .Callable("Filter")
- .Add(0, std::move(list))
- .Lambda(1)
- .Param("row")
- .Callable(0, "Exists")
- .Arg(0, "row")
- .Seal()
- .Seal()
- .Seal()
- .Build();
- }
- return list;
- }
- template <bool Squeeze>
- TExprNode::TPtr MakeDictForJoin(TExprNode::TPtr&& list, bool payload, bool multi, TExprContext& ctx) {
- return payload ?
- ctx.Builder(list->Pos())
- .Callable(Squeeze ? "SqueezeToDict" : "ToDict")
- .Add(0, std::move(list))
- .Lambda(1)
- .Param("row")
- .Callable("Nth")
- .Arg(0, "row")
- .Atom(1, 0U)
- .Seal()
- .Seal()
- .Lambda(2)
- .Param("row")
- .Callable("Nth")
- .Arg(0, "row")
- .Atom(1, 1U)
- .Seal()
- .Seal()
- .List(3)
- .Atom(0, multi ? "Many" : "One", TNodeFlags::Default)
- .Atom(1, "Hashed", TNodeFlags::Default)
- .Do([&](TExprNodeBuilder& parent) -> TExprNodeBuilder& {
- if constexpr (Squeeze)
- parent.Atom(2, "Compact", TNodeFlags::Default);
- return parent;
- })
- .Seal()
- .Seal()
- .Build():
- ctx.Builder(list->Pos())
- .Callable(Squeeze ? "SqueezeToDict" : "ToDict")
- .Add(0, std::move(list))
- .Lambda(1)
- .Param("row")
- .Arg("row")
- .Seal()
- .Lambda(2)
- .Param("stub")
- .Callable("Void").Seal()
- .Seal()
- .List(3)
- .Atom(0, multi ? "Many" : "One", TNodeFlags::Default)
- .Atom(1, "Hashed", TNodeFlags::Default)
- .Do([&](TExprNodeBuilder& parent) -> TExprNodeBuilder& {
- if constexpr (Squeeze)
- parent.Atom(2, "Compact", TNodeFlags::Default);
- return parent;
- })
- .Seal()
- .Seal()
- .Build();
- }
- template TExprNode::TPtr MakeDictForJoin<true>(TExprNode::TPtr&& list, bool payload, bool multi, TExprContext& ctx);
- template TExprNode::TPtr MakeDictForJoin<false>(TExprNode::TPtr&& list, bool payload, bool multi, TExprContext& ctx);
- TExprNode::TPtr MakeCrossJoin(TPositionHandle pos, TExprNode::TPtr left, TExprNode::TPtr right, TExprContext& ctx) {
- return ctx.Builder(pos)
- .List()
- .Atom(0, "Cross")
- .Add(1, left)
- .Add(2, right)
- .List(3)
- .Seal()
- .List(4)
- .Seal()
- .List(5)
- .Seal()
- .Seal()
- .Build();
- }
- TExprNode::TPtr PreparePredicate(TExprNode::TPtr predicate, TExprContext& ctx) {
- auto originalPredicate = predicate;
- bool isPg = false;
- if (predicate->IsCallable("ToPg")) {
- isPg = true;
- predicate = predicate->ChildPtr(0);
- }
- if (!predicate->IsCallable("Or")) {
- return originalPredicate;
- }
- if (predicate->ChildrenSize() == 1) {
- return originalPredicate;
- }
- // try to extract common And parts from Or
- TVector<TExprNode::TListType> andParts;
- for (ui32 i = 0; i < predicate->ChildrenSize(); ++i) {
- TExprNode::TListType res;
- bool isPg;
- GatherAndTerms(predicate->ChildPtr(i), res, isPg, ctx);
- YQL_ENSURE(!isPg); // direct child for Or
- andParts.emplace_back(std::move(res));
- }
- THashMap<const TExprNode*, ui32> commonParts;
- for (ui32 j = 0; j < andParts[0].size(); ++j) {
- commonParts[andParts[0][j].Get()] = j;
- }
- for (ui32 i = 1; i < andParts.size(); ++i) {
- THashSet<const TExprNode*> found;
- for (ui32 j = 0; j < andParts[i].size(); ++j) {
- found.insert(andParts[i][j].Get());
- }
- // remove
- for (auto it = commonParts.begin(); it != commonParts.end();) {
- if (found.contains(it->first)) {
- ++it;
- } else {
- commonParts.erase(it++);
- }
- }
- }
- if (commonParts.size() == 0) {
- return originalPredicate;
- }
- // rebuild commonParts in order of original And
- TVector<ui32> idx;
- for (const auto& x : commonParts) {
- idx.push_back(x.second);
- }
- Sort(idx);
- TExprNode::TListType andArgs;
- for (ui32 i : idx) {
- andArgs.push_back(andParts[0][i]);
- }
- TExprNode::TListType orArgs;
- for (ui32 i = 0; i < andParts.size(); ++i) {
- TExprNode::TListType restAndArgs;
- for (ui32 j = 0; j < andParts[i].size(); ++j) {
- if (commonParts.contains(andParts[i][j].Get())) {
- continue;
- }
- restAndArgs.push_back(andParts[i][j]);
- }
- if (restAndArgs.size() >= 1) {
- orArgs.push_back(ctx.NewCallable(predicate->Pos(), "And", std::move(restAndArgs)));
- }
- }
- if (orArgs.size() >= 1) {
- andArgs.push_back(ctx.NewCallable(predicate->Pos(), "Or", std::move(orArgs)));
- }
- auto ret = ctx.NewCallable(predicate->Pos(), "And", std::move(andArgs));
- if (isPg) {
- ret = ctx.NewCallable(predicate->Pos(), "ToPg", { ret });
- }
- return ret;
- }
- void GatherAndTermsImpl(const TExprNode::TPtr& predicate, TExprNode::TListType& andTerms, TExprContext& ctx) {
- auto pred = PreparePredicate(predicate, ctx);
- if (!pred->IsCallable("And")) {
- andTerms.emplace_back(pred);
- return;
- }
- for (ui32 i = 0; i < pred->ChildrenSize(); ++i) {
- GatherAndTermsImpl(pred->ChildPtr(i), andTerms, ctx);
- }
- }
- void GatherAndTerms(const TExprNode::TPtr& predicate, TExprNode::TListType& andTerms, bool& isPg, TExprContext& ctx) {
- isPg = false;
- if (predicate->IsCallable("ToPg")) {
- isPg = true;
- GatherAndTermsImpl(predicate->HeadPtr(), andTerms, ctx);
- } else {
- GatherAndTermsImpl(predicate, andTerms, ctx);
- }
- }
- TExprNode::TPtr FuseAndTerms(TPositionHandle position, const TExprNode::TListType& andTerms, const TExprNode::TPtr& exclude, bool isPg, TExprContext& ctx) {
- TExprNode::TPtr prevAndNode = nullptr;
- TNodeSet added;
- for (const auto& otherAndTerm : andTerms) {
- if (otherAndTerm == exclude) {
- continue;
- }
- if (!added.insert(otherAndTerm.Get()).second) {
- continue;
- }
- if (!prevAndNode) {
- prevAndNode = otherAndTerm;
- } else {
- prevAndNode = ctx.NewCallable(position, "And", { prevAndNode, otherAndTerm });
- }
- }
- if (isPg) {
- return ctx.NewCallable(position, "ToPg", { prevAndNode });
- } else {
- return prevAndNode;
- }
- }
- bool IsEquality(TExprNode::TPtr predicate, TExprNode::TPtr& left, TExprNode::TPtr& right) {
- if (predicate->IsCallable("Coalesce")) {
- if (predicate->Tail().IsCallable("Bool") && IsFalse(predicate->Tail().Head().Content())) {
- predicate = predicate->HeadPtr();
- } else {
- return false;
- }
- }
- if (predicate->IsCallable("FromPg")) {
- predicate = predicate->HeadPtr();
- }
- if (predicate->IsCallable("==")) {
- left = predicate->ChildPtr(0);
- right = predicate->ChildPtr(1);
- return true;
- }
- if (predicate->IsCallable("PgResolvedOp") &&
- (predicate->Head().Content() == "=")) {
- left = predicate->ChildPtr(2);
- right = predicate->ChildPtr(3);
- return true;
- }
- return false;
- }
- void GatherJoinInputs(const TExprNode::TPtr& expr, const TExprNode& row,
- const TParentsMap& parentsMap, const THashMap<TString, TString>& backRenameMap,
- const TJoinLabels& labels, TSet<ui32>& inputs, TSet<TStringBuf>& usedFields) {
- usedFields.clear();
- if (!HaveFieldsSubset(expr, row, usedFields, parentsMap, false)) {
- const auto inputStructType = RemoveOptionalType(row.GetTypeAnn())->Cast<TStructExprType>();
- for (const auto& i : inputStructType->GetItems()) {
- usedFields.insert(i->GetName());
- }
- }
- for (auto x : usedFields) {
- // rename used fields
- if (auto renamed = backRenameMap.FindPtr(x)) {
- x = *renamed;
- }
- TStringBuf part1;
- TStringBuf part2;
- SplitTableName(x, part1, part2);
- inputs.insert(*labels.FindInputIndex(part1));
- if (inputs.size() == labels.Inputs.size()) {
- break;
- }
- }
- }
- } // namespace NYql
|