diff --git a/.commitlintrc.js b/.commitlintrc.js index b8ea5c9c8..de4e878d1 100644 --- a/.commitlintrc.js +++ b/.commitlintrc.js @@ -37,7 +37,7 @@ const Configuration = { "scope-enum": [ RuleConfigSeverity.Error, "always", - [...trackedRootDirectories, ...siloSubdirectories, "silo"], + [...trackedRootDirectories, ...siloSubdirectories, "silo", "main", "deps"], ], }, }; diff --git a/documentation/query_documentation.md b/documentation/query_documentation.md index 1f1bde174..048f62c1f 100644 --- a/documentation/query_documentation.md +++ b/documentation/query_documentation.md @@ -285,6 +285,84 @@ default.filter(pango_lineage = 'B.1.1.7').phyloSubtree('usherTree') {"subtreeNewick": "((key_83:0.00027051)NODE_0000077:3.291e-05,(...)NODE_0000079:1e-06)NODE_0000076;", "missingNodeCount": 0} ``` +### `join(left, right, on [, type := ...])` + +Combines two pipelines on an equality condition (equi-join). `join` can be called as a standalone function or with piped syntax. + +The `on` argument is an equality between a left column and a right column. Multiple key pairs are combined with `&&`; a row pair joins only when all key pairs are equal: + +``` +join( + default.project({primaryKey, country}), + default.map({pk := primaryKey, ctry := country}).project({pk, ctry}), + primaryKey = pk +) +``` + +Or equivalently using piped syntax: + +``` +default.project({primaryKey, country}) + .join(default.map({pk := primaryKey, ctry := country}).project({pk, ctry}), primaryKey = pk) +``` + +Named arguments are also supported: + +``` +join(left := , right := , on := primaryKey = pk) +``` + +Multiple equalities: + +``` +join(, , primaryKey = pk && country = ctry) +``` + +**Join type:** the optional `type` named argument selects the join kind (default `inner`): + +| `type` | Kept rows | Output columns | +| ----------- | --------------------------------------------- | -------------- | +| `inner` | matching pairs only (default) | left + right | +| `left` | all left rows; right null-filled if unmatched | left + right | +| `right` | all right rows; left null-filled if unmatched | left + right | +| `full` | all rows from both sides, null-filled | left + right | +| `leftSemi` | left rows that have a match | left only | +| `rightSemi` | right rows that have a match | right only | +| `leftAnti` | left rows without a match | left only | +| `rightAnti` | right rows without a match | right only | + +``` +join(, , primaryKey = pk, type := left) +``` + +**Output schema:** the left input's columns followed by the right input's columns (semi/anti joins output only the kept side's columns). + +**Restrictions:** + +- The two inputs must have disjoint column names. A name occurring on both sides would make column references ambiguous and produce a result with duplicate column names, so such a query is rejected — rename one side (e.g. via `map()`) or drop the duplicates (e.g. via `project()`) before joining. + +``` +join( + default.project({primaryKey, country}), + default.project({primaryKey, country}), + primaryKey = primaryKey +) +``` + + fails because `primaryKey` and `country` are present in both inputs; give the two sides distinct names instead (`primaryKey = pk`). + +- `filter()` cannot be applied to the output of a `join()`. A predicate above a join is not pushed into a join input, because doing so is not semantics-preserving in general (for example, pushing into the null-supplying side of an outer join would keep null-extended rows that the predicate should have removed). Apply the filter to one of the join inputs instead: + +``` +join( + default.filter(country='CH').project({primaryKey, country}), + default.map({pk := primaryKey}).project({pk}), + primaryKey = pk +) +``` + +**Output:** the joined rows. The order of rows is not guaranteed; use `orderBy(...)` for a deterministic order. + ### `unionAll(left, right)` Concatenates the output of two pipelines. `unionAll` can be called as a standalone function or with piped syntax: diff --git a/src/silo/query_engine/operator_visitor.h b/src/silo/query_engine/operator_visitor.h index 426552078..629c95e03 100644 --- a/src/silo/query_engine/operator_visitor.h +++ b/src/silo/query_engine/operator_visitor.h @@ -9,6 +9,7 @@ #include "silo/query_engine/operators/fetch_node.h" #include "silo/query_engine/operators/filter_node.h" #include "silo/query_engine/operators/insertions_node.h" +#include "silo/query_engine/operators/join_node.h" #include "silo/query_engine/operators/map_node.h" #include "silo/query_engine/operators/most_recent_common_ancestor_node.h" #include "silo/query_engine/operators/mutations_node.h" @@ -79,6 +80,8 @@ decltype(auto) visit(QueryNode& node, Func&& func) { return std::forward(func)(static_cast(node)); case NodeKind::UNION_ALL: return std::forward(func)(static_cast(node)); + case NodeKind::JOIN: + return std::forward(func)(static_cast(node)); case NodeKind::SCHEMA: return std::forward(func)(static_cast(node)); case NodeKind::BITMAP_AGGREGATION: diff --git a/src/silo/query_engine/operators/join_node.cpp b/src/silo/query_engine/operators/join_node.cpp new file mode 100644 index 000000000..b555fce6a --- /dev/null +++ b/src/silo/query_engine/operators/join_node.cpp @@ -0,0 +1,108 @@ +#include "silo/query_engine/operators/join_node.h" + +#include + +#include +#include +#include +#include + +#include "silo/common/panic.h" + +namespace silo::query_engine::operators { + +using arrow::acero::JoinType; + +JoinNode::JoinNode( + QueryNodePtr left, + QueryNodePtr right, + std::vector left_keys, + std::vector right_keys, + JoinType join_type +) + : left(std::move(left)), + right(std::move(right)), + left_keys(std::move(left_keys)), + right_keys(std::move(right_keys)), + join_type(join_type) { + // Equi-join key vectors are paired positionally (left_keys[i] == right_keys[i]), so they + // must have the same length + SILO_ASSERT_EQ(this->left_keys.size(), this->right_keys.size()); +} + +std::string_view joinTypeToString(JoinType join_type) { + switch (join_type) { + case JoinType::INNER: + return "inner"; + case JoinType::LEFT_OUTER: + return "left"; + case JoinType::RIGHT_OUTER: + return "right"; + case JoinType::FULL_OUTER: + return "full"; + case JoinType::LEFT_SEMI: + return "leftSemi"; + case JoinType::RIGHT_SEMI: + return "rightSemi"; + case JoinType::LEFT_ANTI: + return "leftAnti"; + case JoinType::RIGHT_ANTI: + return "rightAnti"; + } + SILO_UNREACHABLE(); +} + +std::vector JoinNode::getOutputSchema() const { + // Semi/anti joins act as a filter on one input and emit only that side's columns. + if (join_type == JoinType::LEFT_SEMI || join_type == JoinType::LEFT_ANTI) { + return left->getOutputSchema(); + } + if (join_type == JoinType::RIGHT_SEMI || join_type == JoinType::RIGHT_ANTI) { + return right->getOutputSchema(); + } + auto output = left->getOutputSchema(); + auto right_schema = right->getOutputSchema(); + output.insert(output.end(), right_schema.begin(), right_schema.end()); + return output; +} + +arrow::Result JoinNode::addToExecPlan( + arrow::acero::ExecPlan& plan, + const std::map>& tables, + const config::QueryOptions& query_options +) const { + ARROW_ASSIGN_OR_RAISE(auto* left_node, left->addToExecPlan(plan, tables, query_options)); + ARROW_ASSIGN_OR_RAISE(auto* right_node, right->addToExecPlan(plan, tables, query_options)); + + std::vector left_key_refs; + left_key_refs.reserve(left_keys.size()); + for (const auto& key : left_keys) { + left_key_refs.emplace_back(key.name); + } + std::vector right_key_refs; + right_key_refs.reserve(right_keys.size()); + for (const auto& key : right_keys) { + right_key_refs.emplace_back(key.name); + } + + const arrow::acero::HashJoinNodeOptions options{ + join_type, std::move(left_key_refs), std::move(right_key_refs) + }; + return arrow::acero::MakeExecNode("hashjoin", &plan, {left_node, right_node}, options); +} + +nlohmann::json JoinNode::toJson() const { + nlohmann::json keys = nlohmann::json::array(); + for (size_t i = 0; i < left_keys.size(); ++i) { + keys.push_back({{"left", left_keys[i].name}, {"right", right_keys[i].name}}); + } + return { + {"type", nodeKindToString(kind())}, + {"joinType", joinTypeToString(join_type)}, + {"on", std::move(keys)}, + {"left", left->toJson()}, + {"right", right->toJson()}, + }; +} + +} // namespace silo::query_engine::operators diff --git a/src/silo/query_engine/operators/join_node.h b/src/silo/query_engine/operators/join_node.h new file mode 100644 index 000000000..e8a3b2a2a --- /dev/null +++ b/src/silo/query_engine/operators/join_node.h @@ -0,0 +1,58 @@ +#pragma once + +#include +#include +#include + +#include +#include +#include + +#include "silo/query_engine/operators/query_node.h" +#include "silo/schema/database_schema.h" +#include "silo/storage/table.h" + +namespace silo::query_engine::operators { + +/// Joins two child pipelines on an equality condition (equi-join), backed by +/// Arrow's hash-join. `left_keys[i]` is compared for equality against +/// `right_keys[i]`; all key pairs must match for a row pair to join. +/// +/// The output schema is the left child's columns followed by the right child's +/// columns, except for semi/anti joins which output only the columns of the +/// side they keep. Overlapping column names between the two sides are passed +/// through as-is (mirroring SQL `SELECT *`); rename or project them downstream +/// if distinct names are required. +class JoinNode final : public QueryNode { + public: + QueryNodePtr left; + QueryNodePtr right; + std::vector left_keys; + std::vector right_keys; + arrow::acero::JoinType join_type; + + JoinNode( + QueryNodePtr left, + QueryNodePtr right, + std::vector left_keys, + std::vector right_keys, + arrow::acero::JoinType join_type + ); + + [[nodiscard]] std::vector getOutputSchema() const override; + + [[nodiscard]] arrow::Result addToExecPlan( + arrow::acero::ExecPlan& plan, + const std::map>& tables, + const config::QueryOptions& query_options + ) const override; + + [[nodiscard]] NodeKind kind() const override { return NodeKind::JOIN; } + + [[nodiscard]] nlohmann::json toJson() const override; +}; + +/// Human-readable name for a join type, used in query-plan JSON and diagnostics. +[[nodiscard]] std::string_view joinTypeToString(arrow::acero::JoinType join_type); + +} // namespace silo::query_engine::operators diff --git a/src/silo/query_engine/operators/join_node.test.cpp b/src/silo/query_engine/operators/join_node.test.cpp new file mode 100644 index 000000000..7890a8b8a --- /dev/null +++ b/src/silo/query_engine/operators/join_node.test.cpp @@ -0,0 +1,342 @@ +#include + +#include "silo/test/query_fixture.test.h" + +namespace { +using silo::ReferenceGenomes; +using silo::test::QueryTestData; +using silo::test::QueryTestScenario; + +nlohmann::json createData(const std::string& primaryKey, const std::string& country) { + return { + {"primaryKey", primaryKey}, + {"country", country}, + {"segment1", {{"sequence", "T"}, {"insertions", nlohmann::json::array()}}}, + {"gene1", nullptr}, + {"unaligned_segment1", nullptr} + }; +} + +const std::vector DATA = { + createData("id_0", "CH"), + createData("id_1", "DE"), + createData("id_2", "CH"), + createData("id_3", "DE"), +}; + +const auto DATABASE_CONFIG = + R"( +defaultNucleotideSequence: "segment1" +schema: + instanceName: "dummy name" + metadata: + - name: "primaryKey" + type: "string" + - name: "country" + type: "string" + primaryKey: "primaryKey" +)"; + +const auto REFERENCE_GENOMES = ReferenceGenomes{ + {{"segment1", "A"}}, + {{"gene1", "*"}}, +}; + +const QueryTestData TEST_DATA{ + .ndjson_input_data = DATA, + .database_config = DATABASE_CONFIG, + .reference_genomes = REFERENCE_GENOMES +}; + +// Inner join on a unique key: every row matches exactly its renamed copy. +const QueryTestScenario JOIN_INNER_ON_KEY_SCENARIO = { + .name = "JOIN_INNER_ON_KEY", + .query = R"(join( + default.project({primaryKey, country}), + default.map({pk := primaryKey, ctry := country}).project({pk, ctry}), + primaryKey = pk + ).orderBy({asc(primaryKey)}))", + .expected_query_result = nlohmann::json( + {{{"primaryKey", "id_0"}, {"country", "CH"}, {"pk", "id_0"}, {"ctry", "CH"}}, + {{"primaryKey", "id_1"}, {"country", "DE"}, {"pk", "id_1"}, {"ctry", "DE"}}, + {{"primaryKey", "id_2"}, {"country", "CH"}, {"pk", "id_2"}, {"ctry", "CH"}}, + {{"primaryKey", "id_3"}, {"country", "DE"}, {"pk", "id_3"}, {"ctry", "DE"}}} + ) +}; + +// Piped syntax: left.join(right, on) instead of join(left, right, on). +const QueryTestScenario JOIN_PIPED_SYNTAX_SCENARIO = { + .name = "JOIN_PIPED_SYNTAX", + .query = R"( + default.project({primaryKey, country}) + .join(default.map({pk := primaryKey, ctry := country}).project({pk, ctry}), primaryKey = pk) + .orderBy({asc(primaryKey)}) + )", + .expected_query_result = nlohmann::json( + {{{"primaryKey", "id_0"}, {"country", "CH"}, {"pk", "id_0"}, {"ctry", "CH"}}, + {{"primaryKey", "id_1"}, {"country", "DE"}, {"pk", "id_1"}, {"ctry", "DE"}}, + {{"primaryKey", "id_2"}, {"country", "CH"}, {"pk", "id_2"}, {"ctry", "CH"}}, + {{"primaryKey", "id_3"}, {"country", "DE"}, {"pk", "id_3"}, {"ctry", "DE"}}} + ) +}; + +// Explicit `type := inner` with a filter on the left input. +const QueryTestScenario JOIN_EXPLICIT_INNER_SCENARIO = { + .name = "JOIN_EXPLICIT_INNER", + .query = R"(join( + default.filter(country='CH').project({primaryKey, country}), + default.map({pk := primaryKey, ctry := country}).project({pk, ctry}), + primaryKey = pk, + type := inner + ).orderBy({asc(primaryKey)}))", + .expected_query_result = nlohmann::json( + {{{"primaryKey", "id_0"}, {"country", "CH"}, {"pk", "id_0"}, {"ctry", "CH"}}, + {{"primaryKey", "id_2"}, {"country", "CH"}, {"pk", "id_2"}, {"ctry", "CH"}}} + ) +}; + +// Join on a non-key column yields the cross product of matching groups. +const QueryTestScenario JOIN_MANY_TO_MANY_SCENARIO = { + .name = "JOIN_MANY_TO_MANY", + .query = R"(join( + default.filter(country='DE').project({primaryKey, country}), + default.filter(country='DE').map({pk := primaryKey, ctry := country}).project({pk, ctry}), + country = ctry + ).orderBy({asc(primaryKey), asc(pk)}))", + .expected_query_result = nlohmann::json( + {{{"primaryKey", "id_1"}, {"country", "DE"}, {"pk", "id_1"}, {"ctry", "DE"}}, + {{"primaryKey", "id_1"}, {"country", "DE"}, {"pk", "id_3"}, {"ctry", "DE"}}, + {{"primaryKey", "id_3"}, {"country", "DE"}, {"pk", "id_1"}, {"ctry", "DE"}}, + {{"primaryKey", "id_3"}, {"country", "DE"}, {"pk", "id_3"}, {"ctry", "DE"}}} + ) +}; + +// Left outer join: unmatched left rows keep null values for the right columns. +const QueryTestScenario JOIN_LEFT_OUTER_SCENARIO = { + .name = "JOIN_LEFT_OUTER", + .query = R"(join( + default.project({primaryKey, country}), + default.filter(country='CH').map({pk := primaryKey, ctry := country}).project({pk, ctry}), + primaryKey = pk, + type := left + ).orderBy({asc(primaryKey)}))", + .expected_query_result = nlohmann::json( + {{{"primaryKey", "id_0"}, {"country", "CH"}, {"pk", "id_0"}, {"ctry", "CH"}}, + {{"primaryKey", "id_1"}, {"country", "DE"}, {"pk", nullptr}, {"ctry", nullptr}}, + {{"primaryKey", "id_2"}, {"country", "CH"}, {"pk", "id_2"}, {"ctry", "CH"}}, + {{"primaryKey", "id_3"}, {"country", "DE"}, {"pk", nullptr}, {"ctry", nullptr}}} + ) +}; + +// Left semi join: keeps left rows that have a match, outputs only left columns. +const QueryTestScenario JOIN_LEFT_SEMI_SCENARIO = { + .name = "JOIN_LEFT_SEMI", + .query = R"(join( + default.project({primaryKey, country}), + default.filter(country='CH').map({pk := primaryKey, ctry := country}).project({pk, ctry}), + primaryKey = pk, + type := leftSemi + ).orderBy({asc(primaryKey)}))", + .expected_query_result = nlohmann::json( + {{{"primaryKey", "id_0"}, {"country", "CH"}}, {{"primaryKey", "id_2"}, {"country", "CH"}}} + ) +}; + +// Left anti join: keeps left rows WITHOUT a match, outputs only left columns. +const QueryTestScenario JOIN_LEFT_ANTI_SCENARIO = { + .name = "JOIN_LEFT_ANTI", + .query = R"(join( + default.project({primaryKey, country}), + default.filter(country='CH').map({pk := primaryKey, ctry := country}).project({pk, ctry}), + primaryKey = pk, + type := leftAnti + ).orderBy({asc(primaryKey)}))", + .expected_query_result = nlohmann::json( + {{{"primaryKey", "id_1"}, {"country", "DE"}}, {{"primaryKey", "id_3"}, {"country", "DE"}}} + ) +}; + +// Right outer join: unmatched right rows keep null values for the left columns. +const QueryTestScenario JOIN_RIGHT_OUTER_SCENARIO = { + .name = "JOIN_RIGHT_OUTER", + .query = R"(join( + default.filter(country='CH').project({primaryKey, country}), + default.map({pk := primaryKey, ctry := country}).project({pk, ctry}), + primaryKey = pk, + type := right + ).orderBy({asc(pk)}))", + .expected_query_result = nlohmann::json( + {{{"primaryKey", "id_0"}, {"country", "CH"}, {"pk", "id_0"}, {"ctry", "CH"}}, + {{"primaryKey", nullptr}, {"country", nullptr}, {"pk", "id_1"}, {"ctry", "DE"}}, + {{"primaryKey", "id_2"}, {"country", "CH"}, {"pk", "id_2"}, {"ctry", "CH"}}, + {{"primaryKey", nullptr}, {"country", nullptr}, {"pk", "id_3"}, {"ctry", "DE"}}} + ) +}; + +// Full outer join with disjoint keys on the two inputs: every left row and every right row +// appears, each null-extended on the side it has no match on. +const QueryTestScenario JOIN_FULL_OUTER_SCENARIO = { + .name = "JOIN_FULL_OUTER", + .query = R"(join( + default.filter(country='CH').project({primaryKey, country}), + default.filter(country='DE').map({pk := primaryKey, ctry := country}).project({pk, ctry}), + primaryKey = pk, + type := full + ).orderBy({asc(primaryKey), asc(pk)}))", + .expected_query_result = nlohmann::json( + {{{"primaryKey", nullptr}, {"country", nullptr}, {"pk", "id_1"}, {"ctry", "DE"}}, + {{"primaryKey", nullptr}, {"country", nullptr}, {"pk", "id_3"}, {"ctry", "DE"}}, + {{"primaryKey", "id_0"}, {"country", "CH"}, {"pk", nullptr}, {"ctry", nullptr}}, + {{"primaryKey", "id_2"}, {"country", "CH"}, {"pk", nullptr}, {"ctry", nullptr}}} + ) +}; + +// Right semi join: keeps right rows that have a match, outputs only right columns. +const QueryTestScenario JOIN_RIGHT_SEMI_SCENARIO = { + .name = "JOIN_RIGHT_SEMI", + .query = R"(join( + default.filter(country='CH').project({primaryKey, country}), + default.map({pk := primaryKey, ctry := country}).project({pk, ctry}), + primaryKey = pk, + type := rightSemi + ).orderBy({asc(pk)}))", + .expected_query_result = + nlohmann::json({{{"pk", "id_0"}, {"ctry", "CH"}}, {{"pk", "id_2"}, {"ctry", "CH"}}}) +}; + +// Right anti join: keeps right rows WITHOUT a match, outputs only right columns. +const QueryTestScenario JOIN_RIGHT_ANTI_SCENARIO = { + .name = "JOIN_RIGHT_ANTI", + .query = R"(join( + default.filter(country='CH').project({primaryKey, country}), + default.map({pk := primaryKey, ctry := country}).project({pk, ctry}), + primaryKey = pk, + type := rightAnti + ).orderBy({asc(pk)}))", + .expected_query_result = + nlohmann::json({{{"pk", "id_1"}, {"ctry", "DE"}}, {{"pk", "id_3"}, {"ctry", "DE"}}}) +}; + +// A filter above the join cannot be pushed into a single join input safely (which input a +// predicate belongs to is not derivable, and outer-join / column-less cases are not +// semantics-preserving), so it is rejected. Apply the filter to a join input instead. +const QueryTestScenario JOIN_DOWNSTREAM_FILTER_SCENARIO = { + .name = "JOIN_DOWNSTREAM_FILTER", + .query = R"(join( + default.project({primaryKey, country}), + default.map({pk := primaryKey, ctry := country}).project({pk, ctry}), + primaryKey = pk + ).filter(country='CH').orderBy({asc(primaryKey)}))", + .expected_error_message = + "filter() cannot be applied to the output of join(); a filter above a join cannot be " + "pushed into a join input safely. Apply the filter to one of the join inputs instead." +}; + +// GroupBy applied to the join result. +const QueryTestScenario JOIN_WITH_GROUPBY_SCENARIO = { + .name = "JOIN_WITH_GROUPBY", + .query = R"(join( + default.project({primaryKey, country}), + default.map({pk := primaryKey, ctry := country}).project({pk, ctry}), + primaryKey = pk + ).groupBy({count := count()}, {country}).orderBy({asc(country)}))", + .expected_query_result = + nlohmann::json({{{"country", "CH"}, {"count", 2}}, {{"country", "DE"}, {"count", 2}}}) +}; + +// Multiple equalities combined with `&&`. +const QueryTestScenario JOIN_MULTI_KEY_SCENARIO = { + .name = "JOIN_MULTI_KEY", + .query = R"(join( + default.project({primaryKey, country}), + default.map({pk := primaryKey, ctry := country}).project({pk, ctry}), + primaryKey = pk && country = ctry + ).orderBy({asc(primaryKey)}))", + .expected_query_result = nlohmann::json( + {{{"primaryKey", "id_0"}, {"country", "CH"}, {"pk", "id_0"}, {"ctry", "CH"}}, + {{"primaryKey", "id_1"}, {"country", "DE"}, {"pk", "id_1"}, {"ctry", "DE"}}, + {{"primaryKey", "id_2"}, {"country", "CH"}, {"pk", "id_2"}, {"ctry", "CH"}}, + {{"primaryKey", "id_3"}, {"country", "DE"}, {"pk", "id_3"}, {"ctry", "DE"}}} + ) +}; + +// Inputs sharing column names are rejected: the join output would contain duplicate names. +const QueryTestScenario JOIN_OVERLAPPING_COLUMNS_SCENARIO = { + .name = "JOIN_OVERLAPPING_COLUMNS", + .query = R"(join( + default.project({primaryKey, country}), + default.project({primaryKey, country}), + primaryKey = primaryKey + ))", + .expected_query_result = {}, + .expected_error_message = + "join() requires the two inputs to have disjoint column names, but the column(s) " + "[primaryKey, country] are present in both. Rename one side (e.g. via map()) before joining." +}; + +// Only some of the columns overlap; the join is still rejected. +const QueryTestScenario JOIN_PARTIALLY_OVERLAPPING_COLUMNS_SCENARIO = { + .name = "JOIN_PARTIALLY_OVERLAPPING_COLUMNS", + .query = R"(join( + default.project({primaryKey, country}), + default.map({pk := primaryKey}).project({pk, country}), + primaryKey = pk + ))", + .expected_query_result = {}, + .expected_error_message = + "join() requires the two inputs to have disjoint column names, but the column(s) [country] " + "are present in both. Rename one side (e.g. via map()) before joining." +}; + +// An on-expression referencing a column that exists in neither input. +const QueryTestScenario JOIN_UNKNOWN_COLUMN_SCENARIO = { + .name = "JOIN_UNKNOWN_COLUMN", + .query = R"(join( + default.project({primaryKey}), + default.map({pk := primaryKey}).project({pk}), + primaryKey = doesNotExist + ))", + .expected_query_result = {}, + .expected_error_message = "join() on-expression references unknown column 'doesNotExist'" +}; + +// An unknown join type symbol. +const QueryTestScenario JOIN_INVALID_TYPE_SCENARIO = { + .name = "JOIN_INVALID_TYPE", + .query = R"(join( + default.project({primaryKey}), + default.map({pk := primaryKey}).project({pk}), + primaryKey = pk, + type := sideways + ))", + .expected_query_result = {}, + .expected_error_message = + "invalid join type 'sideways'. Valid types are: inner, left, right, full, leftSemi, " + "rightSemi, leftAnti, rightAnti" +}; +} // namespace + +QUERY_TEST( + JoinTest, + TEST_DATA, + ::testing::Values( + JOIN_INNER_ON_KEY_SCENARIO, + JOIN_PIPED_SYNTAX_SCENARIO, + JOIN_EXPLICIT_INNER_SCENARIO, + JOIN_MANY_TO_MANY_SCENARIO, + JOIN_LEFT_OUTER_SCENARIO, + JOIN_LEFT_SEMI_SCENARIO, + JOIN_LEFT_ANTI_SCENARIO, + JOIN_RIGHT_OUTER_SCENARIO, + JOIN_FULL_OUTER_SCENARIO, + JOIN_RIGHT_SEMI_SCENARIO, + JOIN_RIGHT_ANTI_SCENARIO, + JOIN_DOWNSTREAM_FILTER_SCENARIO, + JOIN_WITH_GROUPBY_SCENARIO, + JOIN_MULTI_KEY_SCENARIO, + JOIN_OVERLAPPING_COLUMNS_SCENARIO, + JOIN_PARTIALLY_OVERLAPPING_COLUMNS_SCENARIO, + JOIN_UNKNOWN_COLUMN_SCENARIO, + JOIN_INVALID_TYPE_SCENARIO + ) +); diff --git a/src/silo/query_engine/operators/query_node.cpp b/src/silo/query_engine/operators/query_node.cpp index 899180c91..a11d97dae 100644 --- a/src/silo/query_engine/operators/query_node.cpp +++ b/src/silo/query_engine/operators/query_node.cpp @@ -54,6 +54,8 @@ std::string_view nodeKindToString(NodeKind kind) { return "Map"; case NodeKind::UNION_ALL: return "UnionAll"; + case NodeKind::JOIN: + return "Join"; case NodeKind::SCHEMA: return "Schema"; case NodeKind::BITMAP_AGGREGATION: diff --git a/src/silo/query_engine/operators/query_node.h b/src/silo/query_engine/operators/query_node.h index f26354a3b..18a4c0440 100644 --- a/src/silo/query_engine/operators/query_node.h +++ b/src/silo/query_engine/operators/query_node.h @@ -37,6 +37,7 @@ enum class NodeKind : uint8_t { TABLE_SCAN, COUNT_FILTER, UNION_ALL, + JOIN, SCHEMA, BITMAP_AGGREGATION, }; diff --git a/src/silo/query_engine/optimizer/column_narrowing_pass.cpp b/src/silo/query_engine/optimizer/column_narrowing_pass.cpp index 634d50647..ecb016d30 100644 --- a/src/silo/query_engine/optimizer/column_narrowing_pass.cpp +++ b/src/silo/query_engine/optimizer/column_narrowing_pass.cpp @@ -4,6 +4,7 @@ #include #include "silo/query_engine/operators/aggregate_node.h" +#include "silo/query_engine/operators/join_node.h" #include "silo/query_engine/operators/map_node.h" #include "silo/query_engine/operators/order_by_node.h" #include "silo/query_engine/operators/project_node.h" @@ -171,6 +172,21 @@ operators::QueryNodePtr ColumnNarrowingPass::operator()(operators::SchemaNode& n return nullptr; } +// NOLINTNEXTLINE(misc-no-recursion,readability-make-member-function-const) +operators::QueryNodePtr ColumnNarrowingPass::operator()(operators::JoinNode& node) { + // A join reads columns from both inputs (at minimum its key columns), and the mapping + // between the join's own required columns and each input's columns is not a simple + // subset relation. Rather than track that, recurse into each branch with a fresh pass + // seeded with that branch's complete output schema: no column is pruned directly below + // the join (keeping the join keys and all outputs intact), while structural + // simplifications deeper in each branch still run. + ColumnNarrowingPass left_pass{node.left->getOutputSchema()}; + left_pass.propagateToNode(node.left); + ColumnNarrowingPass right_pass{node.right->getOutputSchema()}; + right_pass.propagateToNode(node.right); + return nullptr; +} + // NOLINTNEXTLINE(misc-no-recursion,readability-make-member-function-const) operators::QueryNodePtr ColumnNarrowingPass::operator()(operators::UnionAllNode& node) { // Narrow columns in each child independently using the same required set. Each branch needs diff --git a/src/silo/query_engine/optimizer/column_narrowing_pass.h b/src/silo/query_engine/optimizer/column_narrowing_pass.h index 3298523af..b87e767a9 100644 --- a/src/silo/query_engine/optimizer/column_narrowing_pass.h +++ b/src/silo/query_engine/optimizer/column_narrowing_pass.h @@ -13,6 +13,7 @@ class ProjectNode; class MapNode; class OrderByNode; class UnionAllNode; +class JoinNode; class SchemaNode; } // namespace silo::query_engine::operators @@ -42,6 +43,7 @@ class ColumnNarrowingPass : public PipelinePassBase { operators::QueryNodePtr operator()(operators::MapNode& node); operators::QueryNodePtr operator()(operators::OrderByNode& node); operators::QueryNodePtr operator()(operators::UnionAllNode& node); + operators::QueryNodePtr operator()(operators::JoinNode& node); operators::QueryNodePtr operator()(operators::SchemaNode& node); }; diff --git a/src/silo/query_engine/optimizer/filter_pushdown_pass.cpp b/src/silo/query_engine/optimizer/filter_pushdown_pass.cpp index b58117432..a796a9985 100644 --- a/src/silo/query_engine/optimizer/filter_pushdown_pass.cpp +++ b/src/silo/query_engine/optimizer/filter_pushdown_pass.cpp @@ -1,10 +1,13 @@ #include "silo/query_engine/optimizer/filter_pushdown_pass.h" +#include + #include "silo/common/aa_symbols.h" #include "silo/common/nucleotide_symbols.h" #include "silo/query_engine/illegal_query_exception.h" #include "silo/query_engine/operators/filter_node.h" #include "silo/query_engine/operators/insertions_node.h" +#include "silo/query_engine/operators/join_node.h" #include "silo/query_engine/operators/most_recent_common_ancestor_node.h" #include "silo/query_engine/operators/mutations_node.h" #include "silo/query_engine/operators/phylo_subtree_node.h" @@ -92,6 +95,30 @@ operators::QueryNodePtr FilterPushdownPass::operator()(operators::SchemaNode& no return nullptr; } +// NOLINTNEXTLINE(misc-no-recursion) +operators::QueryNodePtr FilterPushdownPass::operator()(operators::JoinNode& node) { + // A filter sitting above a join is not turned into a pre-join filter. Which input a + // predicate belongs to could be derived from freeIUs(), but pushing is only + // semantics-preserving for some combinations: pushing into the null-supplying side of an + // outer join changes the result (null-extended rows would no longer be filtered out), as + // does pushing a predicate that references no column at all. Rather than push unsafely, + // reject any filter above join() and point the user at the inputs. + CHECK_SILO_QUERY( + current_filters.empty(), + "filter() cannot be applied to the output of join(); a filter above a join cannot be " + "pushed into a join input safely. Apply the filter to one of the join inputs instead." + ); + + // No filters to carry across, but the child subtrees may still contain FilterNodes of + // their own (e.g. `join(default.filter(...), ...)`); push those down within each input + // using fresh passes so no state leaks between the two branches. + FilterPushdownPass left_pass; + FilterPushdownPass right_pass; + left_pass.propagateToNode(node.left); + right_pass.propagateToNode(node.right); + return nullptr; +} + // NOLINTNEXTLINE(misc-no-recursion) operators::QueryNodePtr FilterPushdownPass::operator()(operators::UnionAllNode& node) { // Push parent filters into both children. Clone for right, move originals into left. diff --git a/src/silo/query_engine/optimizer/filter_pushdown_pass.h b/src/silo/query_engine/optimizer/filter_pushdown_pass.h index a50367d1a..16e2e7c19 100644 --- a/src/silo/query_engine/optimizer/filter_pushdown_pass.h +++ b/src/silo/query_engine/optimizer/filter_pushdown_pass.h @@ -14,6 +14,7 @@ class InsertionsNode; class PhyloSubtreeNode; class MostRecentCommonAncestorNode; class UnionAllNode; +class JoinNode; class SchemaNode; } // namespace silo::query_engine::operators @@ -39,6 +40,8 @@ class FilterPushdownPass : public PipelinePassBase { operators::QueryNodePtr operator()(operators::SchemaNode& node); operators::QueryNodePtr operator()(operators::UnionAllNode& node); + + operators::QueryNodePtr operator()(operators::JoinNode& node); }; } // namespace silo::query_engine::optimizer diff --git a/src/silo/query_engine/optimizer/filter_pushdown_pass.test.cpp b/src/silo/query_engine/optimizer/filter_pushdown_pass.test.cpp index 08b74cda0..8f411bb05 100644 --- a/src/silo/query_engine/optimizer/filter_pushdown_pass.test.cpp +++ b/src/silo/query_engine/optimizer/filter_pushdown_pass.test.cpp @@ -6,7 +6,11 @@ #include #include +#include + +#include "silo/query_engine/illegal_query_exception.h" #include "silo/query_engine/operators/filter_node.h" +#include "silo/query_engine/operators/join_node.h" #include "silo/query_engine/operators/map_node.h" #include "silo/query_engine/operators/project_node.h" #include "silo/query_engine/operators/table_scan_node.h" @@ -204,4 +208,46 @@ TEST(FilterPushdownPass, pushesFilterIntoBothUnionAllBranches) { EXPECT_EQ(right_scan->filter->toString(), "And(true & true & true)"); } +// --- FilterNode(JoinNode(...)) --- + +operators::QueryNodePtr makeJoin(operators::QueryNodePtr left, operators::QueryNodePtr right) { + return std::make_unique( + std::move(left), + std::move(right), + std::vector{}, + std::vector{}, + arrow::acero::JoinType::INNER + ); +} + +// A filter sitting above a join cannot be pushed into a single input safely, so it is +// rejected rather than silently mis-attributed to one side. +TEST(FilterPushdownPass, rejectsFilterAboveJoin) { + auto join = makeJoin(makeScan(), makeScan()); + auto filter_node = std::make_unique(std::move(join), makeDummyFilter()); + + EXPECT_THROW( + { FilterPushdownPass::run(std::move(filter_node)); }, + silo::query_engine::IllegalQueryException + ); +} + +// Filters that live *inside* a join input are still pushed down into that input's scan; +// only filters stacked on top of the join itself are rejected. +TEST(FilterPushdownPass, pushesFiltersInsideJoinInputsIntoScans) { + auto join = makeJoin(makeFilteredScan(false), makeScan()); + + auto result = FilterPushdownPass::run(std::move(join)); + + ASSERT_EQ(result->kind(), operators::NodeKind::JOIN); + auto* join_node = dynamic_cast(result.get()); + ASSERT_EQ(join_node->left->kind(), operators::NodeKind::TABLE_SCAN); + ASSERT_EQ(join_node->right->kind(), operators::NodeKind::TABLE_SCAN); + auto* left_scan = dynamic_cast(join_node->left.get()); + auto* right_scan = dynamic_cast(join_node->right.get()); + // left: branch filter (false) + scan filter (true); right: only its scan filter (true) + EXPECT_EQ(left_scan->filter->toString(), "And(false & true)"); + EXPECT_EQ(right_scan->filter->toString(), "And(true)"); +} + } // namespace diff --git a/src/silo/query_engine/optimizer/pipeline_pass_base.h b/src/silo/query_engine/optimizer/pipeline_pass_base.h index 781a3dc0c..2ed427930 100644 --- a/src/silo/query_engine/optimizer/pipeline_pass_base.h +++ b/src/silo/query_engine/optimizer/pipeline_pass_base.h @@ -6,6 +6,7 @@ #include "silo/query_engine/operators/aggregate_node.h" #include "silo/query_engine/operators/fetch_node.h" #include "silo/query_engine/operators/filter_node.h" +#include "silo/query_engine/operators/join_node.h" #include "silo/query_engine/operators/map_node.h" #include "silo/query_engine/operators/order_by_node.h" #include "silo/query_engine/operators/project_node.h" @@ -114,9 +115,9 @@ class PipelinePassBase { return nullptr; } - // Default propagation for the two-child UnionAll operator. Both branches are - // walked by the same pass instance, so this default is only correct for - // stateless passes. A stateful pass MUST override this. + // Default propagation for the UnionAll operator. Both branches are walked by the same + // pass instance, so this default is only correct for stateless passes. + // A stateful pass MUST override this. // NOLINTNEXTLINE(misc-no-recursion) operators::QueryNodePtr operator()(operators::UnionAllNode& node) { propagateToNode(node.left); @@ -124,6 +125,16 @@ class PipelinePassBase { return nullptr; } + // Default propagation for the two-child Join operator. Both branches are walked + // by the same pass instance, so this default is only correct for stateless passes. + // A stateful pass MUST override this. + // NOLINTNEXTLINE(misc-no-recursion) + operators::QueryNodePtr operator()(operators::JoinNode& node) { + propagateToNode(node.left); + propagateToNode(node.right); + return nullptr; + } + template operators::QueryNodePtr operator()(T& /*node*/) { return nullptr; diff --git a/src/silo/query_engine/saneql/ast_to_query.cpp b/src/silo/query_engine/saneql/ast_to_query.cpp index c33bb821c..363e9058c 100644 --- a/src/silo/query_engine/saneql/ast_to_query.cpp +++ b/src/silo/query_engine/saneql/ast_to_query.cpp @@ -19,6 +19,7 @@ #include "silo/query_engine/operators/aggregate_node.h" #include "silo/query_engine/operators/fetch_node.h" #include "silo/query_engine/operators/filter_node.h" +#include "silo/query_engine/operators/join_node.h" #include "silo/query_engine/operators/map_node.h" #include "silo/query_engine/operators/order_by_node.h" #include "silo/query_engine/operators/project_node.h" @@ -1379,6 +1380,201 @@ operators::QueryNodePtr handlePhyloSubtree( ); } +namespace { + +std::optional findColumnByName( + const std::vector& schema, + const std::string& name +) { + auto found = std::ranges::find_if(schema, [&](const auto& col) { return col.name == name; }); + if (found == schema.end()) { + return std::nullopt; + } + return *found; +} + +struct JoinKeys { + std::vector left; + std::vector right; +}; + +/// Resolves an identifier appearing in a join condition against exactly one of the two +/// join inputs. Errors if the name is present on neither side or both sides +enum class JoinSide : uint8_t { LEFT, RIGHT }; + +struct ResolvedJoinColumn { + JoinSide side; + schema::ColumnIdentifier column; +}; + +ResolvedJoinColumn resolveJoinColumn( + const ast::Expression& expression, + const std::vector& left_schema, + const std::vector& right_schema +) { + CHECK_SILO_QUERY( + std::holds_alternative(expression.value), + "join() on-expression must compare column identifiers, got '{}' at {}:{}", + expression.toString(), + expression.location.line, + expression.location.column + ); + const auto name = extractIdentifierName(expression); + auto in_left = findColumnByName(left_schema, name); + auto in_right = findColumnByName(right_schema, name); + CHECK_SILO_QUERY( + !(in_left.has_value() && in_right.has_value()), + "join() on-expression references column '{}', which exists in both inputs and is therefore " + "ambiguous. Rename one side (e.g. via map()) before joining.", + name + ); + CHECK_SILO_QUERY( + in_left.has_value() || in_right.has_value(), + "join() on-expression references unknown column '{}'", + name + ); + if (in_left.has_value()) { + return {.side = JoinSide::LEFT, .column = in_left.value()}; + } + return {.side = JoinSide::RIGHT, .column = in_right.value()}; +} + +// NOLINTNEXTLINE(misc-no-recursion) +void collectJoinKeys( + const ast::Expression& on_expression, + const std::vector& left_schema, + const std::vector& right_schema, + JoinKeys& keys +) { + CHECK_SILO_QUERY( + std::holds_alternative(on_expression.value), + "join() on-expression must be an equality between a left and a right column, or a " + "conjunction (&&) of such equalities, at {}:{}", + on_expression.location.line, + on_expression.location.column + ); + const auto& binary = std::get(on_expression.value); + if (binary.op == ast::BinaryOp::AND) { + collectJoinKeys(*binary.left, left_schema, right_schema, keys); + collectJoinKeys(*binary.right, left_schema, right_schema, keys); + return; + } + CHECK_SILO_QUERY( + binary.op == ast::BinaryOp::EQUALS, + "join() on-expression only supports equality (=) comparisons, optionally combined with " + "'&&', at {}:{}", + on_expression.location.line, + on_expression.location.column + ); + auto first = resolveJoinColumn(*binary.left, left_schema, right_schema); + auto second = resolveJoinColumn(*binary.right, left_schema, right_schema); + CHECK_SILO_QUERY( + first.side != second.side, + "join() on-expression equality must reference one column from each input, but both '{}' and " + "'{}' resolve to the same input at {}", + binary.left->toString(), + binary.right->toString(), + on_expression.location.toString() + ); + CHECK_SILO_QUERY( + first.column.type == second.column.type, + "join() on-expression equality must reference equal column types from each input, but " + "'{}' and '{}' have mismatching types {} and {} at {}", + binary.left->toString(), + binary.right->toString(), + schema::columnTypeToString(first.column.type), + schema::columnTypeToString(second.column.type), + on_expression.location.toString() + ); + if (first.side == JoinSide::LEFT) { + keys.left.push_back(first.column); + keys.right.push_back(second.column); + } else { + keys.left.push_back(second.column); + keys.right.push_back(first.column); + } +} + +arrow::acero::JoinType parseJoinType(const BoundArguments& args) { + if (!args.has("type")) { + return arrow::acero::JoinType::INNER; + } + const auto* type_expr = args.get("type"); + const auto name = extractIdentifierName(*type_expr); + if (name == "inner") { + return arrow::acero::JoinType::INNER; + } + if (name == "left") { + return arrow::acero::JoinType::LEFT_OUTER; + } + if (name == "right") { + return arrow::acero::JoinType::RIGHT_OUTER; + } + if (name == "full") { + return arrow::acero::JoinType::FULL_OUTER; + } + if (name == "leftSemi") { + return arrow::acero::JoinType::LEFT_SEMI; + } + if (name == "rightSemi") { + return arrow::acero::JoinType::RIGHT_SEMI; + } + if (name == "leftAnti") { + return arrow::acero::JoinType::LEFT_ANTI; + } + if (name == "rightAnti") { + return arrow::acero::JoinType::RIGHT_ANTI; + } + throw IllegalQueryException( + "invalid join type '{}'. Valid types are: inner, left, right, full, leftSemi, rightSemi, " + "leftAnti, rightAnti", + name + ); +} + +} // namespace + +// NOLINTNEXTLINE(misc-no-recursion) +operators::QueryNodePtr handleJoin( + const BoundArguments& args, + const Tables& tables, + const ChildConverter& convert_child +) { + auto left = convert_child(args.at("left"), tables); + auto right = convert_child(args.at("right"), tables); + + auto left_schema = left->getOutputSchema(); + auto right_schema = right->getOutputSchema(); + + // A join concatenates the columns of both inputs, so a name occurring on both sides would + // produce a result with duplicate column names and make the on-expression ambiguous. + std::vector overlapping_names; + for (const auto& left_column : left_schema) { + if (findColumnByName(right_schema, left_column.name).has_value()) { + overlapping_names.push_back(left_column.name); + } + } + CHECK_SILO_QUERY( + overlapping_names.empty(), + "join() requires the two inputs to have disjoint column names, but the column(s) [{}] are " + "present in both. Rename one side (e.g. via map()) before joining.", + fmt::join(overlapping_names, ", ") + ); + + JoinKeys keys; + collectJoinKeys(args.at("on"), left_schema, right_schema, keys); + CHECK_SILO_QUERY( + !keys.left.empty(), + "join() on-expression must contain at least one equality between a left and a right column" + ); + + const auto join_type = parseJoinType(args); + + return std::make_unique( + std::move(left), std::move(right), std::move(keys.left), std::move(keys.right), join_type + ); +} + // NOLINTNEXTLINE(misc-no-recursion) operators::QueryNodePtr handleUnionAll( const BoundArguments& args, @@ -1525,6 +1721,10 @@ FunctionRegistry::FunctionRegistry() { ); registerFunction("unionAll", {{pos("left"), pos("right")}}, handleUnionAll); + + registerFunction( + "join", {{pos("left"), pos("right"), pos("on"), named("type", false)}}, handleJoin + ); } FunctionRegistry& FunctionRegistry::instance() { diff --git a/src/silo/query_engine/saneql/ast_to_query.test.cpp b/src/silo/query_engine/saneql/ast_to_query.test.cpp index feb672725..39c8e02c4 100644 --- a/src/silo/query_engine/saneql/ast_to_query.test.cpp +++ b/src/silo/query_engine/saneql/ast_to_query.test.cpp @@ -596,6 +596,104 @@ TEST(AstToQueryNOf, matchExactlyNotBoolThrows) { ); } +// --- collectJoinKeys --- + +TEST(AstToQueryJoin, onExpressionNotABinaryExpressionThrows) { + auto tables = makeTablesWithDefault(); + EXPECT_THAT( + [&tables]() { + (void)parseAndConvertToQueryTree( + "join(default.project({id}), default.map({pk := id}).project({pk}), id)", tables + ); + }, + ThrowsMessage(::testing::HasSubstr( + "join() on-expression must be an equality between a left and a right column, or a " + "conjunction (&&) of such equalities" + )) + ); +} + +TEST(AstToQueryJoin, conjunctOfOnExpressionNotABinaryExpressionThrows) { + // The same check, but reached through the recursive descent into a '&&' conjunction. + auto tables = makeTablesWithDefault(); + EXPECT_THAT( + [&tables]() { + (void)parseAndConvertToQueryTree( + "join(default.project({id, date}), default.map({pk := id}).project({pk}), id = pk && " + "date)", + tables + ); + }, + ThrowsMessage(::testing::HasSubstr( + "join() on-expression must be an equality between a left and a right column, or a " + "conjunction (&&) of such equalities" + )) + ); +} + +TEST(AstToQueryJoin, onExpressionNotAnEqualityThrows) { + auto tables = makeTablesWithDefault(); + EXPECT_THAT( + [&tables]() { + (void)parseAndConvertToQueryTree( + "join(default.project({id}), default.map({pk := id}).project({pk}), id <> pk)", tables + ); + }, + ThrowsMessage(::testing::HasSubstr( + "join() on-expression only supports equality (=) comparisons, optionally combined with " + "'&&'" + )) + ); +} + +TEST(AstToQueryJoin, onExpressionEqualityOfTwoColumnsOfTheSameInputThrows) { + auto tables = makeTablesWithDefault(); + EXPECT_THAT( + [&tables]() { + (void)parseAndConvertToQueryTree( + "join(default.project({id, date}), default.map({pk := id}).project({pk}), id = date)", + tables + ); + }, + ThrowsMessage(::testing::HasSubstr( + "join() on-expression equality must reference one column from each input, but both 'id' " + "and 'date' resolve to the same input" + )) + ); +} + +TEST(AstToQueryJoin, onExpressionEqualityOfMismatchingColumnTypesThrows) { + auto tables = makeTablesWithDefault(); + EXPECT_THAT( + [&tables]() { + (void)parseAndConvertToQueryTree( + "join(default.project({id}), default.map({num := 3}).project({num}), id = num)", tables + ); + }, + ThrowsMessage(::testing::HasSubstr( + "join() on-expression equality must reference equal column types from each input, but " + "'id' and 'num' have mismatching types STRING and INT64" + )) + ); +} + +// --- resolveJoinColumn --- + +TEST(AstToQueryJoin, onExpressionComparingANonIdentifierThrows) { + auto tables = makeTablesWithDefault(); + EXPECT_THAT( + [&tables]() { + (void)parseAndConvertToQueryTree( + "join(default.project({id}), default.map({pk := id}).project({pk}), id = 'value')", + tables + ); + }, + ThrowsMessage( + ::testing::HasSubstr("join() on-expression must compare column identifiers") + ) + ); +} + // --- convertExpression --- TEST(AstToQueryConvertExpression, nonIdentifierNonFunctionCallThrows) {