Add ALTER INDEX WHERE construction

Author: knizhnik

src/backend/commands/indexcmds.c

@@ -34,6 +34,7 @@
 #include "commands/tablespace.h"
 #include "mb/pg_wchar.h"
 #include "miscadmin.h"
+#include "funcapi.h"
 #include "nodes/nodeFuncs.h"
 #include "optimizer/clauses.h"
 #include "optimizer/planner.h"
@@ -52,6 +53,9 @@
 #include "utils/snapmgr.h"
 #include "utils/syscache.h"
 #include "utils/tqual.h"
+#include "utils/ruleutils.h"
+#include "executor/executor.h"
+#include "executor/spi.h"
 
 
 /* non-export function prototypes */
@@ -280,33 +284,40 @@ CheckIndexCompatible(Oid oldId,
 	return ret;
 }
 
-#if 0
 void
-AlterIndex(Oid relationId, IndexStmt *stmt, Oid indexRelationId)
+AlterIndex(Oid indexRelationId, IndexStmt *stmt)
 {
 	char* select;
+	Oid heapRelationId;
 	IndexUniqueCheck checkUnique;
-	bool		satisfiesConstraint;
 	Datum		values[INDEX_MAX_KEYS];
 	bool		isnull[INDEX_MAX_KEYS];
 	Relation heapRelation;
 	Relation indexRelation;
     SPIPlanPtr plan;
     Portal portal;
 	HeapTuple tuple;
-	TupleDesc tupdesc;
+	HeapTuple updatedTuple;
 	TupleTableSlot *slot;
 	ItemPointer tupleid;
 	IndexInfo  *indexInfo;
 	EState *estate;
+	Oid	namespaceId;
+	Relation	pg_index;
+	List*       deparseCtx;
 
 	Assert(stmt->whereClause);
+	CheckPredicate((Expr *) stmt->whereClause);
+
+	/* Open the target index relation */
+	indexRelation = index_open(indexRelationId, RowExclusiveLock);
+	namespaceId = RelationGetNamespace(indexRelation);
 
 	/* Open and lock the parent heap relation */
-	heapRelation = heap_openrv(stmt->relation, ShareUpdateExclusiveLock);
+	heapRelationId = IndexGetRelation(indexRelationId, false);
+	heapRelation = heap_open(heapRelationId, ShareLock);
 
-	/* And the target index relation */
-	indexRelation = index_open(indexRelationId, RowExclusiveLock);
+	pg_index = heap_open(IndexRelationId, RowExclusiveLock);
 
 	indexInfo = BuildIndexInfo(indexRelation);
 	Assert(indexInfo->ii_Predicate);
@@ -319,58 +330,40 @@ AlterIndex(Oid relationId, IndexStmt *stmt, Oid indexRelationId)
 
 	checkUnique = indexRelation->rd_index->indisunique ? UNIQUE_CHECK_YES : UNIQUE_CHECK_NO;
 
-    SPI_connect();
-	select = psprintf("select * from %s where %s and not (%s)",
-					  quote_qualified_identifier(get_namespace_name(RelationGetNamespace(heapRelation)),
-												 get_rel_name(relationId)),
-					  nodeToString(indexInfo->ii_Predicate),
-					  nodeToString(stmt->whereClause)
-		);
-	plan = SPI_parepare(select, 0, NULL); 
-	if (plan == NULL) {
-		ereport(ERROR,
-				(errcode(ERRCODE_INVALID_CURSOR_STATE),
-				 errmsg("Failed to preapre statement ", select)));
-	} 
-	portal = SPI_cursor_open(NULL, plan, NULL, NULL, true);
-	if (portal == NULL) { 
-		ereport(ERROR,
-				(errcode(ERRCODE_INVALID_CURSOR_STATE),
-				 errmsg("Failed to open cursor for ", select)));
-	}	
-	while (true)
-	{
-        SPI_cursor_fetch(portal, true, 1);
-        if (!SPI_processed) { 
-			break;
-		}
-		tuple = SPI_tuptable->vals[0];
-		tupdesc = SPI_tuptable->tupdesc;
-		slot = TupleDescGetSlot(tupdesc);
-		tupleid = &tuple->t_datat->t_ctid;
+	/* Update pg_index tuple */
+	tuple = SearchSysCacheCopy1(INDEXRELID, ObjectIdGetDatum(indexRelationId));
+	if (!HeapTupleIsValid(tuple))
+		elog(ERROR, "cache lookup failed for index %u", indexRelationId);
 
-		/* delete tuple from index */
-	}
-    SPI_cursor_close(portal);
+	Assert(Natts_pg_index <= INDEX_MAX_KEYS);
+	heap_deform_tuple(tuple, RelationGetDescr(pg_index), values, isnull);
+	values[Anum_pg_index_indpred - 1] = CStringGetTextDatum(nodeToString(stmt->whereClause));
+	updatedTuple = heap_form_tuple(RelationGetDescr(pg_index), values, isnull);
+	simple_heap_update(pg_index, &tuple->t_self, updatedTuple);
+	CatalogUpdateIndexes(pg_index, updatedTuple);
+	heap_freetuple(updatedTuple);
+	heap_freetuple(tuple);
 
+	slot = MakeSingleTupleTableSlot(RelationGetDescr(heapRelation));
 
+    SPI_connect();
+	deparseCtx = deparse_context_for(RelationGetRelationName(heapRelation), heapRelationId);
 	select = psprintf("select * from %s where %s and not (%s)",
-					  quote_qualified_identifier(get_namespace_name(RelationGetNamespace(heapRelation)),
-												 get_rel_name(relationId)),
-					  nodeToString(stmt->whereClause),
-					  nodeToString(indexInfo->ii_Predicate)
-		);
-	plan = SPI_parepare(select, 0, NULL); 
+					  quote_qualified_identifier(get_namespace_name(namespaceId),
+					                             get_rel_name(heapRelationId)),
+ 		              deparse_expression(stmt->whereClause, deparseCtx, false, false),
+					  deparse_expression((Node*)make_ands_explicit(indexInfo->ii_Predicate), deparseCtx, false, false));
+	plan = SPI_prepare(select, 0, NULL); 
 	if (plan == NULL) {
 		ereport(ERROR,
 				(errcode(ERRCODE_INVALID_CURSOR_STATE),
-				 errmsg("Failed to preapre statement ", select)));
+				 errmsg("Failed to preapre statement %s", select)));
 	} 
 	portal = SPI_cursor_open(NULL, plan, NULL, NULL, true);
 	if (portal == NULL) { 
 		ereport(ERROR,
 				(errcode(ERRCODE_INVALID_CURSOR_STATE),
-				 errmsg("Failed to open cursor for ", select)));
+				 errmsg("Failed to open cursor for %s", select)));
 	}	
 	while (true)
 	{
@@ -379,40 +372,34 @@ AlterIndex(Oid relationId, IndexStmt *stmt, Oid indexRelationId)
 			break;
 		}
 		tuple = SPI_tuptable->vals[0];
