handled the 5 task types

Author: mbakli

demo_queries/berlinmod/queries.sql

@@ -204,7 +204,6 @@ ORDER BY t.PointId, t.PeriodId, v.Licence;
 -- Q16) List the pairs of licences for vehicles, the first from Licences1, the second from
 -- Licences2, where the corresponding vehicles are both present within a region from Regions1
 -- during a period from Periods1, but do not meet each other there and then.
--- NOT CURRENTLY SUPPORTED: self-join on trips_16t with no distribution-column equi-join.
 -----------------------------------------------------------------------------------------------------------------------
 SELECT p.PeriodId, p.Period, r.RegionId,
   l1.Licence AS Licence1, l2.Licence AS Licence2

src/executor/multi_phase_executor.c

@@ -541,14 +541,33 @@ ConstructPredicatePushDownQuery(PlanTask *plan, char * query_string, MultiPhaseE
     multiPhaseExecutor->tasks = lappend(multiPhaseExecutor->tasks, task);
 }
 
-/* GetTaskType renders task's ExecTaskType as a human-readable label for EXPLAIN output. */
+/*
+ * GetTaskType renders task's ExecTaskType as a human-readable label for
+ * EXPLAIN output. Every ExecTaskType that ExplainPlanStrategies can
+ * actually iterate over (multiPhaseExecutor->tasks -- never the separate
+ * coordTasks list, so INTERMEDIATEScan/FINALScan don't reach here today)
+ * needs a case; falling off the end without returning left this Datum
+ * uninitialized, and the caller's DatumGetCString/appendStringInfo("%s")
+ * would then dereference whatever garbage pointer was left in the return
+ * register -- e.g. for a PushDownScan task (a single distributed table
+ * joined against reference tables only, no self-join), crashing EXPLAIN
+ * outright.
+ */
 extern Datum
 GetTaskType(ExecutorTask *task)
 {
     if (task->taskType == NeighborTilingScan)
         return CStringGetDatum("Neighbor Scan");
     else if (task->taskType == SelfTilingScan)
         return CStringGetDatum("Self Tiling Scan");
+    else if (task->taskType == PushDownScan)
+        return CStringGetDatum("Push Down Scan");
+    else if (task->taskType == INTERMEDIATEScan)
+        return CStringGetDatum("Intermediate Scan");
+    else if (task->taskType == FINALScan)
+        return CStringGetDatum("Final Scan");
+    else
+        return CStringGetDatum("Unknown Scan");
 }
 
 /*

src/planner/distributed_mobilitydb_explain.c

@@ -23,6 +23,7 @@
 #include "utils/planner_utils.h"
 #include "catalog/nodes.h"
 #include "planner/planner_strategies.h"
+#include "catalog/table_ops.h"
 #include <tcop/tcopprot.h>
 #include <executor/spi.h>
 #include <utils/builtins.h>
@@ -320,10 +321,11 @@ ExplainOneTask(ExecutorTask *task, STMultirelation *base,ExplainState *es, int i
 }
 
 /*
- * GetLocalQuery pins query_string to one representative tile by appending a
+ * GetLocalQuery pins query_string to one representative tile by adding a
  * literal `<alias>.tile_key = rand_tile` predicate for every range table
- * entry it references, so Citus' shard pruning narrows each table down to
- * the single matching shard instead of planning across all of them.
+ * entry that actually has a tile_key column, so Citus' shard pruning
+ * narrows each such table down to the single matching shard instead of
+ * planning across all of them.
  */
 static char *
 GetLocalQuery(char *query_string, Oid base, ExecTaskType taskType, int rand_tile)
