Merge pull request #23 from mbakli/feature_upgrade

Feature upgrade: auto-clean catalog & bug fixes

Author: mbakli

CMakeLists.txt

@@ -59,42 +59,28 @@ find_program(POSTGRESQL_EXECUTABLE NAMES postgres
         )
 
 
-find_path(PostgreSQL_TYPE_INCLUDE_DIR
-        NAMES catalog/pg_type.h
-        PATHS
-        # Look in other places.
-        ${PostgreSQL_ROOT_DIRECTORIES}
-        PATH_SUFFIXES
-        postgresql
-        pgsql/server
-        postgresql/server
-        include/server
-        # Help the user find it if we cannot.
-        DOC "The ${PostgreSQL_INCLUDE_DIR_MESSAGE}"
-        )
-
-find_package(PostgreSQL)
+# Resolve include directories from the same pg_config used for --bindir above,
+# instead of find_package(PostgreSQL) (which only locates the client libpq
+# headers and can pick a different, unrelated PostgreSQL installation when
+# several major versions are installed side by side).
+execute_process(
+        COMMAND ${POSTGRESQL_PG_CONFIG} --includedir-server
+        OUTPUT_STRIP_TRAILING_WHITESPACE
+        OUTPUT_VARIABLE PostgreSQL_SERVER_INCLUDE_DIR)
+execute_process(
+        COMMAND ${POSTGRESQL_PG_CONFIG} --includedir
+        OUTPUT_STRIP_TRAILING_WHITESPACE
+        OUTPUT_VARIABLE PostgreSQL_INCLUDE_DIR)
 
-if (PostgreSQL_INCLUDE_DIRS)
-    message(STATUS "POSTGRES_INCLUDE_PATH ${PostgreSQL_INCLUDE_DIRS}")
-    set (PostgreSQL_INCLUDE_DIRS /usr/local/pgsql/include)
-    include_directories(SYSTEM ${PostgreSQL_INCLUDE_DIRS}/server)
-    message(STATUS "POSTGRES_SERVER_PATH ${PostgreSQL_INCLUDE_DIRS}/server")
-else()
-    set (PostgreSQL_INCLUDE_DIRS /usr/local/pgsql/include)
-    message(STATUS "POSTGRES_INCLUDE_PATH ${PostgreSQL_INCLUDE_DIRS}")
-    include_directories(SYSTEM ${PostgreSQL_INCLUDE_DIR}/server)
-    message(STATUS "POSTGRES_SERVER_PATH ${PostgreSQL_INCLUDE_DIR}/server")
+if (NOT PostgreSQL_SERVER_INCLUDE_DIR)
+    message(FATAL_ERROR "pg_config --includedir-server failed to return a value. Please check your PostgreSQL installation!")
 endif()
 
-include_directories(SYSTEM ${PostgreSQL_INCLUDE_DIRS})
+message(STATUS "POSTGRES_INCLUDE_PATH ${PostgreSQL_INCLUDE_DIR}")
+message(STATUS "POSTGRES_SERVER_PATH ${PostgreSQL_SERVER_INCLUDE_DIR}")
+include_directories(SYSTEM ${PostgreSQL_SERVER_INCLUDE_DIR})
 include_directories(SYSTEM ${PostgreSQL_INCLUDE_DIR})
 
-if (${PostgreSQL_VERSION_STRING} GREATER ${PG_MIN_MAJOR_VERSION})
-    message(STATUS "PostgreSQL version: ${PostgreSQL_VERSION_STRING}")
-    include_directories(SYSTEM ${PostgreSQL_INCLUDE_DIR}/${PG_MAX_MAJOR_VERSION}/server)
-endif()
-
 #---------------------------------------------
 # library directory
 #---------------------------------------------
@@ -144,6 +130,13 @@ if(PROJ_INCLUDE_DIRS)
     message(STATUS "POSTGIS_PROJ_VERSION: ${POSTGIS_PROJ_VERSION}")
 endif()
 
+# liblwgeom.h (used for GBOX) is PostGIS' own internal header and is not
+# shipped by any postgresql-*-postgis-3 package; it must be sourced from a
+# PostGIS/MobilityDB checkout. See README for how to place a copy under
+# /usr/local/include/postgis-lwgeom/liblwgeom (alongside its required
+# ../postgis_config.h).
+include_directories(SYSTEM /usr/local/include/postgis-lwgeom/liblwgeom)
+
 #-------------------
 # add the MobilityDB link
 #-------------------
@@ -180,8 +173,12 @@ foreach (subdir ${DistMobilityDB_SOURCE_NAMES} )
 endforeach()
 
 
