Join query

engine/CMakeLists.txt

@@ -29,9 +29,12 @@ add_library(engine
     src/evaluator/evaluator.cpp
     src/evaluator/knn.cpp
     src/query/knn.cpp
+    src/query/frechet_helper.cpp
     src/evaluator/where_at.cpp
     src/query/range_query.cpp
     src/evaluator/range.cpp
+    src/query/join.cpp
+    src/evaluator/join.cpp
 )
 
 target_include_directories(engine PUBLIC include)
@@ -62,6 +65,7 @@ add_executable(engine_tests
     tests/parallel_test.cpp
     tests/knn_test.cpp
     tests/config_test.cpp
+    tests/join_test.cpp
 )
 target_link_libraries(engine_tests PRIVATE engine GTest::gtest_main Arrow::arrow_shared Parquet::parquet_shared)
 gtest_discover_tests(engine_tests)

engine/configs/geolife.json

@@ -29,5 +29,9 @@
         "hit_radius_m": 100.0,
         "min_tolerance_m": 25.0,
         "max_tolerance_m": 500.0
+    },
+    "join": {
+        "query_count": 0,
+        "eps_m": 500.0
     }
 }

engine/include/backend/engine/evaluator/evaluator.hpp

@@ -1,8 +1,10 @@
 #pragma once
 
 #include "backend/engine/core/trajectory_db.hpp"
+#include "backend/engine/evaluator/join.hpp"
 #include "backend/engine/evaluator/knn.hpp"
 #include "backend/engine/evaluator/where_at.hpp"
+#include "backend/engine/query/join.hpp"
 #include "backend/engine/query/knn.hpp"
 #include "backend/engine/query/range_query.hpp"
 
