feat(planner): lookup join for inner equi-joins on node ID

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 {

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<LogicalPlan>& 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

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);
 };

src/include/planner/planner.h

@@ -238,6 +238,18 @@ class LBUG_API Planner {
     bool tryPlanPackedINLJoin(const binder::SubqueryGraph& subgraph,
         const binder::SubqueryGraph& otherSubgraph,
         const std::vector<std::shared_ptr<binder::NodeExpression>>& 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, 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<binder::NodeExpression>& joinNode,
+        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<std::shared_ptr<binder::NodeExpression>>& joinNodes, bool flipPlan);

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; }
 

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<LogicalPlan>& buildPlans) {
     uint64_t cost = 0ul;

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<PropertyExpression>().getRawVariableName() + ")";
 }
 
+void LogicalPlanUtil::encodeQueryPrimaryKeyLookup(LogicalOperator* logicalOperator,
+    std::string& encodeString) {
+    auto& lookup = logicalOperator->constCast<LogicalQueryPrimaryKeyLookup>();
+    encodeString += "LKP(" + lookup.getExpressionsForPrinting() + ")";
+}
+
 void LogicalPlanUtil::encodeFilter(LogicalOperator*, std::string& encodedString) {
     encodedString += "Filter()";
 }

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);
     }
 }
 

src/planner/plan/plan_join_order.cpp

@@ -1,7 +1,10 @@
+#include <algorithm>
 #include <cmath>
 
 #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/join_type.h"
@@ -14,8 +17,12 @@
 #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"
+#include "transaction/transaction.h"
 
 using namespace lbug::binder;
 using namespace lbug::common;
@@ -838,6 +845,132 @@ 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<LogicalScanNodeTable>();
+    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<NodeExpression>& joinNode,
+    const expression_vector& predicates, const SubqueryGraph& newSubgraph, uint64_t maxCost) {
+    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];
+    if (tableID != joinNode->getTableIDs()[0]) {
+        return false;
+    }
+    // Per-tuple probes only pay off for small outer cardinalities (INLJ precedent),
+    // and only when probing is cheaper than the scan it eliminates: the outer row
+    // count must be below both the absolute probe budget and the inner table size.
+    // The second condition is what keeps full-coverage joins on the hash path: the
+    // metadata rewrites (COUNT_REL_TABLE, REL_DEGREE_TABLE, reachability) match on
+    // HASH_JOIN legs over unfiltered scans as proof of full-table coverage, and a
+    // lookup plan consumes that scan. When the outer side already spans (most of)
+    // the inner table, probing buys nothing and must not hide the join.
+    // 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;
+    }
+    if (outerCard >= innerPlan.getCardinality()) {
+        return false;
+    }
+    storage::NodeTable* table = nullptr;
+    try {
+        table = storage::StorageManager::Get(*clientContext)
+                    ->getTable(tableID)
+                    ->ptrCast<storage::NodeTable>();
+    } 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;
+    }
+    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>& expression) {
+                             if (expression->constCast<PropertyExpression>().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());
+    // 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<PropertyExpression>();
+            const auto& propName = prop.getPropertyName();
+            if (!tableEntry->containsProperty(propName) && !tableEntry->containsProperty("data")) {
+                return false;
+            }
+        }
+    }
+    const auto lookupCost = CostModel::computeLookupJoinCost(outerPlan, outerCard);
+    if (lookupCost >= maxCost) {
+        return false;
+    }
+    auto lookup = std::make_shared<LogicalQueryPrimaryKeyLookup>(tableID, joinNode->getInternalID(),
+        properties, key, outputGroupPos, plan.getLastOperator());
+    lookup->computeFactorizedSchema();
+    lookup->setCardinality(outerCard);
+    plan.setLastOperator(std::move(lookup));
+    appendFilters(predicates, plan);
+    plan.setCost(lookupCost);
+    context.addPlan(newSubgraph, std::move(plan));
+    return true;
+}
+
 void Planner::planInnerHashJoin(const SubqueryGraph& subgraph, const SubqueryGraph& otherSubgraph,
     const std::vector<std::shared_ptr<NodeExpression>>& joinNodes, bool flipPlan) {
     auto newSubgraph = subgraph;
@@ -851,6 +984,17 @@ 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 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());
             if (CostModel::computeHashJoinCost(joinNodeIDs, leftPlan, rightPlan,

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<ValueVector*> vectorsToRead,
     uint64_t& numTuplesRead) {
-    auto numTuplesToRead =
-        ft.getTableSchema()->getNumUnFlatColumns() != 0 ?
-            1 :
-            std::min<uint64_t>(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;
 }

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<common::nodeID_t>(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;
     }

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<common::nodeID_t>(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