-set(POSTGRESQL_DYNLIB_DIR /usr/local/pgsql/lib)
-set(POSTGRESQL_SHARE_DIR /usr/local/pgsql/share)
+# Install into the actual pg_config-reported directories for this PostgreSQL
+# installation (LIBRARY_INSTALL_PATH/SHARE_DIR, computed above), rather than
+# an author-specific /usr/local/pgsql source-build layout. Note SHARE_DIR
+# already has "/extension" appended (see the extension-directory block above).
+set(POSTGRESQL_DYNLIB_DIR ${LIBRARY_INSTALL_PATH})
+set(POSTGRESQL_SHARE_DIR ${SHARE_DIR})
 
 set(MOBILITYDB_LIB_NAME "${CMAKE_PROJECT_NAME}-1.0")
 add_library(${MOBILITYDB_LIB_NAME} MODULE ${PROJECT_OBJECTS})
@@ -196,4 +193,4 @@ add_custom_target(control ALL DEPENDS ${CMAKE_BINARY_DIR}/distributed_mobilitydb
 install(TARGETS ${MOBILITYDB_LIB_NAME} DESTINATION "${POSTGRESQL_DYNLIB_DIR}")
 install(FILES "${CMAKE_BINARY_DIR}/distributed_mobilitydb.control"
         "${CMAKE_BINARY_DIR}/${DISTRIBUTED_MOBILITYDB_EXTENSION_FILE}"
-        DESTINATION "${POSTGRESQL_SHARE_DIR}/extension")
+        DESTINATION "${POSTGRESQL_SHARE_DIR}")

include/catalog/nodes.h

@@ -23,6 +23,9 @@ extern Datum GetDBName();
 /* Coordinator/worker node info (host, port, role) for the local backend. */
 extern TaskNode *GetNodeInfo();
 
+/* Node (host, port) actually hosting relationId's shard for rand_tile. */
+extern TaskNode *GetShardHostNode(Oid relationId, int rand_tile);
+
 /* Table id of the tile assigned to a randomly-picked worker for relationId. */
 extern char* GetRandomTileId(Oid relationId, ExecTaskType taskType, int rand_tile);
 

include/executor/tile_tasks.h

@@ -26,8 +26,8 @@ extern char *taskQuery (List *tasks, ExecTaskType taskType);
 /* Rebalances relid's data into numTiles tiles, materializing into reshuffledTable. */
 extern void RearrangeTiles(Oid relid, int numTiles, char *reshuffledTable);
 
-/* Appends the tile-key projection for a distributed (spatiotemporal) relation to query_string. */
-extern Datum AddTilingKey(STMultirelationCatalog tblCatalog, Alias *alias , char * query_string);
+/* Appends the tile-key equi-join predicate between alias and otherAlias for a distributed (spatiotemporal) relation to query_string. */
+extern Datum AddTilingKey(STMultirelationCatalog tblCatalog, Alias *alias, Alias *otherAlias, char * query_string);
 
 /* Appends the tile-key projection for a non-spatiotemporal (plain) rte to query_string. */
 extern Datum AddNonStRteTilingKey(Rte *tbl, Alias *alias ,char * query_string);

include/multirelation/multirelation_utils.h

@@ -92,6 +92,11 @@ typedef struct STMultirelations
     int stCount;
     int nonStCount;
     int length;
+    /* Count of nonStCount entries that are Citus reference tables -- these
+     * are already replicated to every node, so they never need reshuffling
+     * and shouldn't count as a "different distributed table" when deciding
+     * whether a join needs the NonColocation strategy. */
+    int refCount;
 } STMultirelations;
 
 /* True if relationId is registered as a distributed spatiotemporal table. */

include/planner/distributed_mobilitydb_explain.h

@@ -50,10 +50,9 @@ typedef struct SpatiotemporalScanState
 } SpatiotemporalScanState;
 
 
-/* A query paired with its (possibly rewritten) SQL text, for EXPLAIN output. */
+/* The (possibly rewritten) SQL text of the query being explained. */
 typedef struct DistributedQueryExplain
 {
-    Query *query;
     char *query_string;
 
 }DistributedQueryExplain;

include/planner/predicate_management.h

@@ -75,9 +75,20 @@ extern bool IsDistanceOperation(Oid operationId);
 /* True if operationId is a registered intersection operator (e.g. eIntersects). */
 extern bool IsIntersectionOperation(Oid operationId);
 
