diff options
Diffstat (limited to 'src/backend/commands/cluster.c')
| -rw-r--r-- | src/backend/commands/cluster.c | 580 |
1 files changed, 298 insertions, 282 deletions
diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c index 65a9c041643..2b18cb46df0 100644 --- a/src/backend/commands/cluster.c +++ b/src/backend/commands/cluster.c @@ -1,20 +1,20 @@ /*------------------------------------------------------------------------- * * cluster.c-- - * Paul Brown's implementation of cluster index. + * Paul Brown's implementation of cluster index. * - * I am going to use the rename function as a model for this in the - * parser and executor, and the vacuum code as an example in this - * file. As I go - in contrast to the rest of postgres - there will - * be BUCKETS of comments. This is to allow reviewers to understand - * my (probably bogus) assumptions about the way this works. - * [pbrown '94] + * I am going to use the rename function as a model for this in the + * parser and executor, and the vacuum code as an example in this + * file. As I go - in contrast to the rest of postgres - there will + * be BUCKETS of comments. This is to allow reviewers to understand + * my (probably bogus) assumptions about the way this works. + * [pbrown '94] * * Copyright (c) 1994-5, Regents of the University of California * * * IDENTIFICATION - * $Header: /cvsroot/pgsql/src/backend/commands/cluster.c,v 1.13 1997/08/19 21:30:45 momjian Exp $ + * $Header: /cvsroot/pgsql/src/backend/commands/cluster.c,v 1.14 1997/09/07 04:40:36 momjian Exp $ * *------------------------------------------------------------------------- */ @@ -47,307 +47,323 @@ #include <optimizer/internal.h> #ifndef NO_SECURITY #include <utils/acl.h> -#endif /* !NO_SECURITY */ +#endif /* !NO_SECURITY */ static Relation copy_heap(Oid OIDOldHeap); -static void copy_index(Oid OIDOldIndex, Oid OIDNewHeap); -static void rebuildheap(Oid OIDNewHeap, Oid OIDOldHeap, Oid OIDOldIndex); +static void copy_index(Oid OIDOldIndex, Oid OIDNewHeap); +static void rebuildheap(Oid OIDNewHeap, Oid OIDOldHeap, Oid OIDOldIndex); /* * cluster * - * Check that the relation is a relation in the appropriate user - * ACL. I will use the same security that limits users on the - * renamerel() function. + * Check that the relation is a relation in the appropriate user + * ACL. I will use the same security that limits users on the + * renamerel() function. * - * Check that the index specified is appropriate for the task - * ( ie it's an index over this relation ). This is trickier. + * Check that the index specified is appropriate for the task + * ( ie it's an index over this relation ). This is trickier. * - * Create a list of all the other indicies on this relation. Because - * the cluster will wreck all the tids, I'll need to destroy bogus - * indicies. The user will have to re-create them. Not nice, but - * I'm not a nice guy. The alternative is to try some kind of post - * destroy re-build. This may be possible. I'll check out what the - * index create functiond want in the way of paramaters. On the other - * hand, re-creating n indicies may blow out the space. + * Create a list of all the other indicies on this relation. Because + * the cluster will wreck all the tids, I'll need to destroy bogus + * indicies. The user will have to re-create them. Not nice, but + * I'm not a nice guy. The alternative is to try some kind of post + * destroy re-build. This may be possible. I'll check out what the + * index create functiond want in the way of paramaters. On the other + * hand, re-creating n indicies may blow out the space. * - * Create new (temporary) relations for the base heap and the new - * index. - * - * Exclusively lock the relations. - * - * Create new clustered index and base heap relation. + * Create new (temporary) relations for the base heap and the new + * index. + * + * Exclusively lock the relations. + * + * Create new clustered index and base heap relation. * */ void cluster(char oldrelname[], char oldindexname[]) { - Oid OIDOldHeap, OIDOldIndex, OIDNewHeap; - - Relation OldHeap, OldIndex; - Relation NewHeap; - - char NewIndexName[NAMEDATALEN]; - char NewHeapName[NAMEDATALEN]; - char saveoldrelname[NAMEDATALEN]; - char saveoldindexname[NAMEDATALEN]; - - - /* Save the old names because they will get lost when the old relations - * are destroyed. - */ - strcpy(saveoldrelname, oldrelname); - strcpy(saveoldindexname, oldindexname); - - /* - * - * I'm going to force all checking back into the commands.c function. - * - * Get the list if indicies for this relation. If the index we want - * is among them, do not add it to the 'kill' list, as it will be - * handled by the 'clean up' code which commits this transaction. - * - * I'm not using the SysCache, because this will happen but - * once, and the slow way is the sure way in this case. - * - */ - /* - * Like vacuum, cluster spans transactions, so I'm going to handle it in - * the same way. - */ - - /* matches the StartTransaction in PostgresMain() */ - - OldHeap = heap_openr(oldrelname); - if (!RelationIsValid(OldHeap)) { - elog(WARN, "cluster: unknown relation: \"%s\"", - oldrelname); - } - OIDOldHeap = OldHeap->rd_id; /* Get OID for the index scan */ - - OldIndex=index_openr(oldindexname);/* Open old index relation */ - if (!RelationIsValid(OldIndex)) { - elog(WARN, "cluster: unknown index: \"%s\"", - oldindexname); - } - OIDOldIndex = OldIndex->rd_id; /* OID for the index scan */ - - heap_close(OldHeap); - index_close(OldIndex); - - /* - * I need to build the copies of the heap and the index. The Commit() - * between here is *very* bogus. If someone is appending stuff, they will - * get the lock after being blocked and add rows which won't be present in - * the new table. Bleagh! I'd be best to try and ensure that no-one's - * in the tables for the entire duration of this process with a pg_vlock. - */ - NewHeap = copy_heap(OIDOldHeap); - OIDNewHeap = NewHeap->rd_id; - strcpy(NewHeapName,NewHeap->rd_rel->relname.data); - - - /* To make the new heap visible (which is until now empty). */ - CommandCounterIncrement(); - - rebuildheap(OIDNewHeap, OIDOldHeap, OIDOldIndex); - - /* To flush the filled new heap (and the statistics about it). */ - CommandCounterIncrement(); - - /* Create new index over the tuples of the new heap. */ - copy_index(OIDOldIndex, OIDNewHeap); - sprintf(NewIndexName, "temp_%x", OIDOldIndex); - - /* - * make this really happen. Flush all the buffers. - * (Believe me, it is necessary ... ended up in a mess without it.) - */ - CommitTransactionCommand(); - StartTransactionCommand(); - - - /* Destroy old heap (along with its index) and rename new. */ - heap_destroy(oldrelname); - - renamerel(NewHeapName, saveoldrelname); - TypeRename(NewHeapName, saveoldrelname); - - renamerel(NewIndexName, saveoldindexname); - - /* - * Again flush all the buffers. - */ - CommitTransactionCommand(); - StartTransactionCommand(); + Oid OIDOldHeap, + OIDOldIndex, + OIDNewHeap; + + Relation OldHeap, + OldIndex; + Relation NewHeap; + + char NewIndexName[NAMEDATALEN]; + char NewHeapName[NAMEDATALEN]; + char saveoldrelname[NAMEDATALEN]; + char saveoldindexname[NAMEDATALEN]; + + + /* + * Save the old names because they will get lost when the old + * relations are destroyed. + */ + strcpy(saveoldrelname, oldrelname); + strcpy(saveoldindexname, oldindexname); + + /* + * I'm going to force all checking back into the commands.c function. + * + * Get the list if indicies for this relation. If the index we want is + * among them, do not add it to the 'kill' list, as it will be handled + * by the 'clean up' code which commits this transaction. + * + * I'm not using the SysCache, because this will happen but once, and the + * slow way is the sure way in this case. + * + */ + + /* + * Like vacuum, cluster spans transactions, so I'm going to handle it + * in the same way. + */ + + /* matches the StartTransaction in PostgresMain() */ + + OldHeap = heap_openr(oldrelname); + if (!RelationIsValid(OldHeap)) + { + elog(WARN, "cluster: unknown relation: \"%s\"", + oldrelname); + } + OIDOldHeap = OldHeap->rd_id;/* Get OID for the index scan */ + + OldIndex = index_openr(oldindexname); /* Open old index relation */ + if (!RelationIsValid(OldIndex)) + { + elog(WARN, "cluster: unknown index: \"%s\"", + oldindexname); + } + OIDOldIndex = OldIndex->rd_id; /* OID for the index scan */ + + heap_close(OldHeap); + index_close(OldIndex); + + /* + * I need to build the copies of the heap and the index. The Commit() + * between here is *very* bogus. If someone is appending stuff, they + * will get the lock after being blocked and add rows which won't be + * present in the new table. Bleagh! I'd be best to try and ensure + * that no-one's in the tables for the entire duration of this process + * with a pg_vlock. + */ + NewHeap = copy_heap(OIDOldHeap); + OIDNewHeap = NewHeap->rd_id; + strcpy(NewHeapName, NewHeap->rd_rel->relname.data); + + + /* To make the new heap visible (which is until now empty). */ + CommandCounterIncrement(); + + rebuildheap(OIDNewHeap, OIDOldHeap, OIDOldIndex); + + /* To flush the filled new heap (and the statistics about it). */ + CommandCounterIncrement(); + + /* Create new index over the tuples of the new heap. */ + copy_index(OIDOldIndex, OIDNewHeap); + sprintf(NewIndexName, "temp_%x", OIDOldIndex); + + /* + * make this really happen. Flush all the buffers. (Believe me, it is + * necessary ... ended up in a mess without it.) + */ + CommitTransactionCommand(); + StartTransactionCommand(); + + + /* Destroy old heap (along with its index) and rename new. */ + heap_destroy(oldrelname); + + renamerel(NewHeapName, saveoldrelname); + TypeRename(NewHeapName, saveoldrelname); + + renamerel(NewIndexName, saveoldindexname); + + /* + * Again flush all the buffers. + */ + CommitTransactionCommand(); + StartTransactionCommand(); } -static Relation +static Relation copy_heap(Oid OIDOldHeap) { - char NewName[NAMEDATALEN]; - TupleDesc OldHeapDesc, tupdesc; - Oid OIDNewHeap; - Relation NewHeap, OldHeap; - - /* - * Create a new heap relation with a temporary name, which has the - * same tuple description as the old one. - */ - sprintf(NewName,"temp_%x", OIDOldHeap); - - OldHeap= heap_open(OIDOldHeap); - OldHeapDesc= RelationGetTupleDescriptor(OldHeap); - - /* - * Need to make a copy of the tuple descriptor, heap_create modifies - * it. - */ - - tupdesc = CreateTupleDescCopy(OldHeapDesc); - - OIDNewHeap=heap_create(NewName, - NULL, - OldHeap->rd_rel->relarch, - OldHeap->rd_rel->relsmgr, - tupdesc); - - if (!OidIsValid(OIDNewHeap)) - elog(WARN,"clusterheap: cannot create temporary heap relation\n"); - - NewHeap=heap_open(OIDNewHeap); - - heap_close(NewHeap); - heap_close(OldHeap); - - return NewHeap; + char NewName[NAMEDATALEN]; + TupleDesc OldHeapDesc, + tupdesc; + Oid OIDNewHeap; + Relation NewHeap, + OldHeap; + + /* + * Create a new heap relation with a temporary name, which has the + * same tuple description as the old one. + */ + sprintf(NewName, "temp_%x", OIDOldHeap); + + OldHeap = heap_open(OIDOldHeap); + OldHeapDesc = RelationGetTupleDescriptor(OldHeap); + + /* + * Need to make a copy of the tuple descriptor, heap_create modifies + * it. + */ + + tupdesc = CreateTupleDescCopy(OldHeapDesc); + + OIDNewHeap = heap_create(NewName, + NULL, + OldHeap->rd_rel->relarch, + OldHeap->rd_rel->relsmgr, + tupdesc); + + if (!OidIsValid(OIDNewHeap)) + elog(WARN, "clusterheap: cannot create temporary heap relation\n"); + + NewHeap = heap_open(OIDNewHeap); + + heap_close(NewHeap); + heap_close(OldHeap); + + return NewHeap; } static void copy_index(Oid OIDOldIndex, Oid OIDNewHeap) { - Relation OldIndex, NewHeap; - HeapTuple Old_pg_index_Tuple, Old_pg_index_relation_Tuple, pg_proc_Tuple; - IndexTupleForm Old_pg_index_Form; - Form_pg_class Old_pg_index_relation_Form; - Form_pg_proc pg_proc_Form; - char *NewIndexName; - AttrNumber *attnumP; - int natts; - FuncIndexInfo * finfo; - - NewHeap = heap_open(OIDNewHeap); - OldIndex = index_open(OIDOldIndex); - - /* - * OK. Create a new (temporary) index for the one that's already - * here. To do this I get the info from pg_index, re-build the - * FunctInfo if I have to, and add a new index with a temporary - * name. - */ - Old_pg_index_Tuple = - SearchSysCacheTuple(INDEXRELID, - ObjectIdGetDatum(OldIndex->rd_id), - 0,0,0); - - Assert(Old_pg_index_Tuple); - Old_pg_index_Form = (IndexTupleForm)GETSTRUCT(Old_pg_index_Tuple); - - Old_pg_index_relation_Tuple = - SearchSysCacheTuple(RELOID, - ObjectIdGetDatum(OldIndex->rd_id), - 0,0,0); - - Assert(Old_pg_index_relation_Tuple); - Old_pg_index_relation_Form = - (Form_pg_class)GETSTRUCT(Old_pg_index_relation_Tuple); - - NewIndexName = palloc(NAMEDATALEN); /* XXX */ - sprintf(NewIndexName, "temp_%x", OIDOldIndex); /* Set the name. */ - - /* - * Ugly as it is, the only way I have of working out the number of - * attribues is to count them. Mostly there'll be just one but - * I've got to be sure. - */ - for (attnumP = &(Old_pg_index_Form->indkey[0]), natts = 0; - *attnumP != InvalidAttrNumber; - attnumP++, natts++); - - /* - * If this is a functional index, I need to rebuild the functional - * component to pass it to the defining procedure. - */ - if (Old_pg_index_Form->indproc != InvalidOid) { - finfo = (FuncIndexInfo *) palloc(sizeof(FuncIndexInfo)); - FIgetnArgs(finfo) = natts; - FIgetProcOid(finfo) = Old_pg_index_Form->indproc; - - pg_proc_Tuple = - SearchSysCacheTuple(PROOID, - ObjectIdGetDatum(Old_pg_index_Form->indproc), - 0,0,0); - - Assert(pg_proc_Tuple); - pg_proc_Form = (Form_pg_proc)GETSTRUCT(pg_proc_Tuple); - namecpy(&(finfo->funcName), &(pg_proc_Form->proname)); - } else { - finfo = (FuncIndexInfo *) NULL; - natts = 1; - } - - index_create((NewHeap->rd_rel->relname).data, - NewIndexName, - finfo, - NULL, /* type info is in the old index */ - Old_pg_index_relation_Form->relam, - natts, - Old_pg_index_Form->indkey, - Old_pg_index_Form->indclass, - (uint16)0, (Datum) NULL, NULL, - Old_pg_index_Form->indislossy, - Old_pg_index_Form->indisunique); - - heap_close(OldIndex); - heap_close(NewHeap); + Relation OldIndex, + NewHeap; + HeapTuple Old_pg_index_Tuple, + Old_pg_index_relation_Tuple, + pg_proc_Tuple; + IndexTupleForm Old_pg_index_Form; + Form_pg_class Old_pg_index_relation_Form; + Form_pg_proc pg_proc_Form; + char *NewIndexName; + AttrNumber *attnumP; + int natts; + FuncIndexInfo *finfo; + + NewHeap = heap_open(OIDNewHeap); + OldIndex = index_open(OIDOldIndex); + + /* + * OK. Create a new (temporary) index for the one that's already here. + * To do this I get the info from pg_index, re-build the FunctInfo if + * I have to, and add a new index with a temporary name. + */ + Old_pg_index_Tuple = + SearchSysCacheTuple(INDEXRELID, + ObjectIdGetDatum(OldIndex->rd_id), + 0, 0, 0); + + Assert(Old_pg_index_Tuple); + Old_pg_index_Form = (IndexTupleForm) GETSTRUCT(Old_pg_index_Tuple); + + Old_pg_index_relation_Tuple = + SearchSysCacheTuple(RELOID, + ObjectIdGetDatum(OldIndex->rd_id), + 0, 0, 0); + + Assert(Old_pg_index_relation_Tuple); + Old_pg_index_relation_Form = + (Form_pg_class) GETSTRUCT(Old_pg_index_relation_Tuple); + + NewIndexName = palloc(NAMEDATALEN); /* XXX */ + sprintf(NewIndexName, "temp_%x", OIDOldIndex); /* Set the name. */ + + /* + * Ugly as it is, the only way I have of working out the number of + * attribues is to count them. Mostly there'll be just one but I've + * got to be sure. + */ + for (attnumP = &(Old_pg_index_Form->indkey[0]), natts = 0; + *attnumP != InvalidAttrNumber; + attnumP++, natts++); + + /* + * If this is a functional index, I need to rebuild the functional + * component to pass it to the defining procedure. + */ + if (Old_pg_index_Form->indproc != InvalidOid) + { + finfo = (FuncIndexInfo *) palloc(sizeof(FuncIndexInfo)); + FIgetnArgs(finfo) = natts; + FIgetProcOid(finfo) = Old_pg_index_Form->indproc; + + pg_proc_Tuple = + SearchSysCacheTuple(PROOID, + ObjectIdGetDatum(Old_pg_index_Form->indproc), + 0, 0, 0); + + Assert(pg_proc_Tuple); + pg_proc_Form = (Form_pg_proc) GETSTRUCT(pg_proc_Tuple); + namecpy(&(finfo->funcName), &(pg_proc_Form->proname)); + } + else + { + finfo = (FuncIndexInfo *) NULL; + natts = 1; + } + + index_create((NewHeap->rd_rel->relname).data, + NewIndexName, + finfo, + NULL, /* type info is in the old index */ + Old_pg_index_relation_Form->relam, + natts, + Old_pg_index_Form->indkey, + Old_pg_index_Form->indclass, + (uint16) 0, (Datum) NULL, NULL, + Old_pg_index_Form->indislossy, + Old_pg_index_Form->indisunique); + + heap_close(OldIndex); + heap_close(NewHeap); } static void rebuildheap(Oid OIDNewHeap, Oid OIDOldHeap, Oid OIDOldIndex) { - Relation LocalNewHeap, LocalOldHeap, LocalOldIndex; - IndexScanDesc ScanDesc; - RetrieveIndexResult ScanResult; - ItemPointer HeapTid; - HeapTuple LocalHeapTuple; - Buffer LocalBuffer; - Oid OIDNewHeapInsert; - - /* - * Open the relations I need. Scan through the OldHeap on the OldIndex and - * insert each tuple into the NewHeap. - */ - LocalNewHeap=(Relation)heap_open(OIDNewHeap); - LocalOldHeap=(Relation)heap_open(OIDOldHeap); - LocalOldIndex=(Relation)index_open(OIDOldIndex); - - ScanDesc=index_beginscan(LocalOldIndex, false, 0, (ScanKey) NULL); - - while ((ScanResult = - index_getnext(ScanDesc, ForwardScanDirection)) != NULL) { - - HeapTid = &ScanResult->heap_iptr; - LocalHeapTuple = heap_fetch(LocalOldHeap, 0, HeapTid, &LocalBuffer); - OIDNewHeapInsert = - heap_insert(LocalNewHeap, LocalHeapTuple); - pfree(ScanResult); - ReleaseBuffer(LocalBuffer); - } - index_endscan(ScanDesc); - - index_close(LocalOldIndex); - heap_close(LocalOldHeap); - heap_close(LocalNewHeap); + Relation LocalNewHeap, + LocalOldHeap, + LocalOldIndex; + IndexScanDesc ScanDesc; + RetrieveIndexResult ScanResult; + ItemPointer HeapTid; + HeapTuple LocalHeapTuple; + Buffer LocalBuffer; + Oid OIDNewHeapInsert; + + /* + * Open the relations I need. Scan through the OldHeap on the OldIndex + * and insert each tuple into the NewHeap. + */ + LocalNewHeap = (Relation) heap_open(OIDNewHeap); + LocalOldHeap = (Relation) heap_open(OIDOldHeap); + LocalOldIndex = (Relation) index_open(OIDOldIndex); + + ScanDesc = index_beginscan(LocalOldIndex, false, 0, (ScanKey) NULL); + + while ((ScanResult = + index_getnext(ScanDesc, ForwardScanDirection)) != NULL) + { + + HeapTid = &ScanResult->heap_iptr; + LocalHeapTuple = heap_fetch(LocalOldHeap, 0, HeapTid, &LocalBuffer); + OIDNewHeapInsert = + heap_insert(LocalNewHeap, LocalHeapTuple); + pfree(ScanResult); + ReleaseBuffer(LocalBuffer); + } + index_endscan(ScanDesc); + + index_close(LocalOldIndex); + heap_close(LocalOldHeap); + heap_close(LocalNewHeap); } - |