Add query test cases for LOOKUP join (#18279)

Author: yashmayya

pinot-query-planner/src/main/java/org/apache/pinot/query/routing/WorkerManager.java

@@ -149,7 +149,6 @@ private void assignWorkersToNonRootFragment(PlanFragment fragment, DispatchableP
         metadata.setWorkerIdToServerInstanceMap(workerIdToServerInstanceMap);
         metadata.setPartitionFunction(childMetadata.getPartitionFunction());
         // Fake a segments map so that the worker can be correctly identified as leaf stage
-        // TODO: Add a query test for LOOKUP join
         Map<String, List<String>> segmentsMap = Map.of(TableType.OFFLINE.name(), List.of());
         Map<Integer, Map<String, List<String>>> workerIdToSegmentsMap =
             Maps.newHashMapWithExpectedSize(workerIdToServerInstanceMap.size());

pinot-query-runtime/src/test/resources/queries/LookupJoin.json

@@ -1,6 +1,6 @@
 {
   "lookup_join_basic": {
-    "comment": "Tests lookup join with a dimension table. V1 and V2 pass end-to-end. Lite mode is skipped because the broker lacks DimensionTableDataManager. The V2 path is additionally validated at the planner level by LookupJoinPlanTest.",
+    "comment": "Tests lookup join with a dimension table: inner/left join, fact-side filter, and aggregation above the join. V1 and V2 pass end-to-end. Lite mode is skipped because the broker lacks DimensionTableDataManager. The V2 path is additionally validated at the planner level by LookupJoinPlanTest. Exercises WorkerManager.assignWorkersToNonRootFragment lookup-join branch on the V1 path.",
     "tables": {
       "fact_tbl": {
         "schema": [
@@ -41,6 +41,38 @@
           [3, "gamma"]
         ],
         "ignoreLiteMode": true
+      },
+      {
+        "description": "Left outer lookup join - unmatched fact row (dim_key=4) preserved with null dim column",
+        "sql": "SELECT /*+ joinOptions(join_strategy='lookup') */ {fact_tbl}.dim_key, {dim_tbl}.name FROM {fact_tbl} LEFT JOIN {dim_tbl} ON {fact_tbl}.dim_key = {dim_tbl}.id",
+        "outputs": [
+          [1, "alpha"],
+          [1, "alpha"],
+          [2, "beta"],
+          [3, "gamma"],
+          [4, null]
+        ],
+        "ignoreLiteMode": true
+      },
+      {
+        "description": "Lookup join with filter on fact-side column - filter applied before the join",
+        "sql": "SELECT /*+ joinOptions(join_strategy='lookup') */ {fact_tbl}.dim_key, {fact_tbl}.metric, {dim_tbl}.name FROM {fact_tbl} JOIN {dim_tbl} ON {fact_tbl}.dim_key = {dim_tbl}.id WHERE {fact_tbl}.metric >= 200",
+        "outputs": [
+          [2, 200, "beta"],
+          [3, 300, "gamma"],
+          [1, 400, "alpha"]
+        ],
+        "ignoreLiteMode": true
+      },
+      {
+        "description": "Aggregation above lookup join - counts matched fact rows per dim name",
+        "sql": "SELECT /*+ joinOptions(join_strategy='lookup') */ {dim_tbl}.name, COUNT(*) FROM {fact_tbl} JOIN {dim_tbl} ON {fact_tbl}.dim_key = {dim_tbl}.id GROUP BY {dim_tbl}.name",
+        "outputs": [
+          ["alpha", 2],
+          ["beta", 1],
+          ["gamma", 1]
+        ],
+        "ignoreLiteMode": true
       }
     ]
   }