-/* Computes the query's search bounding box from tbls and the range/distance predicate opExpr. */
-extern Datum get_query_range(STMultirelations *tbls, OpExpr *opExpr);
+/*
+ * Extracts the callable identifier and argument list from a WHERE-clause
+ * predicate node, regardless of whether MobilityDB/PostGIS exposed it as an
+ * infix operator (OpExpr, e.g. `&&`) or a plain function call (FuncExpr,
+ * e.g. eDwithin(...), ST_Intersects(...)) -- both forms appear in practice
+ * and pg_spatiotemporal_join_operations is keyed by either an operator or a
+ * function oid. Returns false (leaving *oid/*args unset) for any other node
+ * type.
+ */
+extern bool GetPredicateOidAndArgs(Node *clause, Oid *oid, List **args);
 
-/* True if opExpr's search box spans enough tiles of tbls to warrant rebalancing first. */
-extern bool CheckTileRebalancerActivation(STMultirelations *tbls, OpExpr *opExpr, Datum box);
+/* Computes the query's search bounding box from tbls and the range/distance predicate clause. */
+extern Datum get_query_range(STMultirelations *tbls, Node *clause);
+
+/* True if clause's search box spans enough tiles of tbls to warrant rebalancing first. */
+extern bool CheckTileRebalancerActivation(STMultirelations *tbls, Node *clause, Datum box);
 #endif /* PREDICATE_MANAGMENT_H */

include/utils/helper_functions.h

@@ -38,4 +38,10 @@ extern char *toLower(char *str);
 
 /* True if val is a NULL/zero Datum (no value set). */
 extern bool IsDatumEmpty(Datum val);
+
+/* Renders datum (of the given PostgreSQL type oid) as a palloc'd C string via its type's output function. */
+extern char *DatumToString(Datum datum, Oid typeoid);
+
+/* Runs query via SPI, erroring out if its result status doesn't match expectedSpiOk (an SPI_OK_* constant). */
+extern void ExecuteQueryViaSPI(char *query, int expectedSpiOk);
 #endif /* HELPER_FUNCTIONS_H */

sql/helper_functions/CMakeLists.txt

@@ -2,6 +2,7 @@ SET(LOCAL_FILES
         write_to_catalog.sql
         colocation_utils.sql
         catalog_info_check.sql
+        catalog_cleanup.sql
         )
 
 foreach (f ${LOCAL_FILES})

sql/helper_functions/catalog_cleanup.sql

@@ -0,0 +1,39 @@
+--------------------------------------------------------------------------------------------------------------------------------------------------------
+-- Catalog Cleanup
+--------------------------------------------------------------------------------------------------------------------------------------------------------
+-- Dropping a distributed spatiotemporal table (e.g. DROP TABLE ... CASCADE)
+-- only removes the table itself and its Citus shards; Postgres has no
+-- dependency link between it and the rows this extension keeps in
+-- pg_dist_spatiotemporal_tables/pg_dist_spatiotemporal_tiles, so those rows
+-- are silently orphaned otherwise. This event trigger removes them
+-- automatically whenever a registered table is dropped.
+CREATE OR REPLACE FUNCTION distributed_mobilitydb_drop_cleanup()
+    RETURNS event_trigger AS $$
+DECLARE
+    obj record;
+    dropped_table_id integer;
+BEGIN
+    IF to_regclass('pg_dist_spatiotemporal_tables') IS NULL THEN
+        -- The extension itself (and its catalog tables) is being dropped;
+        -- nothing left to clean up.
+        RETURN;
+    END IF;
+    FOR obj IN SELECT * FROM pg_event_trigger_dropped_objects()
+               WHERE object_type = 'table'
+    LOOP
+        SELECT id INTO dropped_table_id
+        FROM pg_dist_spatiotemporal_tables
+        WHERE tableName = obj.object_name;
+
+        IF dropped_table_id IS NOT NULL THEN
+            DELETE FROM pg_dist_spatiotemporal_tiles WHERE table_id = dropped_table_id;
+            DELETE FROM pg_dist_spatiotemporal_tables WHERE id = dropped_table_id;
+        END IF;
+    END LOOP;
+END;
+$$ LANGUAGE 'plpgsql';
+
+DROP EVENT TRIGGER IF EXISTS distributed_mobilitydb_drop_cleanup_trigger;
+CREATE EVENT TRIGGER distributed_mobilitydb_drop_cleanup_trigger
+    ON sql_drop
+    EXECUTE FUNCTION distributed_mobilitydb_drop_cleanup();

sql/helper_functions/colocation_utils.sql

@@ -34,47 +34,38 @@ END;
 $$;
 
 ----------------------------------------------------------------------------------------------------------------------
