Merge pull request #22 from mbakli/develop

Improved query planning

Author: mbakli

README.md

@@ -17,9 +17,18 @@
 #include "postgres.h"
 #include "executor/executor_tasks.h"
 
+/* Name of the database the current backend is connected to. */
 extern Datum GetDBName();
+
+/* Coordinator/worker node info (host, port, role) for the local backend. */
 extern TaskNode *GetNodeInfo();
+
+/* Table id of the tile assigned to a randomly-picked worker for relationId. */
 extern char* GetRandomTileId(Oid relationId, ExecTaskType taskType, int rand_tile);
+
+/* Looks up the tiling method used to distribute relationId; -1 if not distributed. */
 extern int TilingSearch(Oid relationId);
+
+/* Number of tiles the given distributed relation was split into. */
 extern int GetNumTiles(Oid relationId);
 #endif /* NODESS_H */

include/catalog/nodes.h

@@ -17,6 +17,12 @@
 
 /* ----------------
  *  pg_dist_spatiotemporal_dist_functions: Compiler constants
+ *
+ *  Catalog table that maps a distributable aggregate/function (id) to the
+ *  three-phase function set used to run it across the cluster: the
+ *  per-tile `worker` function, the `combiner` that merges partial results,
+ *  and the `final` function that produces the end result. `sexec_id` ties
+ *  the row to the pg_execution_run entry for the query that registered it.
  * ----------------
  */
 

include/catalog/pg_dist_spatiotemporal_dist_functions.h

@@ -17,6 +17,12 @@
 
 /* ----------------
  *  pg_dist_spatiotemporal_tables: Compiler constants
+ *
+ *  Catalog table holding one row per table that has been distributed via
+ *  create_spatiotemporal_distributed_table(). Records the source relation
+ *  (tbloid/tablename), how it was tiled (numtiles, tilingmethod,
+ *  tilingtype, granularity, disjoint, tilekey), and which column drives
+ *  the distribution (distcol/distcoltype/ismobilitydb/shapesegmented/srid).
  * ----------------
  */
 

include/catalog/pg_dist_spatiotemporal_tables.h

@@ -17,6 +17,12 @@
 
 /* ----------------
  *  pg_dist_spatiotemporal_tiles: Compiler constants
+ *
+ *  Catalog table with one row per tile produced for a distributed table
+ *  (table_id references pg_dist_spatiotemporal_tables). Stores the tile's
+ *  key and its bounding boxes in both MobilityDB (mobdb_bbox) and PostGIS
+ *  (postgis_bbox) representations, plus per-tile cardinality used by the
+ *  planner for load balancing (num_shapes, num_points).
  * ----------------
  */
 

include/catalog/pg_dist_spatiotemporal_tiles.h

@@ -17,6 +17,11 @@
 
 /* ----------------
  *  pg_execution_run: Compiler constants
+ *
+ *  Catalog table that assigns a unique execution id (id) to each
+ *  distributed query run, paired with its serialized executor state
+ *  (sexec) so distributed_functions rows can be traced back to the run
+ *  that produced them.
  * ----------------
  */
 

include/catalog/pg_execution_run.h

@@ -17,19 +17,32 @@
 
 #include "postgres.h"
 
-
-
 /* constants for columnar.options */
 #define Anum_pg_spatiotemporal_join_operations_oid 3
 #define Anum_pg_spatiotemporal_join_operations_distance 4
 
+/* Full pg_dist_spatiotemporal_tables row for relationId, as a Datum array. */
 extern Datum * GetDistributedTableMetadata(Oid relationId);
+
+/* True if relationId's tiles have been rebalanced/reshuffled since creation. */
 extern bool IsReshuffledTable(Oid relationId);
+
+/* PostgreSQL type oid of the column driving relationId's distribution. */
 extern int DistributedColumnType(Oid relationId);
+
+/* Name of the spatiotemporal (MobilityDB) column used to distribute relationId. */
 extern  char * GetSpatiotemporalCol(Oid relationId);
+
+/* Serialized global (cross-tile) index metadata for relid. */
 extern char * GetGlobalIndexInfo(Oid relid);
+
+/* Resolves a relation oid from its (possibly schema-qualified) name. */
 extern Oid RelationId(const char *relationName);
+
+/* True if relationId is registered in pg_dist_spatiotemporal_tables. */
 extern bool IsDistributedSpatiotemporalTable(Oid relationId);
+
+/* Name of the geometry/shape column used to distribute relationId. */
 char *GetShapeCol(Oid relationId);
 
 #endif /* TABLE_OPS_H */

include/catalog/table_ops.h

@@ -15,6 +15,14 @@
 #ifndef COORDINATOR_OPERATIONS_H
 #define COORDINATOR_OPERATIONS_H
 
+/*
+ * CoordinatorOperation
+ *
+ * The coordinator-side half of a distributed aggregate/expression: the
+ * `intermediate_op` combines the per-tile results collected from the
+ * workers, and `final_op` transforms that combined result into the value
+ * returned to the client.
+ */
 typedef struct CoordinatorOperation
 {
     Datum intermediate_op;

include/distributed_functions/coordinator_operations.h

@@ -26,22 +26,41 @@
 #define Anum_DistFun_combiner 3
 #define Anum_DistFun_final 4
 
-
+/*
+ * QOperation
+ *
+ * A single query operation (e.g. an aggregate call) being rewritten for
+ * distributed execution: `op` is the operation's Datum representation,
+ * `col` the column/argument it applies to, and `alias` the output alias
+ * it must be projected under in the rewritten targetlist.
+ */
 typedef struct QOperation
 {
     Datum op;
     Alias *alias;
     Datum col;
 } QOperation;
 
+/*
+ * DistributedFunction
+ *
+ * Binds a query targetlist entry to the worker/coordinator function pair
+ * (looked up via pg_dist_spatiotemporal_dist_functions) that implements it
+ * across the distributed plan.
+ */
 typedef struct DistributedFunction
 {
     CoordinatorOperation *coordinatorOp;
     WorkerOperation *workerOp;
     TargetEntry *targetEntry;
 } DistributedFunction;
 
+/* Resolves and attaches the worker/coordinator functions for a targetlist entry. */
 extern DistributedFunction *addDistributedFunction(TargetEntry *operation);
+
+/* True if targetEntry's expression is a registered distributable function. */
 extern bool IsDistFunc(TargetEntry *targetEntry);
+
+/* Builds a QOperation pairing a distributed op (des) with its argument column (cur). */
 extern QOperation * AddQOperation(Datum des, Datum cur);
 #endif /* DISTRIBUTED_FUNCTION_H */

include/distributed_functions/distributed_function.h

@@ -15,6 +15,13 @@
 #ifndef WORKER_OPERATIONS_H
 #define WORKER_OPERATIONS_H
 
+/*
+ * WorkerOperation
+ *
+ * The worker-side half of a distributed aggregate/expression: `op` is the
+ * per-tile computation run on each worker before its partial result is
+ * shipped back to the coordinator for combination (see CoordinatorOperation).
+ */
 typedef struct WorkerOperation
 {
     Datum op;