@@ -15,13 +17,15 @@ struct Workloads {
     const query::KnnWorkload *knn = nullptr; // a pointer, as it also carries k and the query db
     std::span<const query::WhereAtQuery> where_at = {};
     std::span<const query::RangeQuery> range = {};
+    const query::JoinWorkload *join = nullptr; // a pointer, as it also carries eps and the queries
 };
 
 // Kendall's tau is in [-1, 1] and the others in [0, 1], or NaN if it could not be computed
 struct Scoretable {
     KnnScores knn;
     double range_recall;
     WhereAtScores where_at;
+    JoinScores join;
 };
 
 struct EvaluationResult {
@@ -41,6 +45,7 @@ class Evaluator {
     KnnScores get_knn_scores() const;
     double get_range_recall() const;
     WhereAtScores get_where_at_scores() const;
+    JoinScores get_join_scores() const;
 
   private:
     const core::TrajectoryDb &m_simplified;

engine/include/backend/engine/evaluator/join.hpp

@@ -0,0 +1,23 @@
+#pragma once
+
+#include "backend/engine/core/trajectory_db.hpp"
+#include "backend/engine/query/join.hpp"
+
+#include <cstddef>
+#include <limits>
+
+namespace evaluator {
+
+// Averages over queries. Precision skips queries with an empty simplified result, recall those
+// with an empty reference, F1 those where both are empty. NaN if no query counts.
+struct JoinScores {
+    double f1 = std::numeric_limits<double>::quiet_NaN();
+    double precision = std::numeric_limits<double>::quiet_NaN();
+    double recall = std::numeric_limits<double>::quiet_NaN();
+    double eps_m = std::numeric_limits<double>::quiet_NaN();
+    std::size_t query_count = 0;
+};
+
+JoinScores evaluate_join(const core::TrajectoryDb &compressed, const query::JoinWorkload &workload);
+
+} // namespace evaluator

engine/include/backend/engine/pipeline/config.hpp

@@ -1,6 +1,7 @@
 #pragma once
 
 #include "backend/engine/evaluator/where_at.hpp"
+#include "backend/engine/query/join.hpp"
 #include "backend/engine/query/knn.hpp"
 #include "backend/engine/query/range_query.hpp"
 #include "backend/engine/query/where_at_query.hpp"
@@ -26,6 +27,7 @@ struct Config {
     query::RangeConfig range;
     query::WhereAtConfig where_at;
     evaluator::WhereAtScoringConfig where_at_scoring;
+    query::JoinConfig join;
 };
 
 // Reads a JSON config. Only data_path is required, a missing key keeps its default.

engine/include/backend/engine/query/frechet_helper.hpp

@@ -0,0 +1,34 @@
+#pragma once
+
+#include "backend/engine/core/trajectory_db.hpp"
+
+#include <vector>
+
+namespace query {
+
+// Squared instead of std::hypot, which is slow. Fréchet only takes max and min of distances,
+// Square still has order
+inline double squared_point_dist(double a_x, double a_y, double b_x, double b_y) {
+    const double dx = a_x - b_x;
+    const double dy = a_y - b_y;
+    return dx * dx + dy * dy;
+}
+
+// Squared discrete Fréchet distance. Values below max_squared_dist are exact, anything else is
+// at or above it.
+double get_squared_frechet_distance(const core::TrajectoryView &query,
+                                    const core::TrajectoryView &reference, double max_squared_dist,
+                                    std::vector<double> &prev, std::vector<double> &curr);
+
+double squared_endpoint_lower_bound(const core::TrajectoryView &query,
+                                    const core::TrajectoryView &reference);
+
+double frechet_distance(const core::TrajectoryView &a, const core::TrajectoryView &b);
+
+// Whether the discrete Fréchet distance is at most max_dist, up to a relative 1e-12 for
+// rounding. Stops as soon as it is not.
+bool frechet_within(const core::TrajectoryView &a, const core::TrajectoryView &b, double max_dist);
+
+double endpoint_lower_bound(const core::TrajectoryView &a, const core::TrajectoryView &b);
+
+} // namespace query

engine/include/backend/engine/query/join.hpp

@@ -0,0 +1,37 @@
+#pragma once
+
+#include "backend/engine/core/trajectory_db.hpp"
+
+#include <cstddef>
+#include <cstdint>
+#include <vector>
+
+namespace query {
+
+struct JoinConfig {
+    std::size_t query_count = 10; // 0 = use every trajectory in the query pool
+    double eps_m = 500.0;         // join threshold in metres, must be positive
+};
+
+struct JoinQuery {
+    std::size_t query_index;
+    std::vector<std::size_t> reference;
+    std::vector<std::size_t> candidates;
+};
+
+struct JoinWorkload {
+    core::TrajectoryDb queries;
+    std::vector<JoinQuery> entries;
+    double eps_m = 0.0;
+};
+
+// Samples query_count trajectories from query_pool (the held-out split) and stores their join
+// result in original as the reference answer
+JoinWorkload generate_join_queries(const core::TrajectoryDb &original,
+                                   const core::TrajectoryDb &query_pool,
+                                   const JoinConfig &config = {}, std::uint64_t seed = 42);
+
+std::vector<std::size_t> join(const core::TrajectoryDb &trajectories,
+                              const core::TrajectoryView &query, double eps);
+
+} // namespace query

engine/src/evaluator/evaluator.cpp

@@ -25,7 +25,8 @@ EvaluationResult Evaluator::evaluate() const {
     auto range = evaluate_range(m_simplified, m_workloads.range);
     return {.scores = {.knn = knn.scores,
                        .range_recall = range.recall,
-                       .where_at = get_where_at_scores()},
+                       .where_at = get_where_at_scores(),
+                       .join = get_join_scores()},
             .knn = std::move(knn.answers),
             .range = std::move(range.answers)};
 }
@@ -42,4 +43,8 @@ WhereAtScores Evaluator::get_where_at_scores() const {
     return evaluate_where_at(m_simplified, m_workloads.where_at, m_where_at_config);
 }
 
+JoinScores Evaluator::get_join_scores() const {
+    return m_workloads.join ? evaluate_join(m_simplified, *m_workloads.join) : JoinScores{};
+}
+
 } // namespace evaluator

engine/src/evaluator/join.cpp

@@ -0,0 +1,73 @@
+#include "backend/engine/evaluator/join.hpp"
+#include "backend/engine/core/parallel.hpp"
+#include "backend/engine/query/frechet_helper.hpp"
+
+#include <algorithm>
+#include <cmath>
+#include <cstddef>
+#include <limits>
+#include <vector>
+
+namespace evaluator {
+
+namespace {
+
+struct Counts {
+    std::size_t found = 0;
+    std::size_t hits = 0;
+};
+
+double average(const std::vector<double> &values) {
+    double sum = 0.0;
+    std::size_t counted = 0;
+    for (const double value : values) {
+        if (!std::isnan(value)) {
+            sum += value;
+            ++counted;
+        }
+    }
+    return counted == 0 ? std::numeric_limits<double>::quiet_NaN()
+                        : sum / static_cast<double>(counted);
+}
+
+} // namespace
+
+JoinScores evaluate_join(const core::TrajectoryDb &compressed,
+                         const query::JoinWorkload &workload) {
+    const std::size_t n = workload.entries.size();
+    const double nan = std::numeric_limits<double>::quiet_NaN();
+    std::vector<double> f1(n, nan), precision(n, nan), recall(n, nan);
+
+    core::parallel_for(n, [&](std::size_t i) {
+        const auto &entry = workload.entries[i];
+        const auto query = workload.queries.trajectory(entry.query_index);
+
+        // Simplification keeps endpoints, so the original candidates cover every possible match
+        Counts counts;
+        for (const auto j : entry.candidates) {
+            if (!query::frechet_within(query, compressed.trajectory(j), workload.eps_m))
+                continue;
+            ++counts.found;
+            if (std::ranges::binary_search(entry.reference, j))
+                ++counts.hits;
+        }
+
+        const auto hits = static_cast<double>(counts.hits);
+        const auto found = static_cast<double>(counts.found);
+        const auto truth = static_cast<double>(entry.reference.size());
+        if (counts.found != 0)
+            precision[i] = hits / found;
+        if (!entry.reference.empty())
+            recall[i] = hits / truth;
+        if (counts.found != 0 || !entry.reference.empty())
+            f1[i] = 2.0 * hits / (found + truth);
+    });
+
+    return {.f1 = average(f1),
+            .precision = average(precision),
+            .recall = average(recall),
+            .eps_m = workload.eps_m,
+            .query_count = n};
+}
+
+} // namespace evaluator

