From e93bf1f5eb1dd1c792ff6f23fa96c518fe743f15 Mon Sep 17 00:00:00 2001 From: Remy Boutonnet Date: Wed, 21 Oct 2026 13:24:41 +0200 Subject: [PATCH 1/3] Cypher: implement NEXT --- query/AST/CMakeLists.txt | 1 + query/AST/CypherAST.h | 9 +- query/AST/CypherASTDumper.cpp | 13 + query/AST/SinglePartQuery.h | 1 + query/AST/stmt/CallSubqueryStmt.h | 4 + query/AST/stmt/NextSegmentStmt.cpp | 19 ++ query/AST/stmt/NextSegmentStmt.h | 29 +++ query/AST/stmt/Stmt.cpp | 35 +-- query/AST/stmt/Stmt.h | 5 + query/analyzer/CypherAnalyzer.cpp | 93 +++++-- query/analyzer/CypherAnalyzer.h | 25 +- query/analyzer/ReadStmtAnalyzer.cpp | 1 + query/ir/codegen/DBProgramGenerator.cpp | 309 ++++++++++++++++++++---- query/ir/codegen/DBProgramGenerator.h | 41 +++- query/ir/dialect/db/DBOps.cpp | 31 +++ query/ir/dialect/db/DBOps.td | 66 ++++- query/ir/dialect/db/DBTypes.td | 10 + query/ir/lowering/DBLowering.cpp | 64 ++++- query/ir/lowering/DBLowering.h | 5 + query/parser/CypherLexer.l | 1 + query/parser/CypherParser.y | 57 ++++- query/parser/ParserUtils.cpp | 111 +++++++++ query/parser/ParserUtils.h | 23 ++ test/query/ir/CMakeLists.txt | 1 + test/query/ir/next/CMakeLists.txt | 5 + test/query/ir/next/NextQueryTest.cpp | 153 ++++++++++++ test/query/ir/next/NextWriteTest.cpp | 32 +++ 27 files changed, 1048 insertions(+), 96 deletions(-) create mode 100644 query/AST/stmt/NextSegmentStmt.cpp create mode 100644 query/AST/stmt/NextSegmentStmt.h create mode 100644 test/query/ir/next/CMakeLists.txt create mode 100644 test/query/ir/next/NextQueryTest.cpp create mode 100644 test/query/ir/next/NextWriteTest.cpp diff --git a/query/AST/CMakeLists.txt b/query/AST/CMakeLists.txt index 07741f7928..8134409465 100644 --- a/query/AST/CMakeLists.txt +++ b/query/AST/CMakeLists.txt @@ -43,6 +43,7 @@ set(ast_sources stmt/CallSubqueryStmt.cpp stmt/CreateStmt.cpp stmt/MergeStmt.cpp + stmt/NextSegmentStmt.cpp stmt/DeleteStmt.cpp stmt/ReturnStmt.cpp stmt/ShortestPathStmt.cpp diff --git a/query/AST/CypherAST.h b/query/AST/CypherAST.h index a5ca5b0c83..9a5fc9622f 100644 --- a/query/AST/CypherAST.h +++ b/query/AST/CypherAST.h @@ -71,6 +71,7 @@ class CallStmt; class CallSubqueryStmt; class CreateStmt; class MergeStmt; +class NextSegmentStmt; class SetStmt; class RemoveStmt; class DeleteStmt; @@ -180,6 +181,7 @@ class CypherAST { friend CallSubqueryStmt; friend CreateStmt; friend MergeStmt; + friend NextSegmentStmt; friend SetStmt; friend RemoveStmt; friend DeleteStmt; @@ -238,6 +240,10 @@ class CypherAST { const QueryCommands& queries() const { return _queries; } + // Moves a command out of the top-level list and into the one a holder reads it + // through, once that holder is built over it + void nestQuery(QueryCommand* query); + const std::vector& getExpressions() const { return _expressions; } const std::vector& getEntityPatterns() const { return _entityPatterns; } @@ -321,9 +327,6 @@ class CypherAST { void addStmtContainer(StmtContainer* container); void addQuery(QueryCommand* query); - // Moves a command out of the top-level list and into the one a holder reads it - // through, once that holder is built over it - void nestQuery(QueryCommand* query); void addDeclContext(DeclContext* ctxt); void addVarDecl(VarDecl* decl); void addNodePatternData(NodePatternData* data); diff --git a/query/AST/CypherASTDumper.cpp b/query/AST/CypherASTDumper.cpp index 8775a6a715..6693449ba2 100644 --- a/query/AST/CypherASTDumper.cpp +++ b/query/AST/CypherASTDumper.cpp @@ -15,6 +15,7 @@ #include "stmt/ReturnStmt.h" #include "stmt/CallStmt.h" #include "stmt/CallSubqueryStmt.h" +#include "stmt/NextSegmentStmt.h" #include "stmt/DeleteStmt.h" #include "stmt/SetStmt.h" #include "stmt/Limit.h" @@ -357,6 +358,18 @@ void CypherASTDumper::dumpQueryBody(std::ostream& out, const SinglePartQuery* qu } break; + case Stmt::Kind::NEXT_SEGMENT: { + const NextSegmentStmt* segment = static_cast(stmt); + out << " _" << std::hex << query << " ||--o{ _" << std::hex << segment << " : \"\"\n"; + out << " _" << std::hex << segment << " {\n"; + out << " ASTType NEXT_SEGMENT\n"; + out << " }\n"; + for (const CallSubqueryStmt::Branch& branch : segment->getBody()->branches()) { + dump(out, branch._query); + } + } + break; + case Stmt::Kind::UNWIND: { const UnwindStmt* unwStmt = static_cast(stmt); const Expr* arg = unwStmt->arg(); diff --git a/query/AST/SinglePartQuery.h b/query/AST/SinglePartQuery.h index 9ecff0b2d7..0d5a0ff6f2 100644 --- a/query/AST/SinglePartQuery.h +++ b/query/AST/SinglePartQuery.h @@ -19,6 +19,7 @@ class SinglePartQuery : public QueryCommand { const StmtContainer* getStmts() const { return _stmts; } const ReturnStmt* getReturnStmt() const { return _returnStmt; } + ReturnStmt* getReturnStmt() { return _returnStmt; } // A CALL subquery counts when its own body does bool writesToTheGraph() const; diff --git a/query/AST/stmt/CallSubqueryStmt.h b/query/AST/stmt/CallSubqueryStmt.h index de7cc01e14..79c1898df7 100644 --- a/query/AST/stmt/CallSubqueryStmt.h +++ b/query/AST/stmt/CallSubqueryStmt.h @@ -11,6 +11,7 @@ class CypherAST; class Expr; class SinglePartQuery; class Symbol; +class VarDecl; // CALL (a, b) { ... }: a query of its own, run once per row in flight, reading the // variables it imports and nothing else of the scope around it @@ -23,12 +24,15 @@ class CallSubqueryStmt final : public Stmt { // imports. Each branch imports through a leading WITH of its own, so the imports are // held per branch. A WHEN branch holds its predicate; the ELSE branch holds none. A // `{ WHEN ... }` side of a UNION holds the CALL it is written as, as UnionQuery's does. + // The analyzer binds each import to a declaration of the body, which a later clause of + // the body may name again. struct Branch { SinglePartQuery* _query {nullptr}; bool _all {false}; Imports _imports; Expr* _condition {nullptr}; CallSubqueryStmt* _whenCall {nullptr}; + std::vector _importDecls; }; using Branches = std::vector; diff --git a/query/AST/stmt/NextSegmentStmt.cpp b/query/AST/stmt/NextSegmentStmt.cpp new file mode 100644 index 0000000000..10e8ed071b --- /dev/null +++ b/query/AST/stmt/NextSegmentStmt.cpp @@ -0,0 +1,19 @@ +#include "NextSegmentStmt.h" + +#include "CypherAST.h" + +using namespace db; + +NextSegmentStmt::NextSegmentStmt(CallSubqueryStmt* body) + : _body(body) +{ +} + +NextSegmentStmt::~NextSegmentStmt() { +} + +NextSegmentStmt* NextSegmentStmt::create(CypherAST* ast, CallSubqueryStmt* body) { + NextSegmentStmt* stmt = new NextSegmentStmt(body); + ast->addStmt(stmt); + return stmt; +} diff --git a/query/AST/stmt/NextSegmentStmt.h b/query/AST/stmt/NextSegmentStmt.h new file mode 100644 index 0000000000..4e76b3a2fb --- /dev/null +++ b/query/AST/stmt/NextSegmentStmt.h @@ -0,0 +1,29 @@ +#pragma once + +#include "Stmt.h" + +namespace db { + +class CallSubqueryStmt; +class CypherAST; + +// A UNION or a WHEN that is one segment of a NEXT chain, or a segment ending on an +// updating clause that NEXT follows. It reads the rows the segments before it returned +// and its RETURN replaces them. A UNION reads the whole table in every branch; a WHEN +// decides per row, as the CALL (*) { WHEN ... } its body is written as would. +class NextSegmentStmt final : public Stmt { +public: + static NextSegmentStmt* create(CypherAST* ast, CallSubqueryStmt* body); + + Kind getKind() const final { return Kind::NEXT_SEGMENT; } + + CallSubqueryStmt* getBody() const { return _body; } + +private: + CallSubqueryStmt* _body {nullptr}; + + explicit NextSegmentStmt(CallSubqueryStmt* body); + ~NextSegmentStmt() final; +}; + +} diff --git a/query/AST/stmt/Stmt.cpp b/query/AST/stmt/Stmt.cpp index 86028dd91e..a95434a2f5 100644 --- a/query/AST/stmt/Stmt.cpp +++ b/query/AST/stmt/Stmt.cpp @@ -2,6 +2,7 @@ #include "SinglePartQuery.h" #include "CallSubqueryStmt.h" +#include "NextSegmentStmt.h" using namespace db; @@ -27,6 +28,7 @@ bool Stmt::isUpdating(Kind kind) { case Kind::UNWIND: case Kind::WITH: case Kind::CALL_SUBQUERY: + case Kind::NEXT_SEGMENT: return false; break; } @@ -34,38 +36,40 @@ bool Stmt::isUpdating(Kind kind) { return false; } -bool Stmt::isUpdating(const Stmt* stmt) { +const CallSubqueryStmt* Stmt::getSubqueryBody(const Stmt* stmt) { const Kind kind = stmt->getKind(); if (kind == Kind::CALL_SUBQUERY) { - const CallSubqueryStmt* subquery = static_cast(stmt); - - return !subquery->isReturning() && writesToTheGraph(stmt); + return static_cast(stmt); + } else if (kind == Kind::NEXT_SEGMENT) { + return static_cast(stmt)->getBody(); } - return isUpdating(kind); + return nullptr; } -bool Stmt::writesToTheGraph(const Stmt* stmt) { - const Kind kind = stmt->getKind(); +bool Stmt::isUpdating(const Stmt* stmt) { + if (const CallSubqueryStmt* subquery = getSubqueryBody(stmt)) { + return !subquery->isReturning() && subquery->writesToTheGraph(); + } - if (kind == Kind::CALL_SUBQUERY) { - const CallSubqueryStmt* subquery = static_cast(stmt); + return isUpdating(stmt->getKind()); +} +bool Stmt::writesToTheGraph(const Stmt* stmt) { + if (const CallSubqueryStmt* subquery = getSubqueryBody(stmt)) { return subquery->writesToTheGraph(); } - return isUpdating(kind); + return isUpdating(stmt->getKind()); } bool Stmt::isReading(const Stmt* stmt) { - const Kind kind = stmt->getKind(); - - if (kind == Kind::CALL_SUBQUERY) { + if (getSubqueryBody(stmt)) { return readsTheGraph(stmt); } - return !isUpdating(kind); + return !isUpdating(stmt->getKind()); } bool Stmt::readsTheGraph(const Stmt* stmt) { @@ -73,7 +77,8 @@ bool Stmt::readsTheGraph(const Stmt* stmt) { switch (kind) { case Kind::CALL_SUBQUERY: - return static_cast(stmt)->readsTheGraph(); + case Kind::NEXT_SEGMENT: + return getSubqueryBody(stmt)->readsTheGraph(); break; case Kind::MATCH: diff --git a/query/AST/stmt/Stmt.h b/query/AST/stmt/Stmt.h index feb5da499c..1dc53e9631 100644 --- a/query/AST/stmt/Stmt.h +++ b/query/AST/stmt/Stmt.h @@ -2,6 +2,7 @@ namespace db { +class CallSubqueryStmt; class CypherAST; class Stmt { @@ -23,6 +24,7 @@ class Stmt { UNWIND, WITH, CALL_SUBQUERY, + NEXT_SEGMENT, }; // A clause that writes to the graph. Cypher orders a query part's clauses reading @@ -48,6 +50,9 @@ class Stmt { // nothing an updating clause above it could hide static bool isReading(const Stmt* stmt); + // The query a CALL subquery or a NEXT segment runs, null for any other clause + static const CallSubqueryStmt* getSubqueryBody(const Stmt* stmt); + virtual Kind getKind() const = 0; protected: diff --git a/query/analyzer/CypherAnalyzer.cpp b/query/analyzer/CypherAnalyzer.cpp index 6ea146b64c..528f6273f2 100644 --- a/query/analyzer/CypherAnalyzer.cpp +++ b/query/analyzer/CypherAnalyzer.cpp @@ -66,6 +66,7 @@ #include "stmt/Skip.h" #include "stmt/CallStmt.h" #include "stmt/CallSubqueryStmt.h" +#include "stmt/NextSegmentStmt.h" #include "stmt/Limit.h" #include "CreateNodePropertyIndexQuery.h" #include "CreateEdgePropertyIndexQuery.h" @@ -219,10 +220,27 @@ void CypherAnalyzer::analyzeQueryBody(const SinglePartQuery* query, bool returnR if (stmts) { throwOnReadAfterUpdate(stmts); + const Stmt* previous = nullptr; + for (Stmt* stmt : stmts->stmts()) { const Stmt::Kind kind = stmt->getKind(); - if (kind == Stmt::Kind::CALL_SUBQUERY) { + // A segment is a part of its own, so a part opens after it as after a WITH + const bool followsASegment = previous && previous->getKind() == Stmt::Kind::NEXT_SEGMENT; + if (followsASegment) { + returnMandatory = returnRequired; + } + + previous = stmt; + + if (kind == Stmt::Kind::NEXT_SEGMENT) { + NextSegmentStmt* segment = static_cast(stmt); + analyze(segment); + + if (Stmt::isUpdating(segment)) { + returnMandatory = false; + } + } else if (kind == Stmt::Kind::CALL_SUBQUERY) { CallSubqueryStmt* subquery = static_cast(stmt); analyze(subquery); @@ -364,7 +382,7 @@ void CypherAnalyzer::throwOnReadAfterUpdate(const StmtContainer* stmts) const { for (const Stmt* stmt : stmts->stmts()) { const Stmt::Kind kind = stmt->getKind(); - if (kind == Stmt::Kind::WITH) { + if (kind == Stmt::Kind::WITH || kind == Stmt::Kind::NEXT_SEGMENT) { hasWritten = false; continue; } @@ -477,19 +495,52 @@ void CypherAnalyzer::setScope(DeclContext* scope) { void CypherAnalyzer::analyze(CallSubqueryStmt* subquery) { CallSubqueryStmt::Branches& branches = subquery->branches(); - const bool conditional = subquery->isConditional(); + const bool hasScopeClause = subquery->hasScopeClause(); if (subquery->importsEverything()) { importEveryVariable(subquery); } for (CallSubqueryStmt::Branch& branch : branches) { - if (!subquery->hasScopeClause()) { + if (!hasScopeClause) { importThroughLeadingWith(branch); } - analyzeSubqueryBranch(branch, subquery->hasScopeClause()); + analyzeSubqueryBranch(branch, hasScopeClause ? BranchImports::SCOPE_CLAUSE : BranchImports::LEADING_WITH); + } + + analyzeBranchColumns(subquery); + + if (subquery->isReturning()) { + publishSubqueryReturn(subquery); } +} + +// The segment reads the table the segments before it returned and its columns replace it, +// so what it returns is the whole scope after it +void CypherAnalyzer::analyze(NextSegmentStmt* segment) { + CallSubqueryStmt* body = segment->getBody(); + + importEveryVariable(body); + + for (CallSubqueryStmt::Branch& branch : body->branches()) { + analyzeSubqueryBranch(branch, BranchImports::NEXT_SEGMENT); + } + + analyzeBranchColumns(body); + + DeclContext* scope = DeclContext::create(_ast, _ctxt); + + if (body->isReturning()) { + publishBranchReturns(body, scope); + } + + setScope(scope); +} + +void CypherAnalyzer::analyzeBranchColumns(const CallSubqueryStmt* subquery) const { + const CallSubqueryStmt::Branches& branches = subquery->branches(); + const bool conditional = subquery->isConditional(); for (size_t index = 1; index < branches.size(); index++) { if (conditional) { @@ -498,26 +549,27 @@ void CypherAnalyzer::analyze(CallSubqueryStmt* subquery) { analyzeUnionColumns(branches.front()._query, branches[index]._query); } } - - if (subquery->isReturning()) { - publishSubqueryReturn(subquery); - } } -void CypherAnalyzer::analyzeSubqueryBranch(const CallSubqueryStmt::Branch& branch, bool hasScopeClause) { +void CypherAnalyzer::analyzeSubqueryBranch(CallSubqueryStmt::Branch& branch, BranchImports imports) { // The body reads the imported variables and nothing else of the scope around it, so // its context is seeded with a declaration per import and the body resolves in that const SinglePartQuery* body = branch._query; DeclContext* const outer = _ctxt; DeclContext* const inner = body->getDeclContext(); + const bool predicateReadsImports = imports != BranchImports::LEADING_WITH; + // What the scope clause names is readable everywhere in the body. A body importing // through a leading WITH carries nothing this way: that WITH is an ordinary - // projection, and an ordinary WITH below it descopes what it does not project. + // projection, and an ordinary WITH below it descopes what it does not project. Nor + // does a segment, whose WITH reads the whole table and groups it on its own keys. std::vector outerImports; - if (hasScopeClause) { + if (predicateReadsImports) { importIntoBranch(branch, outer, inner); + } + if (imports == BranchImports::SCOPE_CLAUSE) { for (const Symbol* import : branch._imports) { outerImports.push_back(import->getName()); } @@ -542,7 +594,7 @@ void CypherAnalyzer::analyzeSubqueryBranch(const CallSubqueryStmt::Branch& branc // A leading WITH runs only once the WHEN predicate has picked its branch, so the // predicate does not read what that WITH imports - if (!hasScopeClause) { + if (!predicateReadsImports) { importIntoBranch(branch, outer, inner); } @@ -553,7 +605,9 @@ void CypherAnalyzer::analyzeSubqueryBranch(const CallSubqueryStmt::Branch& branc setScope(outer); } -void CypherAnalyzer::importIntoBranch(const CallSubqueryStmt::Branch& branch, DeclContext* outer, DeclContext* inner) { +void CypherAnalyzer::importIntoBranch(CallSubqueryStmt::Branch& branch, DeclContext* outer, DeclContext* inner) { + branch._importDecls.clear(); + for (const Symbol* import : branch._imports) { const std::string_view name = import->getName(); @@ -567,6 +621,8 @@ void CypherAnalyzer::importIntoBranch(const CallSubqueryStmt::Branch& branch, De imported->setIsQuantifiedPath(decl->isQuantifiedPath()); _declSources[imported] = decl; _exprAnalyzer->importCSVSource(imported, decl); + + branch._importDecls.push_back(imported); } } @@ -946,7 +1002,14 @@ void CypherAnalyzer::publishSubqueryReturn(const CallSubqueryStmt* subquery) { } } - publishProjection(projection, _ctxt); + publishBranchReturns(subquery, _ctxt); +} + +void CypherAnalyzer::publishBranchReturns(const CallSubqueryStmt* subquery, DeclContext* scope) { + const CallSubqueryStmt::Branches& branches = subquery->branches(); + Projection* projection = branches.front()._query->getReturnStmt()->getProjection(); + + publishProjection(projection, scope); for (size_t branchIndex = 1; branchIndex < branches.size(); branchIndex++) { Projection* branchProjection = branches[branchIndex]._query->getReturnStmt()->getProjection(); diff --git a/query/analyzer/CypherAnalyzer.h b/query/analyzer/CypherAnalyzer.h index ff834a9821..babba14af8 100644 --- a/query/analyzer/CypherAnalyzer.h +++ b/query/analyzer/CypherAnalyzer.h @@ -42,6 +42,7 @@ class ReturnStmt; class Stmt; class StmtContainer; class WithStmt; +class NextSegmentStmt; class ExistsExpr; class CountSubqueryExpr; class CollectSubqueryExpr; @@ -73,6 +74,7 @@ class CypherAnalyzer { void analyze(const ReturnStmt* returnSt); void analyze(WithStmt* withSt); void analyze(CallSubqueryStmt* subquery); + void analyze(NextSegmentStmt* segment); void analyze(const ConditionalQuery* query); void analyze(const LoadGraphQuery* loadGraph); void analyze(const CreateGraphQuery* createGraph); @@ -107,6 +109,19 @@ class CypherAnalyzer { private: using DeclSet = std::unordered_set; + // How a branch of a subquery body reads the variables it imports + enum class BranchImports { + // Named by the scope clause: the WHEN predicate and every clause of the body read them + SCOPE_CLAUSE, + + // Named by a leading WITH, which runs once the WHEN predicate has picked the branch + LEADING_WITH, + + // The table a NEXT segment reads: the WHEN predicate reads it, and a WITH of the + // body descopes what it does not project + NEXT_SEGMENT, + }; + CypherAST* _ast {nullptr}; GraphView _graphView; const GraphMetadata& _graphMetadata; @@ -158,8 +173,10 @@ class CypherAnalyzer { void importEveryVariable(CallSubqueryStmt* subquery) const; // Analyzes one query of a CALL body under its own scope, seeded with what it imports - void analyzeSubqueryBranch(const CallSubqueryStmt::Branch& branch, bool hasScopeClause); - void importIntoBranch(const CallSubqueryStmt::Branch& branch, DeclContext* outer, DeclContext* inner); + void analyzeSubqueryBranch(CallSubqueryStmt::Branch& branch, BranchImports imports); + + void analyzeBranchColumns(const CallSubqueryStmt* subquery) const; + void importIntoBranch(CallSubqueryStmt::Branch& branch, DeclContext* outer, DeclContext* inner); void throwOnPatternPredicateVariable(const Pattern* pattern, const DeclContext* outer) const; @@ -169,6 +186,10 @@ class CypherAnalyzer { // Declares what a returning body publishes in the scope around the CALL, rejecting a // name that scope already holds void publishSubqueryReturn(const CallSubqueryStmt* subquery); + + // Declares in @param scope the columns the branches of a body return, under the + // declarations the first branch makes + void publishBranchReturns(const CallSubqueryStmt* subquery, DeclContext* scope); void publishUnionBranchReturn(Projection* branchProjection, const Projection* firstProjection); // Adds to a barrier of a subquery body the imports it does not project, so what the diff --git a/query/analyzer/ReadStmtAnalyzer.cpp b/query/analyzer/ReadStmtAnalyzer.cpp index 6dff3ae770..ae3c391fa1 100644 --- a/query/analyzer/ReadStmtAnalyzer.cpp +++ b/query/analyzer/ReadStmtAnalyzer.cpp @@ -128,6 +128,7 @@ void ReadStmtAnalyzer::analyze(Stmt* stmt) { case Stmt::Kind::RETURN: case Stmt::Kind::WITH: case Stmt::Kind::CALL_SUBQUERY: + case Stmt::Kind::NEXT_SEGMENT: throw FatalException("Attempted to evaluate invalid read statement."); break; } diff --git a/query/ir/codegen/DBProgramGenerator.cpp b/query/ir/codegen/DBProgramGenerator.cpp index 82085b026e..5cfe739403 100644 --- a/query/ir/codegen/DBProgramGenerator.cpp +++ b/query/ir/codegen/DBProgramGenerator.cpp @@ -110,6 +110,7 @@ #include "expr/UnaryExpr.h" #include "stmt/CreateStmt.h" #include "stmt/MergeStmt.h" +#include "stmt/NextSegmentStmt.h" #include "stmt/DeleteStmt.h" #include "expr/ExprChain.h" #include "stmt/Limit.h" @@ -439,9 +440,11 @@ bool statementCarriesRows(const Stmt* stmt) { // A shortest path answers for the whole source and target columns with one // distance and one path, so its rows are not the rows it read. A RETURN is the - // body's tail, read off getReturnStmt() rather than walked here. + // body's tail, read off getReturnStmt() rather than walked here. A NEXT segment + // replaces the rows it read with the ones it returns. case Stmt::Kind::SHORTESTPATH: case Stmt::Kind::RETURN: + case Stmt::Kind::NEXT_SEGMENT: return false; break; } @@ -2058,6 +2061,10 @@ void DBProgramGenerator::generateQueryParts(const SinglePartQuery* query) { generatePartStatements(stmts.subspan(partBegin, index - partBegin)); generateWith(static_cast(stmt)); partBegin = index + 1; + } else if (stmt->getKind() == Stmt::Kind::NEXT_SEGMENT) { + generatePartStatements(stmts.subspan(partBegin, index - partBegin)); + generateNextSegment(static_cast(stmt)); + partBegin = index + 1; } else if (closesPartOnItsCut(stmt, stmts.subspan(index + 1))) { generatePartStatements(stmts.subspan(partBegin, index + 1 - partBegin)); publishInFlightColumns(); @@ -2137,7 +2144,7 @@ bool DBProgramGenerator::closesPartOnItsCut(const Stmt* stmt, std::spangetKind(); - if (kind == Stmt::Kind::WITH) { + if (kind == Stmt::Kind::WITH || kind == Stmt::Kind::NEXT_SEGMENT) { return false; } else if (kind == Stmt::Kind::MATCH || kind == Stmt::Kind::CALL_SUBQUERY) { return true; @@ -5704,6 +5711,217 @@ bool DBProgramGenerator::subqueryCarriesRows(const SinglePartQuery* body) { } void DBProgramGenerator::generateCallSubquery(const CallSubqueryStmt* subquery) { + SubqueryResults results; + emitCallSubquery(subquery, results); + + if (!subquery->isReturning()) { + return; + } + + llvm::SmallVector published = results._inputs; + published.append(results._returned.begin(), results._returned.end()); + published.append(results._constants.begin(), results._constants.end()); + + rebindScopeKeepingWrittenEntities(published); + + for (auto& [decl, written] : results._returnedEntities) { + _part._writtenEntities[decl] = std::move(written); + } +} + +// A segment reads the table the segments before it returned, and what it returns replaces +// that table +void DBProgramGenerator::generateNextSegment(const NextSegmentStmt* segment) { + const CallSubqueryStmt* body = segment->getBody(); + + if (body->isUnion()) { + generateUnionSegment(body); + } else if (body->isConditional()) { + generateConditionalSegment(body); + } else { + generateUpdatingSegment(body); + } +} + +// Every branch reads the whole table, so the table is held once and each branch walks it +// from the start +void DBProgramGenerator::generateUnionSegment(const CallSubqueryStmt* body) { + const CallSubqueryStmt::Branches& branches = body->branches(); + const size_t dedupedBranches = body->getDedupedBranchCount(); + + const mlir::Location loc = _opBuilder.getUnknownLoc(); + mlir::Block* const outerBlock = _opBuilder.getInsertionBlock(); + + llvm::SmallVector scopeColumns; + collectPublishedColumns(scopeColumns); + + llvm::SmallVector inputs; + llvm::SmallVector constants; + splitConstantColumns(scopeColumns, inputs, constants); + + llvm::SmallVector inputColumns; + llvm::SmallVector inputTypes; + for (const PublishedColumn& input : inputs) { + inputColumns.push_back(input._column); + inputTypes.push_back(input._column.getType()); + } + + mlir::Value rows; + if (!inputColumns.empty()) { + const mlir::db::MaterializedType rowsType = mlir::db::MaterializedType::get(_mlirCtxt); + rows = _opBuilder.create(loc, rowsType, inputColumns).getRows(); + } + + mlir::Value distinctSet; + if (dedupedBranches > 0) { + distinctSet = _opBuilder.create(loc).getSet(); + } + + SubqueryBranchResults branchResults; + std::vector> branchScopes(branches.size()); + std::vector importedEntities(branches.size()); + + for (size_t branchIndex = 0; branchIndex < branches.size(); branchIndex++) { + mlir::Block* const block = openSubqueryBranch(branchResults); + + llvm::SmallVector scanned; + if (rows) { + _opBuilder.setInsertionPointToStart(block); + mlir::db::ScanMaterialized scan = _opBuilder.create(loc, inputTypes, rows); + scanned.assign(scan.getColumns().begin(), scan.getColumns().end()); + } + + collectSubqueryBranchScope(branches[branchIndex], + inputs, + scanned, + mlir::Value(), + constants, + /*bindsHidden=*/false, + branchScopes[branchIndex], + importedEntities[branchIndex]); + } + + std::unordered_set outerImportedCSVRows; + std::swap(_importedCSVRows, outerImportedCSVRows); + + for (size_t branchIndex = 0; branchIndex < branches.size(); branchIndex++) { + rebindSubqueryBranchScope(branchScopes[branchIndex], importedEntities[branchIndex]); + + mlir::Block* const block = &branchResults._regions[branchIndex]->front(); + generateSubqueryBranch(branches[branchIndex]._query, block, outerBlock, branchResults); + } + + llvm::SmallVector yielded; + CarriedEntities returnedEntities; + generateUnionOverBranches(distinctSet, dedupedBranches, outerBlock, branchResults, yielded, returnedEntities); + + std::swap(_importedCSVRows, outerImportedCSVRows); + + rebindScope(yielded); + + for (auto& [decl, written] : returnedEntities) { + _part._writtenEntities[decl] = std::move(written); + } +} + +// A WHEN decides per row, as the CALL (*) its body is written as does, and what it returns +// replaces the row: a name the row held and the branch returns again is bound afresh +void DBProgramGenerator::generateConditionalSegment(const CallSubqueryStmt* body) { + SubqueryResults results; + emitCallSubquery(body, results); + + rebindScope(results._returned); + + for (auto& [decl, written] : results._returnedEntities) { + _part._writtenEntities[decl] = std::move(written); + } +} + +// A segment ending on a write returns nothing, so the segment after it starts from a single +// row, as a query does: the segment is the body of a CALL () run once, which reads the +// table it is handed as a UNION segment's branches do +void DBProgramGenerator::generateUpdatingSegment(const CallSubqueryStmt* body) { + const CallSubqueryStmt::Branch& branch = body->branches().front(); + const SinglePartQuery* query = branch._query; + + const mlir::Location loc = _opBuilder.getUnknownLoc(); + mlir::Block* const outerBlock = _opBuilder.getInsertionBlock(); + + llvm::SmallVector scopeColumns; + collectPublishedColumns(scopeColumns); + + llvm::SmallVector inputs; + llvm::SmallVector constants; + splitConstantColumns(scopeColumns, inputs, constants); + + llvm::SmallVector inputColumns; + llvm::SmallVector inputTypes; + for (const PublishedColumn& input : inputs) { + inputColumns.push_back(input._column); + inputTypes.push_back(input._column.getType()); + } + + mlir::Region bodyRegion; + mlir::Block* const bodyBlock = new mlir::Block(); // Region destructor frees it + bodyRegion.push_back(bodyBlock); + + llvm::SmallVector scanned; + if (!inputColumns.empty()) { + const mlir::db::MaterializedType rowsType = mlir::db::MaterializedType::get(_mlirCtxt); + const mlir::Value rows = _opBuilder.create(loc, rowsType, inputColumns).getRows(); + + _opBuilder.setInsertionPointToStart(bodyBlock); + mlir::db::ScanMaterialized scan = _opBuilder.create(loc, inputTypes, rows); + scanned.assign(scan.getColumns().begin(), scan.getColumns().end()); + } + + llvm::SmallVector scope; + CarriedEntities importedEntities; + collectSubqueryBranchScope(branch, + inputs, + scanned, + mlir::Value(), + constants, + /*bindsHidden=*/false, + scope, + importedEntities); + + std::unordered_set outerImportedCSVRows; + std::swap(_importedCSVRows, outerImportedCSVRows); + + _opBuilder.setInsertionPointToEnd(bodyBlock); + rebindSubqueryBranchScope(scope, importedEntities); + generateQueryParts(query); + _opBuilder.create(loc, mlir::Value(), mlir::ValueRange {}); + + std::swap(_importedCSVRows, outerImportedCSVRows); + + _opBuilder.setInsertionPointToEnd(outerBlock); + + mlir::db::CallSubquery callOp = _opBuilder.create(loc, + mlir::TypeRange {}, + mlir::ValueRange {}, + /*unit=*/true, + subqueryCarriesRows(query), + /*optional=*/false); + callOp.getBody().takeBody(bodyRegion); + + rebindScope({}); +} + +void DBProgramGenerator::splitConstantColumns(llvm::ArrayRef columns, + llvm::SmallVectorImpl& rowColumns, + llvm::SmallVectorImpl& constants) { + for (const PublishedColumn& column : columns) { + if (yieldsConstantColumn(column._column, _constantColumns)) { + constants.push_back(column); + } else { + rowColumns.push_back(column); + } + } +} + +void DBProgramGenerator::emitCallSubquery(const CallSubqueryStmt* subquery, SubqueryResults& results) { const CallSubqueryStmt::Branches& branches = subquery->branches(); const bool returning = subquery->isReturning(); const bool optional = returning && subquery->isOptional(); @@ -5722,14 +5940,7 @@ void DBProgramGenerator::generateCallSubquery(const CallSubqueryStmt* subquery) // is no input of the body: an imported one is read where it is bound llvm::SmallVector inputs; llvm::SmallVector constants; - - for (const PublishedColumn& column : scopeColumns) { - if (yieldsConstantColumn(column._column, _constantColumns)) { - constants.push_back(column); - } else { - inputs.push_back(column); - } - } + splitConstantColumns(scopeColumns, inputs, constants); const mlir::Location loc = _opBuilder.getUnknownLoc(); mlir::Block* const outerBlock = _opBuilder.getInsertionBlock(); @@ -5883,34 +6094,39 @@ void DBProgramGenerator::generateCallSubquery(const CallSubqueryStmt* subquery) return; } - const mlir::ResultRange results = callOp.getResults(); + const mlir::ResultRange callResults = callOp.getResults(); size_t resultIndex = 0; - llvm::SmallVector published; if (!carriesScope) { for (const PublishedColumn& input : inputs) { - published.push_back({input._decl, input._name, results[resultIndex]}); + results._inputs.push_back({input._decl, input._name, callResults[resultIndex]}); resultIndex++; } } - for (const PublishedColumn& column : yielded) { - published.push_back({column._decl, column._name, results[resultIndex]}); + // A body carrying the scope yields the inputs ahead of its own columns + for (size_t yieldedIndex = 0; yieldedIndex < yielded.size(); yieldedIndex++) { + const PublishedColumn& column = yielded[yieldedIndex]; + const PublishedColumn bound {column._decl, column._name, callResults[resultIndex]}; resultIndex++; - } - published.append(constants.begin(), constants.end()); + if (carriesScope && yieldedIndex < inputs.size()) { + results._inputs.push_back(bound); + } else { + results._returned.push_back(bound); + } + } - rebindScopeKeepingWrittenEntities(published); + results._constants.assign(constants.begin(), constants.end()); for (auto& [decl, written] : returnedEntities) { if (optional) { written._labels.clear(); written._edgeType.clear(); } - - _part._writtenEntities[decl] = std::move(written); } + + results._returnedEntities = std::move(returnedEntities); } void DBProgramGenerator::collectSubqueryBranchScope(const CallSubqueryStmt::Branch& branch, @@ -5921,22 +6137,15 @@ void DBProgramGenerator::collectSubqueryBranchScope(const CallSubqueryStmt::Bran bool bindsHidden, llvm::SmallVectorImpl& scope, CarriedEntities& importedEntities) const { - const DeclContext* bodyContext = branch._query->getDeclContext(); - const CallSubqueryStmt::Imports& imports = branch._imports; + const std::vector& importDecls = branch._importDecls; - const auto importedDecl = [&imports, bodyContext](std::string_view name) -> const VarDecl* { - const auto namesIt = [name](const Symbol* import) { - return import->getName() == name; - }; - - if (!std::ranges::any_of(imports, namesIt)) { - return nullptr; - } + const auto importedDecl = [&importDecls](std::string_view name) -> const VarDecl* { + const auto importIt = std::ranges::find(importDecls, name, &VarDecl::getName); - return bodyContext->getDecl(name); + return importIt != importDecls.end() ? *importIt : nullptr; }; - // The body reads an import under the declaration its own context holds for it. A body + // The body reads an import under the declaration the analyzer bound it to. A body // carrying the scope holds every input under a hidden name too, so the input comes back // as the body's rows left it whatever its clauses publish, and a write reading no import // still has rows to write one of per. A body run per row reads the row it is handed. @@ -6001,9 +6210,20 @@ void DBProgramGenerator::generateSubqueryUnion(llvm::ArrayRef& yielded, + CarriedEntities& returnedEntities) { + const size_t branchCount = branchResults._regions.size(); + yieldEveryPendingMask(branchResults); // The branches share one distinct set, so each dedups on the same columns: its masks @@ -6016,10 +6236,12 @@ void DBProgramGenerator::generateSubqueryUnion(llvm::ArrayRef(loc, branchResults._resultTypes, branches.size()); + mlir::db::Union unionOp = _opBuilder.create(_opBuilder.getUnknownLoc(), + branchResults._resultTypes, + branchCount); const mlir::MutableArrayRef unionBranches = unionOp.getBranches(); - for (size_t branchIndex = 0; branchIndex < branches.size(); branchIndex++) { + for (size_t branchIndex = 0; branchIndex < branchCount; branchIndex++) { unionBranches[branchIndex].takeBody(*branchResults._regions[branchIndex]); } @@ -6050,7 +6272,7 @@ void DBProgramGenerator::generateSubqueryConditional(llvm::ArrayRef& region = branchResults._regions.emplace_back(std::make_unique()); + mlir::Block* const block = new mlir::Block(); // Region destructor frees it + region->push_back(block); + + return block; +} + void DBProgramGenerator::generateSubqueryBranch(const SinglePartQuery* query, + mlir::Block* block, mlir::Block* bodyBlock, SubqueryBranchResults& branchResults) { const mlir::Location loc = _opBuilder.getUnknownLoc(); - std::unique_ptr& region = branchResults._regions.emplace_back(std::make_unique()); - mlir::Block* const block = new mlir::Block(); // Region destructor frees it - region->push_back(block); - _unionBranchRowBlocks[block] = bodyBlock; - _opBuilder.setInsertionPointToStart(block); + _opBuilder.setInsertionPointToEnd(block); generateQueryParts(query); llvm::SmallVector& branchColumns = branchResults._branchColumns.emplace_back(); diff --git a/query/ir/codegen/DBProgramGenerator.h b/query/ir/codegen/DBProgramGenerator.h index 749c83558c..24df0c0bf2 100644 --- a/query/ir/codegen/DBProgramGenerator.h +++ b/query/ir/codegen/DBProgramGenerator.h @@ -88,6 +88,7 @@ class ConditionalQuery; class Stmt; class WhereClause; class WithStmt; +class NextSegmentStmt; class YieldClause; class YieldItems; @@ -380,6 +381,16 @@ class DBProgramGenerator { // inputs beside what the body returned, take the scope over void generateCallSubquery(const CallSubqueryStmt* subquery); + void generateNextSegment(const NextSegmentStmt* segment); + void generateUnionSegment(const CallSubqueryStmt* body); + void generateConditionalSegment(const CallSubqueryStmt* body); + void generateUpdatingSegment(const CallSubqueryStmt* body); + + // A constant holds one value standing for every row rather than rows of its own + void splitConstantColumns(llvm::ArrayRef columns, + llvm::SmallVectorImpl& rowColumns, + llvm::SmallVectorImpl& constants); + // Whether the body's rows carry the columns it was given through to its RETURN: true // unless a clause of it aggregates, sorts, cuts or dedups the rows, which is when the // body has to run one input row at a time instead @@ -532,6 +543,19 @@ class DBProgramGenerator { using CarriedEntities = llvm::SmallVector>; + // What a db.call_subquery hands back, bound to its results: the rows in flight it was + // given, the columns its body returned, and the constants, which it never took + struct SubqueryResults { + llvm::SmallVector _inputs; + llvm::SmallVector _returned; + llvm::SmallVector _constants; + CarriedEntities _returnedEntities; + }; + + // Emits the db.call_subquery of @param subquery, leaving the scope as it was, and fills + // @param results when the body returns + void emitCallSubquery(const CallSubqueryStmt* subquery, SubqueryResults& results); + // The scope one branch of a CALL body opens with: the imports it names, read through // @param inputArguments or where they are bound as constants, and - for a body carrying // the scope - every input under a hidden name @@ -575,12 +599,25 @@ class DBProgramGenerator { mlir::Block* bodyBlock, std::vector>& regions); - // Generates one branch into a region ending in a db.yield of what its RETURN publishes, - // none for a branch ending on a write + // Opens the region of one more branch, its single block empty + mlir::Block* openSubqueryBranch(SubqueryBranchResults& branchResults); + + // Generates one branch into @param block, opened by openSubqueryBranch, ending in a + // db.yield of what its RETURN publishes, none for a branch ending on a write void generateSubqueryBranch(const SinglePartQuery* query, + mlir::Block* block, mlir::Block* bodyBlock, SubqueryBranchResults& branchResults); + // Emits the db.union holding the branches built and fills @param yielded as + // generateSubqueryUnion does + void generateUnionOverBranches(mlir::Value distinctSet, + size_t dedupedBranches, + mlir::Block* bodyBlock, + SubqueryBranchResults& branchResults, + llvm::SmallVectorImpl& yielded, + CarriedEntities& returnedEntities); + // Makes every branch yield the mask of each entity a MERGE wrote in any of them, or a // CREATE wrote in some of them void yieldEveryPendingMask(SubqueryBranchResults& branchResults); diff --git a/query/ir/dialect/db/DBOps.cpp b/query/ir/dialect/db/DBOps.cpp index c5bf4340a9..35ad0f5031 100644 --- a/query/ir/dialect/db/DBOps.cpp +++ b/query/ir/dialect/db/DBOps.cpp @@ -1034,6 +1034,37 @@ void Union::print(OpAsmPrinter& printer) { printer.printOptionalAttrDict((*this)->getAttrs()); } +LogicalResult Materialize::verify() { + if (getColumns().empty()) { + return emitOpError("requires at least one column"); + } + + return success(); +} + +LogicalResult ScanMaterialized::verify() { + Materialize materialize = getRows().getDefiningOp(); + if (!materialize) { + return emitOpError("requires rows a db.materialize holds"); + } + + const OperandRange heldColumns = materialize.getColumns(); + const ResultRange columns = getColumns(); + if (columns.size() != heldColumns.size()) { + return emitOpError("expects ") << heldColumns.size() + << " results, one per column the db.materialize holds, but has " + << columns.size(); + } + + for (size_t columnIndex = 0; columnIndex < columns.size(); columnIndex++) { + if (columns[columnIndex].getType() != heldColumns[columnIndex].getType()) { + return emitOpError("result ") << columnIndex << " must have the type of the column it reads"; + } + } + + return success(); +} + // A union emits one result table, so every branch has to end in a db.output and all of // them have to name the same columns. The names are the query's, checked by the analyzer; // what is re-checked here is the shape a hand-written module can still get wrong. diff --git a/query/ir/dialect/db/DBOps.td b/query/ir/dialect/db/DBOps.td index 506e892341..cc6303beb5 100644 --- a/query/ir/dialect/db/DBOps.td +++ b/query/ir/dialect/db/DBOps.td @@ -2346,6 +2346,62 @@ def DistinctSet : TuringOp<"distinct_set", []> { let assemblyFormat = "attr-dict `:` qualified(type($set))"; } +/** +* Materialize +*/ +def Materialize : TuringOp<"materialize", [Pure]> { + let summary = "Hold every row in flight for several dataflows to read in turn"; + let description = [{ + The table a NEXT hands a UNION segment, every branch of which reads all of it. The + columns in flight are held once and each branch walks them with a db.scan_materialized + of its own, as a common table expression is scanned once per reference: + + %t = db.materialize(%name, %age) : (!db.column, !db.column) -> !db.materialized + %n = db.union { + %name1, %age1 = db.scan_materialized(%t) : (!db.materialized) -> (!db.column, !db.column) + ... + db.yield %c : !db.column + }, { + %name2, %age2 = db.scan_materialized(%t) : (!db.materialized) -> (!db.column, !db.column) + ... + db.yield %l : !db.column + } : !db.column + + A pipeline breaker: a scan starts once the dataflow feeding the op has run to its end. + The columns are row-aligned, the rows of one relation, and no constant is among them: + a constant stands for every row, and a branch reads it where it is bound. + }]; + + let arguments = (ins Variadic:$columns); + let results = (outs MaterializedType:$rows); + + let assemblyFormat = [{ + `(` $columns `)` attr-dict `:` functional-type($columns, $rows) + }]; + + let hasVerifier = 1; +} + +/** +* ScanMaterialized +*/ +def ScanMaterialized : TuringOp<"scan_materialized", [Pure]> { + let summary = "Walk every row a db.materialize holds"; + let description = [{ + A source like a scan: it opens a loop over the rows `rows` holds, in the order they + were held, one result per column the db.materialize took, of the same type. + }]; + + let arguments = (ins MaterializedType:$rows); + let results = (outs Variadic:$columns); + + let assemblyFormat = [{ + `(` $rows `)` attr-dict `:` functional-type($rows, $columns) + }]; + + let hasVerifier = 1; +} + /** * Union */ @@ -2378,10 +2434,12 @@ def Union : TuringOp<"union", [NoTerminator]> { the concatenation is the sink seeing each nest's rows in turn. Not Pure: its branches emit the result, so it must survive dead-code elimination as db.output does. - The body of a db.call_subquery is a UNION when the CALL's is. The rows the branches - produce then feed the rest of the body rather than the result table, so the op has one - result per column and each branch ends in a db.yield of its columns. The branches read - the body's arguments, the columns of the row the body runs for: + The body of a db.call_subquery is a UNION when the CALL's is, and so is a UNION + segment of a NEXT chain. The rows the branches produce then feed what follows rather + than the result table, so the op has one result per column and each branch ends in a + db.yield of its columns. The branches of a CALL body read the body's arguments, the + columns of the row the body runs for; those of a segment read the table in flight + through a db.scan_materialized each: %z = db.union { %s, %e, %et, %i = db.get_out_edges(%pIn, {}) : ... diff --git a/query/ir/dialect/db/DBTypes.td b/query/ir/dialect/db/DBTypes.td index 7ba5826dc4..ac8141f52c 100644 --- a/query/ir/dialect/db/DBTypes.td +++ b/query/ir/dialect/db/DBTypes.td @@ -90,4 +90,14 @@ def DistinctSetType : TuringType<"DistinctSet", "distinct_set"> { }]; } +// The rows in flight held once for several dataflows to read in turn: the branches of a +// UNION that a NEXT hands the whole table it returned. +def MaterializedType : TuringType<"Materialized", "materialized"> { + let summary = "Rows held by db.materialize for each db.scan_materialized to read"; + let description = [{ + Produced by db.materialize and read by every db.scan_materialized that walks the + rows it holds. + }]; +} + #endif // TURINGDB_TYPES diff --git a/query/ir/lowering/DBLowering.cpp b/query/ir/lowering/DBLowering.cpp index c5861c771d..d73e6ee785 100644 --- a/query/ir/lowering/DBLowering.cpp +++ b/query/ir/lowering/DBLowering.cpp @@ -953,7 +953,8 @@ bool opensSourceLoop(mlir::Operation* operation) { mlir::db::GetOutEdgesByTypeAndLabel, mlir::db::GetInEdgesByTypeAndLabel, mlir::db::ExplorePaths, - mlir::db::CallProcedure>(operation); + mlir::db::CallProcedure, + mlir::db::ScanMaterialized>(operation); } // An EXISTS, a COUNT or a COLLECT: a subquery standing in an expression, one value per @@ -1442,6 +1443,10 @@ void DBLowering::lowerOperationByKind(mlir::Operation& operation) { lowerConditional(conditional); } else if (mlir::db::DistinctSet distinctSet = mlir::dyn_cast(operation)) { lowerDistinctSet(distinctSet); + } else if (mlir::db::Materialize materialize = mlir::dyn_cast(operation)) { + lowerMaterialize(materialize); + } else if (mlir::db::ScanMaterialized scan = mlir::dyn_cast(operation)) { + lowerScanMaterialized(scan); } else if (mlir::db::OptionalMatch optionalMatch = mlir::dyn_cast(operation)) { lowerOptionalMatch(optionalMatch); } else if (mlir::db::CallSubquery callSubquery = mlir::dyn_cast(operation)) { @@ -4419,6 +4424,51 @@ void DBLowering::lowerDistinctSet(mlir::db::DistinctSet distinctSet) { _valueMap[distinctSet.getSet()] = _builder.create(_builder.getUnknownLoc()).getState(); } +// The buffer is emptied each step of the root and filled in the loop producing the rows, so +// a scan placed after that loop finds every row of the step +void DBLowering::lowerMaterialize(mlir::db::Materialize materialize) { + const mlir::Location loc = _builder.getUnknownLoc(); + + llvm::SmallVector chunks; + for (const mlir::Value column : materialize.getColumns()) { + chunks.push_back(mapValue(column)); + } + + rowAlignBufferedChunks(chunks); + + _builder.setInsertionPointToStart(_rootBlock); + const mlir::Value state = _builder.create(loc).getState(); + + setInsertionInto(deepestOwnerBlock(chunks, _rootBlock)); + _builder.create(loc, state, chunks); + + _valueMap[materialize.getRows()] = state; +} + +void DBLowering::lowerScanMaterialized(mlir::db::ScanMaterialized scan) { + const mlir::Value state = mapValue(scan.getRows()); + + nl::UnionCollect collect; + for (mlir::Operation* user : state.getUsers()) { + if (nl::UnionCollect stateCollect = mlir::dyn_cast(user)) { + collect = stateCollect; + break; + } + } + + if (!collect) { + throw IRException("db.scan_materialized reads rows no db.materialize collected"); + } + + const llvm::SmallVector chunkTypes(collect.getColumns().getTypes()); + const nl::IteratorType iteratorType = nl::IteratorType::get(_builder.getContext(), chunkTypes); + + setInsertionInto(_rootBlock); + nl::UnionDrain drain = _builder.create(_builder.getUnknownLoc(), iteratorType, state); + + buildLoopForSource(drain.getResult(), scan.getOperation()); +} + void DBLowering::lowerRemoveDuplicates(mlir::db::RemoveDuplicates distinct) { // The nl chunks the deduped columns lowered to; these are what the filter // reads to build each row's key, and gathers the survivors from. @@ -5181,11 +5231,13 @@ void DBLowering::visitProducer(mlir::Value column, // A sort or a grouped aggregate accumulates the whole relation before emitting any of // it, so the loops feeding it have to see every row: the walk stops and the limit - // budgets the emit loop alone. An optional match accumulates one step of the rows it - // joins onto rather than the relation, and emits them in input order, so once the - // budget is spent no later step can contribute a row - the walk carries on and bounds - // the loops feeding it too. - const bool accumulatesTheRelation = mlir::isa(definingOp); + // budgets the emit loop alone. So does a materialize, whose scans are the emit loops. + // An optional match accumulates one step of the rows it joins onto rather than the + // relation, and emits them in input order, so once the budget is spent no later step + // can contribute a row - the walk carries on and bounds the loops feeding it too. + const bool accumulatesTheRelation = mlir::isa(definingOp); const bool breaksPipeline = accumulatesTheRelation || reducesToOneRow(definingOp); frame._reachedALoop = opensLoop || isCrossProduct || isHashJoin || emitsThroughLoop || loopsOverSingleRows; diff --git a/query/ir/lowering/DBLowering.h b/query/ir/lowering/DBLowering.h index f7d01c5434..72479a2f3d 100644 --- a/query/ir/lowering/DBLowering.h +++ b/query/ir/lowering/DBLowering.h @@ -428,6 +428,11 @@ class DBLowering { // rows in, hoisted where every branch's filter can reach it void lowerDistinctSet(mlir::db::DistinctSet distinctSet); + void lowerMaterialize(mlir::db::Materialize materialize); + + // Opens a loop over the rows a db.materialize collected, after the loop collecting them + void lowerScanMaterialized(mlir::db::ScanMaterialized scan); + // Brings every branch's result columns to the one type each column of the result // carries, once the branches have been lowered and their types are known. // @param columnNames names the columns for a diagnostic, and is null where they have none. diff --git a/query/parser/CypherLexer.l b/query/parser/CypherLexer.l index 2ac9f388b2..5257780618 100644 --- a/query/parser/CypherLexer.l +++ b/query/parser/CypherLexer.l @@ -156,6 +156,7 @@ using Token = db::YCypherParser::token; "PUSH" { KEYWORD(PUSH) } "PULL" { KEYWORD(PULL) } "NEW" { KEYWORD(NEW) } +"NEXT" { KEYWORD(NEXT) } "CSV" { KEYWORD(CSV) } "GML" { KEYWORD(GML) } "PARQUET" { KEYWORD(PARQUET) } diff --git a/query/parser/CypherParser.y b/query/parser/CypherParser.y index ac2073aaab..9fcfee459a 100644 --- a/query/parser/CypherParser.y +++ b/query/parser/CypherParser.y @@ -247,6 +247,7 @@ %token PUSH %token PULL %token NEW +%token NEXT %token FAIL %token CSV %token GML @@ -444,6 +445,10 @@ %type foreachSt %type foreachUpdates %type callSubqueryBody +%type nextSegment +%type segmentBeforeNext +%type nextHead +%type nextChain %type conditionalBranches %type whenBranches %type conditionalBranchQuery @@ -520,6 +525,43 @@ regularQuery $$ = ConditionalQuery::create(ast, body); LOC($$, @$); } + | nextChain { $$ = ParserUtils::createNextQuery(ast, $1, @$); } + ; + +nextChain + : nextHead nextSegment { + $$ = std::move($1); + $$.push_back(std::move($2)); + } + ; + +nextHead + : segmentBeforeNext { $$.push_back(std::move($1)); } + | nextHead segmentBeforeNext { + $$ = std::move($1); + $$.push_back(std::move($2)); + } + ; + +segmentBeforeNext + : nextSegment NEXT { + if (ParserUtils::returnsUnaliasedItem($1)) { + scanner.syntaxError(@1, "Expression in a RETURN followed by NEXT must be aliased (use AS)"); + } + + $$ = std::move($1); + } + ; + +nextSegment + : singlePartQuery { $$.push_back({$1, false, {}}); } + | unionOperand unionList { + $$.push_back({$1._query, false, {}, nullptr, $1._whenCall}); + for (const UnionQuery::Branch& branch : $2) { + $$.push_back({branch._query, branch._all, {}, nullptr, branch._whenCall}); + } + } + | conditionalBranches { $$ = std::move($1); } ; explainQuery @@ -1210,14 +1252,8 @@ callSubquerySt ; callSubqueryBody - : singlePartQuery { $$.push_back({$1, false, {}}); } - | unionOperand unionList { - $$.push_back({$1._query, false, {}, nullptr, $1._whenCall}); - for (const UnionQuery::Branch& branch : $2) { - $$.push_back({branch._query, branch._all, {}, nullptr, branch._whenCall}); - } - } - | conditionalBranches { $$ = std::move($1); } + : nextSegment { $$ = std::move($1); } + | nextChain { $$.push_back({ParserUtils::createNextQuery(ast, $1, @$), false, {}}); } ; conditionalBranches @@ -1240,6 +1276,7 @@ conditionalBranchQuery : singlePartQuery { $$ = $1; } | OBRACE singlePartQuery CBRACE { $$ = $2; } | OBRACE unionOperand unionList CBRACE { scanner.notImplemented(@$, "UNION inside a WHEN branch"); } + | OBRACE nextChain CBRACE { $$ = ParserUtils::createNextQuery(ast, $2, @2); } ; existsBody @@ -1258,6 +1295,7 @@ existsBody $$.push_back(branch._query); } } + | nextChain { $$.push_back(ParserUtils::createNextQuery(ast, $1, @$)); } ; countBody @@ -1276,6 +1314,7 @@ countBody $$.push_back(branch); } } + | nextChain { $$.push_back({ParserUtils::createNextQuery(ast, $1, @$), false}); } ; collectBody @@ -1830,6 +1869,7 @@ unionOperand scanner.syntaxError(@2, "A WHEN combined by UNION must name each column it returns with AS"); } } + | OBRACE nextChain CBRACE { $$ = {ParserUtils::createNextQuery(ast, $2, @2), false, nullptr}; } ; subqueryExist @@ -2180,6 +2220,7 @@ symbol | DATETIMES { $$ = Symbol::create(ast, $1); } | DURATIONS { $$ = Symbol::create(ast, $1); } | FILTER { $$ = Symbol::create(ast, $1); } + | NEXT { $$ = Symbol::create(ast, $1); } | EXTRACT { $$ = Symbol::create(ast, $1); } | EMBEDDING { $$ = Symbol::create(ast, $1); } | TYPE { $$ = Symbol::create(ast, $1); } diff --git a/query/parser/ParserUtils.cpp b/query/parser/ParserUtils.cpp index e7e8f1ab7f..52e6878698 100644 --- a/query/parser/ParserUtils.cpp +++ b/query/parser/ParserUtils.cpp @@ -23,9 +23,11 @@ #include "stmt/CallStmt.h" #include "stmt/ReturnStmt.h" #include "stmt/MatchStmt.h" +#include "stmt/NextSegmentStmt.h" #include "stmt/SetStmt.h" #include "stmt/StmtContainer.h" #include "stmt/UnwindStmt.h" +#include "stmt/WithStmt.h" #include "CypherAST.h" #include "EdgePattern.h" #include "FunctionInvocation.h" @@ -322,6 +324,115 @@ bool ParserUtils::createWhenOperand(CypherAST* ast, return true; } +bool ParserUtils::returnsUnaliasedItem(const CallSubqueryStmt::Branches& segment) { + for (const CallSubqueryStmt::Branch& branch : segment) { + const ReturnStmt* returnStmt = branch._query->getReturnStmt(); + if (!returnStmt) { + continue; + } + + for (const Projection::ReturnItem& item : returnStmt->getProjection()->items()) { + const Expr* expr = std::get(item); + const bool named = expr->getKind() == Expr::Kind::SYMBOL || !expr->getName().empty(); + + if (!named) { + return true; + } + } + } + + return false; +} + +SinglePartQuery* ParserUtils::createNextQuery(CypherAST* ast, + const NextSegments& segments, + const SourceLocation& location) { + SourceManager* sourceManager = ast->getSourceManager(); + + StmtContainer* stmts = StmtContainer::create(ast); + sourceManager->setLocation(stmts, location); + + SinglePartQuery* query = SinglePartQuery::create(ast); + query->setStmts(stmts); + sourceManager->setLocation(query, location); + + for (size_t index = 0; index < segments.size(); index++) { + const CallSubqueryStmt::Branches& segment = segments[index]; + const bool endsTheChain = index + 1 == segments.size(); + + SinglePartQuery* first = segment.front()._query; + ReturnStmt* returnStmt = first->getReturnStmt(); + + const bool composite = segment.size() > 1 || segment.front()._condition; + + if (!composite && (returnStmt || endsTheChain)) { + ast->nestQuery(first); + + if (const StmtContainer* segmentStmts = first->getStmts()) { + for (Stmt* stmt : segmentStmts->stmts()) { + stmts->add(stmt); + } + } + + if (endsTheChain) { + query->setReturnStmt(returnStmt); + } else { + WithStmt* with = WithStmt::create(ast, returnStmt->getProjection()); + sourceManager->setLocation(with, location); + stmts->add(with); + } + + continue; + } + + CallSubqueryStmt* body = CallSubqueryStmt::create(ast, segment); + sourceManager->setLocation(body, location); + + NextSegmentStmt* segmentStmt = NextSegmentStmt::create(ast, body); + sourceManager->setLocation(segmentStmt, location); + stmts->add(segmentStmt); + + if (endsTheChain && returnStmt) { + query->setReturnStmt(createSegmentReturn(ast, returnStmt->getProjection(), location)); + } + } + + return query; +} + +ReturnStmt* ParserUtils::createSegmentReturn(CypherAST* ast, + const Projection* projection, + const SourceLocation& location) { + SourceManager* sourceManager = ast->getSourceManager(); + + Projection* columns = Projection::create(ast); + sourceManager->setLocation(columns, location); + + if (projection->isReturningAll()) { + columns->setReturnAll(); + } + + for (const Projection::ReturnItem& item : projection->items()) { + const Expr* expr = std::get(item); + + std::string_view name = expr->getName(); + if (name.empty() && expr->getKind() == Expr::Kind::SYMBOL) { + name = static_cast(expr)->getSymbol()->getName(); + } else if (name.empty()) { + name = sourceManager->getStringRepr(std::bit_cast(expr)); + } + + SymbolExpr* column = SymbolExpr::create(ast, Symbol::create(ast, name)); + sourceManager->setLocation(column, location); + columns->addExpr(column); + } + + ReturnStmt* returnStmt = ReturnStmt::create(ast, columns); + sourceManager->setLocation(returnStmt, location); + + return returnStmt; +} + CallSubqueryStmt* ParserUtils::createForeach(CypherAST* ast, Symbol* variable, Expr* list, diff --git a/query/parser/ParserUtils.h b/query/parser/ParserUtils.h index 2598f950c8..917e9e65df 100644 --- a/query/parser/ParserUtils.h +++ b/query/parser/ParserUtils.h @@ -24,6 +24,8 @@ class Pattern; class Parameter; class PatternElement; class PropertyExpr; +class Projection; +class ReturnStmt; class SetStmt; class SinglePartQuery; class StmtContainer; @@ -41,6 +43,10 @@ class ParserUtils { SourceLocation _rightOperandLocation; }; + // The queries a NEXT chains, in order: each a single query, the branches of a UNION or + // those of a WHEN + using NextSegments = std::vector; + ParserUtils() = delete; ~ParserUtils() = delete; @@ -79,6 +85,17 @@ class ParserUtils { const SourceLocation& location, UnionQuery::Branch& operand); + // Whether an item of a segment's RETURN has neither an alias nor a variable to name the + // column the next segment reads it through + static bool returnsUnaliasedItem(const CallSubqueryStmt::Branches& segment); + + // `a RETURN x NEXT b` is `a WITH x b`. A UNION or a WHEN segment, and one ending on an + // updating clause that NEXT follows, is a NextSegmentStmt; ending the chain, its + // columns are what the query returns. + static SinglePartQuery* createNextQuery(CypherAST* ast, + const NextSegments& segments, + const SourceLocation& location); + // `FOREACH (x IN list | updates)` is `CALL (*) { UNWIND list AS x updates }` static CallSubqueryStmt* createForeach(CypherAST* ast, Symbol* variable, @@ -148,6 +165,12 @@ class ParserUtils { private: static void listExprToFloatVector(const ListLiteral* list, std::vector& out); + + // `RETURN `, read back by name from the scope the + // segment that projects them publishes + static ReturnStmt* createSegmentReturn(CypherAST* ast, + const Projection* projection, + const SourceLocation& location); }; } diff --git a/test/query/ir/CMakeLists.txt b/test/query/ir/CMakeLists.txt index 6671b1d234..8e41478df7 100644 --- a/test/query/ir/CMakeLists.txt +++ b/test/query/ir/CMakeLists.txt @@ -73,6 +73,7 @@ add_subdirectory(map) add_subdirectory(match) add_subdirectory(memory) add_subdirectory(merge) +add_subdirectory(next) add_subdirectory(null) add_subdirectory(optional_match) add_subdirectory(parameter) diff --git a/test/query/ir/next/CMakeLists.txt b/test/query/ir/next/CMakeLists.txt new file mode 100644 index 0000000000..143d35d64f --- /dev/null +++ b/test/query/ir/next/CMakeLists.txt @@ -0,0 +1,5 @@ +# Queries chained with NEXT. +add_turing_test(test_query_ir_next_query NextQueryTest.cpp) + +# NEXT chains whose segments write. +add_turing_test(test_query_ir_next_write NextWriteTest.cpp) diff --git a/test/query/ir/next/NextQueryTest.cpp b/test/query/ir/next/NextQueryTest.cpp new file mode 100644 index 0000000000..e1ea31bb1a --- /dev/null +++ b/test/query/ir/next/NextQueryTest.cpp @@ -0,0 +1,153 @@ +#include + +#include "WriteQueryTest.h" + +using namespace db; +using namespace turing::test; + +// Queries chained with NEXT: each segment reads the whole table the segment before it +// returned, under the names it returned and nothing else +class NextQueryTest : public WriteQueryTest { +}; + +TEST_F(NextQueryTest, readsWhatTheSegmentBeforeReturned) { + expectRows("MATCH (p:Person) WHERE p.isFrench RETURN p.name AS name NEXT RETURN name", + {{"Remy"}, {"Adam"}, {"Maxime"}, {"Luc"}}); +} + +TEST_F(NextQueryTest, rejectsAVariableTheSegmentBeforeDidNotReturn) { + expectError("MATCH (p:Person) RETURN p.name AS name NEXT RETURN p", "Variable 'p' not found"); +} + +TEST_F(NextQueryTest, rejectsAnUnaliasedExpressionBeforeNext) { + expectError("MATCH (p:Person) RETURN p.name NEXT RETURN 1 AS one", + "Expression in a RETURN followed by NEXT must be aliased (use AS)"); +} + +TEST_F(NextQueryTest, chainsThreeSegments) { + expectRows("UNWIND [1, 2, 3] AS x RETURN x NEXT RETURN x * 10 AS y NEXT RETURN sum(y) AS total", + {{"60"}}); +} + +// 15 INTERESTED_IN edges, 3 of them to Gym and 2 each to Bio, Computers and Cooking +TEST_F(NextQueryTest, aggregatesOverEveryRowOfTheTable) { + expectRows("MATCH (p:Person)-[:INTERESTED_IN]->(i:Interest) RETURN p, i NEXT RETURN count(*) AS n", + {{"15"}}); + + expectRowsInOrder("MATCH (p:Person)-[:INTERESTED_IN]->(i:Interest) RETURN i.name AS interest " + "NEXT RETURN interest, count(*) AS n ORDER BY n DESC, interest LIMIT 3", + {{"Gym", "3"}, {"Bio", "2"}, {"Computers", "2"}}); +} + +TEST_F(NextQueryTest, matchesFromAnEntityTheSegmentBeforeReturned) { + expectRows("MATCH (p:Person) WHERE p.name = 'Remy' RETURN p AS person " + "NEXT MATCH (person)-[:INTERESTED_IN]->(i:Interest) RETURN i.name AS name", + {{"Ghosts"}, {"Computers"}, {"Eighties"}}); +} + +TEST_F(NextQueryTest, passesEveryVariableOfAReturnStar) { + expectRows("MATCH (p:Person) WHERE p.name = 'Remy' WITH p.name AS name, p.age AS age RETURN * " + "NEXT RETURN name, age", + {{"Remy", "32"}}); +} + +// The French are Remy, Adam, Maxime and Luc; the PhDs Remy, Adam, Luc and Martina +TEST_F(NextQueryTest, readsTheWholeResultOfAUnion) { + expectRows("MATCH (p:Person) WHERE p.isFrench RETURN p.name AS name " + "UNION MATCH (p:Person) WHERE p.hasPhD RETURN p.name AS name " + "NEXT RETURN count(*) AS n", + {{"5"}}); + + expectRows("MATCH (p:Person) WHERE p.isFrench RETURN p.name AS name " + "UNION ALL MATCH (p:Person) WHERE p.hasPhD RETURN p.name AS name " + "NEXT RETURN count(*) AS n", + {{"8"}}); +} + +TEST_F(NextQueryTest, runsEveryBranchOfAUnionOverTheWholeTable) { + expectRows("MATCH (p:Person) RETURN p.name AS name, p.isFrench AS french " + "NEXT RETURN count(*) AS n " + "UNION ALL WITH name, french WHERE french RETURN count(name) AS n", + {{"8"}, {"4"}}); +} + +TEST_F(NextQueryTest, dedupsAUnionAcrossTheWholeTable) { + expectRows("UNWIND [1, 1, 2] AS x RETURN x NEXT RETURN x UNION RETURN x + 1 AS x", + {{"1"}, {"2"}, {"3"}}); +} + +TEST_F(NextQueryTest, cutsOneBranchOfAUnionWithoutCuttingTheOthers) { + expectRows("MATCH (p:Person) RETURN p.name AS name " + "NEXT RETURN name ORDER BY name LIMIT 2 " + "UNION ALL RETURN name ORDER BY name DESC LIMIT 1", + {{"Adam"}, {"Cyrus"}, {"Suhas"}}); + + expectRows("MATCH (p:Person) RETURN p.name AS name " + "NEXT RETURN name LIMIT 2 UNION ALL RETURN name " + "NEXT RETURN count(*) AS n", + {{"10"}}); +} + +TEST_F(NextQueryTest, feedsAUnionIntoAnotherUnion) { + expectRows("UNWIND [1, 2] AS x RETURN x UNION ALL UNWIND [3] AS x RETURN x " + "NEXT RETURN sum(x) AS x UNION ALL RETURN max(x) AS x " + "NEXT RETURN sum(x) AS total", + {{"9"}}); +} + +// Remy, Adam, Luc and Martina have a PhD +TEST_F(NextQueryTest, decidesAWhenPerRow) { + expectRows("MATCH (p:Person) RETURN p.name AS name, p.hasPhD AS phd " + "NEXT WHEN phd THEN RETURN name, 'doctor' AS title ELSE RETURN name, 'student' AS title", + {{"Remy", "doctor"}, {"Adam", "doctor"}, {"Luc", "doctor"}, {"Martina", "doctor"}, + {"Maxime", "student"}, {"Suhas", "student"}, {"Cyrus", "student"}, {"Doruk", "student"}}); +} + +TEST_F(NextQueryTest, rebindsANameAWhenReturnsAgain) { + expectRows("MATCH (p:Person) RETURN p AS person " + "NEXT WHEN person.isFrench THEN RETURN person.name AS person ELSE RETURN 'other' AS person", + {{"Remy"}, {"Adam"}, {"Maxime"}, {"Luc"}, {"other"}, {"other"}, {"other"}, {"other"}}); +} + +TEST_F(NextQueryTest, aggregatesWhatAWhenReturned) { + expectRows("MATCH (p:Person) RETURN p.hasPhD AS phd " + "NEXT WHEN phd THEN RETURN 'doctor' AS title ELSE RETURN 'student' AS title " + "NEXT RETURN title, count(*) AS n", + {{"doctor", "4"}, {"student", "4"}}); +} + +TEST_F(NextQueryTest, startsFromAStandaloneWhen) { + expectRows("WHEN true THEN RETURN 1 AS x ELSE RETURN 2 AS x NEXT RETURN x + 1 AS y", {{"2"}}); +} + +TEST_F(NextQueryTest, groupsPerRowInsideACall) { + expectRows("MATCH (p:Person) " + "CALL (p) { MATCH (p)-[:INTERESTED_IN]->(i:Interest) RETURN i NEXT RETURN count(i) AS n } " + "RETURN p.name AS name, n", + {{"Remy", "3"}, {"Adam", "2"}, {"Maxime", "2"}, {"Luc", "2"}, + {"Martina", "1"}, {"Suhas", "2"}, {"Cyrus", "2"}, {"Doruk", "1"}}); +} + +TEST_F(NextQueryTest, countsWhatAChainedBodyReturns) { + expectRows("MATCH (p:Person) WHERE p.isFrench " + "RETURN p.name AS name, " + "COUNT { MATCH (p)-[:INTERESTED_IN]->(i:Interest) RETURN i NEXT RETURN i } AS n", + {{"Remy", "3"}, {"Adam", "2"}, {"Maxime", "2"}, {"Luc", "2"}}); +} + +TEST_F(NextQueryTest, chainsInsideAUnionOperand) { + expectRows("{ MATCH (p:Person) RETURN p.name AS name NEXT RETURN name ORDER BY name LIMIT 1 } " + "UNION ALL " + "{ MATCH (i:Interest) RETURN i.name AS name NEXT RETURN name ORDER BY name LIMIT 1 }", + {{"Adam"}, {"Animals"}}); +} + +TEST_F(NextQueryTest, chainsInsideAWhenBranch) { + expectRows("WHEN true THEN { MATCH (p:Person) RETURN p.age AS age NEXT RETURN max(age) AS a } " + "ELSE { RETURN 0 AS a }", + {{"32"}}); +} + +TEST_F(NextQueryTest, keepsNextAVariableName) { + expectRows("MATCH (p:Person) WHERE p.name = 'Remy' RETURN p.name AS next", {{"Remy"}}); +} diff --git a/test/query/ir/next/NextWriteTest.cpp b/test/query/ir/next/NextWriteTest.cpp new file mode 100644 index 0000000000..cc46906ef2 --- /dev/null +++ b/test/query/ir/next/NextWriteTest.cpp @@ -0,0 +1,32 @@ +#include + +#include "WriteQueryTest.h" + +using namespace db; +using namespace turing::test; + +// NEXT chains whose segments write +class NextWriteTest : public WriteQueryTest { +}; + +TEST_F(NextWriteTest, writesOverTheRowsTheSegmentBeforeReturned) { + applyWrite("MATCH (p:Person) WHERE p.isFrench RETURN p NEXT SET p.seen = true"); + + expectRows("MATCH (p:Person) WHERE p.seen RETURN p.name", + {{"Remy"}, {"Adam"}, {"Maxime"}, {"Luc"}}); +} + +// A segment ending on a write returns nothing, so the one after it starts from one row +TEST_F(NextWriteTest, startsAfreshAfterASegmentThatWrites) { + expectWriteRows("UNWIND [1, 2, 3] AS x CREATE (:Tag {x: x}) NEXT RETURN 1 AS one", {{"1"}}); + + expectRows("MATCH (t:Tag) RETURN t.x", {{"1"}, {"2"}, {"3"}}); +} + +// Remy, Adam, Luc and Martina have a PhD +TEST_F(NextWriteTest, writesInTheWhenBranchEachRowTakes) { + applyWrite("MATCH (p:Person) RETURN p " + "NEXT WHEN p.hasPhD THEN { SET p.title = 'doctor' } ELSE { SET p.title = 'student' }"); + + expectRows("MATCH (p:Person) RETURN p.title, count(*)", {{"doctor", "4"}, {"student", "4"}}); +} From 66920d1339b306203b3590f215d5f9fbd959edf5 Mon Sep 17 00:00:00 2001 From: Remy Boutonnet Date: Wed, 21 Oct 2026 13:30:25 +0200 Subject: [PATCH 2/3] MLIR: rename the nl union buffer to a row buffer --- query/ir/dialect/nl/NLOps.cpp | 4 +- query/ir/dialect/nl/NLOps.td | 57 +++++++++-------- query/ir/dialect/nl/NLTypes.td | 29 +++++---- query/ir/exec/NLExecutor.cpp | 12 ++-- query/ir/exec/NLExecutor.h | 14 ++-- query/ir/exec/NLProgram.cpp | 4 +- query/ir/exec/NLProgram.h | 54 ++++++++-------- query/ir/exec/NLTranslator.cpp | 64 +++++++++---------- query/ir/exec/NLTranslator.h | 30 ++++----- query/ir/lowering/DBLowering.cpp | 32 +++++----- query/ir/lowering/DBLowering.h | 8 +-- samples/mlir/count_subquery_union.nl.mlir | 8 +-- .../memory/RetainedValuesAcrossChunksTest.cpp | 2 +- 13 files changed, 160 insertions(+), 158 deletions(-) diff --git a/query/ir/dialect/nl/NLOps.cpp b/query/ir/dialect/nl/NLOps.cpp index 7ede39cc9f..9f0577f6d7 100644 --- a/query/ir/dialect/nl/NLOps.cpp +++ b/query/ir/dialect/nl/NLOps.cpp @@ -658,9 +658,9 @@ LogicalResult SortCollect::verify() { return success(); } -// An nl.union_collect must append at least one column: the buffers' row count is read +// An nl.row_collect must append at least one column: the buffers' row count is read // from the first of them. -LogicalResult UnionCollect::verify() { +LogicalResult RowCollect::verify() { if (getColumns().empty()) { return emitOpError("requires at least one column to collect"); } diff --git a/query/ir/dialect/nl/NLOps.td b/query/ir/dialect/nl/NLOps.td index 8d8a9202d3..7dbb89183b 100644 --- a/query/ir/dialect/nl/NLOps.td +++ b/query/ir/dialect/nl/NLOps.td @@ -3970,43 +3970,43 @@ def RowListsResult : NLOp<"row_lists_result", [Pure, RowAlignedThroughOperands]> } /** -* UnionBuffer +* RowBuffer */ -// NOT Pure: each nl.union_buffer is a distinct accumulator, and resetting its buffers +// NOT Pure: each nl.row_buffer is a distinct accumulator, and resetting its buffers // is a side effect (like nl.sort_buffer). -def UnionBuffer : NLOp<"union_buffer", []> { - let summary = "Create (and reset) the accumulator the branches of one UNION collect into"; +def RowBuffer : NLOp<"row_buffer", []> { + let summary = "Create (and reset) a row accumulator"; let description = [{ Sits at the top of the block whose step it covers - the block the branches of a - UNION inside a CALL body are rooted in - so the accumulator is empty again at every - step. The buffers' column types are recovered from the nl.union_collect ops that - feed the handle. + UNION inside a CALL body, or of a NEXT segment, are rooted in - so the accumulator + is empty again at every step. The buffers' column types are recovered from the + nl.row_collect ops that feed the handle. - %state = nl.union_buffer + %state = nl.row_buffer }]; - let results = (outs NLUnionState:$state); + let results = (outs NLRowState:$state); let assemblyFormat = "attr-dict"; } /** -* UnionCollect +* RowCollect */ // NOT Pure: it appends the current chunk's rows to the handle's buffers, so it must run // exactly where it sits (like nl.sort_collect). -def UnionCollect : NLOp<"union_collect", []> { - let summary = "Append this step's chunk of every column of one UNION branch to its accumulator"; +def RowCollect : NLOp<"row_collect", []> { + let summary = "Append this step's chunk of every column to a row accumulator"; let description = [{ - Runs in the innermost loop body of one branch, once per step, and appends the - current chunk of each column that branch yields to the matching buffer. Every - branch has its own nl.union_collect on the same handle, with the same column - types, so the buffers hold the rows of one branch and then of the next. + Runs in the innermost loop body of what produces the rows, once per step, and + appends the current chunk of each column to the matching buffer. Each branch of a + UNION in a CALL body has its own nl.row_collect on the same handle, with the same + column types, so the buffers hold the rows of one branch and then of the next. - nl.union_collect %state, (%name) : !nl.chunk> + nl.row_collect %state, (%name) : !nl.chunk> }]; - let arguments = (ins NLUnionStateHandle:$state, Variadic:$columns); + let arguments = (ins NLRowStateHandle:$state, Variadic:$columns); let assemblyFormat = [{ $state `,` `(` $columns `)` attr-dict `:` qualified(type($columns)) @@ -4015,26 +4015,27 @@ def UnionCollect : NLOp<"union_collect", []> { } /** -* UnionDrain +* RowDrain */ // Pure: given a filled accumulator it deterministically yields the same rows, and // produces no side effect of its own - the emission happens in the driving nl.for. // Mirrors nl.sort: a source op whose loop does the work. -def UnionDrain : NLOp<"union_drain", [Pure]> { - let summary = "Create a database iterator over the rows the branches of a UNION collected"; +def RowDrain : NLOp<"row_drain", [Pure]> { + let summary = "Create a database iterator over the rows a row accumulator collected"; let description = [{ - The emit phase of a UNION inside a CALL body. Consumes a filled nl.union_state and - produces an iterator whose steps yield its rows in the order they were collected, - one chunk per column per step: + The emit phase of a row accumulator. Reads a filled nl.row_state without consuming + it, so each drain of a NEXT segment's table walks every row, and produces an + iterator whose steps yield its rows in the order they were collected, one chunk per + column per step: - %rows = nl.union_drain(%state) : !nl.iter> + %rows = nl.row_drain(%state) : !nl.iter> nl.for %z in %rows : !nl.iter> { ... } - Its first step must find every branch's rows already collected, so place it after - the branches' loop nests, in the block holding the nl.union_buffer. + Its first step must find every row already collected, so place it after the loop + nests that collect, in the block holding the nl.row_buffer. }]; - let arguments = (ins NLUnionStateHandle:$state); + let arguments = (ins NLRowStateHandle:$state); let results = (outs NLIterator:$result); let assemblyFormat = "`(` $state `)` attr-dict `:` qualified(type($result))"; diff --git a/query/ir/dialect/nl/NLTypes.td b/query/ir/dialect/nl/NLTypes.td index e4ab29ae26..a6b30e59fa 100644 --- a/query/ir/dialect/nl/NLTypes.td +++ b/query/ir/dialect/nl/NLTypes.td @@ -178,15 +178,16 @@ def NLRowListsState : NLType<"RowListsState", "row_lists_state"> { }]; } -def NLUnionState : NLType<"UnionState", "union_state"> { - let summary = "A UNION accumulator handle"; +def NLRowState : NLType<"RowState", "row_state"> { + let summary = "A row accumulator handle"; let description = [{ - Stands for the rows the branches of a UNION inside a CALL body produce over - the current step. Produced by nl.union_buffer at the top of the block whose - step it covers - so it is reset once per step - appended to by one - nl.union_collect per branch, and drained by the nl.for over nl.union_drain, - which emits the rows in the order they were collected. The unordered sibling - of !nl.sort_state, fed by several collects rather than one. + Stands for rows held over the current step: the ones the branches of a UNION + inside a CALL body produce, or the table a db.materialize holds for the + branches of a NEXT segment. Produced by nl.row_buffer at the top of the block + whose step it covers - so it is reset once per step - appended to by one + nl.row_collect per producer, and drained by each nl.for over an nl.row_drain, + which emits the rows in the order they were collected. The unordered sibling of + !nl.sort_state. }]; } @@ -489,13 +490,13 @@ def NLOptionalStateHandle : Type< "optional state handle", "::mlir::nl::OptionalStateType">, BuildableType<"::mlir::nl::OptionalStateType::get($_builder.getContext())">; -// Constrains an operand to an !nl.union_state handle, as NLSortStateHandle constrains -// the sort one, and carries the recipe that lets nl.union_collect and nl.union_drain +// Constrains an operand to an !nl.row_state handle, as NLSortStateHandle constrains +// the sort one, and carries the recipe that lets nl.row_collect and nl.row_drain // omit the handle's type. -def NLUnionStateHandle : Type< - CPred<"::llvm::isa<::mlir::nl::UnionStateType>($_self)">, - "union state handle", "::mlir::nl::UnionStateType">, - BuildableType<"::mlir::nl::UnionStateType::get($_builder.getContext())">; +def NLRowStateHandle : Type< + CPred<"::llvm::isa<::mlir::nl::RowStateType>($_self)">, + "row state handle", "::mlir::nl::RowStateType">, + BuildableType<"::mlir::nl::RowStateType::get($_builder.getContext())">; // Constrains an operand to an !nl.exists_state handle, as NLOptionalStateHandle // constrains the optional one, and carries the recipe that lets nl.exists_mark and diff --git a/query/ir/exec/NLExecutor.cpp b/query/ir/exec/NLExecutor.cpp index 017f4a7a61..fdadc88566 100644 --- a/query/ir/exec/NLExecutor.cpp +++ b/query/ir/exec/NLExecutor.cpp @@ -9484,21 +9484,21 @@ void NLExecutor::runSortLoop(NLExecutionContext* context, NLFunctionData* data) } } -void NLExecutor::runUnionReset(NLExecutionContext* context, NLFunctionData* data) { - const NLUnionResetData* reset = static_cast(data); +void NLExecutor::runRowReset(NLExecutionContext* context, NLFunctionData* data) { + const NLRowResetData* reset = static_cast(data); reset->getState()->reset(); } -void NLExecutor::runUnionCollect(NLExecutionContext* context, NLFunctionData* data) { - const NLUnionCollectData* collect = static_cast(data); +void NLExecutor::runRowCollect(NLExecutionContext* context, NLFunctionData* data) { + const NLRowCollectData* collect = static_cast(data); for (const NLSortCollectData::Append& append : collect->appends()) { appendCollectedChunk(append, collect->getState()->values()); } } -void NLExecutor::runUnionLoop(NLExecutionContext* context, NLFunctionData* data) { - NLUnionLoopData* loopData = static_cast(data); +void NLExecutor::runRowLoop(NLExecutionContext* context, NLFunctionData* data) { + NLRowLoopData* loopData = static_cast(data); const size_t totalRows = loopData->getState()->getRowCount(); const NLStmtContainer* loopBody = loopData->getStmts(); diff --git a/query/ir/exec/NLExecutor.h b/query/ir/exec/NLExecutor.h index 17d68f0f0c..8cec5d841b 100644 --- a/query/ir/exec/NLExecutor.h +++ b/query/ir/exec/NLExecutor.h @@ -176,15 +176,15 @@ class NLExecutor { // variables and running the body (the nl.output) per chunk. static void runSortLoop(NLExecutionContext* context, NLFunctionData* data); - // Empty the buffers of a UNION accumulator; runs each time its block runs - static void runUnionReset(NLExecutionContext* context, NLFunctionData* data); + // Empty the buffers of a row accumulator; runs each time its block runs + static void runRowReset(NLExecutionContext* context, NLFunctionData* data); - // Append the current chunk of every column one UNION branch yields to its buffer - static void runUnionCollect(NLExecutionContext* context, NLFunctionData* data); + // Append the current chunk of every column to its buffer + static void runRowCollect(NLExecutionContext* context, NLFunctionData* data); - // The emit phase of a UNION inside a CALL body: re-chunk the collected rows in the - // order they were collected, running the body per chunk - static void runUnionLoop(NLExecutionContext* context, NLFunctionData* data); + // The emit phase of a row accumulator: re-chunk the collected rows in the order they + // were collected, running the body per chunk + static void runRowLoop(NLExecutionContext* context, NLFunctionData* data); // Empty the buffers and matched flags of an OPTIONAL MATCH accumulator and lay its row // tag out over this step's input rows; runs each time its block runs. diff --git a/query/ir/exec/NLProgram.cpp b/query/ir/exec/NLProgram.cpp index 1c169fdd76..57238c249a 100644 --- a/query/ir/exec/NLProgram.cpp +++ b/query/ir/exec/NLProgram.cpp @@ -411,7 +411,7 @@ void NLSortState::reset() { _sorted = false; } -size_t NLUnionState::getRowCount() const { +size_t NLRowState::getRowCount() const { if (_buffers.empty()) { return 0; } @@ -419,7 +419,7 @@ size_t NLUnionState::getRowCount() const { return _buffers.front()->size(); } -void NLUnionState::reset() { +void NLRowState::reset() { for (Column* buffer : _buffers) { buffer->clear(); } diff --git a/query/ir/exec/NLProgram.h b/query/ir/exec/NLProgram.h index f2a7fc1362..2196d69935 100644 --- a/query/ir/exec/NLProgram.h +++ b/query/ir/exec/NLProgram.h @@ -5183,11 +5183,11 @@ class NLCountSubqueryResultData : public NLFunctionData { ColumnVector* _result {nullptr}; }; -// Runtime state of one UNION inside a CALL body: one growing buffer per column, which -// the nl.union_collect of every branch appends to and the nl.for over nl.union_drain reads -// back in collected order. The buffer columns are borrowed: the translator pool-allocates +// Runtime state of one row accumulator: one growing buffer per column, which every +// nl.row_collect appends to and each nl.for over an nl.row_drain reads back in collected +// order. The buffer columns are borrowed: the translator pool-allocates // them in the same arena as the loop columns. -class NLUnionState { +class NLRowState { public: void addBuffer(Column* buffer) { _buffers.push_back(buffer); } @@ -5196,7 +5196,7 @@ class NLUnionState { size_t getRowCount() const; - // Clear every buffer; runs each time nl.union_buffer's block runs + // Clear every buffer; runs each time nl.row_buffer's block runs void reset(); // The values the buffers hold, copied in as NLSortState::values' are @@ -5207,30 +5207,30 @@ class NLUnionState { NLValueStore _values; }; -// nl.union_buffer data: resets an accumulator to empty each time its block runs -class NLUnionResetData : public NLFunctionData { +// nl.row_buffer data: resets an accumulator to empty each time its block runs +class NLRowResetData : public NLFunctionData { public: - NLUnionResetData(NLUnionState* state) + NLRowResetData(NLRowState* state) : _state(state) { } - NLUnionState* getState() const { return _state; } + NLRowState* getState() const { return _state; } private: - NLUnionState* _state {nullptr}; + NLRowState* _state {nullptr}; }; -// nl.union_collect data: appends the current chunk of every column one branch yields to -// the matching buffer of the accumulator, in NLSortCollectData's append shape -class NLUnionCollectData : public NLFunctionData { +// nl.row_collect data: appends the current chunk of every column to the matching buffer +// of the accumulator, in NLSortCollectData's append shape +class NLRowCollectData : public NLFunctionData { public: - NLUnionCollectData(NLUnionState* state) + NLRowCollectData(NLRowState* state) : _state(state) { } - NLUnionState* getState() const { return _state; } + NLRowState* getState() const { return _state; } const std::vector& appends() const { return _appends; } @@ -5239,20 +5239,20 @@ class NLUnionCollectData : public NLFunctionData { } private: - NLUnionState* _state {nullptr}; + NLRowState* _state {nullptr}; std::vector _appends; }; -// nl.for over nl.union_drain data: the emit phase of a UNION inside a CALL body, which -// gathers the collected rows chunk by chunk into the loop variables -class NLUnionLoopData : public NLFunctionData { +// nl.for over nl.row_drain data: the emit phase of a row accumulator, which gathers the +// collected rows chunk by chunk into the loop variables +class NLRowLoopData : public NLFunctionData { public: - NLUnionLoopData(NLUnionState* state) + NLRowLoopData(NLRowState* state) : _state(state) { } - NLUnionState* getState() const { return _state; } + NLRowState* getState() const { return _state; } const std::vector& columns() const { return _columns; } @@ -5269,7 +5269,7 @@ class NLUnionLoopData : public NLFunctionData { const NLStmtContainer* getStmts() const { return &_stmts; } private: - NLUnionState* _state {nullptr}; + NLRowState* _state {nullptr}; NLLimitState* _limit {nullptr}; std::vector _columns; ColumnVector _indices; @@ -5449,10 +5449,10 @@ class NLProgram { return statePtr; } - NLUnionState* allocUnionState() { - auto state = std::make_unique(); - NLUnionState* statePtr = state.get(); - _unionStates.push_back(std::move(state)); + NLRowState* allocRowState() { + auto state = std::make_unique(); + NLRowState* statePtr = state.get(); + _rowStates.push_back(std::move(state)); return statePtr; } @@ -5515,7 +5515,7 @@ class NLProgram { std::vector> _existsStates; std::vector> _countSubqueryStates; std::vector> _rowListsStates; - std::vector> _unionStates; + std::vector> _rowStates; std::vector> _procedureStates; std::unordered_map> _mergeNodeIndexes; NLMergePendingEdges _mergePendingEdges; diff --git a/query/ir/exec/NLTranslator.cpp b/query/ir/exec/NLTranslator.cpp index a90368d2ed..ce80ed5ac8 100644 --- a/query/ir/exec/NLTranslator.cpp +++ b/query/ir/exec/NLTranslator.cpp @@ -917,11 +917,11 @@ void NLTranslator::translateBlock(mlir::Block& block, NLStmtContainer* body) { config._kind = IteratorKind::OptionalDrain; config._optionalState = optionalStateFor(optionalDrain.getState()); _iteratorConfigs[optionalDrain.getResult()] = config; - } else if (nl::UnionDrain unionDrain = mlir::dyn_cast(operation)) { + } else if (nl::RowDrain rowDrain = mlir::dyn_cast(operation)) { IteratorConfig config; - config._kind = IteratorKind::UnionDrain; - config._unionState = unionStateFor(unionDrain.getState()); - _iteratorConfigs[unionDrain.getResult()] = config; + config._kind = IteratorKind::RowDrain; + config._rowState = rowStateFor(rowDrain.getState()); + _iteratorConfigs[rowDrain.getResult()] = config; } else if (nl::ProcedureInit procedureInit = mlir::dyn_cast(operation)) { IteratorConfig config; config._kind = IteratorKind::ProcedureInit; @@ -1103,10 +1103,10 @@ void NLTranslator::translateBlock(mlir::Block& block, NLStmtContainer* body) { translateSkipTruncate(truncate, body); } else if (nl::SortBuffer sortBuffer = mlir::dyn_cast(operation)) { translateSortBuffer(sortBuffer, body); - } else if (nl::UnionBuffer unionBuffer = mlir::dyn_cast(operation)) { - translateUnionBuffer(unionBuffer, body); - } else if (nl::UnionCollect unionCollect = mlir::dyn_cast(operation)) { - translateUnionCollect(unionCollect, body); + } else if (nl::RowBuffer rowBuffer = mlir::dyn_cast(operation)) { + translateRowBuffer(rowBuffer, body); + } else if (nl::RowCollect rowCollect = mlir::dyn_cast(operation)) { + translateRowCollect(rowCollect, body); } else if (nl::SortCollect sortCollect = mlir::dyn_cast(operation)) { translateSortCollect(sortCollect, body); } else if (nl::HashJoinBuffer hashJoinBuffer = mlir::dyn_cast(operation)) { @@ -1270,8 +1270,8 @@ void NLTranslator::translateFor(nl::For forLoop, NLStmtContainer* body) { translateUnwindRangeLoop(config, loopBody, limit, body); } else if (config._kind == IteratorKind::OptionalDrain) { translateOptionalDrainLoop(config, loopBody, limit, body); - } else if (config._kind == IteratorKind::UnionDrain) { - translateUnionLoop(config, loopBody, limit, body); + } else if (config._kind == IteratorKind::RowDrain) { + translateRowLoop(config, loopBody, limit, body); } else if (config._kind == IteratorKind::ProcedureInit) { translateProcedureInitLoop(config, loopBody, limit, body); } else if (config._kind == IteratorKind::CrossProduct) { @@ -4537,16 +4537,16 @@ void NLTranslator::translateSortLoop(const IteratorConfig& config, translateBlock(loopBody, loopData->getStmts()); } -void NLTranslator::translateUnionBuffer(nl::UnionBuffer buffer, NLStmtContainer* body) { - NLUnionState* state = _program->allocUnionState(); - _unionStates[buffer.getState()] = state; +void NLTranslator::translateRowBuffer(nl::RowBuffer buffer, NLStmtContainer* body) { + NLRowState* state = _program->allocRowState(); + _rowStates[buffer.getState()] = state; - NLUnionResetData* resetData = _program->allocFunctionData(state); - body->emplaceStmt(&NLExecutor::runUnionReset, resetData); + NLRowResetData* resetData = _program->allocFunctionData(state); + body->emplaceStmt(&NLExecutor::runRowReset, resetData); } -void NLTranslator::translateUnionCollect(nl::UnionCollect collect, NLStmtContainer* body) { - NLUnionState* state = unionStateFor(collect.getState()); +void NLTranslator::translateRowCollect(nl::RowCollect collect, NLStmtContainer* body) { + NLRowState* state = rowStateFor(collect.getState()); const mlir::OperandRange columns = collect.getColumns(); if (state->buffers().empty()) { @@ -4554,10 +4554,10 @@ void NLTranslator::translateUnionCollect(nl::UnionCollect collect, NLStmtContain state->addBuffer(allocColumnForChunkType(column.getType())); } } else if (state->buffers().size() != columns.size()) { - throw IRException("every nl.union_collect of an accumulator must append the same number of columns"); + throw IRException("every nl.row_collect of an accumulator must append the same number of columns"); } - NLUnionCollectData* data = _program->allocFunctionData(state); + NLRowCollectData* data = _program->allocFunctionData(state); for (size_t columnIndex = 0; columnIndex < columns.size(); columnIndex++) { const mlir::Value column = columns[columnIndex]; @@ -4574,21 +4574,21 @@ void NLTranslator::translateUnionCollect(nl::UnionCollect collect, NLStmtContain appendOwned}); } - body->emplaceStmt(&NLExecutor::runUnionCollect, data); + body->emplaceStmt(&NLExecutor::runRowCollect, data); } -void NLTranslator::translateUnionLoop(const IteratorConfig& config, - mlir::Block& loopBody, - NLLimitState* limit, - NLStmtContainer* body) { - NLUnionState* state = config._unionState; +void NLTranslator::translateRowLoop(const IteratorConfig& config, + mlir::Block& loopBody, + NLLimitState* limit, + NLStmtContainer* body) { + NLRowState* state = config._rowState; const size_t bufferCount = state->buffers().size(); if (loopBody.getNumArguments() != bufferCount) { - throw IRException("nl.union_drain loop must bind one variable per collected column"); + throw IRException("nl.row_drain loop must bind one variable per collected column"); } - NLUnionLoopData* loopData = _program->allocFunctionData(state); + NLRowLoopData* loopData = _program->allocFunctionData(state); loopData->setLimit(limit); loopData->getIndices()->reserve(_program->getChunkSize()); @@ -4604,15 +4604,15 @@ void NLTranslator::translateUnionLoop(const IteratorConfig& config, loopData->addColumn(column); } - body->emplaceStmt(&NLExecutor::runUnionLoop, loopData); + body->emplaceStmt(&NLExecutor::runRowLoop, loopData); translateBlock(loopBody, loopData->getStmts()); } -NLUnionState* NLTranslator::unionStateFor(mlir::Value handle) const { - const auto stateIt = _unionStates.find(handle); - if (stateIt == _unionStates.end()) { - throw IRException("union handle must be produced by an nl.union_buffer"); +NLRowState* NLTranslator::rowStateFor(mlir::Value handle) const { + const auto stateIt = _rowStates.find(handle); + if (stateIt == _rowStates.end()) { + throw IRException("row handle must be produced by an nl.row_buffer"); } return stateIt->second; diff --git a/query/ir/exec/NLTranslator.h b/query/ir/exec/NLTranslator.h index 77c81d386d..c99c9f6ded 100644 --- a/query/ir/exec/NLTranslator.h +++ b/query/ir/exec/NLTranslator.h @@ -89,7 +89,7 @@ class NLTranslator { UnwindRange, ProcedureInit, OptionalDrain, - UnionDrain, + RowDrain, CrossProduct, HashJoinProbe, EachRow, @@ -115,8 +115,8 @@ class NLTranslator { // The accumulator an OptionalDrain iterator drains; null for the other kinds. NLOptionalState* _optionalState {nullptr}; - // The accumulator a UnionDrain iterator drains; null for the other kinds. - NLUnionState* _unionState {nullptr}; + // The accumulator a RowDrain iterator drains; null for the other kinds. + NLRowState* _rowState {nullptr}; // The call a ProcedureInit iterator drives; null for the other kinds. NLProcedureState* _procedureState {nullptr}; @@ -292,7 +292,7 @@ class NLTranslator { // nl.optional_collect and the nl.for over nl.optional_drain find the same buffers and // matched flags llvm::DenseMap _optionalStates; - llvm::DenseMap _unionStates; + llvm::DenseMap _rowStates; llvm::DenseMap _existsStates; llvm::DenseMap _countSubqueryStates; @@ -759,22 +759,22 @@ class NLTranslator { // produced by an nl.optional_buffer translated earlier. NLOptionalState* optionalStateFor(mlir::Value handle) const; - // Translate an nl.union_buffer: allocate the runtime accumulator, map the handle to it + // Translate an nl.row_buffer: allocate the runtime accumulator, map the handle to it // and record the reset statement. The buffers are allocated by the first collect. - void translateUnionBuffer(mlir::nl::UnionBuffer buffer, NLStmtContainer* body); + void translateRowBuffer(mlir::nl::RowBuffer buffer, NLStmtContainer* body); - // Translate the nl.union_collect of one branch: the first allocates the buffers, and - // each records the per-step statement appending its columns to them - void translateUnionCollect(mlir::nl::UnionCollect collect, NLStmtContainer* body); + // Translate an nl.row_collect: the first allocates the buffers, and each records the + // per-step statement appending its columns to them + void translateRowCollect(mlir::nl::RowCollect collect, NLStmtContainer* body); - // Translate the nl.for over an nl.union_drain iterator: one loop variable per buffer, + // Translate the nl.for over an nl.row_drain iterator: one loop variable per buffer, // gathered from it chunk by chunk - void translateUnionLoop(const IteratorConfig& config, - mlir::Block& loopBody, - NLLimitState* limit, - NLStmtContainer* body); + void translateRowLoop(const IteratorConfig& config, + mlir::Block& loopBody, + NLLimitState* limit, + NLStmtContainer* body); - NLUnionState* unionStateFor(mlir::Value handle) const; + NLRowState* rowStateFor(mlir::Value handle) const; // Translate an nl.exists_buffer: allocate the runtime accumulator, map the handle to // it, record this step's input chunks and the row tag column, and record the reset diff --git a/query/ir/lowering/DBLowering.cpp b/query/ir/lowering/DBLowering.cpp index d73e6ee785..3645e50e23 100644 --- a/query/ir/lowering/DBLowering.cpp +++ b/query/ir/lowering/DBLowering.cpp @@ -4062,9 +4062,9 @@ void DBLowering::lowerUnionResults(mlir::db::Union unionOp) { mlir::Block* const root = _rootBlock; _builder.setInsertionPointToStart(root); - const mlir::Value state = _builder.create(loc).getState(); + const mlir::Value state = _builder.create(loc).getState(); - llvm::SmallVector collects; + llvm::SmallVector collects; for (mlir::Region& branch : unionOp.getBranches()) { lowerCollectedBranch(branch, root, state, collects); @@ -4088,7 +4088,7 @@ void DBLowering::lowerConditional(mlir::db::Conditional conditional) { mlir::Value state; if (returning) { _builder.setInsertionPointToStart(root); - state = _builder.create(loc).getState(); + state = _builder.create(loc).getState(); } _constantComputationBlock = root; @@ -4108,7 +4108,7 @@ void DBLowering::lowerConditional(mlir::db::Conditional conditional) { const size_t conditionCount = regions.size() / 2; const bool hasDefault = regions.size() % 2 == 1; - llvm::SmallVector collects; + llvm::SmallVector collects; for (size_t conditionIndex = 0; conditionIndex < conditionCount; conditionIndex++) { for (size_t columnIndex = 0; columnIndex < readColumns.size(); columnIndex++) { @@ -4164,7 +4164,7 @@ void DBLowering::lowerConditionalBranch(mlir::Region& branch, llvm::ArrayRef readColumns, llvm::ArrayRef taken, mlir::Value state, - llvm::SmallVectorImpl& collects) { + llvm::SmallVectorImpl& collects) { const mlir::Location loc = _builder.getUnknownLoc(); mlir::Block* const root = _rootBlock; @@ -4190,7 +4190,7 @@ void DBLowering::lowerConditionalBranch(mlir::Region& branch, void DBLowering::lowerCollectedBranch(mlir::Region& branch, mlir::Block* root, mlir::Value state, - llvm::SmallVectorImpl& collects) { + llvm::SmallVectorImpl& collects) { const mlir::Location loc = _builder.getUnknownLoc(); _innermostLoopBody = nullptr; @@ -4215,7 +4215,7 @@ void DBLowering::lowerCollectedBranch(mlir::Region& branch, // A branch of constants alone lays them out where they are bound, above the // root, which the collect must still run once per step of setInsertionInto(deepestOwnerBlock(chunks, root)); - collects.push_back(_builder.create(loc, state, chunks)); + collects.push_back(_builder.create(loc, state, chunks)); } for (const auto& [column, chunk] : replacedMappings) { @@ -4224,25 +4224,25 @@ void DBLowering::lowerCollectedBranch(mlir::Region& branch, } } -void DBLowering::drainCollectedBranches(llvm::ArrayRef collects, +void DBLowering::drainCollectedBranches(llvm::ArrayRef collects, mlir::Value state, mlir::Block* root, mlir::Operation* holder) { const mlir::Location loc = _builder.getUnknownLoc(); llvm::SmallVector branchColumns; - for (nl::UnionCollect collect : collects) { + for (nl::RowCollect collect : collects) { branchColumns.push_back(collect.getColumnsMutable()); } reconcileBranchResultTypes(branchColumns, mlir::ArrayAttr()); - nl::UnionCollect firstCollect = collects.front(); + nl::RowCollect firstCollect = collects.front(); const llvm::SmallVector chunkTypes(firstCollect.getColumns().getTypes()); const nl::IteratorType iteratorType = nl::IteratorType::get(_builder.getContext(), chunkTypes); setInsertionInto(root); - nl::UnionDrain drain = _builder.create(loc, iteratorType, state); + nl::RowDrain drain = _builder.create(loc, iteratorType, state); buildLoopForSource(drain.getResult(), holder); } @@ -4437,10 +4437,10 @@ void DBLowering::lowerMaterialize(mlir::db::Materialize materialize) { rowAlignBufferedChunks(chunks); _builder.setInsertionPointToStart(_rootBlock); - const mlir::Value state = _builder.create(loc).getState(); + const mlir::Value state = _builder.create(loc).getState(); setInsertionInto(deepestOwnerBlock(chunks, _rootBlock)); - _builder.create(loc, state, chunks); + _builder.create(loc, state, chunks); _valueMap[materialize.getRows()] = state; } @@ -4448,9 +4448,9 @@ void DBLowering::lowerMaterialize(mlir::db::Materialize materialize) { void DBLowering::lowerScanMaterialized(mlir::db::ScanMaterialized scan) { const mlir::Value state = mapValue(scan.getRows()); - nl::UnionCollect collect; + nl::RowCollect collect; for (mlir::Operation* user : state.getUsers()) { - if (nl::UnionCollect stateCollect = mlir::dyn_cast(user)) { + if (nl::RowCollect stateCollect = mlir::dyn_cast(user)) { collect = stateCollect; break; } @@ -4464,7 +4464,7 @@ void DBLowering::lowerScanMaterialized(mlir::db::ScanMaterialized scan) { const nl::IteratorType iteratorType = nl::IteratorType::get(_builder.getContext(), chunkTypes); setInsertionInto(_rootBlock); - nl::UnionDrain drain = _builder.create(_builder.getUnknownLoc(), iteratorType, state); + nl::RowDrain drain = _builder.create(_builder.getUnknownLoc(), iteratorType, state); buildLoopForSource(drain.getResult(), scan.getOperation()); } diff --git a/query/ir/lowering/DBLowering.h b/query/ir/lowering/DBLowering.h index 72479a2f3d..cd0f517bce 100644 --- a/query/ir/lowering/DBLowering.h +++ b/query/ir/lowering/DBLowering.h @@ -383,7 +383,7 @@ class DBLowering { void lowerUnion(mlir::db::Union unionOp); // Lowers a union with results - the body of a CALL - into branches that collect their - // rows into one nl.union_buffer, drained by the loop the rest of the body lowers into + // rows into one nl.row_buffer, drained by the loop the rest of the body lowers into void lowerUnionResults(mlir::db::Union unionOp); // Lowers a WHEN as db.lazy_case is lowered: each condition over the row no earlier @@ -396,16 +396,16 @@ class DBLowering { llvm::ArrayRef readColumns, llvm::ArrayRef taken, mlir::Value state, - llvm::SmallVectorImpl& collects); + llvm::SmallVectorImpl& collects); // Lowers one branch of a union or a WHEN rooted in @param root, collecting what it // yields into @param state when it yields anything void lowerCollectedBranch(mlir::Region& branch, mlir::Block* root, mlir::Value state, - llvm::SmallVectorImpl& collects); + llvm::SmallVectorImpl& collects); - void drainCollectedBranches(llvm::ArrayRef collects, + void drainCollectedBranches(llvm::ArrayRef collects, mlir::Value state, mlir::Block* root, mlir::Operation* holder); diff --git a/samples/mlir/count_subquery_union.nl.mlir b/samples/mlir/count_subquery_union.nl.mlir index 2c8ce050d3..31f942d15a 100644 --- a/samples/mlir/count_subquery_union.nl.mlir +++ b/samples/mlir/count_subquery_union.nl.mlir @@ -8,22 +8,22 @@ func.func @main() { %4 = nl.get_node_properties(%arg0, %2) : !nl.chunk> %5 = nl.each_row{%arg0, %4} : {!nl.chunk, !nl.chunk>} nl.for %arg1, %arg2 in %5 : !nl.iter, !nl.chunk>> { - %6 = nl.union_buffer + %6 = nl.row_buffer %7 = nl.distinct %state, %tag = nl.count_subquery_buffer(%arg1) : {!nl.chunk} %8 = nl.get_out_edges_by_type(%arg1, %1, {}) nl.for %arg3, %arg4, %arg5, %arg6 in %8 : !nl.iter, !nl.chunk, !nl.chunk, !nl.chunk> { %12 = nl.get_node_properties(%arg6, %2) : !nl.chunk> %14 = nl.distinct_filter %7, (%12) : !nl.chunk> - nl.union_collect %6, (%14) : !nl.chunk> + nl.row_collect %6, (%14) : !nl.chunk> } %9 = nl.get_out_edges_by_type(%arg1, %0, {}) nl.for %arg3, %arg4, %arg5, %arg6 in %9 : !nl.iter, !nl.chunk, !nl.chunk, !nl.chunk> { %12 = nl.get_node_properties(%arg6, %2) : !nl.chunk> %14 = nl.distinct_filter %7, (%12) : !nl.chunk> - nl.union_collect %6, (%14) : !nl.chunk> + nl.row_collect %6, (%14) : !nl.chunk> } - %10 = nl.union_drain(%6) : !nl.iter>> + %10 = nl.row_drain(%6) : !nl.iter>> nl.for %arg3 in %10 : !nl.iter>> { nl.count_subquery_tally %state rows(%arg3 : !nl.chunk>) } diff --git a/test/query/ir/memory/RetainedValuesAcrossChunksTest.cpp b/test/query/ir/memory/RetainedValuesAcrossChunksTest.cpp index a891712bf2..7a400c8d29 100644 --- a/test/query/ir/memory/RetainedValuesAcrossChunksTest.cpp +++ b/test/query/ir/memory/RetainedValuesAcrossChunksTest.cpp @@ -119,7 +119,7 @@ TEST_F(RetainedValuesAcrossChunksTest, unitesValuesBuiltPerRow) { "UNION ALL " "UNWIND range(1, 3) AS i RETURN ['b', toString(i)] AS l, 'b' + toString(i) AS s } " "RETURN l, s", - "nl.union_collect", + "nl.row_collect", {{"[a, 1]", "a1"}, {"[a, 2]", "a2"}, {"[a, 3]", "a3"}, From 0df6ac85fac62833e58bc2539e6d88bf81a5a23c Mon Sep 17 00:00:00 2001 From: Remy Boutonnet Date: Wed, 21 Oct 2026 15:45:03 +0200 Subject: [PATCH 3/3] Cypher: fix NEXT segments that write and CALL imports across segments --- query/AST/CypherASTDumper.cpp | 3 +- query/AST/stmt/CallSubqueryStmt.h | 5 +- query/AST/stmt/NextSegmentStmt.h | 10 + query/AST/stmt/Stmt.cpp | 6 +- query/AST/stmt/WithStmt.h | 5 + query/analyzer/CypherAnalyzer.cpp | 79 ++++++- query/analyzer/CypherAnalyzer.h | 7 +- query/ir/codegen/DBProgramGenerator.cpp | 216 ++++++++++-------- query/ir/codegen/DBProgramGenerator.h | 23 +- query/ir/lowering/DBLowering.cpp | 17 +- query/ir/lowering/DBLowering.h | 2 - query/parser/ParserUtils.cpp | 13 +- test/query/ir/next/CMakeLists.txt | 18 ++ .../ir/next/NextAfterWritingSegmentTest.cpp | 23 ++ .../ir/next/NextImportPastBranchWithTest.cpp | 22 ++ test/query/ir/next/NextInUnscopedCallTest.cpp | 19 ++ .../NextKeepsCallImportsAfterWriteTest.cpp | 20 ++ .../ir/next/NextKeepsCallImportsTest.cpp | 28 +++ test/query/ir/next/NextWhenThatWritesTest.cpp | 20 ++ test/query/ir/subquery/CMakeLists.txt | 3 + .../CallWithoutInputsAfterRowsTest.cpp | 18 ++ 21 files changed, 426 insertions(+), 131 deletions(-) create mode 100644 test/query/ir/next/NextAfterWritingSegmentTest.cpp create mode 100644 test/query/ir/next/NextImportPastBranchWithTest.cpp create mode 100644 test/query/ir/next/NextInUnscopedCallTest.cpp create mode 100644 test/query/ir/next/NextKeepsCallImportsAfterWriteTest.cpp create mode 100644 test/query/ir/next/NextKeepsCallImportsTest.cpp create mode 100644 test/query/ir/next/NextWhenThatWritesTest.cpp create mode 100644 test/query/ir/subquery/CallWithoutInputsAfterRowsTest.cpp diff --git a/query/AST/CypherASTDumper.cpp b/query/AST/CypherASTDumper.cpp index 6693449ba2..b01b2d2727 100644 --- a/query/AST/CypherASTDumper.cpp +++ b/query/AST/CypherASTDumper.cpp @@ -364,7 +364,8 @@ void CypherASTDumper::dumpQueryBody(std::ostream& out, const SinglePartQuery* qu out << " _" << std::hex << segment << " {\n"; out << " ASTType NEXT_SEGMENT\n"; out << " }\n"; - for (const CallSubqueryStmt::Branch& branch : segment->getBody()->branches()) { + const CallSubqueryStmt* body = segment->getBody(); + for (const CallSubqueryStmt::Branch& branch : body->branches()) { dump(out, branch._query); } } diff --git a/query/AST/stmt/CallSubqueryStmt.h b/query/AST/stmt/CallSubqueryStmt.h index 79c1898df7..be1da73026 100644 --- a/query/AST/stmt/CallSubqueryStmt.h +++ b/query/AST/stmt/CallSubqueryStmt.h @@ -24,14 +24,15 @@ class CallSubqueryStmt final : public Stmt { // imports. Each branch imports through a leading WITH of its own, so the imports are // held per branch. A WHEN branch holds its predicate; the ELSE branch holds none. A // `{ WHEN ... }` side of a UNION holds the CALL it is written as, as UnionQuery's does. - // The analyzer binds each import to a declaration of the body, which a later clause of - // the body may name again. struct Branch { SinglePartQuery* _query {nullptr}; bool _all {false}; Imports _imports; Expr* _condition {nullptr}; CallSubqueryStmt* _whenCall {nullptr}; + + // The declaration the analyzer binds each import to, which a later clause of the + // body may name again std::vector _importDecls; }; diff --git a/query/AST/stmt/NextSegmentStmt.h b/query/AST/stmt/NextSegmentStmt.h index 4e76b3a2fb..f4557ff8c5 100644 --- a/query/AST/stmt/NextSegmentStmt.h +++ b/query/AST/stmt/NextSegmentStmt.h @@ -1,11 +1,14 @@ #pragma once +#include + #include "Stmt.h" namespace db { class CallSubqueryStmt; class CypherAST; +class VarDecl; // A UNION or a WHEN that is one segment of a NEXT chain, or a segment ending on an // updating clause that NEXT follows. It reads the rows the segments before it returned @@ -19,8 +22,15 @@ class NextSegmentStmt final : public Stmt { CallSubqueryStmt* getBody() const { return _body; } + // What the scope clause of a CALL around a segment that returns nothing imports, + // declared again after it: the segment's table is gone, the CALL's row is not + const std::vector& getCarriedImports() const { return _carriedImports; } + + void addCarriedImport(const VarDecl* decl) { _carriedImports.push_back(decl); } + private: CallSubqueryStmt* _body {nullptr}; + std::vector _carriedImports; explicit NextSegmentStmt(CallSubqueryStmt* body); ~NextSegmentStmt() final; diff --git a/query/AST/stmt/Stmt.cpp b/query/AST/stmt/Stmt.cpp index a95434a2f5..aef4859f40 100644 --- a/query/AST/stmt/Stmt.cpp +++ b/query/AST/stmt/Stmt.cpp @@ -77,8 +77,10 @@ bool Stmt::readsTheGraph(const Stmt* stmt) { switch (kind) { case Kind::CALL_SUBQUERY: - case Kind::NEXT_SEGMENT: - return getSubqueryBody(stmt)->readsTheGraph(); + case Kind::NEXT_SEGMENT: { + const CallSubqueryStmt* subquery = getSubqueryBody(stmt); + return subquery->readsTheGraph(); + } break; case Kind::MATCH: diff --git a/query/AST/stmt/WithStmt.h b/query/AST/stmt/WithStmt.h index ab22d80e1e..1a9a35f9e4 100644 --- a/query/AST/stmt/WithStmt.h +++ b/query/AST/stmt/WithStmt.h @@ -25,13 +25,18 @@ class WithStmt final : public Stmt { // The variables the WHERE reads that the projection drops const std::vector& filterImports() const { return _filterImports; } + // Whether this is the RETURN a NEXT follows, which imports nothing into a CALL body + bool isSegmentReturn() const { return _segmentReturn; } + void setWhere(WhereClause* where) { _where = where; } void addFilterImport(const VarDecl* decl) { _filterImports.push_back(decl); } + void setSegmentReturn(bool segmentReturn) { _segmentReturn = segmentReturn; } private: Projection* _projection {nullptr}; WhereClause* _where {nullptr}; std::vector _filterImports; + bool _segmentReturn {false}; explicit WithStmt(Projection* projection); ~WithStmt() final; diff --git a/query/analyzer/CypherAnalyzer.cpp b/query/analyzer/CypherAnalyzer.cpp index 528f6273f2..ba0ef0f93a 100644 --- a/query/analyzer/CypherAnalyzer.cpp +++ b/query/analyzer/CypherAnalyzer.cpp @@ -516,15 +516,14 @@ void CypherAnalyzer::analyze(CallSubqueryStmt* subquery) { } } -// The segment reads the table the segments before it returned and its columns replace it, -// so what it returns is the whole scope after it void CypherAnalyzer::analyze(NextSegmentStmt* segment) { CallSubqueryStmt* body = segment->getBody(); importEveryVariable(body); + std::vector returnScopes; for (CallSubqueryStmt::Branch& branch : body->branches()) { - analyzeSubqueryBranch(branch, BranchImports::NEXT_SEGMENT); + returnScopes.push_back(analyzeSubqueryBranch(branch, BranchImports::NEXT_SEGMENT)); } analyzeBranchColumns(body); @@ -532,12 +531,65 @@ void CypherAnalyzer::analyze(NextSegmentStmt* segment) { DeclContext* scope = DeclContext::create(_ast, _ctxt); if (body->isReturning()) { + carrySegmentImports(body, returnScopes); publishBranchReturns(body, scope); + } else { + for (const std::string_view import : _subqueryImports) { + const VarDecl* decl = _ctxt->getDecl(import); + if (!decl) { + continue; + } + + VarDecl* carried = scope->getOrCreateNamedVariable(_ast, decl->getType(), import); + carried->setListShape(decl->getListShape()); + carried->setIsQuantifiedPath(decl->isQuantifiedPath()); + _declSources[carried] = decl; + + segment->addCarriedImport(carried); + } } setScope(scope); } +// The branches return one table, so either every one of them carries the imports of the +// CALL around the segment or none does +void CypherAnalyzer::carrySegmentImports(const CallSubqueryStmt* body, std::span returnScopes) { + const CallSubqueryStmt::Branches& branches = body->branches(); + + if (_subqueryImports.empty()) { + return; + } + + // A keyless aggregate, or a WITH of the branch that dropped an import, stops the imports + for (size_t branchIndex = 0; branchIndex < branches.size(); branchIndex++) { + const ReturnStmt* returnStmt = branches[branchIndex]._query->getReturnStmt(); + const Projection* projection = returnStmt->getProjection(); + const DeclContext* returnScope = returnScopes[branchIndex]; + + const bool keyless = projection->isAggregate() && !projection->hasGroupingKeys(); + const auto dropped = [returnScope](std::string_view import) { + return !returnScope->getDecl(import); + }; + + if (keyless || std::ranges::any_of(_subqueryImports, dropped)) { + return; + } + } + + DeclContext* const outer = _ctxt; + + for (size_t branchIndex = 0; branchIndex < branches.size(); branchIndex++) { + ReturnStmt* returnStmt = branches[branchIndex]._query->getReturnStmt(); + Projection* projection = returnStmt->getProjection(); + + setScope(returnScopes[branchIndex]); + carrySubqueryImports(projection, projection->isAggregate(), projection->hasGroupingKeys()); + } + + setScope(outer); +} + void CypherAnalyzer::analyzeBranchColumns(const CallSubqueryStmt* subquery) const { const CallSubqueryStmt::Branches& branches = subquery->branches(); const bool conditional = subquery->isConditional(); @@ -551,7 +603,7 @@ void CypherAnalyzer::analyzeBranchColumns(const CallSubqueryStmt* subquery) cons } } -void CypherAnalyzer::analyzeSubqueryBranch(CallSubqueryStmt::Branch& branch, BranchImports imports) { +DeclContext* CypherAnalyzer::analyzeSubqueryBranch(CallSubqueryStmt::Branch& branch, BranchImports imports) { // The body reads the imported variables and nothing else of the scope around it, so // its context is seeded with a declaration per import and the body resolves in that const SinglePartQuery* body = branch._query; @@ -559,6 +611,7 @@ void CypherAnalyzer::analyzeSubqueryBranch(CallSubqueryStmt::Branch& branch, Bra DeclContext* const inner = body->getDeclContext(); const bool predicateReadsImports = imports != BranchImports::LEADING_WITH; + const bool keepsEnclosingImports = imports == BranchImports::NEXT_SEGMENT; // What the scope clause names is readable everywhere in the body. A body importing // through a leading WITH carries nothing this way: that WITH is an ordinary @@ -575,7 +628,9 @@ void CypherAnalyzer::analyzeSubqueryBranch(CallSubqueryStmt::Branch& branch, Bra } } - std::swap(_subqueryImports, outerImports); + if (!keepsEnclosingImports) { + std::swap(_subqueryImports, outerImports); + } setScope(inner); @@ -600,9 +655,14 @@ void CypherAnalyzer::analyzeSubqueryBranch(CallSubqueryStmt::Branch& branch, Bra analyze(body); - std::swap(_subqueryImports, outerImports); + if (!keepsEnclosingImports) { + std::swap(_subqueryImports, outerImports); + } + DeclContext* const returnScope = _ctxt; setScope(outer); + + return returnScope; } void CypherAnalyzer::importIntoBranch(CallSubqueryStmt::Branch& branch, DeclContext* outer, DeclContext* inner) { @@ -873,6 +933,10 @@ void CypherAnalyzer::importThroughLeadingWith(CallSubqueryStmt::Branch& branch) } const WithStmt* with = static_cast(first); + if (with->isSegmentReturn()) { + return; + } + const Projection* projection = with->getProjection(); const bool plainProjection = !with->getWhere() @@ -1007,7 +1071,8 @@ void CypherAnalyzer::publishSubqueryReturn(const CallSubqueryStmt* subquery) { void CypherAnalyzer::publishBranchReturns(const CallSubqueryStmt* subquery, DeclContext* scope) { const CallSubqueryStmt::Branches& branches = subquery->branches(); - Projection* projection = branches.front()._query->getReturnStmt()->getProjection(); + ReturnStmt* returnStmt = branches.front()._query->getReturnStmt(); + Projection* projection = returnStmt->getProjection(); publishProjection(projection, scope); diff --git a/query/analyzer/CypherAnalyzer.h b/query/analyzer/CypherAnalyzer.h index babba14af8..f7639e30b2 100644 --- a/query/analyzer/CypherAnalyzer.h +++ b/query/analyzer/CypherAnalyzer.h @@ -118,7 +118,7 @@ class CypherAnalyzer { LEADING_WITH, // The table a NEXT segment reads: the WHEN predicate reads it, and a WITH of the - // body descopes what it does not project + // body descopes what it does not project, bar what a CALL around the segment imports NEXT_SEGMENT, }; @@ -173,9 +173,12 @@ class CypherAnalyzer { void importEveryVariable(CallSubqueryStmt* subquery) const; // Analyzes one query of a CALL body under its own scope, seeded with what it imports - void analyzeSubqueryBranch(CallSubqueryStmt::Branch& branch, BranchImports imports); + // @returns the scope the branch's RETURN reads + DeclContext* analyzeSubqueryBranch(CallSubqueryStmt::Branch& branch, BranchImports imports); void analyzeBranchColumns(const CallSubqueryStmt* subquery) const; + + void carrySegmentImports(const CallSubqueryStmt* body, std::span returnScopes); void importIntoBranch(CallSubqueryStmt::Branch& branch, DeclContext* outer, DeclContext* inner); void throwOnPatternPredicateVariable(const Pattern* pattern, const DeclContext* outer) const; diff --git a/query/ir/codegen/DBProgramGenerator.cpp b/query/ir/codegen/DBProgramGenerator.cpp index 5cfe739403..a55979a8a5 100644 --- a/query/ir/codegen/DBProgramGenerator.cpp +++ b/query/ir/codegen/DBProgramGenerator.cpp @@ -5723,58 +5723,32 @@ void DBProgramGenerator::generateCallSubquery(const CallSubqueryStmt* subquery) published.append(results._constants.begin(), results._constants.end()); rebindScopeKeepingWrittenEntities(published); - - for (auto& [decl, written] : results._returnedEntities) { - _part._writtenEntities[decl] = std::move(written); - } + bindReturnedEntities(results._returnedEntities); } -// A segment reads the table the segments before it returned, and what it returns replaces -// that table void DBProgramGenerator::generateNextSegment(const NextSegmentStmt* segment) { const CallSubqueryStmt* body = segment->getBody(); - if (body->isUnion()) { - generateUnionSegment(body); + if (!body->isReturning()) { + generateUpdatingSegment(segment); } else if (body->isConditional()) { generateConditionalSegment(body); } else { - generateUpdatingSegment(body); + generateUnionSegment(body); } } -// Every branch reads the whole table, so the table is held once and each branch walks it -// from the start void DBProgramGenerator::generateUnionSegment(const CallSubqueryStmt* body) { const CallSubqueryStmt::Branches& branches = body->branches(); const size_t dedupedBranches = body->getDedupedBranchCount(); - - const mlir::Location loc = _opBuilder.getUnknownLoc(); mlir::Block* const outerBlock = _opBuilder.getInsertionBlock(); - llvm::SmallVector scopeColumns; - collectPublishedColumns(scopeColumns); - - llvm::SmallVector inputs; - llvm::SmallVector constants; - splitConstantColumns(scopeColumns, inputs, constants); - - llvm::SmallVector inputColumns; - llvm::SmallVector inputTypes; - for (const PublishedColumn& input : inputs) { - inputColumns.push_back(input._column); - inputTypes.push_back(input._column.getType()); - } - - mlir::Value rows; - if (!inputColumns.empty()) { - const mlir::db::MaterializedType rowsType = mlir::db::MaterializedType::get(_mlirCtxt); - rows = _opBuilder.create(loc, rowsType, inputColumns).getRows(); - } + SegmentTable table; + materializeTable(table); mlir::Value distinctSet; if (dedupedBranches > 0) { - distinctSet = _opBuilder.create(loc).getSet(); + distinctSet = _opBuilder.create(_opBuilder.getUnknownLoc()).getSet(); } SubqueryBranchResults branchResults; @@ -5782,20 +5756,14 @@ void DBProgramGenerator::generateUnionSegment(const CallSubqueryStmt* body) { std::vector importedEntities(branches.size()); for (size_t branchIndex = 0; branchIndex < branches.size(); branchIndex++) { - mlir::Block* const block = openSubqueryBranch(branchResults); - llvm::SmallVector scanned; - if (rows) { - _opBuilder.setInsertionPointToStart(block); - mlir::db::ScanMaterialized scan = _opBuilder.create(loc, inputTypes, rows); - scanned.assign(scan.getColumns().begin(), scan.getColumns().end()); - } + scanTable(table, openSubqueryBranch(branchResults), scanned); collectSubqueryBranchScope(branches[branchIndex], - inputs, + table._rowColumns, scanned, mlir::Value(), - constants, + table._constants, /*bindsHidden=*/false, branchScopes[branchIndex], importedEntities[branchIndex]); @@ -5818,80 +5786,72 @@ void DBProgramGenerator::generateUnionSegment(const CallSubqueryStmt* body) { std::swap(_importedCSVRows, outerImportedCSVRows); rebindScope(yielded); - - for (auto& [decl, written] : returnedEntities) { - _part._writtenEntities[decl] = std::move(written); - } + bindReturnedEntities(returnedEntities); } -// A WHEN decides per row, as the CALL (*) its body is written as does, and what it returns -// replaces the row: a name the row held and the branch returns again is bound afresh +// What a branch returns replaces the row, so a name the row held and the branch returns +// again is bound afresh rather than beside the input it came from void DBProgramGenerator::generateConditionalSegment(const CallSubqueryStmt* body) { SubqueryResults results; emitCallSubquery(body, results); rebindScope(results._returned); - - for (auto& [decl, written] : results._returnedEntities) { - _part._writtenEntities[decl] = std::move(written); - } + bindReturnedEntities(results._returnedEntities); } -// A segment ending on a write returns nothing, so the segment after it starts from a single -// row, as a query does: the segment is the body of a CALL () run once, which reads the -// table it is handed as a UNION segment's branches do -void DBProgramGenerator::generateUpdatingSegment(const CallSubqueryStmt* body) { - const CallSubqueryStmt::Branch& branch = body->branches().front(); - const SinglePartQuery* query = branch._query; - +// The segment after this one starts from a single row, so this one is the body of a +// CALL () run once over that row, reading the table it is handed through a scan +void DBProgramGenerator::generateUpdatingSegment(const NextSegmentStmt* segment) { + const CallSubqueryStmt* body = segment->getBody(); const mlir::Location loc = _opBuilder.getUnknownLoc(); mlir::Block* const outerBlock = _opBuilder.getInsertionBlock(); - llvm::SmallVector scopeColumns; - collectPublishedColumns(scopeColumns); - - llvm::SmallVector inputs; - llvm::SmallVector constants; - splitConstantColumns(scopeColumns, inputs, constants); - - llvm::SmallVector inputColumns; - llvm::SmallVector inputTypes; - for (const PublishedColumn& input : inputs) { - inputColumns.push_back(input._column); - inputTypes.push_back(input._column.getType()); - } + SegmentTable table; + materializeTable(table); mlir::Region bodyRegion; mlir::Block* const bodyBlock = new mlir::Block(); // Region destructor frees it bodyRegion.push_back(bodyBlock); llvm::SmallVector scanned; - if (!inputColumns.empty()) { - const mlir::db::MaterializedType rowsType = mlir::db::MaterializedType::get(_mlirCtxt); - const mlir::Value rows = _opBuilder.create(loc, rowsType, inputColumns).getRows(); - - _opBuilder.setInsertionPointToStart(bodyBlock); - mlir::db::ScanMaterialized scan = _opBuilder.create(loc, inputTypes, rows); - scanned.assign(scan.getColumns().begin(), scan.getColumns().end()); - } - - llvm::SmallVector scope; - CarriedEntities importedEntities; - collectSubqueryBranchScope(branch, - inputs, - scanned, - mlir::Value(), - constants, - /*bindsHidden=*/false, - scope, - importedEntities); + scanTable(table, bodyBlock, scanned); std::unordered_set outerImportedCSVRows; std::swap(_importedCSVRows, outerImportedCSVRows); - _opBuilder.setInsertionPointToEnd(bodyBlock); - rebindSubqueryBranchScope(scope, importedEntities); - generateQueryParts(query); + if (body->isConditional()) { + llvm::SmallVector scope; + for (size_t columnIndex = 0; columnIndex < scanned.size(); columnIndex++) { + const PublishedColumn& rowColumn = table._rowColumns[columnIndex]; + scope.push_back({rowColumn._decl, rowColumn._name, scanned[columnIndex]}); + } + + scope.append(table._constants.begin(), table._constants.end()); + + _opBuilder.setInsertionPointToEnd(bodyBlock); + rebindScopeKeepingWrittenEntities(scope); + + SubqueryResults results; + emitCallSubquery(body, results); + } else { + const CallSubqueryStmt::Branch& branch = body->branches().front(); + + llvm::SmallVector scope; + CarriedEntities importedEntities; + collectSubqueryBranchScope(branch, + table._rowColumns, + scanned, + mlir::Value(), + table._constants, + /*bindsHidden=*/false, + scope, + importedEntities); + + _opBuilder.setInsertionPointToEnd(bodyBlock); + rebindSubqueryBranchScope(scope, importedEntities); + generateQueryParts(branch._query); + } + _opBuilder.create(loc, mlir::Value(), mlir::ValueRange {}); std::swap(_importedCSVRows, outerImportedCSVRows); @@ -5902,11 +5862,66 @@ void DBProgramGenerator::generateUpdatingSegment(const CallSubqueryStmt* body) { mlir::TypeRange {}, mlir::ValueRange {}, /*unit=*/true, - subqueryCarriesRows(query), + /*carriesScope=*/false, /*optional=*/false); callOp.getBody().takeBody(bodyRegion); - rebindScope({}); + llvm::SmallVector carried; + for (const VarDecl* decl : segment->getCarriedImports()) { + const std::string_view name = decl->getName(); + const auto columnIt = std::ranges::find(_subqueryRowColumns, name, &PublishedColumn::_name); + bioassert(columnIt != _subqueryRowColumns.end(), "Import '{}' of no CALL row", name); + + carried.push_back({decl, std::string(name), columnIt->_column}); + } + + rebindScope(carried); +} + +void DBProgramGenerator::materializeTable(SegmentTable& table) { + llvm::SmallVector scopeColumns; + collectPublishedColumns(scopeColumns); + + splitConstantColumns(scopeColumns, table._rowColumns, table._constants); + + if (table._rowColumns.empty()) { + return; + } + + llvm::SmallVector rowColumns; + for (const PublishedColumn& rowColumn : table._rowColumns) { + rowColumns.push_back(rowColumn._column); + } + + const mlir::db::MaterializedType rowsType = mlir::db::MaterializedType::get(_mlirCtxt); + table._rows = _opBuilder.create(_opBuilder.getUnknownLoc(), rowsType, rowColumns).getRows(); +} + +void DBProgramGenerator::scanTable(const SegmentTable& table, + mlir::Block* block, + llvm::SmallVectorImpl& scanned) { + if (!table._rows) { + return; + } + + llvm::SmallVector columnTypes; + for (const PublishedColumn& rowColumn : table._rowColumns) { + columnTypes.push_back(rowColumn._column.getType()); + } + + _opBuilder.setInsertionPointToStart(block); + mlir::db::ScanMaterialized scan = _opBuilder.create(_opBuilder.getUnknownLoc(), + columnTypes, + table._rows); + + const mlir::ResultRange columns = scan.getColumns(); + scanned.assign(columns.begin(), columns.end()); +} + +void DBProgramGenerator::bindReturnedEntities(CarriedEntities& returnedEntities) { + for (auto& [decl, written] : returnedEntities) { + _part._writtenEntities[decl] = std::move(written); + } } void DBProgramGenerator::splitConstantColumns(llvm::ArrayRef columns, @@ -5991,6 +6006,14 @@ void DBProgramGenerator::emitCallSubquery(const CallSubqueryStmt* subquery, Subq std::unordered_set outerImportedCSVRows; std::swap(_importedCSVRows, outerImportedCSVRows); + llvm::SmallVector rowColumns(constants.begin(), constants.end()); + for (size_t inputIndex = 0; inputIndex < inputs.size(); inputIndex++) { + const PublishedColumn& input = inputs[inputIndex]; + rowColumns.push_back({input._decl, input._name, inputArguments[inputIndex]}); + } + + std::swap(_subqueryRowColumns, rowColumns); + _opBuilder.setInsertionPointToStart(bodyBlock); llvm::SmallVector yielded; @@ -6089,6 +6112,7 @@ void DBProgramGenerator::emitCallSubquery(const CallSubqueryStmt* subquery, Subq _vdg = std::move(outerGraph); std::swap(_importedCSVRows, outerImportedCSVRows); + std::swap(_subqueryRowColumns, rowColumns); if (!returning) { return; diff --git a/query/ir/codegen/DBProgramGenerator.h b/query/ir/codegen/DBProgramGenerator.h index 24df0c0bf2..99188f7db8 100644 --- a/query/ir/codegen/DBProgramGenerator.h +++ b/query/ir/codegen/DBProgramGenerator.h @@ -321,6 +321,10 @@ class DBProgramGenerator { // handed the field's column with it std::unordered_map _csvFieldRows; + // The columns of the row the CALL body being generated runs for, as its block + // arguments and the constants it reads where they are bound + llvm::SmallVector _subqueryRowColumns; + // The LOAD CSV rows the subquery body being generated imports. A body that groups or // dedups runs one input row at a time, so such a row holds one record there and the // columns of its fields tell its rows apart as the whole row would @@ -384,13 +388,23 @@ class DBProgramGenerator { void generateNextSegment(const NextSegmentStmt* segment); void generateUnionSegment(const CallSubqueryStmt* body); void generateConditionalSegment(const CallSubqueryStmt* body); - void generateUpdatingSegment(const CallSubqueryStmt* body); + void generateUpdatingSegment(const NextSegmentStmt* segment); - // A constant holds one value standing for every row rather than rows of its own void splitConstantColumns(llvm::ArrayRef columns, llvm::SmallVectorImpl& rowColumns, llvm::SmallVectorImpl& constants); + // The table a segment reads: the columns in flight, the ones holding rows behind a + // db.materialize, which is null when none does + struct SegmentTable { + llvm::SmallVector _rowColumns; + llvm::SmallVector _constants; + mlir::Value _rows; + }; + + void materializeTable(SegmentTable& table); + void scanTable(const SegmentTable& table, mlir::Block* block, llvm::SmallVectorImpl& scanned); + // Whether the body's rows carry the columns it was given through to its RETURN: true // unless a clause of it aggregates, sorts, cuts or dedups the rows, which is when the // body has to run one input row at a time instead @@ -556,6 +570,8 @@ class DBProgramGenerator { // @param results when the body returns void emitCallSubquery(const CallSubqueryStmt* subquery, SubqueryResults& results); + void bindReturnedEntities(CarriedEntities& returnedEntities); + // The scope one branch of a CALL body opens with: the imports it names, read through // @param inputArguments or where they are bound as constants, and - for a body carrying // the scope - every input under a hidden name @@ -599,7 +615,6 @@ class DBProgramGenerator { mlir::Block* bodyBlock, std::vector>& regions); - // Opens the region of one more branch, its single block empty mlir::Block* openSubqueryBranch(SubqueryBranchResults& branchResults); // Generates one branch into @param block, opened by openSubqueryBranch, ending in a @@ -609,8 +624,6 @@ class DBProgramGenerator { mlir::Block* bodyBlock, SubqueryBranchResults& branchResults); - // Emits the db.union holding the branches built and fills @param yielded as - // generateSubqueryUnion does void generateUnionOverBranches(mlir::Value distinctSet, size_t dedupedBranches, mlir::Block* bodyBlock, diff --git a/query/ir/lowering/DBLowering.cpp b/query/ir/lowering/DBLowering.cpp index 3645e50e23..39fd9f7089 100644 --- a/query/ir/lowering/DBLowering.cpp +++ b/query/ir/lowering/DBLowering.cpp @@ -3182,7 +3182,12 @@ void DBLowering::lowerSubqueryPerRow(mlir::db::CallSubquery call, mlir::Value yieldedTag; lowerSubqueryBody(call, stepBlock, inputChunks, yieldedChunks, yieldedTag); - if (unit || !_innermostLoopBody) { + if (unit) { + _innermostLoopBody = nullptr; + _innermostCardinality = mlir::Value(); + + setInsertionInto(stepBlock); + } else if (!_innermostLoopBody) { _innermostLoopBody = previousInnermostLoopBody; _innermostCardinality = previousInnermostCardinality; @@ -4424,8 +4429,6 @@ void DBLowering::lowerDistinctSet(mlir::db::DistinctSet distinctSet) { _valueMap[distinctSet.getSet()] = _builder.create(_builder.getUnknownLoc()).getState(); } -// The buffer is emptied each step of the root and filled in the loop producing the rows, so -// a scan placed after that loop finds every row of the step void DBLowering::lowerMaterialize(mlir::db::Materialize materialize) { const mlir::Location loc = _builder.getUnknownLoc(); @@ -4460,13 +4463,7 @@ void DBLowering::lowerScanMaterialized(mlir::db::ScanMaterialized scan) { throw IRException("db.scan_materialized reads rows no db.materialize collected"); } - const llvm::SmallVector chunkTypes(collect.getColumns().getTypes()); - const nl::IteratorType iteratorType = nl::IteratorType::get(_builder.getContext(), chunkTypes); - - setInsertionInto(_rootBlock); - nl::RowDrain drain = _builder.create(_builder.getUnknownLoc(), iteratorType, state); - - buildLoopForSource(drain.getResult(), scan.getOperation()); + drainCollectedBranches({collect}, state, _rootBlock, scan.getOperation()); } void DBLowering::lowerRemoveDuplicates(mlir::db::RemoveDuplicates distinct) { diff --git a/query/ir/lowering/DBLowering.h b/query/ir/lowering/DBLowering.h index cd0f517bce..8655d9cbf3 100644 --- a/query/ir/lowering/DBLowering.h +++ b/query/ir/lowering/DBLowering.h @@ -429,8 +429,6 @@ class DBLowering { void lowerDistinctSet(mlir::db::DistinctSet distinctSet); void lowerMaterialize(mlir::db::Materialize materialize); - - // Opens a loop over the rows a db.materialize collected, after the loop collecting them void lowerScanMaterialized(mlir::db::ScanMaterialized scan); // Brings every branch's result columns to the one type each column of the result diff --git a/query/parser/ParserUtils.cpp b/query/parser/ParserUtils.cpp index 52e6878698..f7e56bd761 100644 --- a/query/parser/ParserUtils.cpp +++ b/query/parser/ParserUtils.cpp @@ -331,7 +331,9 @@ bool ParserUtils::returnsUnaliasedItem(const CallSubqueryStmt::Branches& segment continue; } - for (const Projection::ReturnItem& item : returnStmt->getProjection()->items()) { + const Projection* projection = returnStmt->getProjection(); + + for (const Projection::ReturnItem& item : projection->items()) { const Expr* expr = std::get(item); const bool named = expr->getKind() == Expr::Kind::SYMBOL || !expr->getName().empty(); @@ -360,10 +362,11 @@ SinglePartQuery* ParserUtils::createNextQuery(CypherAST* ast, const CallSubqueryStmt::Branches& segment = segments[index]; const bool endsTheChain = index + 1 == segments.size(); - SinglePartQuery* first = segment.front()._query; + const CallSubqueryStmt::Branch& firstBranch = segment.front(); + SinglePartQuery* first = firstBranch._query; ReturnStmt* returnStmt = first->getReturnStmt(); - const bool composite = segment.size() > 1 || segment.front()._condition; + const bool composite = segment.size() > 1 || firstBranch._condition; if (!composite && (returnStmt || endsTheChain)) { ast->nestQuery(first); @@ -378,6 +381,7 @@ SinglePartQuery* ParserUtils::createNextQuery(CypherAST* ast, query->setReturnStmt(returnStmt); } else { WithStmt* with = WithStmt::create(ast, returnStmt->getProjection()); + with->setSegmentReturn(true); sourceManager->setLocation(with, location); stmts->add(with); } @@ -417,7 +421,8 @@ ReturnStmt* ParserUtils::createSegmentReturn(CypherAST* ast, std::string_view name = expr->getName(); if (name.empty() && expr->getKind() == Expr::Kind::SYMBOL) { - name = static_cast(expr)->getSymbol()->getName(); + const Symbol* symbol = static_cast(expr)->getSymbol(); + name = symbol->getName(); } else if (name.empty()) { name = sourceManager->getStringRepr(std::bit_cast(expr)); } diff --git a/test/query/ir/next/CMakeLists.txt b/test/query/ir/next/CMakeLists.txt index 143d35d64f..118f2df99e 100644 --- a/test/query/ir/next/CMakeLists.txt +++ b/test/query/ir/next/CMakeLists.txt @@ -3,3 +3,21 @@ add_turing_test(test_query_ir_next_query NextQueryTest.cpp) # NEXT chains whose segments write. add_turing_test(test_query_ir_next_write NextWriteTest.cpp) + +# The segment after one that writes over a table starts from one row. +add_turing_test(test_query_ir_next_after_writing_segment NextAfterWritingSegmentTest.cpp) + +# The segment after a WHEN whose branches only write starts from one row. +add_turing_test(test_query_ir_next_when_that_writes NextWhenThatWritesTest.cpp) + +# A CALL's imports stay readable across the segments of a chain in its body. +add_turing_test(test_query_ir_next_keeps_call_imports NextKeepsCallImportsTest.cpp) + +# A chain in an unscoped CALL body opens on a RETURN, not an importing WITH. +add_turing_test(test_query_ir_next_in_unscoped_call NextInUnscopedCallTest.cpp) + +# A CALL's imports stay readable after a segment of its body that writes. +add_turing_test(test_query_ir_next_keeps_call_imports_after_write NextKeepsCallImportsAfterWriteTest.cpp) + +# A CALL's import read after a UNION segment whose branch projects through a WITH. +add_turing_test(test_query_ir_next_import_past_branch_with NextImportPastBranchWithTest.cpp) diff --git a/test/query/ir/next/NextAfterWritingSegmentTest.cpp b/test/query/ir/next/NextAfterWritingSegmentTest.cpp new file mode 100644 index 0000000000..7ad1c01930 --- /dev/null +++ b/test/query/ir/next/NextAfterWritingSegmentTest.cpp @@ -0,0 +1,23 @@ +#include + +#include "WriteQueryTest.h" + +using namespace db; +using namespace turing::test; + +// A segment ending on a write returns nothing, so the segment after it starts from a +// single row however many rows the segment wrote over +class NextAfterWritingSegmentTest : public WriteQueryTest { +}; + +TEST_F(NextAfterWritingSegmentTest, startsFromOneRowAfterWritingOverATable) { + expectWriteRows("UNWIND [1, 2, 3] AS x RETURN x NEXT CREATE (:Q {x: x}) NEXT RETURN 1 AS one", {{"1"}}); + + expectRows("MATCH (q:Q) RETURN q.x", {{"1"}, {"2"}, {"3"}}); +} + +TEST_F(NextAfterWritingSegmentTest, countsOneRowAfterSettingEveryPerson) { + expectWriteRows("MATCH (p:Person) RETURN p NEXT SET p.seen = true NEXT RETURN count(*) AS n", {{"1"}}); + + expectRows("MATCH (p:Person) WHERE p.seen RETURN count(p)", {{"8"}}); +} diff --git a/test/query/ir/next/NextImportPastBranchWithTest.cpp b/test/query/ir/next/NextImportPastBranchWithTest.cpp new file mode 100644 index 0000000000..efc36c68bb --- /dev/null +++ b/test/query/ir/next/NextImportPastBranchWithTest.cpp @@ -0,0 +1,22 @@ +#include + +#include "WriteQueryTest.h" + +using namespace db; +using namespace turing::test; + +// A CALL's import read after a UNION segment one of whose branches projects through a WITH. +// Remy and Adam have an age; the table holds the 8 Persons. +class NextImportPastBranchWithTest : public WriteQueryTest { +}; + +TEST_F(NextImportPastBranchWithTest, readsTheImportAfterABranchWithAWith) { + expectRows("MATCH (p:Person) WHERE p.name = 'Remy' " + "CALL (p) { MATCH (q:Person) RETURN q AS q " + "NEXT WITH q WHERE q.age > 1 RETURN q.name AS n UNION ALL RETURN 'x' AS n " + "NEXT RETURN p.name AS pn, n } " + "RETURN pn, n", + {{"Remy", "Remy"}, {"Remy", "Adam"}, + {"Remy", "x"}, {"Remy", "x"}, {"Remy", "x"}, {"Remy", "x"}, + {"Remy", "x"}, {"Remy", "x"}, {"Remy", "x"}, {"Remy", "x"}}); +} diff --git a/test/query/ir/next/NextInUnscopedCallTest.cpp b/test/query/ir/next/NextInUnscopedCallTest.cpp new file mode 100644 index 0000000000..9ac1a1afd9 --- /dev/null +++ b/test/query/ir/next/NextInUnscopedCallTest.cpp @@ -0,0 +1,19 @@ +#include + +#include "WriteQueryTest.h" + +using namespace db; +using namespace turing::test; + +// A chain in a CALL with no scope clause opens on a RETURN, which imports nothing +class NextInUnscopedCallTest : public WriteQueryTest { +}; + +TEST_F(NextInUnscopedCallTest, chainsFromAReturnOfItsOwn) { + expectRows("CALL { RETURN 1 AS z NEXT RETURN z + 1 AS w } RETURN w", {{"2"}}); +} + +TEST_F(NextInUnscopedCallTest, readsNoVariableOfTheOuterQuery) { + expectError("UNWIND [1, 2] AS y CALL { RETURN y NEXT RETURN y + 1 AS w } RETURN y, w", + "Variable 'y' not found"); +} diff --git a/test/query/ir/next/NextKeepsCallImportsAfterWriteTest.cpp b/test/query/ir/next/NextKeepsCallImportsAfterWriteTest.cpp new file mode 100644 index 0000000000..271097d005 --- /dev/null +++ b/test/query/ir/next/NextKeepsCallImportsAfterWriteTest.cpp @@ -0,0 +1,20 @@ +#include + +#include "WriteQueryTest.h" + +using namespace db; +using namespace turing::test; + +// What the scope clause of a CALL names stays readable after a segment of its body that +// writes and returns nothing +class NextKeepsCallImportsAfterWriteTest : public WriteQueryTest { +}; + +TEST_F(NextKeepsCallImportsAfterWriteTest, readsTheImportAfterASegmentThatWrites) { + expectWriteRows("MATCH (p:Person) WHERE p.name = 'Remy' " + "CALL (p) { UNWIND [1, 2] AS x RETURN x NEXT CREATE (:T {x: x}) NEXT RETURN p.name AS name } " + "RETURN name", + {{"Remy"}}); + + expectRows("MATCH (t:T) RETURN t.x", {{"1"}, {"2"}}); +} diff --git a/test/query/ir/next/NextKeepsCallImportsTest.cpp b/test/query/ir/next/NextKeepsCallImportsTest.cpp new file mode 100644 index 0000000000..9afd388b92 --- /dev/null +++ b/test/query/ir/next/NextKeepsCallImportsTest.cpp @@ -0,0 +1,28 @@ +#include + +#include "WriteQueryTest.h" + +using namespace db; +using namespace turing::test; + +// What the scope clause of a CALL names stays readable in every segment of a chain in +// its body +class NextKeepsCallImportsTest : public WriteQueryTest { +}; + +TEST_F(NextKeepsCallImportsTest, readsTheImportAfterAUnionSegment) { + expectRows("MATCH (p:Person) WHERE p.name = 'Remy' " + "CALL (p) { UNWIND [1, 2] AS x RETURN x UNION ALL UNWIND [3] AS x RETURN x " + "NEXT RETURN p.name AS name, sum(x) AS s } " + "RETURN name, s", + {{"Remy", "6"}}); +} + +TEST_F(NextKeepsCallImportsTest, readsTheImportAfterAWhenSegment) { + expectRows("MATCH (p:Person) WHERE p.name = 'Remy' " + "CALL (p) { UNWIND [1, 2] AS x RETURN x " + "NEXT WHEN x > 1 THEN RETURN x AS y ELSE RETURN 0 AS y " + "NEXT RETURN p.name AS name, sum(y) AS s } " + "RETURN name, s", + {{"Remy", "2"}}); +} diff --git a/test/query/ir/next/NextWhenThatWritesTest.cpp b/test/query/ir/next/NextWhenThatWritesTest.cpp new file mode 100644 index 0000000000..b852aed336 --- /dev/null +++ b/test/query/ir/next/NextWhenThatWritesTest.cpp @@ -0,0 +1,20 @@ +#include + +#include "WriteQueryTest.h" + +using namespace db; +using namespace turing::test; + +// A WHEN segment whose branches only write returns nothing, so the segment after it starts +// from a single row. Remy, Adam, Luc and Martina have a PhD. +class NextWhenThatWritesTest : public WriteQueryTest { +}; + +TEST_F(NextWhenThatWritesTest, startsFromOneRowAfterAWhenThatWrites) { + expectWriteRows("MATCH (p:Person) RETURN p " + "NEXT WHEN p.hasPhD THEN { SET p.t = 1 } ELSE { SET p.t = 2 } " + "NEXT RETURN 1 AS one", + {{"1"}}); + + expectRows("MATCH (p:Person) RETURN p.t, count(*)", {{"1", "4"}, {"2", "4"}}); +} diff --git a/test/query/ir/subquery/CMakeLists.txt b/test/query/ir/subquery/CMakeLists.txt index 9f10b502ae..00e502e296 100644 --- a/test/query/ir/subquery/CMakeLists.txt +++ b/test/query/ir/subquery/CMakeLists.txt @@ -210,3 +210,6 @@ add_turing_test(test_query_ir_subquery_written_entity_matrix SubqueryWrittenEnti # An EXISTS or COUNT subquery over a node the same query created. add_turing_test(test_query_ir_exists_over_created_node ExistsOverCreatedNodeTest.cpp) + +# A CALL () over rows holding only a projected constant leaves those rows in flight. +add_turing_test(test_query_ir_call_without_inputs_after_rows CallWithoutInputsAfterRowsTest.cpp) diff --git a/test/query/ir/subquery/CallWithoutInputsAfterRowsTest.cpp b/test/query/ir/subquery/CallWithoutInputsAfterRowsTest.cpp new file mode 100644 index 0000000000..744c391546 --- /dev/null +++ b/test/query/ir/subquery/CallWithoutInputsAfterRowsTest.cpp @@ -0,0 +1,18 @@ +#include + +#include "WriteQueryTest.h" + +using namespace db; +using namespace turing::test; + +// A CALL () run one row at a time over rows in flight that hold only a projected constant +// leaves those rows in flight +class CallWithoutInputsAfterRowsTest : public WriteQueryTest { +}; + +TEST_F(CallWithoutInputsAfterRowsTest, keepsTheRowsOfAProjectedConstant) { + expectWriteRows("UNWIND [1, 2, 3] AS x WITH 1 AS one " + "CALL () { WITH count(*) AS c CREATE (:T {c: c}) } " + "RETURN one", + {{"1"}, {"1"}, {"1"}}); +}