Merge a04719a into 7b2cda7

create_shards.c

@@ -21,18 +21,15 @@
 #include "create_shards.h"
 #include "ddl_commands.h"
 #include "distribution_metadata.h"
-#include "prune_shard_list.h"
 
 #include <ctype.h>
 #include <limits.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
 
-#include "access/attnum.h"
 #include "access/hash.h"
 #include "access/nbtree.h"
-#include "access/skey.h"
 #include "catalog/namespace.h"
 #include "catalog/pg_class.h"
 #include "catalog/pg_am.h"
@@ -50,7 +47,6 @@
 
 
 /* local function forward declarations */
-static Oid ResolveRelationId(text *relationName);
 static void CheckHashPartitionedTable(Oid distributedTableId);
 static List * ParseWorkerNodeFile(char *workerNodeFilename);
 static int CompareWorkerNodes(const void *leftElement, const void *rightElement);
@@ -315,7 +311,7 @@ master_create_worker_shards(PG_FUNCTION_ARGS)
 
 
 /* Finds the relationId from a potentially qualified relation name. */
-static Oid
+Oid
 ResolveRelationId(text *relationName)
 {
 	List *relationNameList = NIL;

create_shards.h

@@ -16,6 +16,7 @@
 #include "postgres.h"
 #include "c.h"
 #include "fmgr.h"
+#include "postgres_ext.h"
 
 #include "nodes/pg_list.h"
 
@@ -38,9 +39,10 @@ typedef struct WorkerNode
 } WorkerNode;
 
 
-/* utility function declaration shared within this module */
+/* utility functions declaration shared within this module */
 extern List * SortList(List *pointerList,
 					   int (*ComparisonFunction)(const void *, const void *));
+extern Oid ResolveRelationId(text *relationName);
 
 /* function declarations for initializing a distributed table */
 extern Datum master_create_distributed_table(PG_FUNCTION_ARGS);

distribution_metadata.c

@@ -34,7 +34,7 @@
 #include "catalog/pg_type.h"
 #include "commands/sequence.h"
 #include "nodes/makefuncs.h"
-#include "nodes/memnodes.h"
+#include "nodes/memnodes.h" /* IWYU pragma: keep */
 #include "nodes/pg_list.h"
 #include "nodes/primnodes.h"
 #include "storage/lock.h"

expected/repair_shards.out.tmpl

@@ -20,7 +20,11 @@ SELECT 'customer_engagements_20'::regclass::oid AS shardoid;
 \gset
 \o
 SELECT master_copy_shard_placement(20, 'localhost', $PGPORT, '127.0.0.1', $PGPORT);
-ERROR:  shard placement repair not fully implemented
+ master_copy_shard_placement 
+-----------------------------
+
+(1 row)
+
 \o /dev/null
 SELECT 'customer_engagements_20'::regclass::oid AS repairedoid;
 \gset
@@ -53,21 +57,19 @@ CREATE FOREIGN TABLE remote_engagements_30 (
 	created_at date,
 	event_data text
 ) SERVER fake_fdw_server;
-\o /dev/null
-SELECT 'remote_engagements_30'::regclass::oid AS shardoid;
-\gset
-\o
 SELECT master_copy_shard_placement(30, 'localhost', $PGPORT, '127.0.0.1', $PGPORT);
-ERROR:  shard placement repair not fully implemented
-\o /dev/null
-SELECT 'remote_engagements_30'::regclass::oid AS repairedoid;
-\gset
-\o
-SELECT :shardoid != :repairedoid AS shard_recreated; 
- shard_recreated 
------------------
- t
+ERROR:  cannot repair shard
+DETAIL:  Repairing shards backed by foreign tables is currently unsupported.
+INSERT INTO customer_engagements_20 DEFAULT VALUES;
+SELECT worker_copy_shard_placement('customer_engagements_20', 'localhost', $PGPORT);
+ worker_copy_shard_placement 
+-----------------------------
+
+(1 row)
+
+SELECT COUNT(*) FROM customer_engagements_20;
+ count 
+-------
+     2
 (1 row)
 
-SELECT worker_copy_shard_placement('customer_engagements', 'localhost', $PGPORT);
-ERROR:  worker_copy_shard_placement not implemented

pg_shard.c

@@ -120,8 +120,6 @@ static void AcquireExecutorShardLocks(List *taskList, LOCKMODE lockMode);
 static int CompareTasksByShardId(const void *leftElement, const void *rightElement);
 static void ExecuteMultipleShardSelect(DistributedPlan *distributedPlan,
 									   RangeVar *intermediateTable);
-static bool ExecuteTaskAndStoreResults(Task *task, TupleDesc tupleDescriptor,
-									   Tuplestorestate *tupleStore);
 static bool SendQueryInSingleRowMode(PGconn *connection, StringInfo query);
 static bool StoreQueryResult(PGconn *connection, TupleDesc tupleDescriptor,
 							 Tuplestorestate *tupleStore);
@@ -1362,7 +1360,7 @@ ExecuteMultipleShardSelect(DistributedPlan *distributedPlan,
  * the results and stores them in the given tuple store. If the task fails on
  * one of the placements, the function retries it on other placements.
  */
-static bool
+bool
 ExecuteTaskAndStoreResults(Task *task, TupleDesc tupleDescriptor,
 						   Tuplestorestate *tupleStore)
 {

pg_shard.h

@@ -14,10 +14,12 @@
 
 #include "c.h"
 
+#include "access/tupdesc.h"
 #include "nodes/parsenodes.h"
 #include "nodes/pg_list.h"
 #include "nodes/plannodes.h"
 #include "lib/stringinfo.h"
+#include "utils/tuplestore.h"
 
 
 /* prefix used for temporary tables created on the master node */
@@ -85,6 +87,8 @@ typedef struct Task
 /* function declarations for extension loading and unloading */
 extern void _PG_init(void);
 extern void _PG_fini(void);
+extern bool ExecuteTaskAndStoreResults(Task *task, TupleDesc tupleDescriptor,
+									   Tuplestorestate *tupleStore);
 
 
 #endif /* PG_SHARD_H */

prune_shard_list.c

@@ -27,6 +27,7 @@
 #include "catalog/pg_type.h"
 #include "commands/defrem.h"
 #include "nodes/makefuncs.h"
+#include "nodes/memnodes.h" /* IWYU pragma: keep */
 #include "nodes/nodeFuncs.h"
 #include "nodes/nodes.h"
 #include "nodes/primnodes.h"
@@ -40,6 +41,7 @@
 #include "utils/lsyscache.h"
 #include "utils/typcache.h"
 #include "utils/memutils.h"
+#include "utils/palloc.h"
 
 
 /*