Merge cffe56e into 1e0fb1e

create_shards.c

@@ -21,6 +21,7 @@
 #include "create_shards.h"
 #include "ddl_commands.h"
 #include "distribution_metadata.h"
+#include "prune_shard_list.h"
 
 #include <ctype.h>
 #include <limits.h>
@@ -29,8 +30,13 @@
 #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"
+#include "commands/defrem.h"
 #include "lib/stringinfo.h"
 #include "nodes/pg_list.h"
 #include "nodes/primnodes.h"
@@ -50,6 +56,8 @@ static List * ParseWorkerNodeFile(char *workerNodeFilename);
 static int CompareWorkerNodes(const void *leftElement, const void *rightElement);
 static bool ExecuteRemoteCommand(PGconn *connection, const char *sqlCommand);
 static text * IntegerToText(int32 value);
+static Oid SupportFunctionForColumn(Var* partitionColumn, Oid accessMethodId,
+		                            int16 supportFunctionNumber);
 
 
 /* declarations for dynamic loading */
@@ -70,9 +78,9 @@ master_create_distributed_table(PG_FUNCTION_ARGS)
 	char partitionMethod = PG_GETARG_CHAR(2);
 	Oid distributedTableId = ResolveRelationId(tableNameText);
 	char relationKind = '\0';
-	AttrNumber partitionColumnId = InvalidAttrNumber;
 	char *partitionColumnName = text_to_cstring(partitionColumnText);
 	char *tableName = text_to_cstring(tableNameText);
+	Var *partitionColumnVar = NULL;
 
 	/* verify target relation is either regular or foreign table */
 	relationKind = get_rel_relkind(distributedTableId);
@@ -84,17 +92,53 @@ master_create_distributed_table(PG_FUNCTION_ARGS)
 								  "foreign tables.")));
 	}
 
-	/* verify column exists in given table */
-	partitionColumnId = get_attnum(distributedTableId, partitionColumnName);
-	if (partitionColumnId == InvalidAttrNumber)
+	/* ColumnNameToColumn verifies column exists in given table */
+	partitionColumnVar = ColumnNameToColumn(distributedTableId, partitionColumnName);
+
+	if  (partitionMethod != HASH_PARTITION_TYPE)
 	{
-		ereport(ERROR, (errmsg("could not find column: %s", partitionColumnName)));
+		/* we only support hash partitioning right now */
+		ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+				        errmsg("pg_shard only supports hash partitioning")));
 	}
 
-	/* we only support hash partitioning method for now */
-	if (partitionMethod != HASH_PARTITION_TYPE)
+	/* depending on the partition type, check for the existence of support function */
+	if (partitionMethod == HASH_PARTITION_TYPE)
 	{
-		ereport(ERROR, (errmsg("unsupported partition method: %c", partitionMethod)));
+		Oid hashSupportFunction = SupportFunctionForColumn(partitionColumnVar,
+														   HASH_AM_OID, HASHPROC);
+		if (hashSupportFunction == InvalidOid)
+		{
+			Oid partitionColumnTypeId = partitionColumnVar->vartype;
+
+			ereport(ERROR, (errcode(ERRCODE_UNDEFINED_FUNCTION),
+			                errmsg("could not identify a hash function for type %s",
+			                format_type_be(partitionColumnTypeId)),
+							errdetail("Partition column types must have a hash function "
+									  "defined to use hash partitioning.")));
+		}
+	}
+	else if (partitionMethod == RANGE_PARTITION_TYPE)
+	{
+		/*
+		 * Currently we do not support RANGE_PARTITION_TYPE. However, for the
+		 * completeness of the code, we also check operators for range partitioning.
+		 * TODO: Add regression tests for this check when RANGE_PARTITION_TYPE is
+		 * supported.
+		 */
+		Oid btreeSupportFunction = SupportFunctionForColumn(partitionColumnVar,
+															BTREE_AM_OID, BTORDER_PROC);
+		if (btreeSupportFunction == InvalidOid)
+		{
+			Oid partitionColumnTypeId = partitionColumnVar->vartype;
+
+			ereport(ERROR,
+			        (errcode(ERRCODE_UNDEFINED_FUNCTION),
+			         errmsg("could not identify a comparison function for type %s",
+			                format_type_be(partitionColumnTypeId)),
+			         errdetail("Partition column types must have a comparison function "
+                               "defined to use range partitioning.")));
+		}
 	}
 
 	/* insert row into the partition metadata table */
@@ -534,3 +578,40 @@ IntegerToText(int32 value)
 
 	return valueText;
 }