engine/src/pipeline/config.cpp

@@ -84,10 +84,16 @@ void parse_where_at_scoring_config(Config &config, const json &scoring) {
         scoring.value("max_tolerance_m", config.where_at_scoring.max_tolerance_m);
 }
 
+void parse_join_config(Config &config, const json &join) {
+    validate_keys(join, {"query_count", "eps_m"}, "join");
+    config.join.query_count = join.value("query_count", config.join.query_count);
+    config.join.eps_m = join.value("eps_m", config.join.eps_m);
+}
+
 Config parse_config(const json &data) {
     validate_keys(data,
                   {"data_path", "export_results", "export_dir", "seed", "budget_shares",
-                   "algorithms", "split", "knn", "range", "where_at", "where_at_scoring"},
+                   "algorithms", "split", "knn", "range", "where_at", "where_at_scoring", "join"},
                   "root");
     Config config;
 
@@ -117,6 +123,7 @@ Config parse_config(const json &data) {
     parse_range_config(config, data.value("range", json::object()));
     parse_where_at_config(config, data.value("where_at", json::object()));
     parse_where_at_scoring_config(config, data.value("where_at_scoring", json::object()));
+    parse_join_config(config, data.value("join", json::object()));
 
     return config;
 }

engine/src/pipeline/pipeline.cpp

@@ -5,6 +5,7 @@
 #include "backend/engine/evaluator/evaluator.hpp"
 #include "backend/engine/io/parquet_loader.hpp"
 #include "backend/engine/io/workload_export.hpp"
+#include "backend/engine/query/join.hpp"
 #include "backend/engine/query/knn.hpp"
 #include "backend/engine/query/range_query.hpp"
 #include "backend/engine/query/where_at_query.hpp"
@@ -54,6 +55,8 @@ void print_scores(const evaluator::Scoretable &scores) {
     std::cout << "  where-at: within radius=" << scores.where_at.within_radius
               << ", average hit=" << scores.where_at.average_hit << '\n';
     std::cout << "  range recall: " << scores.range_recall << '\n';
+    std::cout << "  join F1: " << scores.join.f1 << ", precision: " << scores.join.precision
+              << ", recall: " << scores.join.recall << '\n';
 }
 
 } // namespace
@@ -71,6 +74,12 @@ void run(const Config &config) {
               << where_at.skipped_trajectories << " trips without eligible time\n";
     const auto range = query::generate_range_queries(database, config.range, config.seed);
     std::cout << "range: " << range.queries.size() << " queries\n";
+    const auto join = query::generate_join_queries(database, queries, config.join, config.seed);
+    std::size_t join_pairs = 0;
+    for (const auto &entry : join.entries)
+        join_pairs += entry.reference.size();
+    std::cout << "join: " << join.entries.size() << " queries, eps=" << join.eps_m << " m, "
+              << join_pairs << " matching pairs\n";
 
     if (config.export_results) {
         io::export_workloads(config.export_dir, database, range.queries, knn_workload);
@@ -87,10 +96,12 @@ void run(const Config &config) {
             std::cout << share * 100 << "% " << magic_enum::enum_name(algorithm) << ": ";
             print_size("compressed", compressed);
 
-            const evaluator::Evaluator evaluator(
-                database, compressed,
-                {.knn = &knn_workload, .where_at = where_at.queries, .range = range.queries},
-                config.where_at_scoring);
+            const evaluator::Evaluator evaluator(database, compressed,
+                                                 {.knn = &knn_workload,
+                                                  .where_at = where_at.queries,
+                                                  .range = range.queries,
+                                                  .join = &join},
+                                                 config.where_at_scoring);
             const auto evaluation = evaluator.evaluate();
             print_scores(evaluation.scores);