From 9f4c47841140d4741e9e1ad13645c24973e0542e Mon Sep 17 00:00:00 2001 From: Arun Sharma Date: Sat, 17 Oct 2026 11:38:07 -0700 Subject: [PATCH 1/4] feat(planner): lookup join for inner equi-joins on node ID When one side of an inner equi-join on a single node ID is a bare single-node scan of that node (pure payload fetch, e.g. LDBC Q1's Person scan) and the other side already binds the node ID with a small estimated cardinality (<= LOOKUP_JOIN_MAX_OUTER_CARDINALITY), replace the full scan + hash join with per-tuple primary-key lookups (LogicalQueryPrimaryKeyLookup, inner semantics: non-matches dropped). Storage: NodeTable::lookupPK resolves INTERNAL_ID keys directly to offsets (table-ID + bounds + visibility checks, local-storage aware) without probing the HASH/ART primary-key index. Q1: top HASH_JOIN(p._ID) + full Person scan + semi-masker becomes a 0.07ms lookup over the small outer (was ~1ms of the ~2.4ms exec). --- src/include/common/constants.h | 4 ++ src/include/planner/planner.h | 8 +++ src/planner/plan/plan_join_order.cpp | 91 ++++++++++++++++++++++++++++ src/storage/table/node_table.cpp | 39 ++++++++++++ 4 files changed, 142 insertions(+) diff --git a/src/include/common/constants.h b/src/include/common/constants.h index c43bcd312..238caa199 100644 --- a/src/include/common/constants.h +++ b/src/include/common/constants.h @@ -177,6 +177,10 @@ struct PlannerKnobs { // Recursive extend runs one graph search per bound source node. Measured per-source setup // cost vs per-output-pair cost is ~1000:1, so seed cardinality dominates direction choice. static constexpr uint64_t RECURSIVE_EXTEND_SOURCE_COST = 1000; + // Lookup join (per-tuple PK lookup instead of scan + hash join) pays off only when the + // outer side is small: each outer row costs one index probe/payload fetch, so cap the + // estimated outer cardinality this rewrite applies to (Q1's Person fetch: ~9 rows). + static constexpr uint64_t LOOKUP_JOIN_MAX_OUTER_CARDINALITY = 1024; }; struct OrderByConstants { diff --git a/src/include/planner/planner.h b/src/include/planner/planner.h index 5d22b460b..d0cba54c4 100644 --- a/src/include/planner/planner.h +++ b/src/include/planner/planner.h @@ -238,6 +238,14 @@ class LBUG_API Planner { bool tryPlanPackedINLJoin(const binder::SubqueryGraph& subgraph, const binder::SubqueryGraph& otherSubgraph, const std::vector>& joinNodes); + // Lookup join: when one side of an inner equi-join on a single node ID is a bare + // single-node table scan of that node (pure payload fetch, e.g. Q1's Person scan) and the + // other side already binds the node ID with a small estimated cardinality, replace the + // full scan + hash join with per-tuple primary-key lookups (inner semantics). + // Returns true when a lookup plan was added (hash plans for the pair are then skipped). + bool tryPlanLookupJoin(const binder::SubqueryGraph& innerSubgraph, const LogicalPlan& innerPlan, + const LogicalPlan& outerPlan, const std::shared_ptr& joinNode, + const binder::expression_vector& predicates, const binder::SubqueryGraph& newSubgraph); void planInnerHashJoin(const binder::SubqueryGraph& subgraph, const binder::SubqueryGraph& otherSubgraph, const std::vector>& joinNodes, bool flipPlan); diff --git a/src/planner/plan/plan_join_order.cpp b/src/planner/plan/plan_join_order.cpp index 037ffe49f..79f841c45 100644 --- a/src/planner/plan/plan_join_order.cpp +++ b/src/planner/plan/plan_join_order.cpp @@ -1,6 +1,8 @@ +#include #include #include "binder/bound_scan_source.h" +#include "binder/expression/property_expression.h" #include "binder/expression_visitor.h" #include "catalog/catalog_entry/catalog_entry_type.h" #include "catalog/catalog_entry/node_table_catalog_entry.h" @@ -14,8 +16,11 @@ #include "planner/join_order/join_plan_solver.h" #include "planner/join_order/join_tree_constructor.h" #include "planner/operator/extend/logical_extend.h" +#include "planner/operator/scan/logical_query_primary_key_lookup.h" #include "planner/operator/scan/logical_scan_node_table.h" #include "planner/planner.h" +#include "storage/storage_manager.h" +#include "storage/table/node_table.h" using namespace lbug::binder; using namespace lbug::common; @@ -838,6 +843,82 @@ bool Planner::tryPlanINLJoin(const SubqueryGraph& subgraph, const SubqueryGraph& return hasAppliedINLJoin; } +// A bare single-node scan of the join node (a leaf SCAN with no filters, i.e. a pure +// payload fetch) qualifies as the inner side of a lookup join. Filtered scans stay on the +// hash path: their predicates are cheaper to apply before/within the join build. +static const LogicalScanNodeTable* getBareSingleNodeScan(const SubqueryGraph& subgraph, + const LogicalPlan& plan, const NodeExpression& joinNode) { + if (subgraph.getNumQueryRels() != 0 || subgraph.queryNodesSelector.count() != 1) { + return nullptr; + } + auto* op = plan.getLastOperator().get(); + if (op == nullptr || op->getOperatorType() != LogicalOperatorType::SCAN_NODE_TABLE) { + return nullptr; + } + auto& scan = op->constCast(); + if (scan.getScanType() != LogicalScanNodeTableType::SCAN) { + return nullptr; + } + if (!scan.getTableDBMap().empty()) { + return nullptr; + } + if (scan.getNodeID()->getUniqueName() != joinNode.getInternalID()->getUniqueName()) { + return nullptr; + } + if (scan.getTableIDs().size() != 1) { + return nullptr; + } + return &scan; +} + +bool Planner::tryPlanLookupJoin(const SubqueryGraph& innerSubgraph, const LogicalPlan& innerPlan, + const LogicalPlan& outerPlan, const std::shared_ptr& joinNode, + const expression_vector& predicates, const SubqueryGraph& newSubgraph) { + if (!clientContext->getClientConfig()->enablePlanOptimizer) { + return false; + } + auto* scan = getBareSingleNodeScan(innerSubgraph, innerPlan, *joinNode); + if (scan == nullptr) { + return false; + } + if (joinNode->getTableIDs().size() != 1) { + return false; + } + auto tableID = scan->getTableIDs()[0]; + auto table = storage::StorageManager::Get(*clientContext) + ->getTable(tableID) + ->ptrCast(); + if (table->tryGetPrimaryKeyIndex() == nullptr) { + return false; + } + const auto outerCard = outerPlan.getCardinality(); + if (outerCard > PlannerKnobs::LOOKUP_JOIN_MAX_OUTER_CARDINALITY) { + return false; + } + auto key = joinNode->getInternalID(); + if (!outerPlan.getSchema()->evaluable(*key)) { + return false; + } + auto plan = outerPlan.copy(); + appendFlattens(plan.getSchema()->getGroupsPosInScope(), plan); + const auto outputGroupPos = plan.getSchema()->getGroupPos(*key); + auto properties = getProperties(*joinNode); + properties.erase(std::remove_if(properties.begin(), properties.end(), + [](const std::shared_ptr& expression) { + return expression->constCast().isInternalID(); + }), + properties.end()); + auto lookup = std::make_shared(tableID, joinNode->getInternalID(), + properties, key, outputGroupPos, plan.getLastOperator()); + lookup->computeFactorizedSchema(); + lookup->setCardinality(outerCard); + plan.setLastOperator(std::move(lookup)); + appendFilters(predicates, plan); + plan.setCost(outerPlan.getCost() + outerCard); + context.addPlan(newSubgraph, std::move(plan)); + return true; +} + void Planner::planInnerHashJoin(const SubqueryGraph& subgraph, const SubqueryGraph& otherSubgraph, const std::vector>& joinNodes, bool flipPlan) { auto newSubgraph = subgraph; @@ -851,6 +932,16 @@ void Planner::planInnerHashJoin(const SubqueryGraph& subgraph, const SubqueryGra getNewlyMatchedExprs(subgraph, otherSubgraph, newSubgraph, context.getWhereExpressions()); for (auto& leftPlan : context.getPlans(subgraph)) { for (auto& rightPlan : context.getPlans(otherSubgraph)) { + // Lookup join replaces the hash join for this pair when one side is a bare + // single-node scan of the join node and the other side is small (e.g. Q1's + // Person payload fetch). Like INL joins, it prunes the hash alternatives. + if (joinNodes.size() == 1 && + (tryPlanLookupJoin(subgraph, leftPlan, rightPlan, joinNodes[0], predicates, + newSubgraph) || + tryPlanLookupJoin(otherSubgraph, rightPlan, leftPlan, joinNodes[0], + predicates, newSubgraph))) { + continue; + } const auto leftProbeCardinality = cardinalityEstimator.estimateHashJoin(joinNodeIDs, leftPlan.getLastOperatorRef(), rightPlan.getLastOperatorRef()); if (CostModel::computeHashJoinCost(joinNodeIDs, leftPlan, rightPlan, diff --git a/src/storage/table/node_table.cpp b/src/storage/table/node_table.cpp index 96fd06a7d..c39a04b8f 100644 --- a/src/storage/table/node_table.cpp +++ b/src/storage/table/node_table.cpp @@ -954,6 +954,45 @@ Index* NodeTable::tryGetPrimaryKeyIndex() const { bool NodeTable::lookupPK(const Transaction* transaction, ValueVector* keyVector, uint64_t vectorPos, offset_t& result) const { + // Lookup joins on node internal IDs (e.g. fetching Person payloads for already-bound + // `p._ID`s) resolve directly to offsets: the offset is embedded in the ID, so no + // primary-key index probe (HASH or ART) is needed -- only table-ID, bounds and + // visibility checks. Uncommitted rows live in local storage past the committed bound + // (`LocalNodeTable::getStartOffset`), with visibility tracked locally. + if (keyVector->dataType.getLogicalTypeID() == LogicalTypeID::INTERNAL_ID) { + if (keyVector->isNull(vectorPos)) { + return false; + } + const auto nodeID = keyVector->getValue(vectorPos); + if (nodeID.tableID != tableID || nodeID.offset == INVALID_OFFSET) { + return false; + } + if (transaction->getLocalStorage()) { + if (const auto localTable = transaction->getLocalStorage()->getLocalTable(*this)) { + auto& localNodeTable = localTable->cast(); + if (nodeID.offset >= localNodeTable.getStartOffset()) { + if (nodeID.offset >= + localNodeTable.getStartOffset() + localNodeTable.getNumTotalRows()) { + return false; + } + if (!localNodeTable.isVisible(transaction, nodeID.offset)) { + return false; + } + result = nodeID.offset; + return true; + } + } + } + if (nodeID.offset >= nodeGroups->getNumTotalRows()) { + return false; + } + // Handles locally-deleted committed rows via the local-visibility branch. + if (!isVisibleNoLock(transaction, nodeID.offset)) { + return false; + } + result = nodeID.offset; + return true; + } if (transaction->getLocalStorage()) { if (const auto localTable = transaction->getLocalStorage()->getLocalTable(*this); localTable && localTable->cast().lookupPK(transaction, keyVector, From c584b80a1ae86146555a13c37e69b41171762ed6 Mon Sep 17 00:00:00 2001 From: Arun Sharma Date: Sat, 17 Oct 2026 13:13:02 -0700 Subject: [PATCH 2/4] fix(planner): allow lookup join to reuse bound node ID schema LogicalQueryPrimaryKeyLookup assumed nodeID/properties are new, but lookup joins use the already-bound join node ID as the key (same unique name), tripping the schema duplicate-insert assert and breaking shell tests. Use insertToGroupAndScopeMayRepeat (same as hash joins on node IDs) and skip properties already bound by the outer plan. Also fix clang-format layout flagged by CI. --- .../scan/logical_query_primary_key_lookup.cpp | 11 +++++++---- src/planner/plan/plan_join_order.cpp | 18 +++++++++++------- 2 files changed, 18 insertions(+), 11 deletions(-) diff --git a/src/planner/operator/scan/logical_query_primary_key_lookup.cpp b/src/planner/operator/scan/logical_query_primary_key_lookup.cpp index 382819252..db726f801 100644 --- a/src/planner/operator/scan/logical_query_primary_key_lookup.cpp +++ b/src/planner/operator/scan/logical_query_primary_key_lookup.cpp @@ -26,17 +26,20 @@ f_group_pos LogicalQueryPrimaryKeyLookup::resolveOutputGroupPos() const { void LogicalQueryPrimaryKeyLookup::computeFactorizedSchema() { copyChildSchema(0); const auto groupPos = resolveOutputGroupPos(); - schema->insertToGroupAndScope(nodeID, groupPos); + // Lookup joins reuse an already-bound node ID as the key (key and nodeID share the + // same unique name), and properties may already be bound by the outer plan. Like hash + // joins on node IDs, allow repeats and keep the existing group in that case. + schema->insertToGroupAndScopeMayRepeat(nodeID, groupPos); for (auto& property : properties) { - schema->insertToGroupAndScope(property, groupPos); + schema->insertToGroupAndScopeMayRepeat(property, groupPos); } } void LogicalQueryPrimaryKeyLookup::computeFlatSchema() { copyChildSchema(0); - schema->insertToGroupAndScope(nodeID, 0); + schema->insertToGroupAndScopeMayRepeat(nodeID, 0); for (auto& property : properties) { - schema->insertToGroupAndScope(property, 0); + schema->insertToGroupAndScopeMayRepeat(property, 0); } } diff --git a/src/planner/plan/plan_join_order.cpp b/src/planner/plan/plan_join_order.cpp index 79f841c45..9cd88fa6b 100644 --- a/src/planner/plan/plan_join_order.cpp +++ b/src/planner/plan/plan_join_order.cpp @@ -904,8 +904,13 @@ bool Planner::tryPlanLookupJoin(const SubqueryGraph& innerSubgraph, const Logica const auto outputGroupPos = plan.getSchema()->getGroupPos(*key); auto properties = getProperties(*joinNode); properties.erase(std::remove_if(properties.begin(), properties.end(), - [](const std::shared_ptr& expression) { - return expression->constCast().isInternalID(); + [&](const std::shared_ptr& expression) { + if (expression->constCast().isInternalID()) { + return true; + } + // Skip properties already bound by the outer plan: the lookup + // only needs to fetch the missing payload columns. + return plan.getSchema()->isExpressionInScope(*expression); }), properties.end()); auto lookup = std::make_shared(tableID, joinNode->getInternalID(), @@ -935,11 +940,10 @@ void Planner::planInnerHashJoin(const SubqueryGraph& subgraph, const SubqueryGra // Lookup join replaces the hash join for this pair when one side is a bare // single-node scan of the join node and the other side is small (e.g. Q1's // Person payload fetch). Like INL joins, it prunes the hash alternatives. - if (joinNodes.size() == 1 && - (tryPlanLookupJoin(subgraph, leftPlan, rightPlan, joinNodes[0], predicates, - newSubgraph) || - tryPlanLookupJoin(otherSubgraph, rightPlan, leftPlan, joinNodes[0], - predicates, newSubgraph))) { + if (joinNodes.size() == 1 && (tryPlanLookupJoin(subgraph, leftPlan, rightPlan, + joinNodes[0], predicates, newSubgraph) || + tryPlanLookupJoin(otherSubgraph, rightPlan, leftPlan, + joinNodes[0], predicates, newSubgraph))) { continue; } const auto leftProbeCardinality = cardinalityEstimator.estimateHashJoin(joinNodeIDs, From 2aa0f38c0c831d25063a23968b0a14b50a0d0c05 Mon Sep 17 00:00:00 2001 From: Arun Sharma Date: Sat, 17 Oct 2026 13:33:41 -0700 Subject: [PATCH 3/4] fix(shell-test): use up-arrow for row movement in prev_history_multiline The third CTRL_P was commented 'move up line' but on Linux CTRL_P navigates history instead of moving a row (row-up-first applies only on Windows and for the up-arrow escape on all platforms). With just this test's own entries in history.txt the extra history-prev clamps to a no-op, leaving the cursor on the bottom row, so the two Downs walk off the recalled entry onto the empty line and ENTER submits nothing until the 30s expect times out. The test only passed when older shared-history entries happened to absorb the overshoot, hence flaky across selections/orders. Send up-arrow for the row-up step so the sequence is independent of history depth. --- tools/shell/test/test_shell_esc_edit.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/tools/shell/test/test_shell_esc_edit.py b/tools/shell/test/test_shell_esc_edit.py index 99bb41a91..9f223af45 100644 --- a/tools/shell/test/test_shell_esc_edit.py +++ b/tools/shell/test/test_shell_esc_edit.py @@ -198,7 +198,11 @@ def test_prev_history_multiline(test) -> None: ) test.send_control_statement(KEY_ACTION.CTRL_P.value) # move up in history test.send_control_statement(KEY_ACTION.CTRL_P.value) # move up in history - test.send_control_statement(KEY_ACTION.CTRL_P.value) # move up line + # Move up a row within the recalled multiline entry. This must be an up-arrow + # (row movement on all platforms): CTRL_P only moves a row on Windows and + # navigates history on Linux, which overshoots past the entry when history + # holds just this test's statements and leaves an empty line behind. + test.send_statement("\x1b[A") # move up line test.send_statement("\x1b[B") test.send_statement("\x1b[B") test.send_finished_statement(KEY_ACTION.ENTER.value) From ed3aa1c70b1d0502d4376e1b0b2870374f15b00d Mon Sep 17 00:00:00 2001 From: Arun Sharma Date: Sat, 17 Oct 2026 17:15:31 -0700 Subject: [PATCH 4/4] fix(planner): cost-competitive lookup join; fix parquet export of flat results Planner (join order): - Lookup join no longer prunes hash alternatives. Both lookup orientations are added alongside the hash candidates, each maxCost-gated, and the cheapest plan wins. - Cost via CostModel::computeLookupJoinCost (outer cost + per-tuple probes). - Eligibility is semantic: bare single-node scan, single-table node, PK index present, key evaluable, payload properties that exist. - Credibility gate (b2): the outer estimate is only trusted when every operator in the outer plan has well-modeled cardinality (scans, filters, projections, un/flattening, order/limit, cross products, unions, inner hash joins, node-label filters, nested lookups). Recursion, grouping/aggregation, non-inner joins, unwinds, intersects and anything else fail closed, as do extend chains deeper than LOOKUP_JOIN_MAX_OUTER_EXTEND_DEPTH (fan-out error compounds per hop while the estimator has no skew model). - Bounded fan-out gate (b3): the outer estimate is discounted by the observed worst/mean fan-out ratio of every extend in the outer plan (ANALYZE CSR degree stats; missing stats fail closed). The lookup fires only if the discounted estimate still fits comfortably inside the eliminated scan. Keeps full-coverage joins on the hash path so COUNT_REL_TABLE/degree/reachability rewrites still match. - encodeJoin renders QUERY_PRIMARY_KEY_LOOKUP as LKP(key) so plan tests can pin the new shape; SingleNodeTwoHopJoins expectations updated (results verified identical with the optimizer on and off). Storage: - ArrowNodeTable/IceDiskNodeTable resolve INTERNAL_ID keys directly to offsets in lookupPK. Parquet writer: - readFromFT bulk reads into fresh flat single-value vectors, which can only ever hold the first tuple (FactorizedTable::lookup asserts !(isFlat && numTuples > 1); release builds silently wrote corrupt files, surfacing as TProtocolException on import). Always read one tuple here; the bulk path could never be correct. Follow-up runtime fallback tracked in Refs #1307. --- src/include/common/constants.h | 6 + src/include/planner/join_order/cost_model.h | 5 + .../planner/operator/logical_plan_util.h | 2 + src/include/planner/planner.h | 12 +- .../storage/table/ice_disk_node_table.h | 2 + src/planner/join_order/cost_model.cpp | 5 + src/planner/operator/logical_plan_util.cpp | 11 + src/planner/plan/plan_join_order.cpp | 212 ++++++++++++++++-- .../writer/parquet/parquet_writer.cpp | 12 +- src/storage/table/arrow_node_table.cpp | 16 ++ src/storage/table/ice_disk_node_table.cpp | 22 ++ test/optimizer/optimizer_test.cpp | 12 +- 12 files changed, 289 insertions(+), 28 deletions(-) diff --git a/src/include/common/constants.h b/src/include/common/constants.h index 238caa199..31e654850 100644 --- a/src/include/common/constants.h +++ b/src/include/common/constants.h @@ -181,6 +181,12 @@ struct PlannerKnobs { // outer side is small: each outer row costs one index probe/payload fetch, so cap the // estimated outer cardinality this rewrite applies to (Q1's Person fetch: ~9 rows). static constexpr uint64_t LOOKUP_JOIN_MAX_OUTER_CARDINALITY = 1024; + // Lookup join eligibility additionally requires the outer plan to be free of + // unmodeled fan-out (no recursion/aggregation/etc.) and caps the extend chain + // depth: fan-out estimation error compounds multiplicatively per hop while the + // estimator has no skew model, so beyond two chained extends the estimate is no + // longer a credible probe budget. + static constexpr uint32_t LOOKUP_JOIN_MAX_OUTER_EXTEND_DEPTH = 2; }; struct OrderByConstants { diff --git a/src/include/planner/join_order/cost_model.h b/src/include/planner/join_order/cost_model.h index aae315f67..94ad0e112 100644 --- a/src/include/planner/join_order/cost_model.h +++ b/src/include/planner/join_order/cost_model.h @@ -22,6 +22,11 @@ class CostModel { const LogicalPlan& probe, const LogicalPlan& build); static uint64_t computeIntersectCost(const LogicalPlan& probePlan, const std::vector& buildPlans); + // Lookup join: one PK probe/payload fetch per outer row on top of the outer plan. + // Like the INL-join precedent, this only pays off for small outer cardinalities, + // which callers enforce via PlannerKnobs::LOOKUP_JOIN_MAX_OUTER_CARDINALITY. + static uint64_t computeLookupJoinCost(const LogicalPlan& outerPlan, + cardinality_t outerCardinality); }; } // namespace planner diff --git a/src/include/planner/operator/logical_plan_util.h b/src/include/planner/operator/logical_plan_util.h index eab7ec5d1..207337da5 100644 --- a/src/include/planner/operator/logical_plan_util.h +++ b/src/include/planner/operator/logical_plan_util.h @@ -18,6 +18,8 @@ class LogicalPlanUtil { static void encodeHashJoin(LogicalOperator* logicalOperator, std::string& encodeString); static void encodeExtend(LogicalOperator* logicalOperator, std::string& encodeString); static void encodeScanNodeTable(LogicalOperator* logicalOperator, std::string& encodeString); + static void encodeQueryPrimaryKeyLookup(LogicalOperator* logicalOperator, + std::string& encodeString); // Encode filter static void encodeFilter(LogicalOperator* logicalOperator, std::string& encodedString); }; diff --git a/src/include/planner/planner.h b/src/include/planner/planner.h index d0cba54c4..cbde82fd9 100644 --- a/src/include/planner/planner.h +++ b/src/include/planner/planner.h @@ -240,12 +240,16 @@ class LBUG_API Planner { const std::vector>& joinNodes); // Lookup join: when one side of an inner equi-join on a single node ID is a bare // single-node table scan of that node (pure payload fetch, e.g. Q1's Person scan) and the - // other side already binds the node ID with a small estimated cardinality, replace the - // full scan + hash join with per-tuple primary-key lookups (inner semantics). - // Returns true when a lookup plan was added (hash plans for the pair are then skipped). + // other side already binds the node ID with a small estimated cardinality, add a + // per-tuple primary-key lookup plan (inner semantics) alongside the hash join + // candidates. The lookup participates in cost competition like every other plan: + // it is only added when cheaper than the current max cost, and the cheapest plan + // wins, so hash alternatives are never pruned. + // Returns true when a lookup plan was added. bool tryPlanLookupJoin(const binder::SubqueryGraph& innerSubgraph, const LogicalPlan& innerPlan, const LogicalPlan& outerPlan, const std::shared_ptr& joinNode, - const binder::expression_vector& predicates, const binder::SubqueryGraph& newSubgraph); + const binder::expression_vector& predicates, const binder::SubqueryGraph& newSubgraph, + uint64_t maxCost); void planInnerHashJoin(const binder::SubqueryGraph& subgraph, const binder::SubqueryGraph& otherSubgraph, const std::vector>& joinNodes, bool flipPlan); diff --git a/src/include/storage/table/ice_disk_node_table.h b/src/include/storage/table/ice_disk_node_table.h index 2a6dc251f..9232553ac 100644 --- a/src/include/storage/table/ice_disk_node_table.h +++ b/src/include/storage/table/ice_disk_node_table.h @@ -74,6 +74,8 @@ class IceDiskNodeTable final : public ColumnarNodeTableBase { common::offset_t offset) const override; bool isVisibleNoLock(const transaction::Transaction* transaction, common::offset_t offset) const override; + bool lookupPK(const transaction::Transaction* transaction, common::ValueVector* keyVector, + uint64_t vectorPos, common::offset_t& result) const override; const std::string& getParquetFilePath() const { return parquetFilePath; } diff --git a/src/planner/join_order/cost_model.cpp b/src/planner/join_order/cost_model.cpp index f2a1c959c..42662435e 100644 --- a/src/planner/join_order/cost_model.cpp +++ b/src/planner/join_order/cost_model.cpp @@ -48,6 +48,11 @@ uint64_t CostModel::computeMarkJoinCost(const binder::expression_vector& joinNod return computeHashJoinCost(joinNodeIDs, probe, build); } +uint64_t CostModel::computeLookupJoinCost(const LogicalPlan& outerPlan, + cardinality_t outerCardinality) { + return outerPlan.getCost() + outerCardinality; +} + uint64_t CostModel::computeIntersectCost(const LogicalPlan& probePlan, const std::vector& buildPlans) { uint64_t cost = 0ul; diff --git a/src/planner/operator/logical_plan_util.cpp b/src/planner/operator/logical_plan_util.cpp index f3c43ff45..6d47dc45f 100644 --- a/src/planner/operator/logical_plan_util.cpp +++ b/src/planner/operator/logical_plan_util.cpp @@ -4,6 +4,7 @@ #include "planner/operator/extend/logical_extend.h" #include "planner/operator/logical_hash_join.h" #include "planner/operator/logical_intersect.h" +#include "planner/operator/scan/logical_query_primary_key_lookup.h" #include "planner/operator/scan/logical_scan_node_table.h" using namespace lbug::binder; @@ -58,6 +59,10 @@ void LogicalPlanUtil::encodeRecursive(LogicalOperator* logicalOperator, std::str case LogicalOperatorType::SCAN_NODE_TABLE: { encodeScanNodeTable(logicalOperator, encodeString); } break; + case LogicalOperatorType::QUERY_PRIMARY_KEY_LOOKUP: { + encodeQueryPrimaryKeyLookup(logicalOperator, encodeString); + encodeRecursive(logicalOperator->getChild(0).get(), encodeString); + } break; case LogicalOperatorType::FILTER: { encodeFilter(logicalOperator, encodeString); encodeRecursive(logicalOperator->getChild(0).get(), encodeString); @@ -102,6 +107,12 @@ void LogicalPlanUtil::encodeScanNodeTable(LogicalOperator* logicalOperator, "(" + scan.getNodeID()->constCast().getRawVariableName() + ")"; } +void LogicalPlanUtil::encodeQueryPrimaryKeyLookup(LogicalOperator* logicalOperator, + std::string& encodeString) { + auto& lookup = logicalOperator->constCast(); + encodeString += "LKP(" + lookup.getExpressionsForPrinting() + ")"; +} + void LogicalPlanUtil::encodeFilter(LogicalOperator*, std::string& encodedString) { encodedString += "Filter()"; } diff --git a/src/planner/plan/plan_join_order.cpp b/src/planner/plan/plan_join_order.cpp index 9cd88fa6b..5097b6a6e 100644 --- a/src/planner/plan/plan_join_order.cpp +++ b/src/planner/plan/plan_join_order.cpp @@ -4,8 +4,10 @@ #include "binder/bound_scan_source.h" #include "binder/expression/property_expression.h" #include "binder/expression_visitor.h" +#include "catalog/catalog.h" #include "catalog/catalog_entry/catalog_entry_type.h" #include "catalog/catalog_entry/node_table_catalog_entry.h" +#include "common/enums/extend_direction_util.h" #include "common/enums/join_type.h" #include "common/enums/rel_direction.h" #include "common/enums/table_type.h" @@ -15,12 +17,15 @@ #include "planner/join_order/cost_model.h" #include "planner/join_order/join_plan_solver.h" #include "planner/join_order/join_tree_constructor.h" +#include "planner/operator/extend/base_logical_extend.h" #include "planner/operator/extend/logical_extend.h" +#include "planner/operator/logical_hash_join.h" #include "planner/operator/scan/logical_query_primary_key_lookup.h" #include "planner/operator/scan/logical_scan_node_table.h" #include "planner/planner.h" #include "storage/storage_manager.h" #include "storage/table/node_table.h" +#include "transaction/transaction.h" using namespace lbug::binder; using namespace lbug::common; @@ -871,9 +876,135 @@ static const LogicalScanNodeTable* getBareSingleNodeScan(const SubqueryGraph& su return &scan; } +// Credibility + bounded-fan-out analysis of a lookup-join outer plan. +// +// A lookup join replaces a full scan + hash join with per-tuple probes, so it must only +// fire when the outer side is provably small. The outer cardinality estimate alone is not +// sufficient: it compounds mean fan-outs and NDV guesses that can be off by orders of +// magnitude on skewed multi-hop patterns (observed: 112K-1.5M actual probes behind a +// <=1024 estimate on LDBC IC4/IC5-style shapes). +// +// (b2) Only operators with well-modeled cardinalities may appear in the outer plan: +// scans (exact counts / PK lookups / HLL-NDV equality / min-max ranges), row-preserving +// or row-reducing ops, and inner joins/cross products/unions of trusted children. +// Anything that guesses row counts wholesale (recursion, grouping/aggregation, +// mark/outer joins, unwinds, intersects, misc optimizer/executor ops) disqualifies the +// outer: fail closed. Extend chains are additionally capped at depth 2, since fan-out +// estimation error compounds multiplicatively per hop while the estimate itself has no +// skew model. +// (b3) Each non-recursive extend multiplies the estimate's uncertainty by the observed +// worst/mean fan-out ratio (ANALYZE CSR degree stats). The lookup fires only if the +// uncertainty-discounted estimate still fits comfortably inside the eliminated scan: +// outerCard * uncertainty < innerCard. Missing degree stats fail closed. +struct LookupOuterAnalysis { + bool trusted = true; + double uncertainty = 1.0; + uint32_t maxExtendDepth = 0; +}; + +static double getExtendFanoutUncertainty(const binder::RelExpression& rel, + common::ExtendDirection direction, const CardinalityEstimator& estimator, bool& ok) { + ok = false; + const auto dirs = direction == common::ExtendDirection::BOTH ? + std::vector{common::RelDataDirection::FWD, + common::RelDataDirection::BWD} : + std::vector{ + common::ExtendDirectionUtil::getRelDataDirection(direction)}; + // Fail closed on unknown fan-out: without ANALYZE CSR degree stats there is no + // bound on how many rows one outer row can fan out to. (A ONE-multiplicity rel + // records maxDegree 1, so functional edges contribute a factor of exactly 1.) + auto worstRatio = 1.0; + auto sawStats = false; + for (auto tableID : rel.getInnerRelTableIDs()) { + for (auto relDir : dirs) { + const auto* stats = estimator.getRelDirectionStats(tableID, relDir); + if (stats == nullptr || stats->maxDegree == 0) { + return 1.0; + } + sawStats = true; + worstRatio = std::max(worstRatio, + static_cast(stats->maxDegree) / std::max(stats->avgDegree, 1.0)); + } + } + if (!sawStats) { + return 1.0; + } + ok = true; + return worstRatio; +} + +static void analyzeLookupOuterPlan(const LogicalOperator* op, const CardinalityEstimator& estimator, + uint32_t extendDepth, LookupOuterAnalysis& analysis) { + if (!analysis.trusted) { + return; + } + using T = LogicalOperatorType; + const auto type = op->getOperatorType(); + switch (type) { + case T::EXTEND: + case T::PACKED_EXTEND: { + analysis.maxExtendDepth = std::max(analysis.maxExtendDepth, extendDepth + 1); + const auto& extend = op->constCast(); + bool ok = false; + const auto factor = + getExtendFanoutUncertainty(*extend.getRel(), extend.getDirection(), estimator, ok); + if (!ok) { + analysis.trusted = false; + return; + } + analysis.uncertainty *= factor; + extendDepth += 1; + break; + } + case T::HASH_JOIN: { + if (op->constCast().getJoinType() != common::JoinType::INNER) { + analysis.trusted = false; + return; + } + break; + } + case T::UNION_ALL: { + // Additive branches: the worst branch uncertainty dominates. + auto worst = 1.0; + for (auto i = 0u; i < op->getNumChildren(); ++i) { + LookupOuterAnalysis branch{true, 1.0, extendDepth}; + analyzeLookupOuterPlan(op->getChild(i).get(), estimator, extendDepth, branch); + if (!branch.trusted) { + analysis.trusted = false; + return; + } + worst = std::max(worst, branch.uncertainty); + analysis.maxExtendDepth = std::max(analysis.maxExtendDepth, branch.maxExtendDepth); + } + analysis.uncertainty *= worst; + return; + } + // NOTE: rel scans have no logical operator of their own; rel traversal always + // appears as EXTEND / PACKED_EXTEND, which is handled above. + // A nested lookup preserves row count up to inner-join drops, so reusing its + // cardinality as an upper bound is conservative (it can only make us fire less). + case T::QUERY_PRIMARY_KEY_LOOKUP: + case T::SCAN_NODE_TABLE: + case T::FILTER: + case T::PROJECTION: + case T::FLATTEN: + case T::ORDER_BY: + case T::LIMIT: + case T::CROSS_PRODUCT: + case T::NODE_LABEL_FILTER: + break; + default: + analysis.trusted = false; + return; + } + for (auto i = 0u; i < op->getNumChildren(); ++i) { + analyzeLookupOuterPlan(op->getChild(i).get(), estimator, extendDepth, analysis); + } +} + bool Planner::tryPlanLookupJoin(const SubqueryGraph& innerSubgraph, const LogicalPlan& innerPlan, const LogicalPlan& outerPlan, const std::shared_ptr& joinNode, - const expression_vector& predicates, const SubqueryGraph& newSubgraph) { + const expression_vector& predicates, const SubqueryGraph& newSubgraph, uint64_t maxCost) { if (!clientContext->getClientConfig()->enablePlanOptimizer) { return false; } @@ -885,16 +1016,27 @@ bool Planner::tryPlanLookupJoin(const SubqueryGraph& innerSubgraph, const Logica return false; } auto tableID = scan->getTableIDs()[0]; - auto table = storage::StorageManager::Get(*clientContext) - ->getTable(tableID) - ->ptrCast(); - if (table->tryGetPrimaryKeyIndex() == nullptr) { + if (tableID != joinNode->getTableIDs()[0]) { return false; } + // Per-tuple probes only pay off for small outer cardinalities (INLJ precedent). + // Checked before touching storage, so virtual/partition table IDs without a + // storage entry never throw here. const auto outerCard = outerPlan.getCardinality(); if (outerCard > PlannerKnobs::LOOKUP_JOIN_MAX_OUTER_CARDINALITY) { return false; } + storage::NodeTable* table = nullptr; + try { + table = storage::StorageManager::Get(*clientContext) + ->getTable(tableID) + ->ptrCast(); + } catch (const std::exception&) { + return false; + } + if (table == nullptr || table->tryGetPrimaryKeyIndex() == nullptr) { + return false; + } auto key = joinNode->getInternalID(); if (!outerPlan.getSchema()->evaluable(*key)) { return false; @@ -913,13 +1055,53 @@ bool Planner::tryPlanLookupJoin(const SubqueryGraph& innerSubgraph, const Logica return plan.getSchema()->isExpressionInScope(*expression); }), properties.end()); + // The lookup exists to fetch payload. With no unfetched properties it adds an + // operator for zero benefit while hiding the equi-join from rewrites that match + // on HASH_JOIN (e.g. COUNT_REL_TABLE over COUNT(*)), so stay on the hash path. + if (properties.empty()) { + return false; + } + // All fetched properties must exist in the inner table; otherwise fall back to + // the hash join (the mapper would assert on a missing column). + { + auto* catalog = catalog::Catalog::Get(*clientContext); + auto* tx = transaction::Transaction::Get(*clientContext); + auto* tableEntry = catalog->getTableCatalogEntry(tx, tableID); + for (auto& expression : properties) { + auto& prop = expression->constCast(); + const auto& propName = prop.getPropertyName(); + if (!tableEntry->containsProperty(propName) && !tableEntry->containsProperty("data")) { + return false; + } + } + } + // (b2/b3) credibility + bounded fan-out: the outer estimate must come from + // well-modeled operators, and probing must still win after discounting it by the + // observed worst/mean fan-out of every extend in the outer plan. Failing either + // means the outer side is not provably small -- fall back to the hash join. + { + LookupOuterAnalysis analysis; + analyzeLookupOuterPlan(plan.getLastOperator().get(), cardinalityEstimator, 0, analysis); + if (!analysis.trusted || + analysis.maxExtendDepth > PlannerKnobs::LOOKUP_JOIN_MAX_OUTER_EXTEND_DEPTH) { + return false; + } + const auto discountedOuter = static_cast(outerCard) * analysis.uncertainty; + if (!(discountedOuter < static_cast(innerPlan.getCardinality()))) { + return false; + } + } + const auto lookupCost = CostModel::computeLookupJoinCost(outerPlan, outerCard); + if (lookupCost >= maxCost) { + return false; + } auto lookup = std::make_shared(tableID, joinNode->getInternalID(), properties, key, outputGroupPos, plan.getLastOperator()); lookup->computeFactorizedSchema(); lookup->setCardinality(outerCard); plan.setLastOperator(std::move(lookup)); appendFilters(predicates, plan); - plan.setCost(outerPlan.getCost() + outerCard); + plan.setCost(lookupCost); context.addPlan(newSubgraph, std::move(plan)); return true; } @@ -937,14 +1119,16 @@ void Planner::planInnerHashJoin(const SubqueryGraph& subgraph, const SubqueryGra getNewlyMatchedExprs(subgraph, otherSubgraph, newSubgraph, context.getWhereExpressions()); for (auto& leftPlan : context.getPlans(subgraph)) { for (auto& rightPlan : context.getPlans(otherSubgraph)) { - // Lookup join replaces the hash join for this pair when one side is a bare - // single-node scan of the join node and the other side is small (e.g. Q1's - // Person payload fetch). Like INL joins, it prunes the hash alternatives. - if (joinNodes.size() == 1 && (tryPlanLookupJoin(subgraph, leftPlan, rightPlan, - joinNodes[0], predicates, newSubgraph) || - tryPlanLookupJoin(otherSubgraph, rightPlan, leftPlan, - joinNodes[0], predicates, newSubgraph))) { - continue; + // Lookup join competes with hash join: when one side is a bare single-node + // scan of the join node and the other side is small (e.g. Q1's Person + // payload fetch), add the per-tuple PK lookup plan alongside the hash + // candidates below. Each candidate is maxCost-gated and the cheapest plan + // wins, so hash alternatives are never pruned. + if (joinNodes.size() == 1) { + tryPlanLookupJoin(subgraph, leftPlan, rightPlan, joinNodes[0], predicates, + newSubgraph, maxCost); + tryPlanLookupJoin(otherSubgraph, rightPlan, leftPlan, joinNodes[0], predicates, + newSubgraph, maxCost); } const auto leftProbeCardinality = cardinalityEstimator.estimateHashJoin(joinNodeIDs, leftPlan.getLastOperatorRef(), rightPlan.getLastOperatorRef()); diff --git a/src/processor/operator/persistent/writer/parquet/parquet_writer.cpp b/src/processor/operator/persistent/writer/parquet/parquet_writer.cpp index 19a10d896..964c5d74d 100644 --- a/src/processor/operator/persistent/writer/parquet/parquet_writer.cpp +++ b/src/processor/operator/persistent/writer/parquet/parquet_writer.cpp @@ -287,10 +287,14 @@ void ParquetWriter::flushRowGroup(PreparedRowGroup& rowGroup) { void ParquetWriter::readFromFT(FactorizedTable& ft, std::vector vectorsToRead, uint64_t& numTuplesRead) { - auto numTuplesToRead = - ft.getTableSchema()->getNumUnFlatColumns() != 0 ? - 1 : - std::min(ft.getNumTuples() - numTuplesRead, DEFAULT_VECTOR_CAPACITY); + // Destination vectors are freshly allocated with a flat single-value state, so they can + // only ever hold one tuple: FactorizedTable::lookup asserts + // !(vector->state->isFlat() && numTuplesToRead > 1), and in release builds a wider read + // silently keeps just the first tuple while numTuplesRead still advances past the rest, + // producing pages that disagree with the footer (unreadable files). Multi-row all-flat + // tables are the norm, not the exception (e.g. flat single-row morsels accumulated over + // many batches), so always read one tuple at a time here. + auto numTuplesToRead = uint64_t{1}; ft.scan(vectorsToRead, numTuplesRead, numTuplesToRead); numTuplesRead += numTuplesToRead; } diff --git a/src/storage/table/arrow_node_table.cpp b/src/storage/table/arrow_node_table.cpp index c77fdc76f..e3a6f27ee 100644 --- a/src/storage/table/arrow_node_table.cpp +++ b/src/storage/table/arrow_node_table.cpp @@ -242,6 +242,22 @@ void ArrowNodeTable::copyArrowMorselToOutputVectors(const ArrowArrayWrapper& bat bool ArrowNodeTable::lookupPK([[maybe_unused]] const transaction::Transaction* transaction, common::ValueVector* keyVector, uint64_t vectorPos, common::offset_t& result) const { + // Lookup joins resolve INTERNAL_ID keys directly to offsets (same fast path as + // NodeTable::lookupPK). Arrow tables have no PK index probe for internal IDs. + if (keyVector->dataType.getLogicalTypeID() == common::LogicalTypeID::INTERNAL_ID) { + if (keyVector->isNull(vectorPos)) { + return false; + } + const auto nodeID = keyVector->getValue(vectorPos); + if (nodeID.tableID != tableID || nodeID.offset == common::INVALID_OFFSET) { + return false; + } + if (!isVisibleNoLock(transaction, nodeID.offset)) { + return false; + } + result = nodeID.offset; + return true; + } if (keyVector->isNull(vectorPos)) { return false; } diff --git a/src/storage/table/ice_disk_node_table.cpp b/src/storage/table/ice_disk_node_table.cpp index 5cba42c69..16ef61511 100644 --- a/src/storage/table/ice_disk_node_table.cpp +++ b/src/storage/table/ice_disk_node_table.cpp @@ -331,5 +331,27 @@ bool IceDiskNodeTable::isVisibleNoLock(const transaction::Transaction* transacti return offset < getTotalRowCount(transaction); } +bool IceDiskNodeTable::lookupPK(const transaction::Transaction* transaction, + common::ValueVector* keyVector, uint64_t vectorPos, common::offset_t& result) const { + // Lookup joins resolve INTERNAL_ID keys directly to offsets (same fast path as + // NodeTable::lookupPK). The base implementation checks in-memory node groups, + // which are empty for disk-resident tables, so handle it here via total row count. + if (keyVector->dataType.getLogicalTypeID() == common::LogicalTypeID::INTERNAL_ID) { + if (keyVector->isNull(vectorPos)) { + return false; + } + const auto nodeID = keyVector->getValue(vectorPos); + if (nodeID.tableID != tableID || nodeID.offset == common::INVALID_OFFSET) { + return false; + } + if (!isVisibleNoLock(transaction, nodeID.offset)) { + return false; + } + result = nodeID.offset; + return true; + } + return NodeTable::lookupPK(transaction, keyVector, vectorPos, result); +} + } // namespace storage } // namespace lbug diff --git a/test/optimizer/optimizer_test.cpp b/test/optimizer/optimizer_test.cpp index 4872cf354..8e2daf130 100644 --- a/test/optimizer/optimizer_test.cpp +++ b/test/optimizer/optimizer_test.cpp @@ -356,18 +356,18 @@ TEST_F(OptimizerTest, SingleNodeTwoHopJoins) { // TODO(Guodong/Xiyang): We should make sure the plan is consistent on all platforms. GTEST_SKIP(); #else + // The payload-fetch joins below are small-outer (a single bound row fanning out to a + // few neighbours) over a larger node table, so the lookup join wins the cost + // competition against the hash join and the bare scans disappear into LKP ops. auto q1 = "MATCH (a:person)-[e:knows]->(b:person)-[e2:knows]->(c:person) WHERE b.ID=0 RETURN a,b,c;"; - ASSERT_STREQ(getEncodedPlan(q1).c_str(), - "HJ(c._ID){S(c)}{HJ(a._ID){S(a)}{E(c)E(a)IndexScan(b)}}"); + ASSERT_STREQ(getEncodedPlan(q1).c_str(), "LKP(c._ID)LKP(a._ID)E(c)E(a)IndexScan(b)"); auto q2 = "MATCH (a:person)-[e:knows]->(b:person)-[e2:knows]->(c:person) WHERE a.ID=0 " "RETURN a,b,c;"; - ASSERT_STREQ(getEncodedPlan(q2).c_str(), - "HJ(b._ID){S(b)}{HJ(c._ID){S(c)}{E(c)E(b)IndexScan(a)}}"); + ASSERT_STREQ(getEncodedPlan(q2).c_str(), "LKP(b._ID)LKP(c._ID)E(c)E(b)IndexScan(a)"); auto q3 = "MATCH (a:person)-[e:knows]->(b:person)-[e2:knows]->(c:person) WHERE c.ID=0 " "RETURN a,b,c;"; - ASSERT_STREQ(getEncodedPlan(q3).c_str(), - "HJ(a._ID){S(a)}{HJ(b._ID){S(b)}{E(a)E(b)IndexScan(c)}}"); + ASSERT_STREQ(getEncodedPlan(q3).c_str(), "LKP(b._ID)LKP(a._ID)E(a)E(b)IndexScan(c)"); #endif }