+
+
+/*
+ *	SupportFunctionForColumn helps to find the support function given a column, an access
+ *	method and id of a support function. This function returns InvalidOid if there is no
+ *	support function associated with the data type of the column. Also, this function
+ *	errors-out if there is no default operator class for the data type of the column.
+ */
+Oid
+SupportFunctionForColumn(Var *partitionColumn, Oid accessMethodId,
+		                 int16 supportFunctionNumber)
+{
+	Oid operatorFamilyId = InvalidOid;
+	Oid supportFunctionOid = InvalidOid;
+	Oid columnOid = partitionColumn->vartype;
+	Oid operatorClassId = GetDefaultOpClass(columnOid, accessMethodId);
+
+	/*
+	 * If data type of the partition column does not have a default operator class,
+	 * we should not continue to get support function. A type with no default op
+	 * class cannot have any support functions.
+	 */
+	if (operatorClassId == InvalidOid)
+	{
+		ereport(ERROR, (errcode(ERRCODE_UNDEFINED_OBJECT),
+						errmsg("data type %s has no default operator class for specified"
+								" partition method", format_type_be(columnOid)),
+						errdetail("Partition column types must have a default operator"
+								  " class defined.")));
+	}
+
+	operatorFamilyId = get_opclass_family(operatorClassId);
+	supportFunctionOid = get_opfamily_proc(operatorFamilyId, columnOid, columnOid,
+										   supportFunctionNumber);
+
+	return supportFunctionOid;
+}

distribution_metadata.c

@@ -58,7 +58,6 @@ static List *ShardIntervalListCache = NIL;
 
 
 /* local function forward declarations */
