Support range data types.

Selectivity estimation functions are missing for some range type operators,
which is a TODO.

Jeff Davis

1 files changed, 441 insertions, 4 deletions

diff --git a/src/backend/commands/typecmds.c b/src/backend/commands/typecmds.c
index 5069c5759ec..91488bbbf5c 100644
--- a/src/backend/commands/typecmds.c
+++ b/src/backend/commands/typecmds.c
@@ -42,7 +42,11 @@
 #include "catalog/pg_constraint.h"
 #include "catalog/pg_depend.h"
 #include "catalog/pg_enum.h"
+#include "catalog/pg_language.h"
 #include "catalog/pg_namespace.h"
+#include "catalog/pg_proc.h"
+#include "catalog/pg_proc_fn.h"
+#include "catalog/pg_range.h"
 #include "catalog/pg_type.h"
 #include "catalog/pg_type_fn.h"
 #include "commands/defrem.h"
@@ -63,6 +67,7 @@
 #include "utils/fmgroids.h"
 #include "utils/lsyscache.h"
 #include "utils/memutils.h"
+#include "utils/rangetypes.h"
 #include "utils/rel.h"
 #include "utils/syscache.h"
 #include "utils/tqual.h"
@@ -87,6 +92,9 @@ static Oid	findTypeSendFunction(List *procname, Oid typeOid);
 static Oid	findTypeTypmodinFunction(List *procname);
 static Oid	findTypeTypmodoutFunction(List *procname);
 static Oid	findTypeAnalyzeFunction(List *procname, Oid typeOid);
+static Oid  findRangeCanonicalFunction(List *procname, Oid typeOid);
+static Oid	findRangeSubOpclass(List *procname, Oid typeOid);
+static Oid findRangeSubtypeDiffFunction(List *procname, Oid typeOid);
 static void	validateDomainConstraint(Oid domainoid, char *ccbin);
 static List *get_rels_with_domain(Oid domainOid, LOCKMODE lockmode);
 static void checkDomainOwner(HeapTuple tup);
@@ -95,6 +103,8 @@ static char *domainAddConstraint(Oid domainOid, Oid domainNamespace,
 					Oid baseTypeOid,
 					int typMod, Constraint *constr,
 					char *domainName);
+static void makeRangeConstructor(char *name, Oid namespace, Oid rettype,
+								 Oid subtype);
 
 
 /*
@@ -643,6 +653,14 @@ RemoveTypeById(Oid typeOid)
 	if (((Form_pg_type) GETSTRUCT(tup))->typtype == TYPTYPE_ENUM)
 		EnumValuesDelete(typeOid);
 
+	/*
+	 * If it is a range type, delete the pg_range entries too; we
+	 * don't bother with making dependency entries for those, so it
+	 * has to be done "by hand" here.
+	 */
+	if (((Form_pg_type) GETSTRUCT(tup))->typtype == TYPTYPE_RANGE)
+		RangeDelete(typeOid);
+
 	ReleaseSysCache(tup);
 
 	heap_close(relation, RowExclusiveLock);