@@ -333,18 +335,43 @@ GetLocalQuery(char *query_string, Oid base, ExecTaskType taskType, int rand_tile
 
     Query *query = ParseQueryString(query_string, NULL, 0);
     List *rangeTableList = ExtractRangeTableEntryList(query);
-    StringInfo pinnedQuery = makeStringInfo();
-    appendStringInfo(pinnedQuery, "%s", query_string);
+    StringInfo tileKeyConditions = makeStringInfo();
 
     ListCell *rangeTableCell = NULL;
     foreach(rangeTableCell, rangeTableList)
     {
         RangeTblEntry *rangeTableEntry = (RangeTblEntry *) lfirst(rangeTableCell);
-        appendStringInfo(pinnedQuery, " AND %s.%s = %d", rangeTableEntry->eref->aliasname,
+        /* Only tables tiled by this extension's own machinery (a
+         * distributed spatiotemporal table, or a plain table reshuffled
+         * to be colocated with one) actually have a tile_key column --
+         * unconditionally pinning every range table entry (as this used
+         * to) added "alias.tile_key = N" for reference tables too (e.g.
+         * vehicles_ref/points_ref), which have no such column at all,
+         * producing "column v.tile_key does not exist" instead of a plan. */
+        if (!IsDistributedSpatiotemporalTable(rangeTableEntry->relid) &&
+            !IsReshuffledTable(rangeTableEntry->relid))
+            continue;
+        appendStringInfo(tileKeyConditions, "%s.%s = %d AND ", rangeTableEntry->eref->aliasname,
                          Var_Catalog_Tile_Key, rand_tile);
     }
 
-    return pinnedQuery->data;
+    if (tileKeyConditions->len == 0)
+        return query_string;
+
+    /*
+     * Inserted right after the query's own WHERE keyword rather than
+     * appended at the very end -- appending unconditionally landed these
+     * AND-joined conditions after a trailing ORDER BY whenever the task
+     * query had one (e.g. Q16's per-tile query), silently folding them
+     * into the ORDER BY expression list instead of the WHERE clause:
+     * "ORDER BY ..., l2.licence AND t1.tile_key = 5 AND ..." parses as one
+     * AND-expression whose left operand is l2.licence (text), producing
+     * "argument of AND must be type boolean, not type text" instead of
+     * pinning the query to one tile.
+     */
+    StringInfo key = makeStringInfo();
+    appendStringInfo(key, "WHERE %s", tileKeyConditions->data);
+    return replaceWord(query_string, "where", key->data);
 }
 
 /*

src/planner/planner_strategies.c

@@ -537,15 +537,53 @@ PredicatePushDownStrategyPlan(DistributedSpatiotemporalQueryPlan *distPlan)
 {
     PlanTask * strategy = (PlanTask *) palloc0(sizeof(PlanTask));
     strategy->type = PredicatePushDown;
-    strategy->tbl1 = (STMultirelation *) list_nth(distPlan->tablesList->tables, 0);
-    strategy->tbl2 = (STMultirelation *) list_nth(distPlan->tablesList->tables, 1);
+    /* tablesList->tables holds Rte wrappers (STRte/CitusRte/LocalRte), not
+     * bare STMultirelation pointers -- list_nth(...)[0]/[1] cast directly
+     * to STMultirelation* (as this used to) read a Citus/local reference
+     * table's Rte wrapper as if it were the spatiotemporal table's own
+     * struct whenever one of the query's other tables sorted before it, a
+     * type confusion that left task->catalog_filtered pointing at garbage
+     * and crashed EXPLAIN's "Task Count: %d" (task->catalog_filtered->
+     * candidates). Find the actual STRte entry instead; PredicatePushDown
+     * only ever needs the one spatiotemporal table its predicate pushes
+     * down onto (see ConstructPredicatePushDownQuery, which only reads
+     * tbl1), so tbl2 is left unset.
+     */
+    ListCell *rangeTableCell = NULL;
+    STMultirelation *stTable = NULL;
+    foreach(rangeTableCell, distPlan->tablesList->tables)
+    {
+        Rte *rteNode = (Rte *) lfirst(rangeTableCell);
+        if (rteNode->RteType == STRte)
+        {
+            stTable = (STMultirelation *) rteNode->rte;
+            break;
+        }
+    }
+    if (stTable == NULL)
+        ereport(ERROR, (errmsg("PredicatePushDown strategy requires a spatiotemporal table")));
+    strategy->tbl1 = stTable;
     strategy->tileKey = (Datum) Var_Catalog_Tile_Key;
     distPlan->strategyPlans = lappend(distPlan->strategyPlans, strategy);
 }
 
-/* AddStrategy appends `type` to distPlan's list of chosen StrategyTypes. */
+/* AddStrategy appends `type` to distPlan's list of chosen StrategyTypes, if
+ * not already present. checkQueryType calls this once per registered
+ * predicate clause, and a query can have more than one predicate that maps
+ * to the same strategy between the same table pair (e.g. Q16's two
+ * ST_Intersects clauses plus an aDisjoint clause all resolve to Colocation)
+ * -- appending unconditionally queued the same strategy's plan/task/query
+ * multiple times, and ConstructGeneralQuery's UNION of one query per
+ * strategies-list entry then UNIONed several copies of a query that already
+ * has its own trailing ORDER BY, which Postgres rejects outright. */
 extern
 void AddStrategy(DistributedSpatiotemporalQueryPlan *distPlan, StrategyType type)
 {
+    ListCell *cell;
+    foreach(cell, distPlan->strategies)
+    {
+        if ((StrategyType) lfirst_int(cell) == type)
+            return;
+    }
     distPlan->strategies = lappend(distPlan->strategies, (Datum *)type);
 }

src/planner/query_parameters.c

@@ -14,6 +14,7 @@
 
 #include "postgres.h"
 #include <commands/explain.h>
