From dec2127a79a17c33466f45ef3e3cc27c4429e115 Mon Sep 17 00:00:00 2001 From: Arun Sharma Date: Sat, 5 Sep 2026 13:42:34 -0700 Subject: [PATCH 1/2] Support directional chain hops in anti-edge count rewrite CountAntiEdgeChain previously required the two triangle chain hops to be BOTH-direction extends, so LSQB q9 (as written in the benchmark, with directed knows edges) never fired the rewrite and fell back to a ~2.2s hash join plan. - Generalize the T - S - A count arithmetic to any FWD/BWD/BOTH combination of chain hop directions and anti-edge match direction. Chain enumeration is row-level (a BOTH hop enumerates FWD rows then BWD rows, no dedup), the anti-edge mark is boolean per key pair so row partners are deduped when subtracting A. - Record the hop/anti-edge directions in LogicalCountAntiEdgeChain and plumb them through the plan mapper into the physical operator. - Fix the mid-node resolution in the optimizer: scanNode->getNodeID() is the internal-ID property (uniqueName has the ._ID suffix), so the mid NodeExpression is now recovered from the chain extends and names are compared as plain variable names (same fix in the suffix scan check). - Fix build errors (const Transaction*, containsOp declaration) and warnings (dangling reference, unused variable). - Remove the LBUG_TRACE_ANTI trace scaffolding. Validation on the LSQB sf-like database: q9.cypher now fires the operator and returns the same count as the join plan (268837983); execution drops from ~2219ms to ~23ms (~95x). Six direction combos were cross-checked against the join-plan ground truth. --- .../optimizer/count_rel_table_optimizer.h | 7 + .../planner/operator/logical_operator.h | 1 + .../scan/logical_count_anti_edge_chain.h | 116 +++++ .../processor/operator/physical_operator.h | 1 + .../operator/scan/count_anti_edge_chain.h | 114 +++++ src/include/processor/plan_mapper.h | 2 + src/optimizer/count_rel_table_optimizer.cpp | 435 ++++++++++++++++++ src/planner/operator/scan/CMakeLists.txt | 1 + .../scan/logical_count_anti_edge_chain.cpp | 20 + src/processor/map/CMakeLists.txt | 1 + .../map/map_count_anti_edge_chain.cpp | 52 +++ src/processor/map/plan_mapper.cpp | 3 + src/processor/operator/physical_operator.cpp | 2 + src/processor/operator/scan/CMakeLists.txt | 1 + .../operator/scan/count_anti_edge_chain.cpp | 349 ++++++++++++++ 15 files changed, 1105 insertions(+) create mode 100644 src/include/planner/operator/scan/logical_count_anti_edge_chain.h create mode 100644 src/include/processor/operator/scan/count_anti_edge_chain.h create mode 100644 src/planner/operator/scan/logical_count_anti_edge_chain.cpp create mode 100644 src/processor/map/map_count_anti_edge_chain.cpp create mode 100644 src/processor/operator/scan/count_anti_edge_chain.cpp diff --git a/src/include/optimizer/count_rel_table_optimizer.h b/src/include/optimizer/count_rel_table_optimizer.h index 916ac63bb..00c55c292 100644 --- a/src/include/optimizer/count_rel_table_optimizer.h +++ b/src/include/optimizer/count_rel_table_optimizer.h @@ -49,6 +49,13 @@ class CountRelTableOptimizer : public LogicalOperatorVisitor { std::shared_ptr tryRewriteExtendChainCount( std::shared_ptr op); + // Rewrite COUNT(*) over the LSQB-q9 shape: a path n0-R-n1-R-n2-... whose first two hops + // are BOTH-direction R extends with an anti-edge filter NOT(EXISTS{n0-R-n2}) (+ optional + // id(n0)<>id(n2)), plus a filter-free suffix chain from n2. Replaced with count arithmetic + // T - A - S (see CountAntiEdgeChain). + std::shared_ptr tryRewriteAntiEdgeChainCount( + std::shared_ptr op); + // Check if the aggregate is a simple (non-distinct) COUNT with no keys. bool isSimpleCount(planner::LogicalOperator* op) const; diff --git a/src/include/planner/operator/logical_operator.h b/src/include/planner/operator/logical_operator.h index 4cd2cfa38..34cba1fe8 100644 --- a/src/include/planner/operator/logical_operator.h +++ b/src/include/planner/operator/logical_operator.h @@ -23,6 +23,7 @@ enum class LogicalOperatorType : uint8_t { ATTACH_DATABASE, COPY_FROM, COPY_TO, + COUNT_ANTI_EDGE_CHAIN, COUNT_EXTEND_CHAIN, COUNT_REL_TABLE, CREATE_GRAPH, diff --git a/src/include/planner/operator/scan/logical_count_anti_edge_chain.h b/src/include/planner/operator/scan/logical_count_anti_edge_chain.h new file mode 100644 index 000000000..92563fefb --- /dev/null +++ b/src/include/planner/operator/scan/logical_count_anti_edge_chain.h @@ -0,0 +1,116 @@ +#pragma once + +#include +#include + +#include "binder/expression/expression.h" +#include "catalog/catalog_entry/rel_group_catalog_entry.h" +#include "common/enums/extend_direction.h" +#include "common/enums/rel_direction.h" +#include "common/types/types.h" +#include "planner/operator/logical_operator.h" +#include "planner/operator/scan/logical_count_extend_chain.h" + +namespace lbug { +namespace planner { + +struct LogicalCountAntiEdgeChainPrintInfo final : OPPrintInfo { + std::string relTableName; + uint64_t numSuffixHops; + + LogicalCountAntiEdgeChainPrintInfo(std::string relTableName, uint64_t numSuffixHops) + : relTableName{std::move(relTableName)}, numSuffixHops{numSuffixHops} {} + + std::string toString() const override { + return "Anti-edge rel: " + relTableName + ", suffix hops: " + std::to_string(numSuffixHops); + } + + std::unique_ptr copy() const override { + return std::make_unique(relTableName, numSuffixHops); + } +}; + +/** + * LogicalCountAntiEdgeChain computes COUNT(*) over a path whose first two hops h1, h2 (same rel + * table R, same node table) are additionally filtered by an anti-edge predicate: + * NOT(EXISTS{MATCH (n0)-[:R]-(n2)}) and optionally id(n0) <> id(n2). Detected by + * CountRelTableOptimizer::tryRewriteAntiEdgeChainCount from plans shaped like LSQB q9: + * + * HASH_JOIN[INNER key=n2] + * FILTER[NOT(EXISTS{...})] + * HASH_JOIN[MARK keys={n0,n1...}] (anti-edge rows as one side) + * chain of R-extends from scan(n1) + anti-edge R-extend over scan(n0) + * suffix chain of extends from scan(n2) + * + * The two chain hops enumerate rows in directions chainN0Dir/chainN2Dir (FWD, BWD or BOTH) and + * the anti-edge match enumerates rows in antiEdgeDir. The count is computed with pure count + * arithmetic (see CountAntiEdgeChain): + * + * count = T - A - S + * T = sum over chain rows (n1,n2) of degChainN0(n1)*D(n2) (all tuples) + * A = sum over anti-edge rows (n0,n2) of N'(n0,n2)*D(n2) (anti-edge present) + * S = sum over mid nodes n1, nodes p of multN0(p)*multN2(p)*D(p) (n0=n2 tuples; only + * when the id<> filter + * is present) + * where D = per-node suffix path counts, degChainN0(n1) = number of n0-hop rows at n1, + * multN0/multN2(p) = number of n0/n2-hop rows between n1 and p, and + * N'(n0,n2) = sum over mid nodes x of multRevN0(n0,x)*multRevN2(n2,x) = number of chain + * enumerations connecting n0 to n2. + */ +class LogicalCountAntiEdgeChain final : public LogicalOperator { + static constexpr LogicalOperatorType type_ = LogicalOperatorType::COUNT_ANTI_EDGE_CHAIN; + +public: + LogicalCountAntiEdgeChain(std::vector suffixHops, + catalog::RelGroupCatalogEntry* antiRelEntry, + std::vector antiRelTableIDs, common::table_id_t midNodeTableID, + common::ExtendDirection chainN0Dir, common::ExtendDirection chainN2Dir, + common::ExtendDirection antiEdgeDir, bool hasNotEquals, + std::shared_ptr countExpr) + : LogicalOperator{type_}, suffixHops{std::move(suffixHops)}, antiRelEntry{antiRelEntry}, + antiRelTableIDs{std::move(antiRelTableIDs)}, midNodeTableID{midNodeTableID}, + chainN0Dir{chainN0Dir}, chainN2Dir{chainN2Dir}, antiEdgeDir{antiEdgeDir}, + hasNotEquals{hasNotEquals}, countExpr{std::move(countExpr)} { + cardinality = 1; + } + + void computeFactorizedSchema() override; + void computeFlatSchema() override; + + std::string getExpressionsForPrinting() const override { return countExpr->toString(); } + + const std::vector& getSuffixHops() const { return suffixHops; } + catalog::RelGroupCatalogEntry* getAntiRelEntry() const { return antiRelEntry; } + const std::vector& getAntiRelTableIDs() const { return antiRelTableIDs; } + common::table_id_t getMidNodeTableID() const { return midNodeTableID; } + common::ExtendDirection getChainN0Dir() const { return chainN0Dir; } + common::ExtendDirection getChainN2Dir() const { return chainN2Dir; } + common::ExtendDirection getAntiEdgeDir() const { return antiEdgeDir; } + bool getHasNotEquals() const { return hasNotEquals; } + std::shared_ptr getCountExpr() const { return countExpr; } + + std::unique_ptr getPrintInfo() const override { + return std::make_unique(antiRelEntry->getName(), + suffixHops.size()); + } + + std::unique_ptr copy() override { + return std::make_unique(suffixHops, antiRelEntry, + antiRelTableIDs, midNodeTableID, chainN0Dir, chainN2Dir, antiEdgeDir, hasNotEquals, + countExpr); + } + +private: + std::vector suffixHops; + catalog::RelGroupCatalogEntry* antiRelEntry; + std::vector antiRelTableIDs; + common::table_id_t midNodeTableID; + common::ExtendDirection chainN0Dir; + common::ExtendDirection chainN2Dir; + common::ExtendDirection antiEdgeDir; + bool hasNotEquals; + std::shared_ptr countExpr; +}; + +} // namespace planner +} // namespace lbug diff --git a/src/include/processor/operator/physical_operator.h b/src/include/processor/operator/physical_operator.h index 62ef95aa3..6d152d1c5 100644 --- a/src/include/processor/operator/physical_operator.h +++ b/src/include/processor/operator/physical_operator.h @@ -48,6 +48,7 @@ enum class PhysicalOperatorType : uint8_t { ATTACH_DATABASE, BATCH_INSERT, COPY_TO, + COUNT_ANTI_EDGE_CHAIN, COUNT_EXTEND_CHAIN, COUNT_REL_TABLE, CREATE_GRAPH, diff --git a/src/include/processor/operator/scan/count_anti_edge_chain.h b/src/include/processor/operator/scan/count_anti_edge_chain.h new file mode 100644 index 000000000..bbf230cc9 --- /dev/null +++ b/src/include/processor/operator/scan/count_anti_edge_chain.h @@ -0,0 +1,114 @@ +#pragma once + +#include + +#include "common/enums/extend_direction.h" +#include "common/enums/rel_direction.h" +#include "common/system_config.h" +#include "common/types/types.h" +#include "processor/data_pos.h" +#include "processor/operator/physical_operator.h" +#include "storage/table/node_table.h" +#include "storage/table/rel_table.h" + +namespace lbug { +namespace processor { + +struct CountAntiEdgeChainPrintInfo final : OPPrintInfo { + std::string relTableName; + uint64_t numSuffixHops; + + CountAntiEdgeChainPrintInfo(std::string relTableName, uint64_t numSuffixHops) + : relTableName{std::move(relTableName)}, numSuffixHops{numSuffixHops} {} + + std::string toString() const override { + return "Anti-edge rel: " + relTableName + ", suffix hops: " + std::to_string(numSuffixHops); + } + + std::unique_ptr copy() const override { + return std::make_unique(relTableName, numSuffixHops); + } +}; + +/** + * CountAntiEdgeChain computes COUNT(*) over LSQB-q9-shaped queries with pure count arithmetic. + * See LogicalCountAntiEdgeChain for the pattern and the T - A - S formula. No hash joins are + * performed and no tuples are materialized: the operator scans the anti-edge rel table to build + * directional adjacency lists, scans the suffix rel tables backward to propagate per-node + * suffix path counts D, then evaluates T, A and S over flat arrays. + * + * The two chain hops enumerate rows in chainN0Dir/chainN2Dir and the anti-edge match in + * antiEdgeDir (each FWD, BWD or BOTH); the arithmetic models exactly that enumeration. + * + * Single-threaded source emitting one row with the final count (int64). + */ +class CountAntiEdgeChain final : public PhysicalOperator { + static constexpr PhysicalOperatorType type_ = PhysicalOperatorType::COUNT_ANTI_EDGE_CHAIN; + +public: + struct Hop { + std::vector relTables; + std::vector scanDirections; + std::vector fromNodeTables; + std::vector toNodeTables; + }; + + CountAntiEdgeChain(std::vector suffixHops, storage::RelTable* antiRelTable, + storage::NodeTable* midNodeTable, common::ExtendDirection chainN0Dir, + common::ExtendDirection chainN2Dir, common::ExtendDirection antiEdgeDir, bool hasNotEquals, + DataPos countOutputPos, physical_op_id id, std::unique_ptr printInfo) + : PhysicalOperator{type_, id, std::move(printInfo)}, suffixHops{std::move(suffixHops)}, + antiRelTable{antiRelTable}, midNodeTable{midNodeTable}, chainN0Dir{chainN0Dir}, + chainN2Dir{chainN2Dir}, antiEdgeDir{antiEdgeDir}, hasNotEquals{hasNotEquals}, + countOutputPos{countOutputPos} {} + + bool isSource() const override { return true; } + bool isParallel() const override { return false; } + + void initLocalStateInternal(ResultSet* resultSet, ExecutionContext* context) override; + + bool getNextTuplesInternal(ExecutionContext* context) override; + + std::unique_ptr copy() override { + return std::make_unique(suffixHops, antiRelTable, midNodeTable, + chainN0Dir, chainN2Dir, antiEdgeDir, hasNotEquals, countOutputPos, id, + printInfo->copy()); + } + +private: + common::offset_t getOffsetUpperBound(storage::NodeTable* nodeTable) const { + const auto numGroups = nodeTable->getNumNodeGroups(); + if (numGroups == 0) { + return 0; + } + return (numGroups - 1) * common::StorageConfig::NODE_GROUP_SIZE + + nodeTable->getNumTuplesInNodeGroup(numGroups - 1); + } + + // Scan one rel table in the given direction with sequential bound-node batches, invoking + // the per-batch callback with (scanState, nodeIDVector, nbrVector). + template + void scanRelRows(storage::RelTable* relTable, common::RelDataDirection direction, + storage::NodeTable* boundNodeTable, transaction::Transaction* transaction, + storage::MemoryManager* memoryManager, Func&& callback); + + // Backward suffix propagation: returns B[0], the per-node suffix path counts over the mid + // node table. + std::vector computeSuffixCounts(transaction::Transaction* transaction, + storage::MemoryManager* memoryManager); + +private: + std::vector suffixHops; + storage::RelTable* antiRelTable; + storage::NodeTable* midNodeTable; + common::ExtendDirection chainN0Dir; + common::ExtendDirection chainN2Dir; + common::ExtendDirection antiEdgeDir; + bool hasNotEquals; + DataPos countOutputPos; + common::ValueVector* countVector = nullptr; + bool hasExecuted = false; +}; + +} // namespace processor +} // namespace lbug diff --git a/src/include/processor/plan_mapper.h b/src/include/processor/plan_mapper.h index a65c59f39..7fd0ba3e7 100644 --- a/src/include/processor/plan_mapper.h +++ b/src/include/processor/plan_mapper.h @@ -93,6 +93,8 @@ class PlanMapper { std::unique_ptr mapCopyTo(const planner::LogicalOperator* logicalOperator); std::unique_ptr mapCountRelTable( const planner::LogicalOperator* logicalOperator); + std::unique_ptr mapCountAntiEdgeChain( + const planner::LogicalOperator* logicalOperator); std::unique_ptr mapCountExtendChain( const planner::LogicalOperator* logicalOperator); std::unique_ptr mapCreateMacro( diff --git a/src/optimizer/count_rel_table_optimizer.cpp b/src/optimizer/count_rel_table_optimizer.cpp index 588d70e64..a9f7782b4 100644 --- a/src/optimizer/count_rel_table_optimizer.cpp +++ b/src/optimizer/count_rel_table_optimizer.cpp @@ -10,6 +10,8 @@ #include "binder/expression/rel_expression.h" #include "catalog/catalog_entry/node_table_catalog_entry.h" #include "catalog/catalog_entry/node_table_id_pair.h" +#include "common/enums/extend_direction_util.h" +#include "common/enums/join_type.h" #include "common/enums/path_semantic.h" #include "common/enums/storage_format.h" #include "function/aggregate/count.h" @@ -26,6 +28,7 @@ #include "planner/operator/logical_order_by.h" #include "planner/operator/logical_path_property_probe.h" #include "planner/operator/logical_projection.h" +#include "planner/operator/scan/logical_count_anti_edge_chain.h" #include "planner/operator/scan/logical_count_extend_chain.h" #include "planner/operator/scan/logical_count_rel_table.h" #include "planner/operator/scan/logical_reachable_count.h" @@ -58,6 +61,423 @@ std::shared_ptr CountRelTableOptimizer::visitOperator( return result; } +static LogicalOperator* skipProjections(LogicalOperator* op); +static bool containsOp(const LogicalOperator* root, const LogicalOperator* target); + +std::shared_ptr CountRelTableOptimizer::tryRewriteAntiEdgeChainCount( + std::shared_ptr op) { + // Target pattern (LSQB q9 shape): COUNT(*) over an inner hash join of + // probe: FILTER[NOT(EXISTS{MATCH (a)-[:R]-(b)})] over + // HASH_JOIN[MARK keys={a,b}] of + // probe: FILTER[id(a)<>id(b)] over chain(a--R--n1--R--b) from scan(n1) + // build: anti-edge R-extend over scan(a) + // build: filter-free chain of extends from scan(n2=b) + // i.e. a path n0-R-n1-R-n2-...-nN whose first two hops are R extends from the scan of the + // middle node n1 (any FWD/BWD/BOTH direction combination), plus an anti-edge between n0 + // and n2 (same rel R), plus a filter-free suffix chain from n2. The count is computed with + // count arithmetic T - A - S (see CountAntiEdgeChain), parameterized by the enumeration + // directions of the two chain hops and of the anti-edge match. + if (op->getOperatorType() != LogicalOperatorType::AGGREGATE) { + return op; + } + auto aggregate = op->ptrCast(); + if (aggregate->hasKeys() || aggregate->getDependentKeys().size() != 0 || + aggregate->getAggregates().size() != 1) { + return op; + } + auto aggExpr = aggregate->getAggregates()[0]; + if (aggExpr->expressionType != ExpressionType::AGGREGATE_FUNCTION) { + return op; + } + auto aggFuncExpr = aggExpr->ptrCast(); + if (aggFuncExpr->getFunction().name != function::CountStarFunction::name || + aggFuncExpr->isDistinct() || aggFuncExpr->getNumChildren() != 0) { + return op; + } + auto transaction = transaction::Transaction::Get(*_context); + if (transaction != nullptr && transaction->isWriteTransaction()) { + return op; + } + + // Collect the subtree. Allowed ops: projections, extends, node-ID hash joins, + // unrestricted scans, and filters (classified below). + std::vector extends; + std::vector joins; + std::vector filters; + std::function collect = [&](LogicalOperator* current) -> bool { + switch (current->getOperatorType()) { + case LogicalOperatorType::PROJECTION: + return collect(current->getChild(0).get()); + case LogicalOperatorType::SCAN_NODE_TABLE: { + auto scan = current->ptrCast(); + if (scan->getScanType() == LogicalScanNodeTableType::PRIMARY_KEY_SCAN) { + return false; + } + for (auto& predicateSet : scan->getPropertyPredicates()) { + if (!predicateSet.isEmpty()) { + return false; + } + } + return true; + } + case LogicalOperatorType::EXTEND: + case LogicalOperatorType::PACKED_EXTEND: + extends.push_back(current->ptrCast()); + return collect(current->getChild(0).get()); + case LogicalOperatorType::HASH_JOIN: + joins.push_back(current->ptrCast()); + return collect(current->getChild(0).get()) && collect(current->getChild(1).get()); + case LogicalOperatorType::FILTER: + filters.push_back(current->ptrCast()); + return collect(current->getChild(0).get()); + default: + return false; + } + }; + if (!collect(op->getChild(0).get())) { + return op; + } + + // Exactly one MARK join and exactly one other (top) join, and >= 4 extends + // (2 triangle hops + anti-edge + >= 1 suffix hop). + LogicalHashJoin* markJoin = nullptr; + LogicalHashJoin* topJoin = nullptr; + for (auto* join : joins) { + if (join->getJoinType() == JoinType::MARK) { + if (markJoin != nullptr) { + return op; + } + markJoin = join; + } else { + if (topJoin != nullptr) { + return op; + } + topJoin = join; + } + } + if (markJoin == nullptr || topJoin == nullptr || extends.size() < 4) { + return op; + } + if (topJoin->getJoinType() != JoinType::INNER || topJoin->getJoinNodeIDs().size() != 1) { + return op; + } + const auto topKey = topJoin->getJoinNodeIDs()[0]; + + // The anti-filter must be NOT(X) where X is the mark expression of the MARK join. + if (!markJoin->hasMark()) { + return op; + } + const auto& markExpr = markJoin->getMark(); + LogicalFilter* antiFilter = nullptr; + for (auto* filter : filters) { + if (filter->getPredicate()->expressionType == ExpressionType::NOT && + *filter->getPredicate()->getChild(0) == *markExpr) { + if (antiFilter != nullptr) { + return op; + } + antiFilter = filter; + } + } + if (antiFilter == nullptr) { + return op; + } + + // All other filters must be NOT_EQUALS between internal-ID properties. + for (auto* filter : filters) { + if (filter == antiFilter) { + continue; + } + const auto& predicate = filter->getPredicate(); + if (predicate->expressionType != ExpressionType::NOT_EQUALS || + predicate->getNumChildren() != 2 || + predicate->getChild(0)->expressionType != ExpressionType::PROPERTY || + predicate->getChild(1)->expressionType != ExpressionType::PROPERTY || + !predicate->getChild(0)->ptrCast()->isInternalID() || + !predicate->getChild(1)->ptrCast()->isInternalID()) { + return op; + } + } + + // Anti-edge endpoints come from the MARK join keys; the rel table comes from the single + // anti-edge extend, which must be one of the MARK join's children. + std::shared_ptr antiNodeA; + std::shared_ptr antiNodeB; + catalog::RelGroupCatalogEntry* antiRelEntry = nullptr; + LogicalExtend* antiEdgeExtend = nullptr; + common::ExtendDirection antiEdgeDir = common::ExtendDirection::BOTH; + LogicalOperator* markProbeChild = nullptr; + { + const auto markKeys = markJoin->getJoinNodeIDs(); + if (markKeys.size() != 2) { + return op; + } + std::vector keyNames; + for (auto& key : markKeys) { + if (key->expressionType != ExpressionType::PROPERTY || + !key->ptrCast()->isInternalID()) { + return op; + } + keyNames.push_back(key->ptrCast()->getVariableName()); + } + if (keyNames[0] == keyNames[1]) { + return op; + } + for (auto ci = 0u; ci < 2; ++ci) { + auto* child = skipProjections(markJoin->getChild(ci).get()); + if (child->getOperatorType() != LogicalOperatorType::EXTEND && + child->getOperatorType() != LogicalOperatorType::PACKED_EXTEND) { + continue; + } + auto ext = child->ptrCast(); + auto rel = ext->getRel(); + if (rel->getNumEntries() != 1) { + continue; + } + auto* relGroupEntry = rel->getEntry(0)->ptrCast(); + if (relGroupEntry->getScanFunction().has_value()) { + continue; + } + auto* extendChild = skipProjections((ext->getChild(0).get())); + if (extendChild->getOperatorType() != LogicalOperatorType::SCAN_NODE_TABLE) { + continue; + } + const auto boundName = ext->getBoundNode()->getUniqueName(); + const auto nbrName = ext->getNbrNode()->getUniqueName(); + const bool match = (boundName == keyNames[0] && nbrName == keyNames[1]) || + (boundName == keyNames[1] && nbrName == keyNames[0]); + if (!match) { + continue; + } + antiRelEntry = relGroupEntry; + antiEdgeExtend = ext; + antiEdgeDir = ext->getDirection(); + markProbeChild = markJoin->getChild(1 - ci).get(); + // Resolve node expressions: the extend endpoints are the nodes named by the keys. + auto nodeByName = [&](const std::string& name) -> std::shared_ptr { + return boundName == name ? ext->getBoundNode() : ext->getNbrNode(); + }; + antiNodeA = nodeByName(keyNames[0]); + antiNodeB = nodeByName(keyNames[1]); + break; + } + if (antiEdgeExtend == nullptr) { + return op; + } + } + + // Parse the MARK join's other child (the prefix chain): exactly 2 extends of the anti rel + // group, both sharing the scan root node, with other endpoints {antiNodeA, antiNodeB}; + // only id(a)<>id(b) NOT_EQUALS filters allowed. The scan root is the middle node n1. The + // hop directions (FWD/BWD/BOTH) are recorded per endpoint for the operator arithmetic. + std::shared_ptr midNode; + common::ExtendDirection chainN0Dir = common::ExtendDirection::BOTH; + common::ExtendDirection chainN2Dir = common::ExtendDirection::BOTH; + { + auto* current = skipProjections(markProbeChild); + std::vector chainExtends; + LogicalScanNodeTable* scanNode = nullptr; + while (true) { + if (current->getOperatorType() == LogicalOperatorType::FILTER) { + const auto predicate = current->ptrCast()->getPredicate(); + if (predicate->expressionType != ExpressionType::NOT_EQUALS || + predicate->getNumChildren() != 2) { + return op; + } + const auto& c0 = predicate->getChild(0); + const auto& c1 = predicate->getChild(1); + const auto isIdProp = [&](const std::shared_ptr& e) { + return e->expressionType == ExpressionType::PROPERTY && + e->ptrCast()->isInternalID(); + }; + if (!isIdProp(c0) || !isIdProp(c1)) { + return op; + } + const auto n0 = antiNodeA->getUniqueName(); + const auto n2 = antiNodeB->getUniqueName(); + const auto v0 = c0->ptrCast()->getVariableName(); + const auto v1 = c1->ptrCast()->getVariableName(); + if (!((v0 == n0 && v1 == n2) || (v0 == n2 && v1 == n0))) { + return op; + } + current = skipProjections(current->getChild(0).get()); + continue; + } + if (current->getOperatorType() == LogicalOperatorType::EXTEND || + current->getOperatorType() == LogicalOperatorType::PACKED_EXTEND) { + chainExtends.push_back(current->ptrCast()); + current = skipProjections(current->getChild(0).get()); + continue; + } + break; + } + if (current->getOperatorType() != LogicalOperatorType::SCAN_NODE_TABLE || + chainExtends.size() != 2) { + return op; + } + scanNode = current->ptrCast(); + // The scan's nodeID is the internal-ID property of the mid node; recover the mid node + // variable name from it and the mid NodeExpression from the chain extends (each + // extension shares the scan root as one endpoint). + if (scanNode->getNodeID()->expressionType != ExpressionType::PROPERTY || + !scanNode->getNodeID()->ptrCast()->isInternalID()) { + return op; + } + const auto midName = + scanNode->getNodeID()->ptrCast()->getVariableName(); + midNode = nullptr; + for (auto* ext : chainExtends) { + if (ext->getBoundNode()->getUniqueName() == midName) { + midNode = ext->getBoundNode(); + break; + } + if (ext->getNbrNode()->getUniqueName() == midName) { + midNode = ext->getNbrNode(); + } + } + if (midNode == nullptr) { + return op; + } + const auto aName = antiNodeA->getUniqueName(); + const auto bName = antiNodeB->getUniqueName(); + // Both extends must share the scan root as one endpoint, with the two anti-edge + // endpoints as the other endpoints (one each). + std::unordered_set others; + std::unordered_map hopDirByOther; + for (auto* ext : chainExtends) { + if (ext->getRel()->getEntry(0)->ptrCast() != antiRelEntry) { + return op; + } + const auto boundName = ext->getBoundNode()->getUniqueName(); + const auto nbrName = ext->getNbrNode()->getUniqueName(); + if (boundName == midName) { + others.insert(nbrName); + hopDirByOther[nbrName] = ext->getDirection(); + } else if (nbrName == midName) { + others.insert(boundName); + hopDirByOther[boundName] = ext->getDirection(); + } else { + return op; + } + } + if (others.size() != 2 || !others.contains(aName) || !others.contains(bName)) { + return op; + } + chainN0Dir = hopDirByOther.at(aName); + chainN2Dir = hopDirByOther.at(bName); + } + + // Parse the top join's other child as the filter-free suffix chain rooted at the scan of + // the top join key node (n2). + std::vector suffixHops; + { + auto* suffixChild = containsOp(topJoin->getChild(0).get(), antiFilter) ? + topJoin->getChild(1).get() : + topJoin->getChild(0).get(); + auto* current = skipProjections(suffixChild); + std::vector chainExtends; + LogicalScanNodeTable* scanNode = nullptr; + while (true) { + if (current->getOperatorType() == LogicalOperatorType::EXTEND || + current->getOperatorType() == LogicalOperatorType::PACKED_EXTEND) { + chainExtends.push_back(current->ptrCast()); + current = skipProjections(current->getChild(0).get()); + continue; + } + break; + } + if (current->getOperatorType() != LogicalOperatorType::SCAN_NODE_TABLE || + chainExtends.empty()) { + return op; + } + scanNode = current->ptrCast(); + const auto cName = topKey->ptrCast()->getVariableName(); + if (scanNode->getNodeID()->expressionType != ExpressionType::PROPERTY || + scanNode->getNodeID()->ptrCast()->getVariableName() != cName) { + return op; + } + // Build suffix hops in n2-outward order (reverse of the top-down walk order). Each hop + // must extend from the previous hop's "to" node (starting at n2 = c). + auto currentName = cName; + for (auto it = chainExtends.rbegin(); it != chainExtends.rend(); ++it) { + auto* ext = *it; + auto rel = ext->getRel(); + if (rel->getNumEntries() != 1) { + return op; + } + auto* relGroupEntry = rel->getEntry(0)->ptrCast(); + if (relGroupEntry->getScanFunction().has_value()) { + return op; + } + if (ext->getDirection() == ExtendDirection::BOTH) { + return op; + } + const auto boundName = ext->getBoundNode()->getUniqueName(); + const auto nbrName = ext->getNbrNode()->getUniqueName(); + const bool fromIsBound = boundName == currentName; + const bool fromIsNbr = nbrName == currentName; + if (!fromIsBound && !fromIsNbr) { + return op; + } + const auto toName = fromIsBound ? nbrName : boundName; + const bool fromIsSrc = + fromIsBound ? ext->extendFromSourceNode() : !ext->extendFromSourceNode(); + const auto scanDirection = fromIsSrc ? RelDataDirection::FWD : RelDataDirection::BWD; + const auto fromIsSrcFinal = fromIsSrc; + CountChainHop hop; + auto matched = false; + const auto fromTableID = fromIsSrcFinal ? + relGroupEntry->getRelEntryInfos()[0].nodePair.srcTableID : + relGroupEntry->getRelEntryInfos()[0].nodePair.dstTableID; + const auto toTableID = fromIsSrcFinal ? + relGroupEntry->getRelEntryInfos()[0].nodePair.dstTableID : + relGroupEntry->getRelEntryInfos()[0].nodePair.srcTableID; + // The from/to node expressions must be single-table and match the rel entry pair. + const NodeExpression* fromNode = nullptr; + const NodeExpression* toNode = nullptr; + if (fromIsBound) { + fromNode = ext->getBoundNode().get(); + toNode = ext->getNbrNode().get(); + } else { + fromNode = ext->getNbrNode().get(); + toNode = ext->getBoundNode().get(); + } + if (fromNode->isMultiLabeled() || toNode->isMultiLabeled() || + fromNode->getNumEntries() != 1 || toNode->getNumEntries() != 1) { + return op; + } + if (fromTableID != fromNode->getTableIDs()[0] || + toTableID != toNode->getTableIDs()[0]) { + return op; + } + hop.relScans.push_back({relGroupEntry->getRelEntryInfos()[0].oid, scanDirection, + fromTableID, toTableID, relGroupEntry->getName()}); + matched = true; + (void)matched; + suffixHops.push_back(std::move(hop)); + currentName = toName; + } + } + + // n0, n1, n2 must all bind the same single node table. + const auto midTableID = midNode->getTableIDs()[0]; + if (antiNodeA->isMultiLabeled() || antiNodeB->isMultiLabeled() || + antiNodeA->getNumEntries() != 1 || antiNodeB->getNumEntries() != 1 || + midNode->isMultiLabeled() || midNode->getNumEntries() != 1) { + return op; + } + if (antiNodeA->getTableIDs()[0] != midTableID || antiNodeB->getTableIDs()[0] != midTableID) { + return op; + } + + std::vector antiRelTableIDs; + antiRelTableIDs.push_back(antiRelEntry->getRelEntryInfos()[0].oid); + auto result = std::make_shared(std::move(suffixHops), antiRelEntry, + std::move(antiRelTableIDs), midTableID, chainN0Dir, chainN2Dir, antiEdgeDir, + true /* hasNotEquals */, aggExpr); + result->computeFlatSchema(); + return result; +} + std::shared_ptr CountRelTableOptimizer::tryRewriteExtendChainCount( std::shared_ptr op) { // Must be a keyless COUNT_STAR aggregate. Single-hop chains are handled by the (cheaper, @@ -577,8 +997,23 @@ bool CountRelTableOptimizer::canOptimize(LogicalOperator* aggregate) const { return true; } +static bool containsOp(const LogicalOperator* root, const LogicalOperator* target) { + if (root == target) { + return true; + } + for (auto i = 0u; i < root->getNumChildren(); ++i) { + if (containsOp(root->getChild(i).get(), target)) { + return true; + } + } + return false; +} + std::shared_ptr CountRelTableOptimizer::visitAggregateReplace( std::shared_ptr op) { + if (auto rewritten = tryRewriteAntiEdgeChainCount(op); rewritten != op) { + return rewritten; + } if (auto rewritten = tryRewriteExtendChainCount(op); rewritten != op) { return rewritten; } diff --git a/src/planner/operator/scan/CMakeLists.txt b/src/planner/operator/scan/CMakeLists.txt index 95b55604c..d1878d0cf 100644 --- a/src/planner/operator/scan/CMakeLists.txt +++ b/src/planner/operator/scan/CMakeLists.txt @@ -1,5 +1,6 @@ add_library(lbug_planner_scan OBJECT + logical_count_anti_edge_chain.cpp logical_count_extend_chain.cpp logical_count_rel_table.cpp logical_expressions_scan.cpp diff --git a/src/planner/operator/scan/logical_count_anti_edge_chain.cpp b/src/planner/operator/scan/logical_count_anti_edge_chain.cpp new file mode 100644 index 000000000..ad3dc28dd --- /dev/null +++ b/src/planner/operator/scan/logical_count_anti_edge_chain.cpp @@ -0,0 +1,20 @@ +#include "planner/operator/scan/logical_count_anti_edge_chain.h" + +namespace lbug { +namespace planner { + +void LogicalCountAntiEdgeChain::computeFactorizedSchema() { + createEmptySchema(); + auto groupPos = schema->createGroup(); + schema->insertToGroupAndScope(countExpr, groupPos); + schema->setGroupAsSingleState(groupPos); +} + +void LogicalCountAntiEdgeChain::computeFlatSchema() { + createEmptySchema(); + auto groupPos = schema->createGroup(); + schema->insertToGroupAndScope(countExpr, groupPos); +} + +} // namespace planner +} // namespace lbug diff --git a/src/processor/map/CMakeLists.txt b/src/processor/map/CMakeLists.txt index f6ee9572e..459a99a80 100644 --- a/src/processor/map/CMakeLists.txt +++ b/src/processor/map/CMakeLists.txt @@ -9,6 +9,7 @@ add_library(lbug_processor_mapper map_aggregate.cpp map_analyze.cpp map_count_rel_table.cpp + map_count_anti_edge_chain.cpp map_count_extend_chain.cpp map_standalone_call.cpp map_table_function_call.cpp diff --git a/src/processor/map/map_count_anti_edge_chain.cpp b/src/processor/map/map_count_anti_edge_chain.cpp new file mode 100644 index 000000000..d5211f206 --- /dev/null +++ b/src/processor/map/map_count_anti_edge_chain.cpp @@ -0,0 +1,52 @@ +#include "main/client_context.h" +#include "planner/operator/scan/logical_count_anti_edge_chain.h" +#include "processor/operator/scan/count_anti_edge_chain.h" +#include "processor/plan_mapper.h" +#include "storage/storage_manager.h" + +using namespace lbug::common; +using namespace lbug::planner; +using namespace lbug::storage; + +namespace lbug { +namespace processor { + +std::unique_ptr PlanMapper::mapCountAntiEdgeChain( + const LogicalOperator* logicalOperator) { + auto& logicalAntiEdgeChain = logicalOperator->constCast(); + auto outSchema = logicalAntiEdgeChain.getSchema(); + auto countOutputPos = getDataPos(*logicalAntiEdgeChain.getCountExpr(), *outSchema); + + auto* storageManager = StorageManager::Get(*clientContext); + std::vector hops; + hops.reserve(logicalAntiEdgeChain.getSuffixHops().size()); + for (auto& logicalHop : logicalAntiEdgeChain.getSuffixHops()) { + CountAntiEdgeChain::Hop hop; + hop.relTables.reserve(logicalHop.relScans.size()); + hop.scanDirections.reserve(logicalHop.relScans.size()); + hop.fromNodeTables.reserve(logicalHop.relScans.size()); + hop.toNodeTables.reserve(logicalHop.relScans.size()); + for (auto& spec : logicalHop.relScans) { + hop.relTables.push_back(storageManager->getTable(spec.relTableID)->ptrCast()); + hop.scanDirections.push_back(spec.scanDirection); + hop.fromNodeTables.push_back( + storageManager->getTable(spec.fromNodeTableID)->ptrCast()); + hop.toNodeTables.push_back( + storageManager->getTable(spec.toNodeTableID)->ptrCast()); + } + hops.push_back(std::move(hop)); + } + + auto* antiRelTable = + storageManager->getTable(logicalAntiEdgeChain.getAntiRelTableIDs()[0])->ptrCast(); + auto* midNodeTable = + storageManager->getTable(logicalAntiEdgeChain.getMidNodeTableID())->ptrCast(); + + return std::make_unique(std::move(hops), antiRelTable, midNodeTable, + logicalAntiEdgeChain.getChainN0Dir(), logicalAntiEdgeChain.getChainN2Dir(), + logicalAntiEdgeChain.getAntiEdgeDir(), logicalAntiEdgeChain.getHasNotEquals(), + countOutputPos, getOperatorID(), logicalAntiEdgeChain.getPrintInfo()); +} + +} // namespace processor +} // namespace lbug diff --git a/src/processor/map/plan_mapper.cpp b/src/processor/map/plan_mapper.cpp index 3847d63e4..74786582e 100644 --- a/src/processor/map/plan_mapper.cpp +++ b/src/processor/map/plan_mapper.cpp @@ -90,6 +90,9 @@ std::unique_ptr PlanMapper::mapOperator(const LogicalOperator* case LogicalOperatorType::COUNT_REL_TABLE: { physicalOperator = mapCountRelTable(logicalOperator); } break; + case LogicalOperatorType::COUNT_ANTI_EDGE_CHAIN: { + physicalOperator = mapCountAntiEdgeChain(logicalOperator); + } break; case LogicalOperatorType::COUNT_EXTEND_CHAIN: { physicalOperator = mapCountExtendChain(logicalOperator); } break; diff --git a/src/processor/operator/physical_operator.cpp b/src/processor/operator/physical_operator.cpp index 6b07bd759..8f0fc1d34 100644 --- a/src/processor/operator/physical_operator.cpp +++ b/src/processor/operator/physical_operator.cpp @@ -29,6 +29,8 @@ std::string PhysicalOperatorUtils::operatorTypeToString(PhysicalOperatorType ope return "BATCH_INSERT"; case PhysicalOperatorType::COPY_TO: return "COPY_TO"; + case PhysicalOperatorType::COUNT_ANTI_EDGE_CHAIN: + return "COUNT_ANTI_EDGE_CHAIN"; case PhysicalOperatorType::COUNT_EXTEND_CHAIN: return "COUNT_EXTEND_CHAIN"; case PhysicalOperatorType::COUNT_REL_TABLE: diff --git a/src/processor/operator/scan/CMakeLists.txt b/src/processor/operator/scan/CMakeLists.txt index 2ee901e7f..1e08b615d 100644 --- a/src/processor/operator/scan/CMakeLists.txt +++ b/src/processor/operator/scan/CMakeLists.txt @@ -1,5 +1,6 @@ add_library(lbug_processor_operator_scan OBJECT + count_anti_edge_chain.cpp count_extend_chain.cpp count_rel_table.cpp primary_key_scan_node_table.cpp diff --git a/src/processor/operator/scan/count_anti_edge_chain.cpp b/src/processor/operator/scan/count_anti_edge_chain.cpp new file mode 100644 index 000000000..2ec91eca0 --- /dev/null +++ b/src/processor/operator/scan/count_anti_edge_chain.cpp @@ -0,0 +1,349 @@ +#include "processor/operator/scan/count_anti_edge_chain.h" + +#include "common/system_config.h" +#include "processor/execution_context.h" +#include "storage/buffer_manager/memory_manager.h" +#include "transaction/transaction.h" + +using namespace lbug::common; +using namespace lbug::storage; +using namespace lbug::transaction; + +namespace lbug { +namespace processor { + +namespace { + +// CSR adjacency of one rel scan direction over the mid node table: offsets has midVecSize + 1 +// entries, nbrs holds the neighbor node offsets. +struct DirAdjacency { + std::vector offsets; + std::vector nbrs; +}; + +// The CSR directions scanned for an extend in the given direction. A BOTH extend enumerates +// the FWD rows followed by the BWD rows of each bound node (no dedup), matching LogicalExtend. +struct DirComponents { + RelDataDirection dirs[2]; + uint8_t numDirs; +}; + +DirComponents componentsOf(ExtendDirection dir) { + switch (dir) { + case ExtendDirection::FWD: + return {{RelDataDirection::FWD}, 1}; + case ExtendDirection::BWD: + return {{RelDataDirection::BWD}, 1}; + case ExtendDirection::BOTH: + return {{RelDataDirection::FWD, RelDataDirection::BWD}, 2}; + default: + UNREACHABLE_CODE; + } +} + +ExtendDirection reverse(ExtendDirection dir) { + switch (dir) { + case ExtendDirection::FWD: + return ExtendDirection::BWD; + case ExtendDirection::BWD: + return ExtendDirection::FWD; + case ExtendDirection::BOTH: + return ExtendDirection::BOTH; + default: + UNREACHABLE_CODE; + } +} + +} // namespace + +void CountAntiEdgeChain::initLocalStateInternal(ResultSet* resultSet, + ExecutionContext* /*context*/) { + countVector = resultSet->getValueVector(countOutputPos).get(); + hasExecuted = false; +} + +template +void CountAntiEdgeChain::scanRelRows(RelTable* relTable, RelDataDirection direction, + NodeTable* boundNodeTable, Transaction* transaction, MemoryManager* memoryManager, + Func&& callback) { + auto nodeIDVector = std::make_shared(LogicalType::INTERNAL_ID(), memoryManager, + std::make_shared()); + auto nbrVector = std::make_shared(LogicalType::INTERNAL_ID(), memoryManager, + std::make_shared()); + std::vector outVectors{nbrVector.get()}; + RelTableScanState scanState(*memoryManager, nodeIDVector.get(), outVectors, nbrVector->state); + scanState.setToTable(transaction, relTable, {NBR_ID_COLUMN_ID}, {}, direction); + scanState.packedMultiParentScan = true; + + const auto offsetUpper = getOffsetUpperBound(boundNodeTable); + for (auto start = offset_t{0}; start < offsetUpper; start += DEFAULT_VECTOR_CAPACITY) { + const auto n = + static_cast(std::min(DEFAULT_VECTOR_CAPACITY, offsetUpper - start)); + nodeIDVector->state->setToUnflat(); + nodeIDVector->state->getSelVectorUnsafe().setToUnfiltered(n); + for (auto k = 0u; k < n; ++k) { + nodeIDVector->setValue(k, nodeID_t{start + k, boundNodeTable->getTableID()}); + } + relTable->initScanState(transaction, scanState); + while (relTable->scan(transaction, scanState)) { + callback(scanState, *nodeIDVector, *nbrVector); + } + } +} + +// Invoke callback(boundOffset, nbrOffset) for every visible row of the current scan batch, +// resolving the bound node of each output row through the packed multi-parent contract (or the +// single-parent flat contract used by the in-memory/local scan paths). +template +static void forEachScanRow(const RelTableScanState& scanState, const ValueVector& nodeIDVector, + const ValueVector& nbrVector, CB&& callback) { + const auto outputSize = scanState.outState->getSelVector().getSelSize(); + if (outputSize == 0) { + return; + } + const auto& boundSel = nodeIDVector.state->getSelVector(); + const auto boundSelSize = boundSel.getSelSize(); + const auto& outSel = scanState.outState->getSelVector(); + if (boundSelSize > 1) { + const auto& packedChildOffsets = scanState.packedChildOffsets; + DASSERT(packedChildOffsets.size() == size_t(boundSelSize) + 1); + DASSERT(packedChildOffsets.back() == outputSize); + for (auto p = 0u; p < boundSelSize; ++p) { + const auto u = nodeIDVector.readNodeOffset(boundSel[p]); + for (auto r = packedChildOffsets[p]; r < packedChildOffsets[p + 1]; ++r) { + callback(u, nbrVector.readNodeOffset(outSel[r])); + } + } + } else { + DASSERT(boundSelSize == 1); + const auto u = nodeIDVector.readNodeOffset(boundSel[0]); + for (auto r = 0u; r < outputSize; ++r) { + callback(u, nbrVector.readNodeOffset(outSel[r])); + } + } +} + +std::vector CountAntiEdgeChain::computeSuffixCounts(Transaction* transaction, + MemoryManager* memoryManager) { + // B[k]: per-node suffix path counts at suffix position k (0 = the n2/mid side, numHops = + // the far end). B[numHops] = ones over the far-end node table; B[k][v] = sum over rows + // (v -> w) of hop[k] (enumerated with the scan direction keyed by the n-side) of B[k+1][w]. + // Only B[0] (= D over the mid node table) feeds the arithmetic; intermediate vectors are + // dropped as we walk backward. + const auto vectorSizeFor = [](NodeTable* t) { + return t->getNumNodeGroups() * StorageConfig::NODE_GROUP_SIZE; + }; + const auto numHops = suffixHops.size(); + std::vector> vecs(numHops + 1); + NodeTable* farTable = suffixHops[numHops - 1].toNodeTables[0]; + vecs[numHops] = std::vector(vectorSizeFor(farTable), 1); + for (auto k = numHops; k > 0; --k) { + auto& hop = suffixHops[k - 1]; + NodeTable* fromTable = hop.fromNodeTables[0]; + vecs[k - 1] = std::vector(vectorSizeFor(fromTable), 0); + auto& Bk = vecs[k - 1]; + auto& Bk1 = vecs[k]; + for (auto i = 0u; i < hop.relTables.size(); ++i) { + scanRelRows(hop.relTables[i], hop.scanDirections[i], hop.fromNodeTables[i], transaction, + memoryManager, + [&](const RelTableScanState& scanState, const ValueVector& nodeIDVector, + const ValueVector& nbrVector) { + forEachScanRow(scanState, nodeIDVector, nbrVector, + [&](offset_t v, offset_t w) { Bk[v] += Bk1[w]; }); + }); + } + vecs[k] = std::vector(); + } + return vecs[0]; +} + +bool CountAntiEdgeChain::getNextTuplesInternal(ExecutionContext* context) { + if (hasExecuted) { + return false; + } + auto transaction = Transaction::Get(*context->clientContext); + auto* memoryManager = MemoryManager::Get(*context->clientContext); + + const auto midVecSize = midNodeTable->getNumNodeGroups() * StorageConfig::NODE_GROUP_SIZE; + const auto offsetUpper = getOffsetUpperBound(midNodeTable); + + // Step 1: D[v] = number of suffix paths starting at v (over the mid node table). + const auto D = computeSuffixCounts(transaction, memoryManager); + + // Step 2: directional adjacency of the anti-edge rel over the mid node table. outAdj[v] + // holds the FWD rows with src=v; inAdj[v] holds the BWD rows with dst=v. All enumeration + // below is row-level: a BOTH hop enumerates the FWD rows followed by the BWD rows of each + // bound node (no dedup), matching the extend operator's semantics. + DirAdjacency outAdj; + DirAdjacency inAdj; + outAdj.offsets.assign(midVecSize + 1, 0); + inAdj.offsets.assign(midVecSize + 1, 0); + auto scanPairs = [&](RelDataDirection dir, auto&& callback) { + scanRelRows(antiRelTable, dir, midNodeTable, transaction, memoryManager, + [&](const RelTableScanState& scanState, const ValueVector& nodeIDVector, + const ValueVector& nbrVector) { + forEachScanRow(scanState, nodeIDVector, nbrVector, callback); + }); + }; + scanPairs(RelDataDirection::FWD, [&](offset_t u, offset_t /*v*/) { ++outAdj.offsets[u + 1]; }); + scanPairs(RelDataDirection::BWD, [&](offset_t u, offset_t /*v*/) { ++inAdj.offsets[u + 1]; }); + for (auto v = 0u; v < midVecSize; ++v) { + outAdj.offsets[v + 1] += outAdj.offsets[v]; + inAdj.offsets[v + 1] += inAdj.offsets[v]; + } + outAdj.nbrs.resize(outAdj.offsets[midVecSize]); + inAdj.nbrs.resize(inAdj.offsets[midVecSize]); + { + std::vector outCursor(outAdj.offsets.begin(), outAdj.offsets.end() - 1); + std::vector inCursor(inAdj.offsets.begin(), inAdj.offsets.end() - 1); + scanPairs(RelDataDirection::FWD, + [&](offset_t u, offset_t v) { outAdj.nbrs[outCursor[u]++] = v; }); + scanPairs(RelDataDirection::BWD, + [&](offset_t u, offset_t v) { inAdj.nbrs[inCursor[u]++] = v; }); + } + const auto& listFor = [&](RelDataDirection dir) -> const DirAdjacency& { + return dir == RelDataDirection::FWD ? outAdj : inAdj; + }; + const auto degreeOf = [&](ExtendDirection dir, offset_t v) -> offset_t { + offset_t deg = 0; + const auto components = componentsOf(dir); + for (auto ci = 0u; ci < components.numDirs; ++ci) { + const auto& lst = listFor(components.dirs[ci]); + deg += lst.offsets[v + 1] - lst.offsets[v]; + } + return deg; + }; + + // Step 3: T = all chain tuples (n0, n1, n2) weighted by D(n2): n2 iterates the n2-hop rows + // in direction chainN2Dir; each such row pairs with every n0-hop row of n1. + int64_t T = 0; + { + const auto n2Components = componentsOf(chainN2Dir); + for (auto ci = 0u; ci < n2Components.numDirs; ++ci) { + const auto& lst = listFor(n2Components.dirs[ci]); + for (auto v = 0u; v < offsetUpper; ++v) { + const auto degN0 = degreeOf(chainN0Dir, v); + if (degN0 == 0) { + continue; + } + for (auto i = lst.offsets[v]; i < lst.offsets[v + 1]; ++i) { + T += static_cast(degN0) * D[lst.nbrs[i]]; + } + } + } + } + + // Step 4: S = tuples with n0 = n2 = p, which are excluded by the id(n0) <> id(n2) + // predicate when it is present. For each mid node n1 and node p, the number of + // (n0-row, n2-row) combinations with both endpoints p is multN2(p) * multN0(p). + int64_t S = 0; + if (hasNotEquals) { + std::vector stampPass(midVecSize, 0); + std::vector stampCnt(midVecSize, 0); + uint32_t pass = 0; + const auto n2Components = componentsOf(chainN2Dir); + const auto n0Components = componentsOf(chainN0Dir); + for (auto v = 0u; v < offsetUpper; ++v) { + ++pass; + for (auto ci = 0u; ci < n2Components.numDirs; ++ci) { + const auto& lst = listFor(n2Components.dirs[ci]); + for (auto i = lst.offsets[v]; i < lst.offsets[v + 1]; ++i) { + const auto p = lst.nbrs[i]; + if (stampPass[p] != pass) { + stampPass[p] = pass; + stampCnt[p] = 0; + } + ++stampCnt[p]; + } + } + for (auto ci = 0u; ci < n0Components.numDirs; ++ci) { + const auto& lst = listFor(n0Components.dirs[ci]); + for (auto i = lst.offsets[v]; i < lst.offsets[v + 1]; ++i) { + const auto p = lst.nbrs[i]; + if (stampPass[p] == pass) { + S += static_cast(stampCnt[p]) * D[p]; + } + } + } + } + } + + // Step 5: A = tuples where the anti-edge row exists. For each anti-edge row (n0, n2) + // enumerated in direction antiEdgeDir, the number of chain tuples through it is + // N'(n0,n2) = sum over mid nodes x of multRevN0(n0,x) * multRevN2(n2,x), weighted by + // D(n2). Computed by stamping n2's rev(chainN2Dir) list once per n2 and scanning each row + // partner n0's rev(chainN0Dir) list against the stamps. + int64_t A = 0; + { + std::vector stampPass(midVecSize, 0); + std::vector stampCnt(midVecSize, 0); + std::vector partnerPass(midVecSize, 0); + uint32_t pass = 0; + uint32_t partnerPassEpoch = 0; + const auto revN0 = reverse(chainN0Dir); + const auto revN2 = reverse(chainN2Dir); + // Anti-edge rows (n0, n2) in direction antiEdgeDir, grouped by n2: FWD rows are + // (n0 -> n2) so n0 is an in-neighbor of n2; BWD rows are (n2 -> n0) so n0 is an + // out-neighbor of n2; BOTH rows are both orientations. + const auto partnerComponents = + componentsOf(antiEdgeDir == ExtendDirection::FWD ? ExtendDirection::BWD : + antiEdgeDir == ExtendDirection::BWD ? ExtendDirection::FWD : + ExtendDirection::BOTH); + const auto revN2Components = componentsOf(revN2); + const auto revN0Components = componentsOf(revN0); + for (auto n2 = 0u; n2 < offsetUpper; ++n2) { + const auto dN2 = D[n2]; + if (dN2 == 0) { + continue; + } + ++pass; + // The anti-edge match is a boolean mark per (n0, n2) key pair, so each distinct + // row partner n0 of n2 is subtracted once even if it appears in both the in and + // out lists (BOTH anti-edge) or via duplicate rows. + ++partnerPassEpoch; + for (auto ci = 0u; ci < revN2Components.numDirs; ++ci) { + const auto& lst = listFor(revN2Components.dirs[ci]); + for (auto i = lst.offsets[n2]; i < lst.offsets[n2 + 1]; ++i) { + const auto x = lst.nbrs[i]; + if (stampPass[x] != pass) { + stampPass[x] = pass; + stampCnt[x] = 0; + } + ++stampCnt[x]; + } + } + for (auto ci = 0u; ci < partnerComponents.numDirs; ++ci) { + const auto& partners = listFor(partnerComponents.dirs[ci]); + for (auto i = partners.offsets[n2]; i < partners.offsets[n2 + 1]; ++i) { + const auto n0 = partners.nbrs[i]; + if (partnerPass[n0] == partnerPassEpoch) { + continue; + } + partnerPass[n0] = partnerPassEpoch; + int64_t numPaths = 0; + for (auto cj = 0u; cj < revN0Components.numDirs; ++cj) { + const auto& lst = listFor(revN0Components.dirs[cj]); + for (auto j = lst.offsets[n0]; j < lst.offsets[n0 + 1]; ++j) { + const auto x = lst.nbrs[j]; + if (stampPass[x] == pass) { + numPaths += static_cast(stampCnt[x]); + } + } + } + A += numPaths * dN2; + } + } + } + } + + const auto count = T - S - A; + + hasExecuted = true; + countVector->state->getSelVectorUnsafe().setToUnfiltered(1); + countVector->setNull(0, false); + countVector->setValue(0, count); + return true; +} + +} // namespace processor +} // namespace lbug From 23a2b3e2b66af1b1f25d34bfd145209b852866a0 Mon Sep 17 00:00:00 2001 From: Arun Sharma Date: Sat, 5 Sep 2026 15:42:24 -0700 Subject: [PATCH 2/2] Generalize LBUG_DUMP_LOGICAL to dump any query's logical plan Move the logical plan tree dumper out of count_rel_table_optimizer.cpp into Optimizer::optimize, which every query goes through. With LBUG_DUMP_LOGICAL set in the environment, the logical plan is printed to stderr as a readable indented tree (one operator per line, with extend directions, join keys/types, filter predicates, aggregate keys and scan targets) both before and after optimization. --- src/optimizer/optimizer.cpp | 88 +++++++++++++++++++++++++++++++++++++ 1 file changed, 88 insertions(+) diff --git a/src/optimizer/optimizer.cpp b/src/optimizer/optimizer.cpp index bc7617d36..5b97c3736 100644 --- a/src/optimizer/optimizer.cpp +++ b/src/optimizer/optimizer.cpp @@ -1,5 +1,10 @@ #include "optimizer/optimizer.h" +#include +#include +#include + +#include "common/enums/extend_direction_util.h" #include "main/client_context.h" #include "optimizer/acc_hash_join_optimizer.h" #include "optimizer/agg_key_dependency_optimizer.h" @@ -21,14 +26,94 @@ #include "optimizer/schema_populator.h" #include "optimizer/top_k_optimizer.h" #include "optimizer/unwind_dedup_optimizer.h" +#include "planner/operator/extend/logical_extend.h" +#include "planner/operator/logical_aggregate.h" #include "planner/operator/logical_explain.h" +#include "planner/operator/logical_filter.h" +#include "planner/operator/logical_hash_join.h" +#include "planner/operator/logical_operator.h" +#include "planner/operator/scan/logical_scan_node_table.h" #include "transaction/transaction.h" namespace lbug { namespace optimizer { +namespace { + +// Prints one operator per line as an indented tree. Enabled by setting LBUG_DUMP_LOGICAL in the +// environment; unlike EXPLAIN LOGICAL this needs no query changes and shows the plan exactly as +// the optimizer sees it, before and after optimization. +void dumpLogicalTree(const planner::LogicalOperator* op, int depth, + std::unordered_set& visited) { + if (depth > 40 || visited.contains(op)) { + for (auto i = 0; i < depth; ++i) { + fprintf(stderr, " "); + } + fprintf(stderr, "...\n"); + return; + } + visited.insert(op); + for (auto i = 0; i < depth; ++i) { + fprintf(stderr, " "); + } + fprintf(stderr, "%s", + planner::LogicalOperatorUtils::logicalOperatorTypeToString(op->getOperatorType()).c_str()); + if (op->getOperatorType() == planner::LogicalOperatorType::EXTEND || + op->getOperatorType() == planner::LogicalOperatorType::PACKED_EXTEND) { + auto& ext = op->constCast(); + fprintf(stderr, " [%s %s bound=%s nbr=%s]", ext.getRel()->detailsToString().c_str(), + common::ExtendDirectionUtil::toString(ext.getDirection()).c_str(), + ext.getBoundNode()->getUniqueName().c_str(), ext.getNbrNode()->getUniqueName().c_str()); + } else if (op->getOperatorType() == planner::LogicalOperatorType::FILTER) { + fprintf(stderr, " [%s]", + op->constCast().getPredicate()->toString().c_str()); + } else if (op->getOperatorType() == planner::LogicalOperatorType::HASH_JOIN) { + auto& join = op->constCast(); + auto jt = join.getJoinType() == common::JoinType::INNER ? "INNER" : + join.getJoinType() == common::JoinType::MARK ? "MARK" : + join.getJoinType() == common::JoinType::LEFT ? "LEFT" : + "COUNT"; + fprintf(stderr, " [%s keys=%s", jt, + join.getJoinNodeIDs().size() == 1 ? join.getJoinNodeIDs()[0]->toString().c_str() : + "..."); + if (join.hasMark()) { + fprintf(stderr, " mark=%s]", join.getMark()->toString().c_str()); + } else { + fprintf(stderr, "]"); + } + } else if (op->getOperatorType() == planner::LogicalOperatorType::AGGREGATE) { + fprintf(stderr, " [keys=%llu aggs=%s]", + (unsigned long long)op->constCast().getKeys().size(), + op->constCast().getAggregates()[0]->toString().c_str()); + } else if (op->getOperatorType() == planner::LogicalOperatorType::SCAN_NODE_TABLE) { + fprintf(stderr, " [%s]", + op->constCast().getNodeID()->toString().c_str()); + } + fprintf(stderr, "\n"); + for (auto i = 0u; i < op->getNumChildren(); ++i) { + dumpLogicalTree(op->getChild(i).get(), depth + 1, visited); + } +} + +void dumpLogicalPlan(const planner::LogicalPlan* plan, const char* label) { + fprintf(stderr, "=== LOGICAL PLAN (%s) ===\n", label); + auto* root = plan->getLastOperator().get(); + if (root == nullptr) { + return; + } + std::unordered_set visited; + dumpLogicalTree(root, 0, visited); + fprintf(stderr, "=== END LOGICAL PLAN ===\n"); +} + +} // namespace + void Optimizer::optimize(planner::LogicalPlan* plan, main::ClientContext* context, const planner::CardinalityEstimator& cardinalityEstimator) { + static const bool dumpLogicalEnabled = getenv("LBUG_DUMP_LOGICAL") != nullptr; + if (dumpLogicalEnabled) { + dumpLogicalPlan(plan, "before optimization"); + } if (context->getClientConfig()->enablePlanOptimizer) { // Factorization structure should be removed before further optimization can be applied. auto removeFactorizationRewriter = RemoveFactorizationRewriter(); @@ -125,6 +210,9 @@ void Optimizer::optimize(planner::LogicalPlan* plan, main::ClientContext* contex auto schemaPopulator = SchemaPopulator{}; schemaPopulator.rewrite(plan); } + if (dumpLogicalEnabled) { + dumpLogicalPlan(plan, "after optimization"); + } } } // namespace optimizer