@@ -724,14 +742,15 @@ DefineDomain(CreateDomainStmt *stmt)
 	basetypeoid = HeapTupleGetOid(typeTup);
 
 	/*
-	 * Base type must be a plain base type, another domain or an enum. Domains
-	 * over pseudotypes would create a security hole.  Domains over composite
-	 * types might be made to work in the future, but not today.
+	 * Base type must be a plain base type, another domain, an enum or a range
+	 * type. Domains over pseudotypes would create a security hole.  Domains
+	 * over composite types might be made to work in the future, but not today.
 	 */
 	typtype = baseType->typtype;
 	if (typtype != TYPTYPE_BASE &&
 		typtype != TYPTYPE_DOMAIN &&
-		typtype != TYPTYPE_ENUM)
+		typtype != TYPTYPE_ENUM &&
+		typtype != TYPTYPE_RANGE)
 		ereport(ERROR,
 				(errcode(ERRCODE_DATATYPE_MISMATCH),
 				 errmsg("\"%s\" is not a valid base type for a domain",
@@ -1135,6 +1154,327 @@ DefineEnum(CreateEnumStmt *stmt)
 }
 
 /*
+ * DefineRange
+ *		Registers a new range type.
+ */
+void
+DefineRange(CreateRangeStmt *stmt)
+{
+	char		*typeName;
+	char		*rangeArrayName;
+	Oid			 typeNamespace;
+	Oid			 typoid;
+	Oid			 rangeArrayOid;
+	List		*parameters = stmt->params;
+
+	ListCell	*lc;
+	List		*rangeSubOpclassName  = NIL;
+	List		*rangeSubtypeDiffName = NIL;
+	List		*rangeCollationName	  = NIL;
+	Oid			 rangeCollation		  = InvalidOid;
+	regproc		 rangeAnalyze		  = InvalidOid;
+	Oid			 rangeSubtype		  = InvalidOid;
+	regproc		 rangeSubOpclass	  = InvalidOid;
+	regproc		 rangeCanonical		  = InvalidOid;
+	regproc		 rangeSubtypeDiff	  = InvalidOid;
+
+	AclResult	 aclresult;
+
+	/* Convert list of names to a name and namespace */
+	typeNamespace = QualifiedNameGetCreationNamespace(stmt->typeName,
+													  &typeName);
+
+	/* Check we have creation rights in target namespace */
+	aclresult = pg_namespace_aclcheck(typeNamespace, GetUserId(), ACL_CREATE);
+	if (aclresult != ACLCHECK_OK)
+		aclcheck_error(aclresult, ACL_KIND_NAMESPACE,
+					   get_namespace_name(typeNamespace));
+
+	/*
+	 * Look to see if type already exists (presumably as a shell; if not,
+	 * TypeCreate will complain).
+	 */
+	typoid = GetSysCacheOid2(TYPENAMENSP,
+							 CStringGetDatum(typeName),
+							 ObjectIdGetDatum(typeNamespace));
+
+	/*
+	 * If it's not a shell, see if it's an autogenerated array type, and if so
+	 * rename it out of the way.
+	 */
+	if (OidIsValid(typoid) && get_typisdefined(typoid))
+	{
+		if (moveArrayTypeName(typoid, typeName, typeNamespace))
+			typoid = InvalidOid;
+	}
+
+	/*
+	 * If it doesn't exist, create it as a shell, so that the OID is known for
+	 * use in the I/O function definitions.
+	 */
+	if (!OidIsValid(typoid))
+	{
+		typoid = TypeShellMake(typeName, typeNamespace, GetUserId());
+		/* Make new shell type visible for modification below */
+		CommandCounterIncrement();
+
+		/*
+		 * If the command was a parameterless CREATE TYPE, we're done ---
+		 * creating the shell type was all we're supposed to do.
+		 */
+		if (parameters == NIL)
+			return;
+	}
+	else
+	{
+		/* Complain if dummy CREATE TYPE and entry already exists */
+		if (parameters == NIL)
+			ereport(ERROR,
+					(errcode(ERRCODE_DUPLICATE_OBJECT),
+					 errmsg("type \"%s\" already exists", typeName)));
+	}
+
+	foreach(lc, stmt->params)
+	{
+		DefElem *defel  = lfirst(lc);
+
+		if (pg_strcasecmp(defel->defname, "subtype") == 0)
+		{
+			if (OidIsValid(rangeSubtype))
+				ereport(ERROR,
+						(errcode(ERRCODE_SYNTAX_ERROR),
+						 errmsg("conflicting or redundant options")));
+			rangeSubtype = typenameTypeId(NULL, defGetTypeName(defel));
+		}
+		else if (pg_strcasecmp(defel->defname, "canonical") == 0)
+		{
+			if (OidIsValid(rangeCanonical))
+				ereport(ERROR,
+						(errcode(ERRCODE_SYNTAX_ERROR),
+						 errmsg("conflicting or redundant options")));
+			rangeCanonical = findRangeCanonicalFunction(
+				defGetQualifiedName(defel), typoid);
+		}
+		else if (pg_strcasecmp(defel->defname, "collation") == 0)
+		{
+			if (rangeCollationName != NIL)
+				ereport(ERROR,
+						(errcode(ERRCODE_SYNTAX_ERROR),
+						 errmsg("conflicting or redundant options")));
+			rangeCollationName = defGetQualifiedName(defel);
+		}
+		else if (pg_strcasecmp(defel->defname, "analyze") == 0)
+		{
+			if (OidIsValid(rangeAnalyze))
+				ereport(ERROR,
+						(errcode(ERRCODE_SYNTAX_ERROR),
+						 errmsg("conflicting or redundant options")));
+			rangeAnalyze = findTypeAnalyzeFunction(defGetQualifiedName(defel),
+												   typoid);
+		}
+		else if (pg_strcasecmp(defel->defname, "subtype_opclass") == 0)
+		{
+			if (rangeSubOpclassName != NIL)
+				ereport(ERROR,
+						(errcode(ERRCODE_SYNTAX_ERROR),
+						 errmsg("conflicting or redundant options")));
+			rangeSubOpclassName = defGetQualifiedName(defel);
+		}
+		else if (pg_strcasecmp(defel->defname, "subtype_diff") == 0)
+		{
+			if (rangeSubtypeDiffName != NIL)
+				ereport(ERROR,
+						(errcode(ERRCODE_SYNTAX_ERROR),
+						 errmsg("conflicting or redundant options")));
+			rangeSubtypeDiffName = defGetQualifiedName(defel);
+		}
+		else
+		{
+			ereport(ERROR,
+					(errcode(ERRCODE_SYNTAX_ERROR),
+					 errmsg("type attribute \"%s\" not recognized",
+							defel->defname)));
+			continue;
+		}
+	}
+
+	if (!OidIsValid(rangeSubtype))
+			ereport(ERROR,
+					(errcode(ERRCODE_SYNTAX_ERROR),
+					 errmsg("type attribute \"subtype\" is required")));
+
+	if (type_is_collatable(rangeSubtype))
+	{
+		if (rangeCollationName == NIL)
+			rangeCollation = get_typcollation(rangeSubtype);
+		else
+			rangeCollation = get_collation_oid(rangeCollationName, false);
+	}
+	else if (rangeCollationName != NIL)
+		ereport(ERROR,
+				(errcode(ERRCODE_WRONG_OBJECT_TYPE),
+				 errmsg("range collation provided but subtype does not support collation")));
+
+	rangeSubOpclass = findRangeSubOpclass(rangeSubOpclassName, rangeSubtype);
+
+	if (rangeSubtypeDiffName != NIL)
+		rangeSubtypeDiff = findRangeSubtypeDiffFunction(
+			rangeSubtypeDiffName, rangeSubtype);
+
+	rangeArrayOid = AssignTypeArrayOid();
+
+	/* Create the pg_type entry */
+	typoid =
+		TypeCreate(InvalidOid,	/* no predetermined type OID */
+				   typeName,	/* type name */
+				   typeNamespace,		/* namespace */
+				   InvalidOid,	/* relation oid (n/a here) */
+				   0,			/* relation kind (ditto) */
+				   GetUserId(), /* owner's ID */
+				   -1,			/* internal size */
+				   TYPTYPE_RANGE,	/* type-type (range type) */
+				   TYPCATEGORY_RANGE,	/* type-category (range type) */
+				   false,		/* range types are never preferred */
+				   DEFAULT_TYPDELIM,	/* array element delimiter */
+				   F_RANGE_IN,	/* input procedure */
+				   F_RANGE_OUT,	/* output procedure */
+				   F_RANGE_RECV, /* receive procedure */
+				   F_RANGE_SEND, /* send procedure */
+				   InvalidOid,	/* typmodin procedure - none */
+				   InvalidOid,	/* typmodout procedure - none */
+				   rangeAnalyze,	/* analyze procedure - default */
+				   InvalidOid,	/* element type ID */
+				   false,		/* this is not an array type */
+				   rangeArrayOid,	/* array type we are about to create */
+				   InvalidOid,	/* base type ID (only for domains) */
+				   NULL,		/* never a default type value */
+				   NULL,		/* binary default isn't sent either */
+				   false,		/* never passed by value */
+				   'i',			/* int alignment */
+				   'x',			/* TOAST strategy always plain */
+				   -1,			/* typMod (Domains only) */
+				   0,			/* Array dimensions of typbasetype */
+				   false,		/* Type NOT NULL */
+				   InvalidOid);	/* typcollation */
+
+	/* create the entry in pg_range */
+	RangeCreate(typoid, rangeSubtype, rangeCollation, rangeSubOpclass,
+				rangeCanonical, rangeSubtypeDiff);
+
+	/*
+	 * Create the array type that goes with it.
+	 */
+	rangeArrayName = makeArrayTypeName(typeName, typeNamespace);
+
+	TypeCreate(rangeArrayOid,	/* force assignment of this type OID */
+			   rangeArrayName,	/* type name */
+			   typeNamespace,	/* namespace */
+			   InvalidOid,		/* relation oid (n/a here) */
+			   0,				/* relation kind (ditto) */
+			   GetUserId(),		/* owner's ID */
+			   -1,				/* internal size (always varlena) */
+			   TYPTYPE_BASE,	/* type-type (base type) */
+			   TYPCATEGORY_ARRAY,		/* type-category (array) */
+			   false,			/* array types are never preferred */
+			   DEFAULT_TYPDELIM,	/* array element delimiter */
+			   F_ARRAY_IN,		/* input procedure */
+			   F_ARRAY_OUT,		/* output procedure */
+			   F_ARRAY_RECV,	/* receive procedure */
+			   F_ARRAY_SEND,	/* send procedure */
+			   InvalidOid,		/* typmodin procedure - none */
+			   InvalidOid,		/* typmodout procedure - none */
+			   InvalidOid,		/* analyze procedure - default */
+			   typoid,		/* element type ID */
+			   true,			/* yes this is an array type */
+			   InvalidOid,		/* no further array type */
+			   InvalidOid,		/* base type ID */
+			   NULL,			/* never a default type value */
+			   NULL,			/* binary default isn't sent either */
+			   false,			/* never passed by value */
+			   'i',				/* align 'i' */
+			   'x',				/* ARRAY is always toastable */
+			   -1,				/* typMod (Domains only) */
+			   0,				/* Array dimensions of typbasetype */
+			   false,			/* Type NOT NULL */
+			   InvalidOid);		/* typcollation */
+
+	pfree(rangeArrayName);
+
+	makeRangeConstructor(typeName, typeNamespace, typoid, rangeSubtype);
+}
+
+/*
+ * Because there may exist several range types over one subtype, the range type
+ * can't be determined from the subtype. This means that constructors can't be
+ * polymorphic, and so we must generate a new constructor for every range type
+ * defined.
+ *
+ * We actually define 4 functions with 0 through 3 arguments. This is just to
+ * offer more convenience for the user.
+ */
+static void
+makeRangeConstructor(char *name, Oid namespace, Oid rangeOid, Oid subtype)
+{
+	ObjectAddress		referenced;
+	Oid					constructorArgTypes[3];
+	int					i;
+
+	referenced.classId	   = TypeRelationId;
+	referenced.objectId	   = rangeOid;
+	referenced.objectSubId = 0;
+
+	constructorArgTypes[0] = subtype;
+	constructorArgTypes[1] = subtype;
+	constructorArgTypes[2] = TEXTOID;
+
+	for (i = 0; i < 4; i++)
+	{
+		oidvector		*constructorArgTypesVector;
+		ObjectAddress	 myself;
+		Oid				 procOid;
+		char			*prosrc[4] = { "range_constructor0",
+									   "range_constructor1",
+									   "range_constructor2",
+									   "range_constructor3"};
+
+		constructorArgTypesVector = buildoidvector(constructorArgTypes, i);
+
+		procOid = ProcedureCreate(
+			name, /* name */
+			namespace, /* namespace */
+			false, /* replace */
+			false, /* return set */
+			rangeOid, /* return type */
+			INTERNALlanguageId, /* language */
+			F_FMGR_INTERNAL_VALIDATOR, /* language validator */
+			prosrc[i], /* prosrc */
+			NULL, /* probin */
+			false, /* agg */
+			false, /* window */
+			false, /* security definer */
+			false, /* strict */
+			PROVOLATILE_IMMUTABLE, /* volatility */
+			constructorArgTypesVector, /* param types */
+			PointerGetDatum(NULL), /* allParameterTypes */
+			PointerGetDatum(NULL), /* parameterModes */
+			PointerGetDatum(NULL), /* parameterNames */
+			NIL, /* parameterDefaults */
+			PointerGetDatum(NULL), /* proconfig */
+			1.0, /* procost */
+			0.0); /* prorows */
+
+		/*
+		 * Make the constructor internally-dependent on the range type so that
+		 * the user doesn't have to treat them as separate objects.
+		 */
+		myself.classId	   = ProcedureRelationId;
+		myself.objectId	   = procOid;
+		myself.objectSubId = 0;
+		recordDependencyOn(&myself, &referenced, DEPENDENCY_INTERNAL);
+	}
+}
+
+/*
  * AlterEnum
  *		Adds a new label to an existing enum.
  */
@@ -1450,6 +1790,103 @@ findTypeAnalyzeFunction(List *procname, Oid typeOid)
 }
 
 /*
+ * Find named btree opclass for subtype, or default btree opclass if
+ * opcname is NIL. This will be used for comparing values of subtype.
+ */
+static Oid
+findRangeSubOpclass(List *opcname, Oid subtype)
+{
+	Oid			opcid;
+
+	if (opcname == NIL)
+	{
+		opcid = GetDefaultOpClass(subtype, BTREE_AM_OID);
+		if (!OidIsValid(opcid))
+		{
+			ereport(ERROR,
+					(errcode(ERRCODE_UNDEFINED_OBJECT),
+					 errmsg("data type %s has no default operator class for access method \"btree\"",
+							format_type_be(subtype)),
+					 errhint("You must specify an operator class for the data type or define a default operator class for the data type.")));
+		}
+		return opcid;
+	}
+
+	opcid = get_opclass_oid(BTREE_AM_OID, opcname, false);
+
+	return opcid;
+}
+
+/*
+ * Used to find a range's 'canonical' function.
+ */
+static Oid
+findRangeSubtypeDiffFunction(List *procname, Oid typeOid)
+{
+	Oid			argList[2];
+	Oid			procOid;
+
+	argList[0] = typeOid;
+	argList[1] = typeOid;
+
+	procOid = LookupFuncName(procname, 2, argList, true);
+
+	if (!OidIsValid(procOid))
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_FUNCTION),
+				 errmsg("function %s does not exist",
+						func_signature_string(procname, 2, NIL, argList))));
+
+	if (get_func_rettype(procOid) != FLOAT8OID)
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+				 errmsg("range subtype diff function %s must return type \"float8\"",
+						func_signature_string(procname, 2, NIL, argList))));
+
+	if (func_volatile(procOid) != PROVOLATILE_IMMUTABLE)
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+				 errmsg("range subtype diff function %s must be immutable",
+						func_signature_string(procname, 2, NIL, argList))));
+
+	return procOid;
+}
+
+/*
+ * Used to find a range's 'canonical' function.
+ */
+static Oid
+findRangeCanonicalFunction(List *procname, Oid typeOid)
+{
+	Oid			argList[1];
+	Oid			procOid;
+
+	argList[0] = typeOid;
+
+	procOid = LookupFuncName(procname, 1, argList, true);
+
+	if (!OidIsValid(procOid))
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_FUNCTION),
+				 errmsg("function %s does not exist",
+						func_signature_string(procname, 1, NIL, argList))));
+
+	if (get_func_rettype(procOid) != typeOid)
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+				 errmsg("range canonical function %s must return range type",
+						NameListToString(procname))));
+
+	if (func_volatile(procOid) != PROVOLATILE_IMMUTABLE)
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+				 errmsg("range canonical function %s must be immutable",
+						func_signature_string(procname, 1, NIL, argList))));
+
+	return procOid;
+}
+
+/*
  *	AssignTypeArrayOid
  *
  *	Pre-assign the type's array OID for use in pg_type.typarray