-		tupdesc = SPI_tuptable->tupdesc;
-		slot = TupleDescGetSlot(tupdesc);
-		tupleid = &tuple->t_datat->t_ctid;
+		tupleid = &tuple->t_data->t_ctid;
+		ExecStoreTuple(tuple, slot, InvalidBuffer, false);
 
 		FormIndexDatum(indexInfo,
 					   slot,
 					   estate,
 					   values,
 					   isnull);
-		satisfiesConstraint =
-			index_insert(indexRelation, /* index relation */
-						 values,	/* array of index Datums */
-						 isnull,	/* null flags */
-						 tupleid,		/* tid of heap tuple */
-						 heapRelation,	/* heap relation */
-						 checkUnique);	/* type of uniqueness check to do */
-
-		if (!satisfiesConstraint)
-		{
-			ereport(ERROR,
-					(errcode(ERRCODE_INTEGRITY_CONSTRAINT_VIOLATION),
-					 errmsg("Index constraint violation")));
-		}
+		index_insert(indexRelation, /* index relation */
+					 values,	/* array of index Datums */
+					 isnull,	/* null flags */
+					 tupleid,		/* tid of heap tuple */
+					 heapRelation,	/* heap relation */
+					 checkUnique);	/* type of uniqueness check to do */
+
 		SPI_freetuple(tuple);
 		SPI_freetuptable(SPI_tuptable);
 	}
     SPI_cursor_close(portal);
     SPI_finish();
 