--- colocate_multirelation colocates one multirelation with another
+-- colocate_shards physically moves table2's shards so each one lands on the
+-- same node as table1's shard sharing the same tile_key range (shardminvalue).
+-- Citus' own colocate_with option doesn't support range-distributed tables,
+-- so this does the move explicitly with citus_move_shard_placement() instead
+-- of relying on Citus' colocation groups.
 ----------------------------------------------------------------------------------------------------------------------
-CREATE OR REPLACE FUNCTION colocate_multirelation(table1 text, table2 text)
+DROP FUNCTION IF EXISTS colocate_multirelation;
+CREATE OR REPLACE FUNCTION colocate_shards(table1 text, table2 text)
     RETURNS boolean
 LANGUAGE plpgsql
 AS $$
 DECLARE
-i integer;
-    shard_info record;
-    node_info record;
-    node text;
-    shardid_test bigint;
+    tile_pair record;
 BEGIN
-    --Move one of them to a new place because one node contains 10 and the other contains 11
-    --It is an enterpise feature
-    --SELECT master_move_shard_placement(102660,'pgxl4', 5432,'pgxl2', 5432);
-    --For every node, update shards information
-    FOR node_info in SELECT * FROM master_get_active_worker_nodes()
+    FOR tile_pair IN
+        SELECT s2.shardid AS moving_shard,
+               n1.nodename AS target_node, n1.nodeport AS target_port,
+               n2.nodename AS source_node, n2.nodeport AS source_port
+        FROM pg_dist_shard s1
+        JOIN pg_dist_placement p1 ON p1.shardid = s1.shardid
+        JOIN pg_dist_node n1 ON n1.groupid = p1.groupid AND n1.noderole = 'primary'
+        JOIN pg_dist_shard s2 ON s2.shardminvalue = s1.shardminvalue
+        JOIN pg_dist_placement p2 ON p2.shardid = s2.shardid
+        JOIN pg_dist_node n2 ON n2.groupid = p2.groupid AND n2.noderole = 'primary'
+        WHERE s1.logicalrelid = table1::regclass
+          AND s2.logicalrelid = table2::regclass
+          AND (n1.nodename, n1.nodeport) IS DISTINCT FROM (n2.nodename, n2.nodeport)
     LOOP
-        i = 0;
-        --Get the shards of every table in every node
-        FOR shard_info in SELECT shard.shardid, shard.shardminvalue, shard.shardmaxvalue
-                          FROM pg_dist_placement AS placement, pg_dist_node AS node, pg_dist_shard As shard
-                          WHERE placement.groupid = node.groupid
-                            AND shard.logicalrelid = table1::regclass
-                            AND placement.shardid = shard.shardid
-                            AND node.noderole = 'primary'
-                            AND nodename=node_info.node_name
-        LOOP
-                SELECT shard.shardid
-                FROM pg_dist_placement AS placement, pg_dist_node AS node, pg_dist_shard As shard
-                WHERE placement.groupid = node.groupid
-                    AND shard.logicalrelid = table2::regclass
-                    AND placement.shardid = shard.shardid
-                    AND node.noderole = 'primary'
-                    AND nodename=node_info.node_name offset i limit 1 INTO shardid_test;
-                --RAISE NOTICE 'Update:%',shardid_test;
-                UPDATE pg_dist_shard SET shardminvalue = shard_info.shardminvalue, shardmaxvalue=shard_info.shardmaxvalue
-                WHERE shardid = shardid_test;
-                i := i + 1;
-        END LOOP;
+        PERFORM citus_move_shard_placement(tile_pair.moving_shard,
+                                           tile_pair.source_node, tile_pair.source_port,
+                                           tile_pair.target_node, tile_pair.target_port,
+                                           'block_writes');
     END LOOP;
     RETURN TRUE;
 END;
@@ -101,11 +92,15 @@ i integer;
     results json;
 BEGIN
     PERFORM create_range_shards(shards, reshuffled_table);
-    --Modify the ranges
-    UPDATE pg_catalog.pg_dist_partition
-    SET colocationid=2
-    WHERE logicalrelid=reshuffled_table::regclass;
-    PERFORM colocate_multirelation(tableName, reshuffled_table);
+    /* create_range_shards() above already assigns reshuffled_table's shards
+     * the correct tile_key-aligned ranges (1..shards), matching tableName's
+     * own tile numbering by convention -- both tables tile the same way.
+     * Citus' shard placement for the newly-created shards is otherwise
+     * independent of tableName's placement, so without this, a tile-key
+     * join between the two tables is a cross-node repartition even though
+     * both sides cover the same tile_key range. colocate_shards() moves
+     * each of reshuffled_table's shards onto tableName's matching-tile node. */
+    PERFORM colocate_shards(tableName, reshuffled_table);
 RETURN true;
 END;
 $$;