diff --git a/common/CMakeLists.txt b/common/CMakeLists.txt index df8cd5a518..1ac6524b96 100644 --- a/common/CMakeLists.txt +++ b/common/CMakeLists.txt @@ -22,6 +22,7 @@ set(common_sources Profiler.cpp ThreadName.cpp HybridString.cpp + SegmentationSolver.cpp log/LogSetup.cpp log/LogUtils.cpp) diff --git a/common/SegmentationSolver.cpp b/common/SegmentationSolver.cpp new file mode 100644 index 0000000000..2174d4c9e2 --- /dev/null +++ b/common/SegmentationSolver.cpp @@ -0,0 +1,95 @@ +#include "SegmentationSolver.h" + +#include + +using namespace db; + +SegmentationSolver::SegmentationSolver() +{ +} + +SegmentationSolver::~SegmentationSolver() { +} + +void SegmentationSolver::clear() { + _lines.clear(); + _firstLiveLine = 0; + _lastSegmentStarts.clear(); +} + +void SegmentationSolver::addBoundary(const Boundary& boundary) { + const size_t index = _lastSegmentStarts.size(); + + double cheapestCost = 0; + if (index == 0) { + _lastSegmentStarts.push_back(0); + } else { + const StartLine& cheapest = findCheapestLine(boundary._position); + cheapestCost = evaluate(cheapest, boundary._position) + boundary._endCost; + _lastSegmentStarts.push_back(cheapest._boundary); + + if (!boundary._cuttable) { + return; + } + } + + const double intercept = cheapestCost + boundary._startCost - boundary._weight * boundary._position; + addLine(StartLine {boundary._weight, intercept, index}); +} + +void SegmentationSolver::getSegmentStarts(std::vector& starts) const { + starts.clear(); + if (_lastSegmentStarts.empty()) { + return; + } + + for (size_t end = _lastSegmentStarts.size() - 1; end > 0; end = _lastSegmentStarts[end]) { + starts.push_back(_lastSegmentStarts[end]); + } + + std::reverse(starts.begin(), starts.end()); +} + +// The lines arrive by falling slope, so the new one is the cheapest for the furthest ends. A +// line before it never is again once the new one undercuts it where it would itself start to +// undercut the line before it. +void SegmentationSolver::addLine(const StartLine& line) { + while (_lines.size() > _firstLiveLine) { + const StartLine& last = _lines.back(); + if (last._slope == line._slope) { + if (last._intercept <= line._intercept) { + return; + } + + _lines.pop_back(); + continue; + } else if (_lines.size() - _firstLiveLine < 2) { + break; + } + + const StartLine& beforeLast = _lines[_lines.size() - 2]; + const double newCrossing = (line._intercept - beforeLast._intercept) * (beforeLast._slope - last._slope); + const double lastCrossing = (last._intercept - beforeLast._intercept) * (beforeLast._slope - line._slope); + if (newCrossing > lastCrossing) { + break; + } + + _lines.pop_back(); + } + + _lines.push_back(line); +} + +// Positions only grow, so a line a later one undercuts here stays undercut +const SegmentationSolver::StartLine& SegmentationSolver::findCheapestLine(double position) { + while (_firstLiveLine + 1 < _lines.size() + && evaluate(_lines[_firstLiveLine + 1], position) <= evaluate(_lines[_firstLiveLine], position)) { + _firstLiveLine++; + } + + return _lines[_firstLiveLine]; +} + +double SegmentationSolver::evaluate(const StartLine& line, double position) { + return line._slope * position + line._intercept; +} diff --git a/common/SegmentationSolver.h b/common/SegmentationSolver.h new file mode 100644 index 0000000000..0d7066b77b --- /dev/null +++ b/common/SegmentationSolver.h @@ -0,0 +1,55 @@ +#pragma once + +#include + +#include + +namespace db { + +// The cheapest cut of boundaries 0..n into segments [start, end), a segment costing +// weight(start) * (position(end) - position(start)) + startCost(start) + endCost(end), with weight +// never rising and position never falling. Each start is a line in the end's position, and the +// cheapest start for every end lies on their lower envelope (the convex hull trick). +class SegmentationSolver { +public: + struct Boundary { + double _position {0}; + double _weight {0}; + double _startCost {0}; + double _endCost {0}; + bool _cuttable {true}; + }; + + SegmentationSolver(); + ~SegmentationSolver(); + + // Forgets the boundaries, keeping the memory for the next sequence + void clear(); + + // A segment starts at the first boundary and one ends at the last. In between, a segment + // may start where `_cuttable` is set. + void addBoundary(const Boundary& boundary); + + // The boundary each segment of the cheapest segmentation starts at, in order + void getSegmentStarts(std::vector& starts) const; + +private: + // A segment starting at `_boundary`: its cost, plus the cheapest segmentation before it, is + // `_slope * position + _intercept` for an end at `position` + struct StartLine { + double _slope {0}; + double _intercept {0}; + size_t _boundary {0}; + }; + + std::vector _lines; + size_t _firstLiveLine {0}; + std::vector _lastSegmentStarts; + + void addLine(const StartLine& line); + const StartLine& findCheapestLine(double position); + + static double evaluate(const StartLine& line, double position); +}; + +} diff --git a/query/ir/codegen/DBPassPipeline.cpp b/query/ir/codegen/DBPassPipeline.cpp index 8093c417da..1d1d1e6f1e 100644 --- a/query/ir/codegen/DBPassPipeline.cpp +++ b/query/ir/codegen/DBPassPipeline.cpp @@ -10,7 +10,7 @@ namespace { using DBPassFactory = std::unique_ptr (*)(const mlir::db::DBPassContext*); -constexpr size_t DB_PASS_COUNT = 52; +constexpr size_t DB_PASS_COUNT = 53; // The optimisation pipeline every query runs through, in order. An EXPLAIN prefix // reporting on a pass walks the same table one pass at a time, which is what keeps the @@ -70,6 +70,7 @@ const std::array dbPassPipeline = { [](const mlir::db::DBPassContext* context) { return mlir::db::createCountFromMetadata(context); }, [](const mlir::db::DBPassContext*) { return mlir::db::createFuseUnwindRange(); }, [](const mlir::db::DBPassContext*) { return mlir::db::createSinkUnwindSource(); }, + [](const mlir::db::DBPassContext* context) { return mlir::db::createStageFilterConjuncts(context); }, }; } diff --git a/query/ir/dialect/db/DBInterfaces.td b/query/ir/dialect/db/DBInterfaces.td index 0d95420dc5..a3e73264e9 100644 --- a/query/ir/dialect/db/DBInterfaces.td +++ b/query/ir/dialect/db/DBInterfaces.td @@ -64,6 +64,22 @@ def ColumnCarrierOpInterface : OpInterface<"ColumnCarrierOpInterface"> { ]; } +def RowCostOpInterface : OpInterface<"RowCostOpInterface"> { + let cppNamespace = "::mlir::db"; + + let description = [{ + An op computing a row of its result at a cost other than a comparison of two scalar + columns, which is 1 on this scale and the cost of every op without the interface. A + random read of node or edge storage costs about 10. + }]; + + let methods = [ + InterfaceMethod< + "The cost of computing one row of the op's result", + "double", "getRowCost", (ins)> + ]; +} + def EdgeTypesOutputOpInterface : OpInterface<"EdgeTypesOutputOpInterface"> { let cppNamespace = "::mlir::db"; diff --git a/query/ir/dialect/db/DBOps.cpp b/query/ir/dialect/db/DBOps.cpp index 9d8c38d29a..2adec99573 100644 --- a/query/ir/dialect/db/DBOps.cpp +++ b/query/ir/dialect/db/DBOps.cpp @@ -35,6 +35,12 @@ namespace storage = mlir::storage; namespace { +constexpr double STORAGE_READ_ROW_COST = 10.0; +constexpr double STRING_TEST_ROW_COST = 3.0; +constexpr double MEMBERSHIP_TEST_ROW_COST = 5.0; +constexpr double CONTAINER_BUILD_ROW_COST = 15.0; +constexpr double VECTOR_FUNCTION_ROW_COST = 40.0; + Type columnElement(Value column) { return ::db::inferredColumnElement(column); } @@ -3401,3 +3407,51 @@ Speculation::Speculatability ModOp::getSpeculatability() { Speculation::Speculatability PowOp::getSpeculatability() { return speculatableOverNumbers(getOperation()); } + +double GetNodeLabelSet::getRowCost() { + return STORAGE_READ_ROW_COST; +} + +double GetEdgeTypes::getRowCost() { + return STORAGE_READ_ROW_COST; +} + +double GetNodeProperties::getRowCost() { + return STORAGE_READ_ROW_COST; +} + +double GetEdgeProperties::getRowCost() { + return STORAGE_READ_ROW_COST; +} + +double StartsWithOp::getRowCost() { + return STRING_TEST_ROW_COST; +} + +double EndsWithOp::getRowCost() { + return STRING_TEST_ROW_COST; +} + +double ContainsOp::getRowCost() { + return STRING_TEST_ROW_COST; +} + +double InOp::getRowCost() { + return MEMBERSHIP_TEST_ROW_COST; +} + +double MakeList::getRowCost() { + return CONTAINER_BUILD_ROW_COST; +} + +double MakeMap::getRowCost() { + return CONTAINER_BUILD_ROW_COST; +} + +double CosineSimilarity::getRowCost() { + return VECTOR_FUNCTION_ROW_COST; +} + +double EuclideanDistance::getRowCost() { + return VECTOR_FUNCTION_ROW_COST; +} diff --git a/query/ir/dialect/db/DBOps.td b/query/ir/dialect/db/DBOps.td index 9b5dc7e988..5e480fa146 100644 --- a/query/ir/dialect/db/DBOps.td +++ b/query/ir/dialect/db/DBOps.td @@ -488,7 +488,7 @@ def UnwindYield : TuringOp<"unwind_yield", [Pure, Terminator, HasParent<"UnwindC /** * MakeList */ -def MakeList : TuringOp<"make_list", [Pure, ConstantThroughOperands]> { +def MakeList : TuringOp<"make_list", [Pure, ConstantThroughOperands, DeclareOpInterfaceMethods]> { let summary = "Build one list per row out of the row's cell in each element column"; let description = [{ @@ -534,7 +534,7 @@ def MakeList : TuringOp<"make_list", [Pure, ConstantThroughOperands]> { /** * MakeMap */ -def MakeMap : TuringOp<"make_map", [Pure, ConstantThroughOperands]> { +def MakeMap : TuringOp<"make_map", [Pure, ConstantThroughOperands, DeclareOpInterfaceMethods]> { let summary = "Build one map per row out of the row's cell in each value column"; let description = [{ @@ -1975,7 +1975,7 @@ def PathElements : TuringOp<"path_elements", [NoMemoryEffect, /** * GetNodeProperties */ -def GetNodeProperties : TuringOp<"get_node_properties", [Pure, ConstantThroughOperands]> { +def GetNodeProperties : TuringOp<"get_node_properties", [Pure, ConstantThroughOperands, DeclareOpInterfaceMethods]> { let summary = "Read one named property of each node as a column of values"; let description = [{ @@ -2031,7 +2031,7 @@ def GetNodeProperties : TuringOp<"get_node_properties", [Pure, ConstantThroughOp /** * GetEdgeProperties */ -def GetEdgeProperties : TuringOp<"get_edge_properties", [Pure, ConstantThroughOperands]> { +def GetEdgeProperties : TuringOp<"get_edge_properties", [Pure, ConstantThroughOperands, DeclareOpInterfaceMethods]> { let summary = "Read one named property of each edge as a column of values"; let description = [{ @@ -2065,7 +2065,7 @@ def GetEdgeProperties : TuringOp<"get_edge_properties", [Pure, ConstantThroughOp }]; } -def GetNodeLabelSet : TuringOp<"get_node_label_set", [Pure]> { +def GetNodeLabelSet : TuringOp<"get_node_label_set", [Pure, DeclareOpInterfaceMethods]> { let summary = "Read the label-set ID of each node as a column"; let description = [{ @@ -2083,7 +2083,7 @@ def GetNodeLabelSet : TuringOp<"get_node_label_set", [Pure]> { }]; } -def GetEdgeTypes : TuringOp<"get_edge_types", [Pure]> { +def GetEdgeTypes : TuringOp<"get_edge_types", [Pure, DeclareOpInterfaceMethods]> { let summary = "Read the type ID of each edge as a column"; let description = [{ @@ -3269,7 +3269,7 @@ def PointWithinBBox : TuringOp<"point_within_bbox", [MayRaise, ConstantThroughOp }]; } -class BinaryFunctionOp : TuringOp { +class BinaryFunctionOp traits = []> : TuringOp { let summary = "Apply a binary function row-wise to two value columns"; let arguments = (ins Column:$lhs, Column:$rhs); @@ -3280,8 +3280,8 @@ class BinaryFunctionOp : TuringOp; -def EuclideanDistance : BinaryFunctionOp<"euclidean_distance">; +def CosineSimilarity : BinaryFunctionOp<"cosine_similarity", [DeclareOpInterfaceMethods]>; +def EuclideanDistance : BinaryFunctionOp<"euclidean_distance", [DeclareOpInterfaceMethods]>; def ConstantOp : TuringOp<"constant", [Pure, ConstantLike, DeclareOpInterfaceMethods]> { let summary = "Produces a constant value in a column"; @@ -3630,19 +3630,19 @@ def LteOp : PredicateOp<"lte", [Pure]> { let summary = "Row-wise less-than-or-equal comparison of two value columns"; } -def StartsWithOp : PredicateOp<"starts_with", [Pure]> { +def StartsWithOp : PredicateOp<"starts_with", [Pure, DeclareOpInterfaceMethods]> { let summary = "Row-wise test of whether each string column value starts with the given prefix"; } -def EndsWithOp : PredicateOp<"ends_with", [Pure]> { +def EndsWithOp : PredicateOp<"ends_with", [Pure, DeclareOpInterfaceMethods]> { let summary = "Row-wise test of whether each string column value ends with the given suffix"; } -def ContainsOp : PredicateOp<"contains", [Pure]> { +def ContainsOp : PredicateOp<"contains", [Pure, DeclareOpInterfaceMethods]> { let summary = "Row-wise test of whether each string column value contains the given substring"; } -def InOp : PredicateOp<"in", [Pure]> { +def InOp : PredicateOp<"in", [Pure, DeclareOpInterfaceMethods]> { let summary = "Row-wise test of whether each value is an element of the given list"; let description = [{ diff --git a/query/ir/dialect/db/DBPasses.cpp b/query/ir/dialect/db/DBPasses.cpp index f174100f2f..9d6c3b9e49 100644 --- a/query/ir/dialect/db/DBPasses.cpp +++ b/query/ir/dialect/db/DBPasses.cpp @@ -3,11 +3,14 @@ #include #include #include +#include +#include #include #include #include #include #include +#include #include "mlir/Dialect/Func/IR/FuncOps.h" #include "mlir/IR/Builders.h" @@ -44,12 +47,14 @@ #include "DBOps.h" #include "BioAssert.h" +#include "SegmentationSolver.h" namespace mlir::db { #define GEN_PASS_DEF_FUSESCANBYLABEL #define GEN_PASS_DEF_PUSHDOWNFILTERS #define GEN_PASS_DEF_MERGEFILTERS +#define GEN_PASS_DEF_STAGEFILTERCONJUNCTS #define GEN_PASS_DEF_FUSEUNWINDEQUALITY #define GEN_PASS_DEF_FUSEUNWINDRANGE #define GEN_PASS_DEF_UNNESTUNWINDCOMPREHENSION @@ -9614,6 +9619,56 @@ class RowEstimator { return kept; } + // A flat OR or AND chain is as long as the query, so the walk keeps its own stack + double getPredicateSelectivity(Value predicate, llvm::function_ref nodeOf, const WalkKnowledge& knowledge) const { + llvm::DenseMap kept; + llvm::SmallVector pending {predicate}; + + while (!pending.empty()) { + const Value current = pending.back(); + if (kept.count(current)) { + pending.pop_back(); + continue; + } + + Operation* const def = current.getDefiningOp(); + if (!isa_and_nonnull(def)) { + kept[current] = getLeafSelectivity(current, nodeOf, knowledge); + pending.pop_back(); + continue; + } + + bool operandsPending = false; + for (const Value operand : def->getOperands()) { + if (!kept.count(operand)) { + pending.push_back(operand); + operandsPending = true; + } + } + + if (operandsPending) { + continue; + } + + const double first = kept.lookup(def->getOperand(0)); + const double second = def->getNumOperands() > 1 ? kept.lookup(def->getOperand(1)) : 0; + + if (isa(def)) { + kept[current] = first * second; + } else if (isa(def)) { + kept[current] = first + second - first * second; + } else if (isa(def)) { + kept[current] = first + second - 2 * first * second; + } else { + kept[current] = 1 - first; + } + + pending.pop_back(); + } + + return kept.lookup(predicate); + } + private: // The edge counts and degree products asked for, read off the graph once each: the walks // the estimate compares ask for the same ones over and over @@ -9903,56 +9958,6 @@ class RowEstimator { return 1; } - - // A flat OR or AND chain is as long as the query, so the walk keeps its own stack - double getPredicateSelectivity(Value predicate, llvm::function_ref nodeOf, const WalkKnowledge& knowledge) const { - llvm::DenseMap kept; - llvm::SmallVector pending {predicate}; - - while (!pending.empty()) { - const Value current = pending.back(); - if (kept.count(current)) { - pending.pop_back(); - continue; - } - - Operation* const def = current.getDefiningOp(); - if (!isa_and_nonnull(def)) { - kept[current] = getLeafSelectivity(current, nodeOf, knowledge); - pending.pop_back(); - continue; - } - - bool operandsPending = false; - for (const Value operand : def->getOperands()) { - if (!kept.count(operand)) { - pending.push_back(operand); - operandsPending = true; - } - } - - if (operandsPending) { - continue; - } - - const double first = kept.lookup(def->getOperand(0)); - const double second = def->getNumOperands() > 1 ? kept.lookup(def->getOperand(1)) : 0; - - if (isa(def)) { - kept[current] = first * second; - } else if (isa(def)) { - kept[current] = first + second - first * second; - } else if (isa(def)) { - kept[current] = first + second - 2 * first * second; - } else { - kept[current] = 1 - first; - } - - pending.pop_back(); - } - - return kept.lookup(predicate); - } }; // The rows an op seeds a factor with, and nothing at all when it seeds none - a fetch, a @@ -12872,6 +12877,501 @@ struct CountFromMetadata : public impl::CountFromMetadataBase const DBPassContext* _context {&defaultPassContext}; }; +// A filter on RowCostOpInterface's scale. It reads its mask on every row it tests and copies each +// column it carries on every row it keeps. Its fixed cost per chunk, spread over the rows of a +// full one, keeps a stage from paying for itself on rows the stages before it all but emptied. +constexpr double FILTER_ROW_COST = 1.0; +constexpr double FILTER_COPY_COST = 1.0; +constexpr double FILTER_STAGE_COST = 0.05; + +// A constant is computed once, not per row +double rowCost(Operation* op) { + if (op->hasTrait()) { + return 0; + } else if (RowCostOpInterface costed = dyn_cast(op)) { + return costed.getRowCost(); + } + + return 1; +} + +// A conjunct of a filter's mask: the share of rows it keeps, its rank, and the cost of the ops +// it is the first conjunct to need. Its cone is a run of the planner's cone ops. +struct RankedConjunct { + Value _value; + double _selectivity {1.0}; + double _rank {0.0}; + double _ownCost {0.0}; + size_t _coneBegin {0}; + size_t _coneEnd {0}; +}; + +// Where an op of the cones is computed: in the stage of the first conjunct needing it, or before +// the filter when an op besides the conjuncts reads it or one of the ops it feeds +struct ConePlacement { + size_t _position {0}; + bool _computedBefore {false}; +}; + +// A value read at a later conjunct than the one it is computed at: the stage boundaries from +// `_crossesFrom` to `_lastRead` have their filter carry it, 0 standing for a value computed +// before the filter +struct StageCrossing { + Value _value; + size_t _crossesFrom {0}; + size_t _lastRead {0}; + bool _carriable {false}; +}; + +// How the count of values crossing a boundary changes from the boundary before +struct CrossingChange { + int64_t _carried {0}; + int64_t _blocked {0}; +}; + +// Where a stage ends in each list of its filter's stages +struct StageEnd { + size_t _conjuncts {0}; + size_t _ops {0}; + size_t _carried {0}; +}; + +// The stages a filter splits into, each over the rows the ones before it keep: its conjuncts, +// the ops a later stage computes after the filter before it, and the columns each stage's filter +// carries beside the filter's own +struct FilterStages { + FilterOp _filter; + llvm::SmallVector _conjuncts; + MaskCone _moved; + llvm::SmallVector _carried; + llvm::SmallVector _stageEnds; +}; + +// Plans the stages of one filter after another, reusing its buffers from one to the next +class StagePlanner { +public: + StagePlanner(const RowEstimator& estimator, const WalkKnowledge& knowledge) + : _estimator(estimator), + _knowledge(knowledge) + { + } + + bool plan(FilterOp filter, FilterStages& stages); + +private: + const RowEstimator& _estimator; + const WalkKnowledge& _knowledge; + ConeCache _cache; + ::db::SegmentationSolver _solver; + llvm::SmallPtrSet _conjunction; + llvm::SmallVector _conjuncts; + MaskCone _cones; + llvm::SmallPtrSet _visited; + llvm::DenseMap _placements; + llvm::SmallVector _pending; + llvm::SmallVector _crossings; + llvm::DenseMap _crossingIndices; + llvm::SmallVector _crossingChanges; + std::vector _stageStarts; + + bool collectConjunction(Value mask); + void rankConjuncts(Block* block); + void placeConeOps(); + void collectCrossings(FilterOp filter); + void chooseStageStarts(size_t columnCount); + void fillStages(FilterOp filter, FilterStages& stages); + + llvm::ArrayRef getCone(const RankedConjunct& conjunct) const { + return llvm::ArrayRef(_cones._ops).slice(conjunct._coneBegin, conjunct._coneEnd - conjunct._coneBegin); + } + + // The position of the conjunct whose stage computes @param op, false where none does + bool findStageOf(Operation* op, size_t& position) const; +}; + +bool StagePlanner::plan(FilterOp filter, FilterStages& stages) { + const Value mask = filter.getMask(); + if (!mask.hasOneUse() || !collectConjunction(mask) || _conjuncts.size() < 2) { + return false; + } + + rankConjuncts(filter->getBlock()); + placeConeOps(); + collectCrossings(filter); + chooseStageStarts(filter.getColumnsToFilter().size()); + if (_stageStarts.size() < 2) { + return false; + } + + fillStages(filter, stages); + return true; +} + +// The conjuncts of @param mask, left to right, and the `and`s joining them. False where one of +// the `and`s has a reader besides the filter: its conjuncts are then computed on every row +// whatever the filter keeps. +bool StagePlanner::collectConjunction(Value mask) { + _conjunction.clear(); + _conjuncts.clear(); + + llvm::SmallVector pending {mask}; + while (!pending.empty()) { + const Value conjunct = pending.pop_back_val(); + AndOp conjunction = conjunct.getDefiningOp(); + if (!conjunction) { + _conjuncts.push_back(RankedConjunct {conjunct}); + continue; + } else if (!conjunct.hasOneUse()) { + return false; + } + + _conjunction.insert(conjunction); + pending.push_back(conjunction.getRhs()); + pending.push_back(conjunction.getLhs()); + } + + return true; +} + +// Orders the conjuncts by Hellerstein's rank, (selectivity - 1) / cost, which puts the cheap and +// selective ones first +void StagePlanner::rankConjuncts(Block* block) { + const auto computedInTheBlock = [block](Operation* op) { + return op->getBlock() == block && computesFromTheRow(op) && !raisesOnSomeRows(op); + }; + + const auto nodeOf = [this](Value column) { + return _cache.climbFilters(column); + }; + + _cones._ops.clear(); + _cones._inputs.clear(); + for (RankedConjunct& conjunct : _conjuncts) { + _visited.clear(); + conjunct._coneBegin = _cones._ops.size(); + collectConePostOrder(conjunct._value, computedInTheBlock, _visited, _cones); + conjunct._coneEnd = _cones._ops.size(); + + double cost = 0; + for (Operation* const op : getCone(conjunct)) { + cost += rowCost(op); + } + + conjunct._selectivity = _estimator.getPredicateSelectivity(conjunct._value, nodeOf, _knowledge); + conjunct._rank = cost == 0 ? -std::numeric_limits::infinity() : (conjunct._selectivity - 1.0) / cost; + } + + std::stable_sort(_conjuncts.begin(), _conjuncts.end(), [](const RankedConjunct& lhs, const RankedConjunct& rhs) { + return lhs._rank < rhs._rank; + }); +} + +void StagePlanner::placeConeOps() { + _placements.clear(); + for (size_t position = 0; position < _conjuncts.size(); position++) { + for (Operation* const op : getCone(_conjuncts[position])) { + _placements.try_emplace(op, ConePlacement {position, false}); + } + } + + _pending.clear(); + for (size_t position = 0; position < _conjuncts.size(); position++) { + for (Operation* const op : getCone(_conjuncts[position])) { + ConePlacement& placement = _placements.find(op)->second; + if (placement._position != position || placement._computedBefore) { + continue; + } + + const bool readByAnotherOp = llvm::any_of(op->getUsers(), [this](Operation* user) { + return !_placements.contains(user) && !_conjunction.contains(user); + }); + + if (readByAnotherOp) { + placement._computedBefore = true; + _pending.push_back(op); + } + } + } + + while (!_pending.empty()) { + Operation* const op = _pending.pop_back_val(); + for (const Value operand : op->getOperands()) { + const auto found = _placements.find(operand.getDefiningOp()); + if (found != _placements.end() && !found->second._computedBefore) { + found->second._computedBefore = true; + _pending.push_back(found->first); + } + } + } + + for (size_t position = 0; position < _conjuncts.size(); position++) { + RankedConjunct& conjunct = _conjuncts[position]; + for (Operation* const op : getCone(conjunct)) { + size_t stage = 0; + if (findStageOf(op, stage) && stage == position) { + conjunct._ownCost += rowCost(op); + } + } + } +} + +bool StagePlanner::findStageOf(Operation* op, size_t& position) const { + const auto found = _placements.find(op); + if (found == _placements.end() || found->second._computedBefore) { + return false; + } + + position = found->second._position; + return true; +} + +// The values a conjunct reads, through its own value or the ops its stage computes for it, that +// a conjunct before it or the code before the filter computes +void StagePlanner::collectCrossings(FilterOp filter) { + _crossings.clear(); + _crossingIndices.clear(); + + Block* const block = filter->getBlock(); + const Operation::operand_range columns = filter.getColumnsToFilter(); + const auto noteRead = [this, block, &columns](Value value, size_t position) { + if (_cache.yieldsConstantColumn(value) || llvm::is_contained(columns, value)) { + return; + } + + size_t stage = 0; + const size_t crossesFrom = findStageOf(value.getDefiningOp(), stage) ? stage + 1 : 0; + if (crossesFrom > position) { + return; + } + + const auto [found, inserted] = _crossingIndices.try_emplace(value, _crossings.size()); + if (inserted) { + const bool carriable = isa(value.getType()) && value.getParentBlock() == block; + _crossings.push_back(StageCrossing {value, crossesFrom, position, carriable}); + } else { + StageCrossing& crossing = _crossings[found->second]; + crossing._lastRead = std::max(crossing._lastRead, position); + } + }; + + for (size_t position = 0; position < _conjuncts.size(); position++) { + const RankedConjunct& conjunct = _conjuncts[position]; + noteRead(conjunct._value, position); + + for (Operation* const op : getCone(conjunct)) { + size_t stage = 0; + if (!findStageOf(op, stage) || stage != position) { + continue; + } + + for (const Value operand : op->getOperands()) { + noteRead(operand, position); + } + } + } +} + +bool crossesBoundary(const StageCrossing& crossing, size_t boundary) { + return std::max(1, crossing._crossesFrom) <= boundary && boundary <= crossing._lastRead; +} + +// Stages of the ranked conjuncts are segments between boundaries: a stage starting at conjunct i +// computes conjuncts i to j - 1 and runs a filter on the rows reaching i, and its filter copies +// the filter's columns and those a later stage reads on the rows reaching j +void StagePlanner::chooseStageStarts(size_t columnCount) { + const size_t count = _conjuncts.size(); + + _crossingChanges.assign(count + 1, CrossingChange {}); + for (const StageCrossing& crossing : _crossings) { + const size_t first = std::max(1, crossing._crossesFrom); + if (first > crossing._lastRead) { + continue; + } + + _crossingChanges[first]._carried++; + _crossingChanges[crossing._lastRead + 1]._carried--; + if (!crossing._carriable) { + _crossingChanges[first]._blocked++; + _crossingChanges[crossing._lastRead + 1]._blocked--; + } + } + + _solver.clear(); + double rows = 1.0; + double spent = 0.0; + int64_t carried = 0; + int64_t blocked = 0; + for (size_t boundary = 0; boundary <= count; boundary++) { + if (boundary > 0) { + const RankedConjunct& conjunct = _conjuncts[boundary - 1]; + const CrossingChange& change = _crossingChanges[boundary]; + rows *= conjunct._selectivity; + spent += conjunct._ownCost; + carried += change._carried; + blocked += change._blocked; + } + + const bool isLast = boundary == count; + const double copiedColumns = static_cast(columnCount + static_cast(carried)); + + ::db::SegmentationSolver::Boundary stageBoundary; + stageBoundary._position = spent; + stageBoundary._weight = rows; + stageBoundary._startCost = rows * FILTER_ROW_COST + FILTER_STAGE_COST; + stageBoundary._endCost = isLast ? 0.0 : rows * FILTER_COPY_COST * copiedColumns; + stageBoundary._cuttable = !isLast && blocked == 0; + _solver.addBoundary(stageBoundary); + } + + _solver.getSegmentStarts(_stageStarts); +} + +void StagePlanner::fillStages(FilterOp filter, FilterStages& stages) { + stages._filter = filter; + for (const RankedConjunct& conjunct : _conjuncts) { + stages._conjuncts.push_back(conjunct._value); + } + + _visited.clear(); + for (size_t index = 0; index < _stageStarts.size(); index++) { + const size_t begin = _stageStarts[index]; + const size_t end = index + 1 < _stageStarts.size() ? _stageStarts[index + 1] : _conjuncts.size(); + + if (begin > 0) { + const auto computedInTheStage = [this, begin, end](Operation* op) { + size_t stage = 0; + return findStageOf(op, stage) && stage >= begin && stage < end; + }; + + for (size_t position = begin; position < end; position++) { + collectConePostOrder(_conjuncts[position]._value, computedInTheStage, _visited, stages._moved); + } + } + + if (end < _conjuncts.size()) { + for (const StageCrossing& crossing : _crossings) { + if (crossesBoundary(crossing, end)) { + stages._carried.push_back(crossing._value); + } + } + } + + stages._stageEnds.push_back(StageEnd {end, stages._moved._ops.size(), stages._carried.size()}); + } +} + +void stageConjuncts(const FilterStages& stages, mlir::RewriterBase& rewriter) { + FilterOp filter = stages._filter; + const Location loc = filter.getLoc(); + const Value mask = filter.getMask(); + const Operation::operand_range columns = filter.getColumnsToFilter(); + const llvm::ArrayRef conjuncts = stages._conjuncts; + const llvm::ArrayRef moved = stages._moved._ops; + const llvm::ArrayRef carried = stages._carried; + + mlir::IRMapping mapping; + rewriter.setInsertionPoint(filter); + + llvm::SmallVector stageMasks; + llvm::SmallVector operands; + llvm::SmallVector types; + const auto addOperand = [&mapping, &operands, &types](Value column) { + const Value operand = mapping.lookupOrDefault(column); + operands.push_back(operand); + types.push_back(operand.getType()); + }; + + StageEnd stageStart; + FilterOp stageFilter; + for (const StageEnd& stageEnd : stages._stageEnds) { + for (Operation* const op : moved.slice(stageStart._ops, stageEnd._ops - stageStart._ops)) { + rewriter.clone(*op, mapping); + } + + stageMasks.clear(); + for (const Value conjunct : conjuncts.slice(stageStart._conjuncts, stageEnd._conjuncts - stageStart._conjuncts)) { + stageMasks.push_back(mapping.lookupOrDefault(conjunct)); + } + + const llvm::ArrayRef stageCarried = carried.slice(stageStart._carried, stageEnd._carried - stageStart._carried); + operands.clear(); + types.clear(); + for (const Value column : columns) { + addOperand(column); + } + + for (const Value column : stageCarried) { + addOperand(column); + } + + const Value stageMask = conjoin(stageMasks, loc, rewriter); + stageFilter = rewriter.create(loc, types, stageMask, operands); + + const ResultRange kept = stageFilter.getFilteredColumns(); + const size_t columnCount = columns.size(); + for (size_t index = 0; index < columnCount; index++) { + mapping.map(columns[index], kept[index]); + } + + for (size_t index = 0; index < stageCarried.size(); index++) { + mapping.map(stageCarried[index], kept[columnCount + index]); + } + + stageStart = stageEnd; + } + + rewriter.replaceOp(filter, stageFilter.getResults()); + eraseUnusedConjunction(mask, rewriter); + + for (Operation* const op : llvm::reverse(moved)) { + eraseIfUnused(op, rewriter); + } +} + +struct StageFilterConjuncts : public impl::StageFilterConjunctsBase { + StageFilterConjuncts() {} + + StageFilterConjuncts(const DBPassContext* context) + : _context(context) + { + } + + void runOnOperation() override { + if (!_context->_view) { + return; + } + + const RowEstimator estimator(*_context->_view); + WalkKnowledge knowledge; + getOperation()->walk([&estimator, &knowledge](Operation* op) { + const std::optional seeded = estimateSourceRows(op, estimator, knowledge); + if (!seeded) { + estimateRowMultiplier(op, estimator, knowledge); + } + }); + + // Every plan is read before any filter is rewritten, and a rewrite erases only its own + // filter and the ops only it reads, so the filters go last to first: a plan names no + // value a later filter's rewrite replaces. + StagePlanner planner(estimator, knowledge); + std::deque plans; + getOperation()->walk([&planner, &plans](FilterOp filter) { + FilterStages& stages = plans.emplace_back(); + if (!planner.plan(filter, stages)) { + plans.pop_back(); + } + }); + + mlir::IRRewriter rewriter(&getContext()); + for (const FilterStages& stages : llvm::reverse(plans)) { + stageConjuncts(stages, rewriter); + } + } + +private: + const DBPassContext* _context {&defaultPassContext}; +}; + } std::unique_ptr createFuseScanEdgesByEndpointLabel(const DBPassContext* context) { @@ -12890,4 +13390,8 @@ std::unique_ptr createRerootPatternAtCheapestNode(const DBPassContext* con return std::make_unique(context); } +std::unique_ptr createStageFilterConjuncts(const DBPassContext* context) { + return std::make_unique(context); +} + } diff --git a/query/ir/dialect/db/DBPasses.h b/query/ir/dialect/db/DBPasses.h index 6c0a785e0c..aa4ee7f0dd 100644 --- a/query/ir/dialect/db/DBPasses.h +++ b/query/ir/dialect/db/DBPasses.h @@ -26,6 +26,8 @@ namespace mlir::db { // fuse_scan_edges_by_endpoint_label reads the graph to walk the index of the end with fewer nodes. // reroot_pattern_at_cheapest_node reads it to start a pattern from the node making the fewest rows, // and to choose where the walk crosses in the rows a filter joins it to. +// stage_filter_conjuncts reads it for the share of rows each conjunct of a filter keeps, and with +// no graph splits no filter. struct DBPassContext { const ::db::GraphView* _view {nullptr}; bool _forcesHashJoin {false}; @@ -37,5 +39,6 @@ std::unique_ptr createFuseScanEdgesByEndpointLabel(const DBPassContext* co std::unique_ptr createFuseHashJoin(const DBPassContext* context); std::unique_ptr createCountFromMetadata(const DBPassContext* context); std::unique_ptr createRerootPatternAtCheapestNode(const DBPassContext* context); +std::unique_ptr createStageFilterConjuncts(const DBPassContext* context); } diff --git a/query/ir/dialect/db/DBPasses.td b/query/ir/dialect/db/DBPasses.td index ff7c82f70c..66e887fa70 100644 --- a/query/ir/dialect/db/DBPasses.td +++ b/query/ir/dialect/db/DBPasses.td @@ -23,6 +23,11 @@ def MergeFilters : Pass<"merge_filters"> { let dependentDialects = ["mlir::db::DB"]; } +def StageFilterConjuncts : Pass<"stage_filter_conjuncts"> { + let summary = "Split a filter into a chain, its cheap and selective conjuncts first, where computing a costly one over the rows they keep saves more than the extra filter costs"; + let dependentDialects = ["mlir::db::DB"]; +} + def FuseUnwindEquality : Pass<"fuse_unwind_equality"> { let summary = "Replace a constant unwind crossed into one equality by that equality's disjunction"; let dependentDialects = ["mlir::db::DB"]; diff --git a/test/common/CMakeLists.txt b/test/common/CMakeLists.txt index 8343d839c7..a409ecd9ed 100644 --- a/test/common/CMakeLists.txt +++ b/test/common/CMakeLists.txt @@ -1,3 +1,4 @@ add_turing_test(test_common_perfstat PerfStatTest.cpp) add_turing_test(test_common_hybridstring HybridStringTest.cpp) add_turing_test(test_common_demon_readiness DemonReadinessTest.cpp) +add_turing_test(test_common_segmentation_solver SegmentationSolverTest.cpp) diff --git a/test/common/SegmentationSolverTest.cpp b/test/common/SegmentationSolverTest.cpp new file mode 100644 index 0000000000..b99f01eb01 --- /dev/null +++ b/test/common/SegmentationSolverTest.cpp @@ -0,0 +1,196 @@ +#include + +#include +#include +#include + +#include +#include +#include + +#include "SegmentationSolver.h" + +using namespace db; + +namespace { + +using Boundary = SegmentationSolver::Boundary; + +double segmentationCost(const std::vector& boundaries, const std::vector& starts) { + double cost = 0; + for (size_t index = 0; index < starts.size(); index++) { + const size_t start = starts[index]; + const size_t end = index + 1 < starts.size() ? starts[index + 1] : boundaries.size() - 1; + const Boundary& first = boundaries[start]; + const Boundary& last = boundaries[end]; + cost += first._weight * (last._position - first._position) + first._startCost + last._endCost; + } + + return cost; +} + +void solve(const std::vector& boundaries, std::vector& starts) { + SegmentationSolver solver; + for (const Boundary& boundary : boundaries) { + solver.addBoundary(boundary); + } + + solver.getSegmentStarts(starts); +} + +// The least cost over every segmentation, trying each set of the cuttable boundaries +double bruteForceCost(const std::vector& boundaries) { + const size_t inner = boundaries.size() - 2; + double best = std::numeric_limits::infinity(); + for (uint64_t cuts = 0; cuts < (uint64_t {1} << inner); cuts++) { + std::vector starts {0}; + bool allowed = true; + for (size_t boundary = 1; boundary <= inner; boundary++) { + if ((cuts >> (boundary - 1)) & 1) { + allowed = allowed && boundaries[boundary]._cuttable; + starts.push_back(boundary); + } + } + + if (allowed) { + best = std::min(best, segmentationCost(boundaries, starts)); + } + } + + return best; +} + +// The least cost by the quadratic dynamic program the solver speeds up +double quadraticCost(const std::vector& boundaries) { + std::vector best(boundaries.size(), std::numeric_limits::infinity()); + best[0] = 0; + for (size_t end = 1; end < boundaries.size(); end++) { + for (size_t start = 0; start < end; start++) { + const bool startsASegment = start == 0 || boundaries[start]._cuttable; + if (!startsASegment) { + continue; + } + + const Boundary& first = boundaries[start]; + const Boundary& last = boundaries[end]; + const double cost = best[start] + first._weight * (last._position - first._position) + first._startCost + last._endCost; + best[end] = std::min(best[end], cost); + } + } + + return best.back(); +} + +// Weights shrink by a factor from @param leastFactor to 1, some of them exactly 1 and, with +// @param dropsToZero, some 0. Positions grow by a step some of which are 0, so lines tie on +// slope and queries tie on position. +void randomBoundaries(std::mt19937& random, + size_t count, + double leastFactor, + bool dropsToZero, + std::vector& boundaries) { + std::uniform_real_distribution unit(0.0, 1.0); + std::uniform_real_distribution factor(leastFactor, 1.0); + std::uniform_int_distribution choice(0, 9); + + boundaries.clear(); + double weight = 1; + double position = 0; + for (size_t index = 0; index <= count; index++) { + if (index > 0) { + const int shape = choice(random); + if (shape == 0 && dropsToZero) { + weight = 0; + } else if (shape >= 3) { + weight *= factor(random); + } + + position += choice(random) < 2 ? 0.0 : 20 * unit(random); + } + + boundaries.push_back(Boundary {position, weight, unit(random), unit(random), choice(random) > 1}); + } +} + +} + +TEST(SegmentationSolverTest, keepsOneSegmentWhereNoCutPays) { + const std::vector boundaries {{0, 1, 1, 0, true}, {1, 1, 1, 0.5, true}, {2, 1, 1, 0, true}}; + + std::vector starts; + solve(boundaries, starts); + + EXPECT_EQ(starts, std::vector({0})); +} + +// One segment costs 1 * 11 + 1 = 12. Cutting at 1 costs 1 * 1 + 1 + 0.2 for the first and +// 0.1 * 10 + 1 for the second, 4.2 in all. +TEST(SegmentationSolverTest, cutsWhereTheWeightDrops) { + const std::vector boundaries {{0, 1, 1, 0, true}, {1, 0.1, 1, 0.2, true}, {11, 0.1, 1, 0, true}}; + + std::vector starts; + solve(boundaries, starts); + + EXPECT_EQ(starts, std::vector({0, 1})); +} + +TEST(SegmentationSolverTest, neverCutsWhereACutIsForbidden) { + const std::vector boundaries {{0, 1, 1, 0, true}, {1, 0.1, 1, 0.2, false}, {11, 0.1, 1, 0, true}}; + + std::vector starts; + solve(boundaries, starts); + + EXPECT_EQ(starts, std::vector({0})); +} + +TEST(SegmentationSolverTest, startsOverAfterClear) { + SegmentationSolver solver; + for (const Boundary& boundary : {Boundary {0, 1, 1, 0, true}, Boundary {1, 0.1, 1, 0.2, true}, Boundary {11, 0.1, 1, 0, true}}) { + solver.addBoundary(boundary); + } + + solver.clear(); + for (const Boundary& boundary : {Boundary {0, 1, 1, 0, true}, Boundary {1, 1, 1, 0.5, true}, Boundary {2, 1, 1, 0, true}}) { + solver.addBoundary(boundary); + } + + std::vector starts; + solver.getSegmentStarts(starts); + + EXPECT_EQ(starts, std::vector({0})); +} + +TEST(SegmentationSolverTest, costsWhatTheBestSegmentationCosts) { + std::mt19937 random(7); + std::vector boundaries; + std::vector starts; + + for (size_t trial = 0; trial < 2000; trial++) { + randomBoundaries(random, 1 + trial % 12, 0.0, true, boundaries); + solve(boundaries, starts); + + ASSERT_FALSE(starts.empty()); + EXPECT_EQ(starts.front(), 0u); + for (size_t index = 1; index < starts.size(); index++) { + EXPECT_LT(starts[index - 1], starts[index]); + EXPECT_TRUE(boundaries[starts[index]]._cuttable); + } + + const double expected = bruteForceCost(boundaries); + EXPECT_NEAR(segmentationCost(boundaries, starts), expected, 1e-9 * (1 + expected)) << "trial " << trial; + } +} + +TEST(SegmentationSolverTest, costsWhatTheQuadraticProgramFindsOnLongSequences) { + std::mt19937 random(11); + std::vector boundaries; + std::vector starts; + + for (size_t trial = 0; trial < 20; trial++) { + randomBoundaries(random, 2000, 0.99, false, boundaries); + solve(boundaries, starts); + + const double expected = quadraticCost(boundaries); + EXPECT_NEAR(segmentationCost(boundaries, starts), expected, 1e-9 * (1 + expected)) << "trial " << trial; + } +} diff --git a/test/query/ir/passes/CMakeLists.txt b/test/query/ir/passes/CMakeLists.txt index 89f48d9f23..55a7b9d67b 100644 --- a/test/query/ir/passes/CMakeLists.txt +++ b/test/query/ir/passes/CMakeLists.txt @@ -208,6 +208,10 @@ add_ir_tests(test_query_ir_merge_filters_over_one_op MergeFiltersOverOneOpTest.c add_ir_tests(test_query_ir_merge_filters_over_one_op_cypher MergeFiltersOverOneOpCypherTest.cpp) +add_ir_tests(test_query_ir_stage_filter_conjuncts StageFilterConjunctsTest.cpp) + +add_ir_tests(test_query_ir_stage_filter_conjuncts_cypher StageFilterConjunctsCypherTest.cpp) + add_ir_tests(test_query_ir_push_down_label_and_type_tests PushDownLabelAndTypeTestsTest.cpp) add_ir_tests(test_query_ir_push_down_label_and_type_tests_cypher PushDownLabelAndTypeTestsCypherTest.cpp) diff --git a/test/query/ir/passes/StageFilterConjunctsCypherTest.cpp b/test/query/ir/passes/StageFilterConjunctsCypherTest.cpp new file mode 100644 index 0000000000..acca59117e --- /dev/null +++ b/test/query/ir/passes/StageFilterConjunctsCypherTest.cpp @@ -0,0 +1,120 @@ +#include + +#include +#include + +#include +#include +#include + +#include "IRTestRows.h" +#include "QueryStatus.h" +#include "WriteQueryTest.h" + +using namespace db; +using namespace turing::test; + +// A filter split into stages keeps the rows the one filter it replaces kept, across chunks +class StageFilterConjunctsCypherTest : public WriteQueryTest { +protected: + // More items than one chunk holds. Item i has v = i % 5 and u = i % 4, an R edge to a + // Target holding the same i, and every third item a second R edge to an Other. + static constexpr int64_t ITEM_COUNT = 2 * 65536 + 5; + + // Enough papers that the Nature ones crossed with the Science ones fill more than a chunk + static constexpr int64_t PAPER_COUNT = 600; + + void createItems() { + applyWrite("UNWIND range(0, " + std::to_string(ITEM_COUNT - 1) + ") AS i " + "CREATE (n:Item {i: i, v: i % 5, u: i % 4})-[:R]->(:Target {i: i})"); + applyWrite("MATCH (n:Item) WHERE n.i % 3 = 0 CREATE (n)-[:R]->(:Other {i: n.i})"); + } + + // Paper i appears in Nature when i is even and in Science otherwise, in year 1990 + i % 30 + // and volume i % 17 + void createPapers() { + applyWrite("UNWIND range(0, " + std::to_string(PAPER_COUNT - 1) + ") AS i " + "CREATE (:Paper {i: i, journal: CASE WHEN i % 2 = 0 THEN 'Nature' ELSE 'Science' END, " + "year: 1990 + i % 30, volume: i % 17})"); + } + + // Prints the first row that differs rather than every row + void expectManyRows(std::string_view query, Rows expected) { + RowSink sink; + const QueryStatus status = runQuery(query, &sink); + ASSERT_TRUE(status.isOk()) << "query: " << query << "\nerror: " << status.getError(); + + Rows rows; + sink.sortedRows(rows); + std::sort(expected.begin(), expected.end()); + + ASSERT_EQ(rows.size(), expected.size()) << "query: " << query; + + const auto [actualIt, expectedIt] = std::mismatch(rows.begin(), rows.end(), expected.begin()); + if (actualIt == rows.end()) { + return; + } + + std::string actualText; + describeRows(Rows {*actualIt}, actualText); + + std::string expectedText; + describeRows(Rows {*expectedIt}, expectedText); + + ADD_FAILURE() << "query: " << query << "\nfirst differing row: " << actualText << "expected: " << expectedText; + } +}; + +TEST_F(StageFilterConjunctsCypherTest, testsTheFarLabelOfTheTypedEdges) { + createItems(); + + Rows targets; + Rows others; + for (int64_t item = 0; item < ITEM_COUNT; item++) { + targets.push_back({std::to_string(item), std::to_string(item)}); + + if (item % 3 == 0) { + others.push_back({std::to_string(item)}); + } + } + + expectManyRows("MATCH (n:Item)-[:R]-(t:Target) RETURN n.i, t.i", targets); + expectManyRows("MATCH (n:Item)-[:R]-(o:Other) RETURN o.i", others); +} + +// t.i is in [n.u, n.v, 7] for item 4 alone among those whose i % 5 is 4: 4 % 5 is 4 +TEST_F(StageFilterConjunctsCypherTest, testsMembershipOverTheRowsTheCheapTestKeeps) { + createItems(); + + Rows kept; + for (int64_t item = 0; item < ITEM_COUNT; item++) { + const bool inTheList = item == item % 4 || item == item % 5 || item == 7; + if (item % 5 == 4 && !inTheList) { + kept.push_back({std::to_string(item)}); + } + } + + expectManyRows("MATCH (n:Item)-[:R]->(t:Target) WHERE t.i % 5 = 4 AND NOT t.i IN [n.u, n.v, 7] RETURN t.i", kept); + expectManyRows("MATCH (n:Item)-[:R]->(t:Target) WHERE NOT t.i IN [n.u, n.v, 7] AND t.i % 5 = 4 RETURN t.i", kept); +} + +TEST_F(StageFilterConjunctsCypherTest, comparesThePairsTheFirstConjunctKeeps) { + createPapers(); + + Rows pairs; + for (int64_t nature = 0; nature < PAPER_COUNT; nature += 2) { + for (int64_t science = 1; science < PAPER_COUNT; science += 2) { + const int64_t yearGap = llabs(nature % 30 - science % 30); + if (yearGap <= 1 && nature % 17 > science % 17) { + pairs.push_back({std::to_string(nature), std::to_string(science)}); + } + } + } + + expectManyRows("MATCH (a:Paper {journal: 'Nature'}), (b:Paper {journal: 'Science'}) " + "WHERE abs(a.year - b.year) <= 1 AND a.volume > b.volume RETURN a.i, b.i", + pairs); + expectManyRows("MATCH (a:Paper {journal: 'Nature'}), (b:Paper {journal: 'Science'}) " + "WHERE a.volume > b.volume AND abs(a.year - b.year) <= 1 RETURN a.i, b.i", + pairs); +} diff --git a/test/query/ir/passes/StageFilterConjunctsTest.cpp b/test/query/ir/passes/StageFilterConjunctsTest.cpp new file mode 100644 index 0000000000..3d7043ad6d --- /dev/null +++ b/test/query/ir/passes/StageFilterConjunctsTest.cpp @@ -0,0 +1,221 @@ +#include + +#include + +#include +#include +#include + +#include "mlir/Dialect/Func/IR/FuncOps.h" +#include "mlir/IR/BuiltinOps.h" +#include "mlir/IR/MLIRContext.h" +#include "mlir/IR/OwningOpRef.h" +#include "llvm/ADT/STLFunctionalExtras.h" +#include "llvm/ADT/SmallVector.h" +#include "llvm/Support/raw_ostream.h" + +#include "DBDialect.h" +#include "DBOps.h" +#include "DBPasses.h" +#include "DBProgramGenerator.h" +#include "NLDialect.h" +#include "StorageDialect.h" + +#include "CypherAST.h" +#include "CypherAnalyzer.h" +#include "CypherParser.h" +#include "Graph.h" +#include "SimpleGraph.h" +#include "SystemAccessor.h" +#include "SystemManager.h" +#include "versioning/Transaction.h" +#include "views/GraphView.h" + +#include "TuringTest.h" +#include "TuringTestEnv.h" + +using namespace db; +using namespace turing::test; + +// A costly conjunct runs over the rows the cheap and selective conjuncts of its filter keep, +// behind a filter of its own, and cheap conjuncts share one mask. Each query compiles with the +// graph's counts in the pass context, as the engine runs it: with none, no filter splits. +class StageFilterConjunctsTest : public TuringTest { +public: + void initialize() override { + _env = TuringTestEnv::create(fs::Path {_outDir} / "turing"); + + SystemAccessor system = _env->getSystemManager().accessUnique(); + _graph = system.createGraph(_graphName); + SimpleGraph::createSimpleGraph(_graph); + + _context.getOrLoadDialect(); + _context.getOrLoadDialect(); + _context.getOrLoadDialect(); + _context.getOrLoadDialect(); + } + +protected: + struct Reads { + size_t _total {0}; + size_t _overFilteredRows {0}; + }; + + void generate(std::string_view query, mlir::OwningOpRef& module) { + SystemAccessor system = _env->getSystemManager().accessUnique(); + const ProcedureManager* procedures = system.getProcedures(); + + const FrozenCommitTx transaction = _graph->openTransaction(); + const GraphView view = transaction.viewGraph(); + + CypherAST ast(procedures, query); + + CypherParser parser(&ast); + parser.parse(query); + + CypherAnalyzer analyzer(&ast, view); + analyzer.analyze(); + + module = mlir::ModuleOp::create(mlir::UnknownLoc::get(&_context)); + mlir::ModuleOp moduleOp = module.get(); + + const mlir::db::DBPassContext passContext {._view = &view}; + DBProgramGenerator generator(&moduleOp, nullptr, passContext); + generator.generate(&ast); + } + + static void render(mlir::ModuleOp module, std::string& program) { + program.clear(); + llvm::raw_string_ostream stream(program); + + mlir::OpPrintingFlags flags; + flags.assumeVerified(); + module.print(stream, flags); + } + + // The ops @param selects, and how many of them read a column a filter keeps + static void countReads(mlir::ModuleOp module, llvm::function_ref selects, Reads& reads) { + module.walk([&selects, &reads](mlir::Operation* op) { + if (!selects(op)) { + return; + } + + reads._total++; + if (op->getOperand(0).getDefiningOp()) { + reads._overFilteredRows++; + } + }); + } + + static bool readsProperty(mlir::Operation* op, std::string_view property) { + mlir::db::GetNodeProperties read = mlir::dyn_cast(op); + return read && read.getProperty() == llvm::StringRef(property); + } + + // The filter whose mask holds @param value, through the `and`s joining it to the others + static mlir::db::FilterOp maskFilter(mlir::Value value) { + mlir::Value conjunct = value; + while (conjunct.hasOneUse()) { + mlir::Operation* const user = *conjunct.getUsers().begin(); + if (mlir::db::FilterOp filter = mlir::dyn_cast(user)) { + return filter; + } else if (!mlir::isa(user)) { + return {}; + } + + conjunct = user->getResult(0); + } + + return {}; + } + + void expectEveryReadOverFilteredRows(std::string_view query, llvm::function_ref selects) { + mlir::OwningOpRef module; + generate(query, module); + + std::string program; + render(*module, program); + + Reads reads; + countReads(*module, selects, reads); + + EXPECT_GT(reads._total, 0u) << query << "\n" << program; + EXPECT_EQ(reads._overFilteredRows, reads._total) << query << "\n" << program; + } + + const std::string _graphName {"simpledb"}; + std::unique_ptr _env; + Graph* _graph {nullptr}; + mlir::MLIRContext _context; +}; + +// Of the edges of a person, only the INTERESTED_IN ones need the label of the node they reach +TEST_F(StageFilterConjunctsTest, readsTheFarNodeLabelsOfTheTypedEdgesAlone) { + expectEveryReadOverFilteredRows("MATCH (a:Person)-[:INTERESTED_IN]-(b:Interest) RETURN count(*)", [](mlir::Operation* op) { + return mlir::isa(op); + }); +} + +// One conjunct compares the ages of a pair, the other its names: the pairs the first keeps +// are the only ones whose names are read +TEST_F(StageFilterConjunctsTest, readsTheSecondPairOfPropertiesOverThePairsTheFirstKeeps) { + const std::string_view query = "MATCH (a:Person), (b:Person) WHERE abs(a.age - b.age) <= 1 AND a.name > b.name RETURN count(*)"; + + mlir::OwningOpRef module; + generate(query, module); + + std::string program; + render(*module, program); + + Reads reads; + countReads(*module, [](mlir::Operation* op) { return mlir::isa(op); }, reads); + + EXPECT_EQ(reads._total, 4u) << query << "\n" << program; + EXPECT_EQ(reads._overFilteredRows, 2u) << query << "\n" << program; +} + +// A range keeps fewer rows than an inequality, so it runs first, whichever conjunct the query +// writes first, and the name is read over the rows it keeps +TEST_F(StageFilterConjunctsTest, testsTheSelectiveConjunctFirst) { + const auto readsName = [](mlir::Operation* op) { + return readsProperty(op, "name"); + }; + + expectEveryReadOverFilteredRows("MATCH (n:Person) WHERE n.name <> 'Remy' AND n.age > 30 RETURN count(n)", readsName); + expectEveryReadOverFilteredRows("MATCH (n:Person) WHERE n.age > 30 AND n.name <> 'Remy' RETURN count(n)", readsName); +} + +// The list each row tests membership in is built over the rows the duration test keeps +TEST_F(StageFilterConjunctsTest, buildsTheListOverTheRowsTheCheapTestKeeps) { + expectEveryReadOverFilteredRows("MATCH (a:Person)-[r]->(b) WHERE r.duration > 15 AND NOT b.name IN [a.name, 'Ghosts'] RETURN count(*)", [](mlir::Operation* op) { + return readsProperty(op, "name"); + }); +} + +// Two comparisons of the two ends of a hop cost less than a filter between them +TEST_F(StageFilterConjunctsTest, keepsCheapConjunctsInOneMask) { + const std::string_view query = "MATCH (a:Person)-[:KNOWS_WELL]->(b:Person) WHERE a <> b AND id(a) < id(b) RETURN count(*)"; + + mlir::OwningOpRef module; + generate(query, module); + + std::string program; + render(*module, program); + + llvm::SmallVector inequalities; + llvm::SmallVector comparisons; + module->walk([&inequalities, &comparisons](mlir::Operation* op) { + if (mlir::isa(op)) { + inequalities.push_back(op->getResult(0)); + } else if (mlir::isa(op)) { + comparisons.push_back(op->getResult(0)); + } + }); + + ASSERT_EQ(inequalities.size(), 1u) << query << "\n" << program; + ASSERT_EQ(comparisons.size(), 1u) << query << "\n" << program; + + const mlir::db::FilterOp filter = maskFilter(inequalities.front()); + ASSERT_TRUE(filter) << query << "\n" << program; + EXPECT_EQ(maskFilter(comparisons.front()), filter) << query << "\n" << program; +}