-	/* Close both the relations, but keep the locks */
+	ExecDropSingleTupleTableSlot(slot);
+	FreeExecutorState(estate);
+
+	heap_close(pg_index, NoLock);
 	heap_close(heapRelation, NoLock);
 	index_close(indexRelation, NoLock);
 }
-#endif
 
 /*
  * DefineIndex

src/backend/nodes/copyfuncs.c

@@ -3127,6 +3127,7 @@ _copyIndexStmt(const IndexStmt *from)
 	COPY_SCALAR_FIELD(transformed);
 	COPY_SCALAR_FIELD(concurrent);
 	COPY_SCALAR_FIELD(if_not_exists);
+	COPY_SCALAR_FIELD(is_alter);
 
 	return newnode;
 }

src/backend/nodes/equalfuncs.c

@@ -1243,6 +1243,7 @@ _equalIndexStmt(const IndexStmt *a, const IndexStmt *b)
 	COMPARE_SCALAR_FIELD(transformed);
 	COMPARE_SCALAR_FIELD(concurrent);
 	COMPARE_SCALAR_FIELD(if_not_exists);
+	COMPARE_SCALAR_FIELD(is_alter);
 
 	return true;
 }

src/backend/nodes/outfuncs.c

@@ -2189,6 +2189,7 @@ _outIndexStmt(StringInfo str, const IndexStmt *node)
 	WRITE_BOOL_FIELD(transformed);
 	WRITE_BOOL_FIELD(concurrent);
 	WRITE_BOOL_FIELD(if_not_exists);
+	WRITE_BOOL_FIELD(is_alter);
 }
 
 static void

src/backend/parser/gram.y

@@ -1801,6 +1801,15 @@ AlterTableStmt:
 					n->nowait = $13;
 					$$ = (Node *)n;
 				}
+        |   ALTER INDEX qualified_name WHERE a_expr
+		        {
+					IndexStmt* n = makeNode(IndexStmt);
+					n->relation = $3;
+					n->whereClause = $5;
+					n->is_alter = true;
+					$$ = (Node *)n;					
+				}
+			
 		|	ALTER INDEX qualified_name alter_table_cmds
 				{
 					AlterTableStmt *n = makeNode(AlterTableStmt);

src/backend/parser/parse_utilcmd.c

@@ -2065,6 +2065,10 @@ transformIndexStmt(Oid relid, IndexStmt *stmt, const char *queryString)
 	 * to its fields without qualification.  Caller is responsible for locking
 	 * relation, but we still need to open it.
 	 */
+	if (stmt->is_alter) 
+	{
+		relid = IndexGetRelation(relid, false);
+	}
 	rel = relation_open(relid, NoLock);
 	rte = addRangeTableEntryForRelation(pstate, rel, NULL, false, true);
 

src/backend/tcop/utility.c

@@ -1257,22 +1257,29 @@ ProcessUtilitySlow(Node *parsetree,
 
 					/* ... and do it */
 					EventTriggerAlterTableStart(parsetree);
-					address =
-						DefineIndex(relid,		/* OID of heap relation */
-									stmt,
-									InvalidOid, /* no predefined OID */
-									false,		/* is_alter_table */
-									true,		/* check_rights */
-									false,		/* skip_build */
-									false);		/* quiet */
-
-					/*
-					 * Add the CREATE INDEX node itself to stash right away;
-					 * if there were any commands stashed in the ALTER TABLE
-					 * code, we need them to appear after this one.
-					 */
-					EventTriggerCollectSimpleCommand(address, secondaryObject,
-													 parsetree);
+					if (stmt->is_alter)
+					{
+						AlterIndex(relid, stmt);
+					}
+					else
+					{
+						address =
+							DefineIndex(relid,		/* OID of heap relation */
+										stmt,
+										InvalidOid, /* no predefined OID */
+										false,		/* is_alter_table */
+										true,		/* check_rights */
+										false,		/* skip_build */
+										false);		/* quiet */
+						
+						/*
+						 * Add the CREATE INDEX node itself to stash right away;
+						 * if there were any commands stashed in the ALTER TABLE
+						 * code, we need them to appear after this one.
+						 */
+						EventTriggerCollectSimpleCommand(address, secondaryObject,
+														 parsetree);
+					}
 					commandCollected = true;
 					EventTriggerAlterTableEnd();
 				}