From 0a65fc14194dce6732551b1e45311dc790307874 Mon Sep 17 00:00:00 2001
From: Esteban Zimanyi <estebanzimanyi@gmail.com>
Date: Sun, 25 Oct 2026 12:52:20 +0100
Subject: [PATCH] Evaluate the topological and position overloads a hundred to
 a query

everyOverloadAnswersUnquoted evaluates the two-argument overloads of the topological and position
families a hundred to a query: the calls are the columns of one query, each column required to
answer a Boolean, and a query that fails is run again call by call through #scalar, so a failing
overload is named by its own query and the innermost cause.

Witness: each overload was a query of its own, and each query is a job the mini-cluster plans and
submits, so the test spent its time on jobs rather than on calls.

Measured: on the CI of main the step building and testing the binding takes 1274 s, against 154 s
for MobilitySpark's, and TopologicalPositionSurfaceTest takes 1022 s of it for its three tests. A
query costs 0.53 s on Flink and 0.08 s on Spark, read from AggregatesCatalogTest (288 queries in
152.7 s, 295 in 23.4 s). On the CI of this change (fork run 40190727763) the three tests of
TopologicalPositionSurfaceTest pass in 116.7 s.

Why: a query per call measures Flink's job submission, not the surface; sharing the jobs keeps
every call and every check while the test costs a tenth of its time.
---
 .../sql/TopologicalPositionSurfaceTest.java   | 62 +++++++++++++++----
 1 file changed, 50 insertions(+), 12 deletions(-)

diff --git a/binding/src/test/java/org/mobilitydb/flink/sql/TopologicalPositionSurfaceTest.java b/binding/src/test/java/org/mobilitydb/flink/sql/TopologicalPositionSurfaceTest.java
index 01b9239..75160fc 100644
--- a/binding/src/test/java/org/mobilitydb/flink/sql/TopologicalPositionSurfaceTest.java
+++ b/binding/src/test/java/org/mobilitydb/flink/sql/TopologicalPositionSurfaceTest.java
@@ -306,11 +306,55 @@ void everyTopologicalNameIsRegistered() {
         assertTrue(missing.isEmpty(), "topological names not registered: " + missing);
     }
 
+    /** The first line of the innermost cause of a failed call. */
+    private static String rootMessage(Exception e) {
+        Throwable root = e;
+        while (root.getCause() != null) {
+            root = root.getCause();
+        }
+        return String.valueOf(root.getMessage()).split("\n")[0];
+    }
+
+    /** The calls one query evaluates: each query is a job on the mini-cluster, whose planning and
+     * submission cost far more than a call, so the calls share their jobs. */
+    private static final int CALLS_PER_QUERY = 100;
+
+    /** The failures of the calls {@code exprs}, evaluated as the columns of one query, each call
+     * of a failing query then named by a query of its own, read by #scalar. No test of the
+     * surface evaluates several calls in one query; this extends #scalar, which reads one. */
+    private static List<String> answerBoolean(List<String> exprs) {
+        List<String> failures = new ArrayList<>();
+        try (CloseableIterator<Row> it =
+                tEnv.executeSql("SELECT " + String.join(", ", exprs)).collect()) {
+            Row row = it.next();
+            for (int i = 0; i < exprs.size(); i++) {
+                if (!(row.getField(i) instanceof Boolean)) {
+                    failures.add("SELECT " + exprs.get(i) + " -> " + row.getField(i));
+                }
+            }
+            return failures;
+        } catch (Exception shared) {
+            for (String expr : exprs) {
+                String sql = "SELECT " + expr;
+                try {
+                    Object answer = scalar(sql);
+                    if (!(answer instanceof Boolean)) {
+                        failures.add(sql + " -> " + answer);
+                    }
+                } catch (Exception e) {
+                    failures.add(sql + " -> " + rootMessage(e));
+                }
+            }
+            return failures;
+        }
+    }
+
     @Test
     void everyOverloadAnswersUnquoted() throws Exception {
         Set<String> names = familyNames();
         int calls = 0;
         List<String> failures = new ArrayList<>();
+        List<String> exprs = new ArrayList<>();
         for (String name : names) {
             Matcher m = FAMILY.matcher(name);
             assertTrue(m.matches());
@@ -321,24 +365,18 @@ void everyOverloadAnswersUnquoted() throws Exception {
                     continue;
                 }
                 Class<?>[] p = eval.getParameterTypes();
-                String sql = name + Arrays.toString(p);
                 try {
-                    sql = "SELECT " + name + "(" + arg(p[0], p[1], z) + ", "
-                            + arg(p[1], p[0], z) + ")";
-                    Object answer = scalar(sql);
-                    if (!(answer instanceof Boolean)) {
-                        failures.add(sql + " -> " + answer);
-                    }
+                    exprs.add(name + "(" + arg(p[0], p[1], z) + ", " + arg(p[1], p[0], z) + ")");
                 } catch (Exception e) {
-                    Throwable root = e;
-                    while (root.getCause() != null) {
-                        root = root.getCause();
-                    }
-                    failures.add(sql + " -> " + String.valueOf(root.getMessage()).split("\n")[0]);
+                    failures.add(name + Arrays.toString(p) + " -> " + rootMessage(e));
                 }
                 calls++;
             }
         }
+        for (int i = 0; i < exprs.size(); i += CALLS_PER_QUERY) {
+            failures.addAll(answerBoolean(
+                    exprs.subList(i, Math.min(i + CALLS_PER_QUERY, exprs.size()))));
+        }
         assertTrue(calls > 0, "no overload of the two families is registered");
         assertTrue(failures.isEmpty(), failures.size() + " of " + calls
                 + " overloads of " + names.size() + " names failed:\n"