Skip to content

Commit e1ca281

Browse files
knizhnikkelvich
knizhnik
authored and
kelvich
committed
Support more restricted predicate in ALTER INDEX clause
1 parent ced7bf4 commit e1ca281

File tree

1 file changed

+92
-59
lines changed

1 file changed

+92
-59
lines changed

‎src/backend/commands/indexcmds.c

Lines changed: 92 additions & 59 deletions
Original file line numberDiff line numberDiff line change
@@ -284,6 +284,30 @@ CheckIndexCompatible(Oid oldId,
284284
return ret;
285285
}
286286

287+
static void
288+
UpdateIndex(Oid indexRelationId, Node* whereClause)
289+
{
290+
Datum values[Natts_pg_index];
291+
bool isnull[Natts_pg_index];
292+
HeapTuple oldTuple;
293+
HeapTuple newTuple;
294+
Relation pg_index;
295+
296+
pg_index = heap_open(IndexRelationId, RowExclusiveLock);
297+
oldTuple = SearchSysCacheCopy1(INDEXRELID, ObjectIdGetDatum(indexRelationId));
298+
if (!HeapTupleIsValid(oldTuple))
299+
elog(ERROR, "cache lookup failed for index %u", indexRelationId);
300+
301+
heap_deform_tuple(oldTuple, RelationGetDescr(pg_index), values, isnull);
302+
values[Anum_pg_index_indpred - 1] = CStringGetTextDatum(nodeToString(whereClause));
303+
newTuple = heap_form_tuple(RelationGetDescr(pg_index), values, isnull);
304+
simple_heap_update(pg_index, &oldTuple->t_self, newTuple);
305+
CatalogUpdateIndexes(pg_index, newTuple);
306+
heap_freetuple(newTuple);
307+
heap_freetuple(oldTuple);
308+
heap_close(pg_index, NoLock);
309+
}
310+
287311
void
288312
AlterIndex(Oid indexRelationId, IndexStmt *stmt)
289313
{
@@ -297,14 +321,15 @@ AlterIndex(Oid indexRelationId, IndexStmt *stmt)
297321
SPIPlanPtr plan;
298322
Portal portal;
299323
HeapTuple tuple;
300-
HeapTuple updatedTuple;
301324
TupleTableSlot *slot;
302325
ItemPointer tupleid;
303326
IndexInfo *indexInfo;
304327
EState *estate;
305328
Oid namespaceId;
306-
Relation pg_index;
307329
List* deparseCtx;
330+
char* oldIndexPredicate;
331+
char* newIndexPredicate;
332+
char* relationName;
308333

309334
Assert(stmt->whereClause);
310335
CheckPredicate((Expr *) stmt->whereClause);
@@ -319,8 +344,6 @@ AlterIndex(Oid indexRelationId, IndexStmt *stmt)
319344
/* indexRelation = index_open(indexRelationId, AccessShareLock); */
320345
namespaceId = RelationGetNamespace(indexRelation);
321346

322-
pg_index = heap_open(IndexRelationId, RowExclusiveLock);
323-
324347
indexInfo = BuildIndexInfo(indexRelation);
325348
Assert(indexInfo->ii_Predicate);
326349
Assert(!indexInfo->ii_ExclusionOps);
@@ -332,75 +355,85 @@ AlterIndex(Oid indexRelationId, IndexStmt *stmt)
332355

333356
checkUnique = indexRelation->rd_index->indisunique ? UNIQUE_CHECK_YES : UNIQUE_CHECK_NO;
334357

335-
/* Update pg_index tuple */
336-
tuple = SearchSysCacheCopy1(INDEXRELID, ObjectIdGetDatum(indexRelationId));
337-
if (!HeapTupleIsValid(tuple))
338-
elog(ERROR, "cache lookup failed for index %u", indexRelationId);
339-
340-
Assert(Natts_pg_index <= INDEX_MAX_KEYS);
341-
heap_deform_tuple(tuple, RelationGetDescr(pg_index), values, isnull);
342-
values[Anum_pg_index_indpred - 1] = CStringGetTextDatum(nodeToString(stmt->whereClause));
343-
updatedTuple = heap_form_tuple(RelationGetDescr(pg_index), values, isnull);
344-
simple_heap_update(pg_index, &tuple->t_self, updatedTuple);
345-
CatalogUpdateIndexes(pg_index, updatedTuple);
346-
heap_freetuple(updatedTuple);
347-
heap_freetuple(tuple);
348-
heap_close(pg_index, NoLock);
349-
350358
slot = MakeSingleTupleTableSlot(RelationGetDescr(heapRelation));
359+
360+
deparseCtx = deparse_context_for(RelationGetRelationName(heapRelation), heapRelationId);
361+
relationName = quote_qualified_identifier(get_namespace_name(namespaceId),
362+
get_rel_name(heapRelationId)),
363+
newIndexPredicate = deparse_expression(stmt->whereClause, deparseCtx, false, false);
364+
oldIndexPredicate = deparse_expression((Node*)make_ands_explicit(indexInfo->ii_Predicate), deparseCtx, false, false);
351365

352366
SPI_connect();
353-
deparseCtx = deparse_context_for(RelationGetRelationName(heapRelation), heapRelationId);
354-
select = psprintf("select * from %s where %s and not (%s)",
355-
quote_qualified_identifier(get_namespace_name(namespaceId),
356-
get_rel_name(heapRelationId)),
357-
deparse_expression(stmt->whereClause, deparseCtx, false, false),
358-
deparse_expression((Node*)make_ands_explicit(indexInfo->ii_Predicate), deparseCtx, false, false));
359-
plan = SPI_prepare(select, 0, NULL);
360-
if (plan == NULL) {
361-
ereport(ERROR,
362-
(errcode(ERRCODE_INVALID_CURSOR_STATE),
363-
errmsg("Failed to preapre statement %s", select)));
364-
}
365-
portal = SPI_cursor_open(NULL, plan, NULL, NULL, true);
366-
if (portal == NULL) {
367+
368+
select = psprintf("select * from %s where %s and not (%s) limit 1",
369+
relationName, oldIndexPredicate, newIndexPredicate);
370+
if (SPI_execute(select, true, 1) != SPI_OK_SELECT)
371+
{
367372
ereport(ERROR,
368373
(errcode(ERRCODE_INVALID_CURSOR_STATE),
369-
errmsg("Failed to open cursor for %s", select)));
374+
errmsg("Failed to execute statement %s", select)));
370375
}
371-
while (true)
372-
{
373-
SPI_cursor_fetch(portal, true, 1);
374-
if (!SPI_processed) {
375-
break;
376-
}
377-
tuple = SPI_tuptable->vals[0];
378-
tupleid = &tuple->t_data->t_ctid;
379-
ExecStoreTuple(tuple, slot, InvalidBuffer, false);
376+
if (SPI_processed) {
377+
/* There is no way in Postgres to exclude records from index, so we have to completelty rebuild index in this case */
378+
bool relpersistence = indexRelation->rd_rel->relpersistence;
379+
index_close(indexRelation, NoLock);
380+
indexRelation->rd_indpred = make_ands_implicit((Expr *) stmt->whereClause);
381+
indexRelation = NULL;
382+
UpdateIndex(indexRelationId, stmt->whereClause);
383+
reindex_index(indexRelationId, false, relpersistence, 0);
384+
} else {
385+
select = psprintf("select * from %s where %s and not (%s)",
386+
relationName, newIndexPredicate, oldIndexPredicate);
387+
plan = SPI_prepare(select, 0, NULL);
388+
if (plan == NULL) {
389+
ereport(ERROR,
390+
(errcode(ERRCODE_INVALID_CURSOR_STATE),
391+
errmsg("Failed to preapre statement %s", select)));
392+
}
393+
portal = SPI_cursor_open(NULL, plan, NULL, NULL, true);
394+
if (portal == NULL) {
395+
ereport(ERROR,
396+
(errcode(ERRCODE_INVALID_CURSOR_STATE),
397+
errmsg("Failed to open cursor for %s", select)));
398+
}
399+
while (true)
400+
{
401+
SPI_cursor_fetch(portal, true, 1);
402+
if (!SPI_processed) {
403+
break;
404+
}
405+
tuple = SPI_tuptable->vals[0];
406+
tupleid = &tuple->t_data->t_ctid;
407+
ExecStoreTuple(tuple, slot, InvalidBuffer, false);
408+
409+
FormIndexDatum(indexInfo,
410+
slot,
411+
estate,
412+
values,
413+
isnull);
414+
index_insert(indexRelation, /* index relation */
415+
values, /* array of index Datums */
416+
isnull, /* null flags */
417+
tupleid, /* tid of heap tuple */
418+
heapRelation, /* heap relation */
419+
checkUnique); /* type of uniqueness check to do */
380420

381-
FormIndexDatum(indexInfo,
382-
slot,
383-
estate,
384-
values,
385-
isnull);
386-
index_insert(indexRelation, /* index relation */
387-
values, /* array of index Datums */
388-
isnull, /* null flags */
389-
tupleid, /* tid of heap tuple */
390-
heapRelation, /* heap relation */
391-
checkUnique); /* type of uniqueness check to do */
392-
393-
SPI_freetuple(tuple);
394-
SPI_freetuptable(SPI_tuptable);
421+
SPI_freetuple(tuple);
422+
SPI_freetuptable(SPI_tuptable);
423+
}
424+
SPI_cursor_close(portal);
425+
426+
UpdateIndex(indexRelationId, stmt->whereClause);
395427
}
396-
SPI_cursor_close(portal);
397428
SPI_finish();
398429

399430
ExecDropSingleTupleTableSlot(slot);
400431
FreeExecutorState(estate);
401432

402433
heap_close(heapRelation, NoLock);
403-
index_close(indexRelation, NoLock);
434+
if (indexRelation) {
435+
index_close(indexRelation, NoLock);
436+
}
404437
}
405438

406439
/*

0 commit comments

Comments
 (0)