+#include <distributed/pg_dist_partition.h>
 #include "planner/predicate_management.h"
 #include "multirelation/multirelation_utils.h"
 #include "planner/distributed_mobilitydb_planner.h"
@@ -68,20 +69,75 @@ static void ExplainMainPredicate(PredicateType predicateType, PredicateInfo * pr
  * touched by the query along with its tiling method, local index, and tile
  * count; repeated references to the same table (self-joins) are skipped
  * after the first.
+ *
+ * The header counts distinguish genuinely-distributed (tiled) spatiotemporal
+ * tables from replicated reference tables -- tablesList->length/diffCount
+ * count every range-table entry or distinct relid regardless of kind, so a
+ * query joining one tiled table against several reference tables (e.g. a
+ * self-join on trips_16t plus 4 reference-table references) used to print
+ * as "Distributed Tables:6" / "different tables: 4", reading as if several
+ * genuinely-sharded tables were involved instead of one.
  */
 static void ExplainDistributedTables(STMultirelations *tablesList, ExplainState *es, int indent_group)
 {
+    List *seenDistributedRelids = NIL;
+    List *seenReferenceRelids = NIL;
+    int distributedOccurrences = 0;
+    int referenceOccurrences = 0;
+    ListCell *countCell = NULL;
+    foreach(countCell, tablesList->tables)
+    {
+        Rte *rteNode = (Rte *) lfirst(countCell);
+        if (rteNode->RteType == STRte)
+        {
+            STMultirelation *st = (STMultirelation *) rteNode->rte;
+            distributedOccurrences++;
+            if (!list_member_oid(seenDistributedRelids, st->catalogTableInfo.table_oid))
+                seenDistributedRelids = lappend_oid(seenDistributedRelids, st->catalogTableInfo.table_oid);
+        }
+        else if (rteNode->RteType == CitusRte)
+        {
+            CitusRteNode *citusRte = (CitusRteNode *) rteNode->rte;
+            /* Reference tables report Citus' "none" partition method
+             * ('n', DISTRIBUTE_BY_NONE) -- the same value plain Citus
+             * local tables report, but a CitusRteNode only ever exists for
+             * a hash/range-distributed table or a reference table (see
+             * analyzeDistributedSpatiotemporalTables), so "not hash, not
+             * range" reliably means "reference table" here. */
+            if (citusRte->partitionMethod != DISTRIBUTE_BY_HASH &&
+                citusRte->partitionMethod != DISTRIBUTE_BY_RANGE)
+            {
+                Oid relid = ((RangeTblEntry *) lfirst(citusRte->rangeTableCell))->relid;
+                referenceOccurrences++;
+                if (!list_member_oid(seenReferenceRelids, relid))
+                    seenReferenceRelids = lappend_oid(seenReferenceRelids, relid);
+            }
+        }
+    }
+
+    int distinctDistributedCount = list_length(seenDistributedRelids);
+
     appendStringInfoSpaces(es->str, es->indent * indent_group);
-    appendStringInfo(es->str, "-> Distributed Tables:%d\n", tablesList->length);
+    appendStringInfo(es->str, "-> Distributed Tables:%d\n", distinctDistributedCount);
     es->indent += indent_group;
     ExplainOpenGroup("TablesInfo", "Distributed Tables Info", true, es);
     appendStringInfoSpaces(es->str, es->indent * indent_group);
-    appendStringInfo(es->str, "Number of similar tables: %d\n", tablesList->simCount);
+    /* "Similar"/"different" here are scoped to genuinely-distributed (tiled)
+     * tables only -- self-joins (e.g. trips_16t as both t1 and t2) count as
+     * "similar", and distinct distributed tables (e.g. two different tiled
+     * tables joined together) count as "different". Reference tables are
+     * reported separately below, never folded into either count. */
+    appendStringInfo(es->str, "Number of similar tables: %d\n",
+                     distributedOccurrences - distinctDistributedCount);
+    appendStringInfoSpaces(es->str, es->indent * indent_group);
+    appendStringInfo(es->str, "Number of different tables: %d\n", distinctDistributedCount);
     appendStringInfoSpaces(es->str, es->indent * indent_group);
-    if (tablesList->diffCount > 1)
-        appendStringInfo(es->str, "Number of different tables: %d\n", tablesList->diffCount);
+    if (referenceOccurrences > 0)
+        appendStringInfo(es->str, "Replicated (reference) tables: %d (%d reference%s)\n",
+                         list_length(seenReferenceRelids), referenceOccurrences,
+                         referenceOccurrences == 1 ? "" : "s");
     else
-        appendStringInfo(es->str, "Number of different tables: %d\n", 0);
+        appendStringInfo(es->str, "Replicated (reference) tables: 0\n");
     appendStringInfoSpaces(es->str, es->indent * indent_group);
     ListCell *rangeTableCell = NULL;
     char * check = NULL;