-static Var * ColumnNameToColumn(Oid relationId, char *columnName);
 static void LoadShardIntervalRow(int64 shardId, Oid *relationId,
 								 char **minValue, char **maxValue);
 static ShardPlacement * TupleToShardPlacement(HeapTuple heapTuple,
@@ -445,7 +444,7 @@ IsDistributedTable(Oid tableId)
  * a Var that represents that column in that relation. This function throws an
  * error if the column doesn't exist or is a system column.
  */
-static Var *
+Var *
 ColumnNameToColumn(Oid relationId, char *columnName)
 {
 	Var *partitionColumn = NULL;

distribution_metadata.h

@@ -59,6 +59,7 @@
 
 /* denotes partition type of the distributed table */
 #define HASH_PARTITION_TYPE 'h'
+#define RANGE_PARTITION_TYPE 'r'
 
 /* human-readable names for addressing columns of partition table */
 #define PARTITION_TABLE_ATTRIBUTE_COUNT 3
@@ -139,6 +140,7 @@ extern List * LoadShardPlacementList(int64 shardId);
 extern Var * PartitionColumn(Oid distributedTableId);
 extern char PartitionType(Oid distributedTableId);
 extern bool IsDistributedTable(Oid tableId);
+extern Var * ColumnNameToColumn(Oid relationId, char *columnName);
 extern void InsertPartitionRow(Oid distributedTableId, char partitionType,
 							   text *partitionKeyText);
 extern void InsertShardRow(Oid distributedTableId, uint64 shardId, char shardStorage,

expected/create_shards.out.tmpl

@@ -10,22 +10,53 @@ CREATE FUNCTION create_table_then_fail(cstring, integer)
 	AS 'pg_shard'
 	LANGUAGE C STRICT;
 -- ===================================================================
+-- create test type, operator, operator family and operator class
+-- ===================================================================
+CREATE TYPE dummy_type AS (
+    i integer
+);
+CREATE FUNCTION dummy_type_function(dummy_type, dummy_type) RETURNS boolean
+AS 'SELECT TRUE;'
+LANGUAGE SQL
+IMMUTABLE
+RETURNS NULL ON NULL INPUT;
+CREATE OPERATOR = (
+    LEFTARG = dummy_type,
+    RIGHTARG = dummy_type,
+    PROCEDURE = dummy_type_function
+);
+CREATE OPERATOR FAMILY dummy_op_family USING hash;
+-- create operator class with no support function
+CREATE OPERATOR CLASS dummy_op_family_class
+DEFAULT FOR TYPE dummy_type USING hash FAMILY dummy_op_family AS
+OPERATOR 1 =;
+-- ===================================================================
 -- test shard creation functionality
 -- ===================================================================
 CREATE TABLE table_to_distribute (
 	name text,
-	id bigint PRIMARY KEY
+	id bigint PRIMARY KEY,
+	json_data json,
+	test_type_data dummy_type
 );
 -- use an index instead of table name
 SELECT master_create_distributed_table('table_to_distribute_pkey', 'id');
 ERROR:  cannot distribute relation: "table_to_distribute_pkey"
 DETAIL:  Distributed relations must be regular or foreign tables.
 -- use a bad column name
 SELECT master_create_distributed_table('table_to_distribute', 'bad_column');
-ERROR:  could not find column: bad_column
+ERROR:  partition column "bad_column" not found
 -- use unsupported partition type
 SELECT master_create_distributed_table('table_to_distribute', 'name', 'r');
-ERROR:  unsupported partition method: r
+ERROR:  pg_shard only supports hash partitioning
+-- use unsupported partition column which does not have any default op class
+SELECT master_create_distributed_table('table_to_distribute', 'json_data');
+ERROR:  data type json has no default operator class for specified partition method
+DETAIL:  Partition column types must have a default operator class defined.
+-- use unsupported partition column which does not have required support functions
+SELECT master_create_distributed_table('table_to_distribute', 'test_type_data');
+ERROR:  could not identify a hash function for type dummy_type
+DETAIL:  Partition column types must have a hash function defined to use hash partitioning.
 -- distribute table and inspect side effects
 SELECT master_create_distributed_table('table_to_distribute', 'name');
  master_create_distributed_table 

prune_shard_list.c

@@ -52,7 +52,6 @@ static List *OperatorIdCache = NIL;
 
 /* local function forward declarations */
 static Oid LookupOperatorByType(Oid typeId, Oid accessMethodId, int16 strategyNumber);
-static Oid GetOperatorByType(Oid typeId, Oid accessMethodId, int16 strategyNumber);
 static bool SimpleOpExpression(Expr *clause);
 static Node * HashableClauseMutator(Node *originalNode, Var *partitionColumn);
 static bool OpExpressionContainsColumn(OpExpr *operatorExpression, Var *partitionColumn);
@@ -305,7 +304,7 @@ LookupOperatorByType(Oid typeId, Oid accessMethodId, int16 strategyNumber)
  * GetOperatorByType returns the operator oid for the given type, access
  * method, and strategy number.
  */
-static Oid
+Oid
 GetOperatorByType(Oid typeId, Oid accessMethodId, int16 strategyNumber)
 {
 	/* Get default operator class from pg_opclass */

prune_shard_list.h

@@ -46,6 +46,7 @@ typedef struct OperatorIdCacheEntry
 extern List * PruneShardList(Oid relationId, List *whereClauseList,
 							 List *shardIntervalList);
 extern OpExpr * MakeOpExpression(Var *variable, int16 strategyNumber);
+extern Oid GetOperatorByType(Oid typeId, Oid accessMethodId, int16 strategyNumber);
 
 
 #endif /* PG_SHARD_PRUNE_SHARD_LIST_H */

sql/create_shards.sql.tmpl

@@ -12,13 +12,40 @@ CREATE FUNCTION create_table_then_fail(cstring, integer)
 	AS 'pg_shard'
 	LANGUAGE C STRICT;
 
+-- ===================================================================
+-- create test type, operator, operator family and operator class
+-- ===================================================================
+CREATE TYPE dummy_type AS (
+    i integer
+);
+
+CREATE FUNCTION dummy_type_function(dummy_type, dummy_type) RETURNS boolean
+AS 'SELECT TRUE;'
+LANGUAGE SQL
+IMMUTABLE
+RETURNS NULL ON NULL INPUT;
+
+CREATE OPERATOR = (
+    LEFTARG = dummy_type,
+    RIGHTARG = dummy_type,
+    PROCEDURE = dummy_type_function
+);
+CREATE OPERATOR FAMILY dummy_op_family USING hash;
+
+-- create operator class with no support function
+CREATE OPERATOR CLASS dummy_op_family_class
+DEFAULT FOR TYPE dummy_type USING hash FAMILY dummy_op_family AS
+OPERATOR 1 =;
+
 -- ===================================================================
 -- test shard creation functionality
 -- ===================================================================
 
 CREATE TABLE table_to_distribute (
 	name text,
-	id bigint PRIMARY KEY
+	id bigint PRIMARY KEY,
+	json_data json,
+	test_type_data dummy_type
 );
 
 -- use an index instead of table name
@@ -30,6 +57,12 @@ SELECT master_create_distributed_table('table_to_distribute', 'bad_column');
 -- use unsupported partition type
 SELECT master_create_distributed_table('table_to_distribute', 'name', 'r');
 
+-- use unsupported partition column which does not have any default op class
+SELECT master_create_distributed_table('table_to_distribute', 'json_data');
+
+-- use unsupported partition column which does not have required support functions
+SELECT master_create_distributed_table('table_to_distribute', 'test_type_data');
+
 -- distribute table and inspect side effects
 SELECT master_create_distributed_table('table_to_distribute', 'name');
 SELECT partition_method, key FROM pgs_distribution_metadata.partition