public inbox for [email protected]
help / color / mirror / Atom feedFrom: Michail Nikolaev <[email protected]>
To: Matthias van de Meent <[email protected]>
Cc: PostgreSQL Hackers <[email protected]>
Cc: Andrey Borodin <[email protected]>
Cc: Melanie Plageman <[email protected]>
Subject: Re: Revisiting {CREATE INDEX, REINDEX} CONCURRENTLY improvements
Date: Sun, 8 Sep 2024 17:18:00 +0200
Message-ID: <CANtu0ojHEVU9U_bxgViRmtqNTJ92LnF+76-yzn4axYjGsK2kqQ@mail.gmail.com> (raw)
In-Reply-To: <CANtu0oh4PwBn_h+4p_MxFigRAyJvF-0nA9Tm5NFRwfsWWjZQiA@mail.gmail.com>
References: <CANtu0oiLc-+7h9zfzOVy2cv2UuYk_5MUReVLnVbOay6OgD_KGg@mail.gmail.com>
<CAEze2WgW6pj48xJhG_YLUE1QS+n9Yv0AZQwaWeb-r+X=HAxU_g@mail.gmail.com>
<CANtu0oizNtPUrPB0Mh+2vyjdijTX=LZvO5_dZN3+NqvE-CFPtw@mail.gmail.com>
<CAEze2Wi3BFLkFBcZ+Brfbr-mGBCcWXcWuHucnCnw5ZOQotc6Eg@mail.gmail.com>
<CANtu0ojRX=osoiXL9JJG6g6qOowXVbVYX+mDsN+2jmFVe=eG7w@mail.gmail.com>
<CAEze2Wg03Ps_StwEhgCdSn7VXY9ZUM=zCrf-m1dRZpTWv6wD_A@mail.gmail.com>
<CANtu0oj66JjAq8xyRSeO=MuRHYS2XsYbhHRRESHtOcLJs=3+Sw@mail.gmail.com>
<CANtu0ogT2Qn7-q_jK6+DqBQvFoTt69eQJDKxJARXV9pdWjd0Gg@mail.gmail.com>
<CANtu0ogXgNkEuxbDRwznAZpxEXRmj3NzOen3y-RGHDwig0YBRw@mail.gmail.com>
<CANtu0oi+FTMqDb+6Bv8w7VHiTFVMB1uAAip_P841WQH+ktPixw@mail.gmail.com>
<CAEze2WgeyVnDb_j4gJQYC4+HcSsYQAdeRA1-F0KDnJ=Y0A_TzA@mail.gmail.com>
<CANtu0oga9zqqEFhdmcWyJTK4d6EGMJsMB_LMgVSE8ar0xVm7Ew@mail.gmail.com>
<CANtu0oirtBK_g4jxtw3jehSop3b0WSQaek5Sv5OGSXwxgcHwZQ@mail.gmail.com>
<CANtu0oijWPRGRpaRR_OvT2R5YALzscvcOTFh-=uZKUpNJmuZtw@mail.gmail.com>
<CAEze2WgHFnYdxkNUmvqxOc-cFUNEYaTqL7+Pei=CtA-ZrTOFyw@mail.gmail.com>
<CANtu0oipL3e8fLnejbH4HnByMW6G_auR4v+ns8j-UHhuPW=9og@mail.gmail.com>
<CANtu0ojmVw8GW5bJknnqSp7Dp1xEuoBewdu2imtQ2tGnWpiWEg@mail.gmail.com>
<CAEze2WgNHTWfw_bP6O0zW_=vi1D-yi1nh6-JDj9kd=8UaB-zLA@mail.gmail.com>
<CANtu0ojA5=rT8BN5==OAiQJZh8CAxD_U8thFhZ3mwrZQ6roNOA@mail.gmail.com>
<CAEze2Wh3eSAnXFdY_6roNPb3WD-YsKbNLiKf=cPmAGHkPUd22w@mail.gmail.com>
<CANtu0og_=ypCbH2ZFayn44i=CL0HAXKW390LfZhQ1F56HoFXtQ@mail.gmail.com>
<CAEze2WghpUS29bJJh5GCZ+WtpO4qWmxiFF-CTWFiP4Qq62G58w@mail.gmail.com>
<CANtu0oiuuGRvYRsH-y0iQjfc+JpT9o4mPUXVkz97+sW9BXA+FA@mail.gmail.com>
<CANtu0og6P+O10XLm-AsoqOhZYEjr8SEHFETadSJ8ifO01YP1qg@mail.gmail.com>
<CANtu0oj0pakvxXhHJhsiKgk=ywY57m623G=OvhJnLVFWe9JCpg@mail.gmail.com>
<CANtu0oiOj65kZqP8ngnsc9O+gywUJATgOjOP6pUARXWsmS9cBQ@mail.gmail.com>
<CANtu0oiT9SPFhs=h8DR4YVPox7TJ6jkfR9JgqT-0L+=uy=Lxng@mail.gmail.com>
<CANtu0ogBOtd9ravu1CUbuZWgq6qvn1rny38PGKDPk9zzQPH8_A@mail.gmail.com>
<CAEze2Wj9SgwOpe_1CWnS_D-txQaQyXArR=dm4DTnha93=yua4g@mail.gmail.com>
<CANtu0ohFr7OzNSbxqBhUpR0mXDYyt0Xt6+=Tbq0EC7as7kr+Lg@mail.gmail.com>
<CANtu0oh4PwBn_h+4p_MxFigRAyJvF-0nA9Tm5NFRwfsWWjZQiA@mail.gmail.com>
Hello, Matthias!
>> - I notice you've added a new argument to
>> heapam_index_build_range_scan. I think this could just as well be
>> implemented by reading the indexInfo->ii_Concurrent field, as the
>> values should be equivalent, right?
> Not always; currently, it is set by ResetSnapshotsAllowed[5].
> We fall back to regular index build if there is a predicate or expression
in the index (which should be considered "safe" according to [6]).
> However, we may remove this check later.
> Additionally, there is no sense in resetting the snapshot if we already
have an xmin assigned to the backend for some reason.
I realized you were right. It's always possible to reset snapshots for
concurrent index building without any limitations related to predicates or
expressions.
Additionally, the PROC_IN_SAFE_IC flag is no longer necessary since
snapshots are rotating quickly, and it's possible to wait for them without
requiring any special exceptions for CREATE/REINDEX INDEX CONCURRENTLY.
Currently, it looks like this [1]. I've also attached a single large patch
just for the case.
I plan to restructure the patch into the following set:
* Introduce catalogXmin as a separate value to calculate the horizon for
the catalog.
* Add the STIR access method.
* Modify concurrent build/reindex to use an aux-index approach without
snapshot rotation.
* Add support for snapshot rotation for non-parallel and non-unique cases.
* Extend support for snapshot rotation in parallel index builds.
* Implement snapshot rotation support for unique indexes.
Best regards,
Mikhail
[1]:
https://github.com/postgres/postgres/compare/master...michail-nikolaev:postgres:new_index_concurrent...
>
Attachments:
[text/x-patch] create_index_concurrently_with_aux_index_or_rotated_snapshots.patch (207.6K, 3-create_index_concurrently_with_aux_index_or_rotated_snapshots.patch)
download | inline diff:
Subject: [PATCH] a lot of refactoring
Ensure the correct determination of index safety to be used with set_indexsafe_procflags during REINDEX CONCURRENTLY
Revert "Revert "backend_catalog_xmin in pg_stat_activity""
revert the revert of catalogXmin
fix resetting snapshot during heapam_index_build_range_scan (snapshot is reset between pages)
apply v3-0002-Modify-the-infer_arbiter_indexes-function-to-cons.patch for test stability
fix unique check for building unique indexes
support for unique indexes
revert ThereAreNoPriorRegisteredSnapshots changes
revert ThereAreNoPriorRegisteredSnapshots changes
do not hold xmin while inserting to the index
rename jam to stir
delete ii_Auxiliary
Revert "introduce PROC->catalogXmin"
Revert "backend_catalog_xmin in pg_stat_activity"
some fixes for jam
few tunes
backend_catalog_xmin in pg_stat_activity
disable snapshot reset for unique indexes
just access method to use as index for validation
support for parallel building with snapshot reset
resetting snapshot during heap scan in the case of serial index build
resetting snapshot during validate_index
introduce PROC->catalogXmin
create index concurrently using auxiliary index
---
Index: src/backend/access/heap/heapam_handler.c
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c
--- a/src/backend/access/heap/heapam_handler.c (revision 2b5f57977f6d16796121d796835c48e4241b4da1)
+++ b/src/backend/access/heap/heapam_handler.c (revision 3dea72b62adc8806917dc459b82ff44d962bcb12)
@@ -41,10 +41,12 @@
#include "storage/bufpage.h"
#include "storage/lmgr.h"
#include "storage/predicate.h"
+#include "storage/proc.h"
#include "storage/procarray.h"
#include "storage/smgr.h"
#include "utils/builtins.h"
#include "utils/rel.h"
+#include "utils/injection_point.h"
static void reform_and_rewrite_tuple(HeapTuple tuple,
Relation OldHeap, Relation NewHeap,
@@ -1191,11 +1193,11 @@
ExprContext *econtext;
Snapshot snapshot;
bool need_unregister_snapshot = false;
+ bool pop_active_snapshot = false;
TransactionId OldestXmin;
BlockNumber previous_blkno = InvalidBlockNumber;
BlockNumber root_blkno = InvalidBlockNumber;
OffsetNumber root_offsets[MaxHeapTuplesPerPage];
-
/*
* sanity checks
*/
@@ -1213,6 +1215,8 @@
* only one of those is requested.
*/
Assert(!(anyvisible && checking_uniqueness));
+ Assert(!(anyvisible && indexInfo->ii_Concurrent));
+ Assert(!indexInfo->ii_Concurrent || !HaveRegisteredOrActiveSnapshot() || scan);
/*
* Need an EState for evaluation of index expressions and partial-index
@@ -1252,17 +1256,22 @@
if (!TransactionIdIsValid(OldestXmin))
{
snapshot = RegisterSnapshot(GetTransactionSnapshot());
- need_unregister_snapshot = true;
+ PushActiveSnapshot(snapshot);
+ need_unregister_snapshot = pop_active_snapshot = !indexInfo->ii_Concurrent;
}
else
+ {
+ Assert(!indexInfo->ii_Concurrent);
snapshot = SnapshotAny;
+ }
scan = table_beginscan_strat(heapRelation, /* relation */
snapshot, /* snapshot */
0, /* number of keys */
NULL, /* scan key */
true, /* buffer access strategy OK */
- allow_sync); /* syncscan OK? */
+ allow_sync, /* syncscan OK? */
+ indexInfo->ii_Concurrent);
}
else
{
@@ -1726,8 +1735,12 @@
table_endscan(scan);
/* we can now forget our snapshot, if set and registered by us */
+ if (pop_active_snapshot)
+ PopActiveSnapshot();
if (need_unregister_snapshot)
UnregisterSnapshot(snapshot);
+ if (indexInfo->ii_Concurrent && !hscan)
+ Assert(!TransactionIdIsValid(MyProc->xmin));
ExecDropSingleTupleTableSlot(slot);
@@ -1740,245 +1753,206 @@
return reltuples;
}
-static void
-heapam_index_validate_scan(Relation heapRelation,
- Relation indexRelation,
- IndexInfo *indexInfo,
+static TransactionId
+heapam_index_validate_scan(Relation table_rel,
+ Relation index_rel,
+ Relation aux_index_rel,
+ struct IndexInfo *index_info,
+ struct IndexInfo *aux_index_info,
Snapshot snapshot,
- ValidateIndexState *state)
+ struct ValidateIndexState *state,
+ struct ValidateIndexState *aux_state)
{
- TableScanDesc scan;
- HeapScanDesc hscan;
- HeapTuple heapTuple;
+ IndexFetchTableData *fetch;
+ TransactionId limitXmin;
+
Datum values[INDEX_MAX_KEYS];
bool isnull[INDEX_MAX_KEYS];
- ExprState *predicate;
- TupleTableSlot *slot;
- EState *estate;
- ExprContext *econtext;
- BlockNumber root_blkno = InvalidBlockNumber;
- OffsetNumber root_offsets[MaxHeapTuplesPerPage];
- bool in_index[MaxHeapTuplesPerPage];
- BlockNumber previous_blkno = InvalidBlockNumber;
+
+ TupleTableSlot *slot;
+ EState *estate;
+ ExprContext *econtext;
/* state variables for the merge */
- ItemPointer indexcursor = NULL;
- ItemPointerData decoded;
- bool tuplesort_empty = false;
+ ItemPointer indexcursor = NULL,
+ auxindexcursor = NULL,
+ prev_indexcursor = NULL;
+ ItemPointerData decoded,
+ auxdecoded,
+ prev_decoded,
+ fetched;
+ bool tuplesort_empty = false,
+ auxtuplesort_empty = false;
+ instr_time snapshotTime,
+ currentTime,
+ elapsed;
+
+ Assert(!HaveRegisteredOrActiveSnapshot());
+ Assert(!TransactionIdIsValid(MyProc->xmin));
+
+ snapshot = RegisterSnapshot(GetLatestSnapshot());
+ PushActiveSnapshot(snapshot);
+ INSTR_TIME_SET_CURRENT(snapshotTime);
+ limitXmin = snapshot->xmin;
/*
* sanity checks
*/
- Assert(OidIsValid(indexRelation->rd_rel->relam));
+ Assert(OidIsValid(index_rel->rd_rel->relam));
+ Assert(OidIsValid(aux_index_rel->rd_rel->relam));
- /*
- * Need an EState for evaluation of index expressions and partial-index
- * predicates. Also a slot to hold the current tuple.
- */
estate = CreateExecutorState();
econtext = GetPerTupleExprContext(estate);
- slot = MakeSingleTupleTableSlot(RelationGetDescr(heapRelation),
- &TTSOpsHeapTuple);
+
+ slot = MakeSingleTupleTableSlot(RelationGetDescr(table_rel),
+ &TTSOpsBufferHeapTuple);
/* Arrange for econtext's scan tuple to be the tuple under test */
econtext->ecxt_scantuple = slot;
- /* Set up execution state for predicate, if any. */
- predicate = ExecPrepareQual(indexInfo->ii_Predicate, estate);
+ fetch = heapam_index_fetch_begin(table_rel);
- /*
- * Prepare for scan of the base relation. We need just those tuples
- * satisfying the passed-in reference snapshot. We must disable syncscan
- * here, because it's critical that we read from block zero forward to
- * match the sorted TIDs.
- */
- scan = table_beginscan_strat(heapRelation, /* relation */
- snapshot, /* snapshot */
- 0, /* number of keys */
- NULL, /* scan key */
- true, /* buffer access strategy OK */
- false); /* syncscan not OK */
- hscan = (HeapScanDesc) scan;
+ ItemPointerSetInvalid(&decoded);
+ ItemPointerSetInvalid(&prev_decoded);
+ ItemPointerSetInvalid(&auxdecoded);
+ ItemPointerSetInvalid(&fetched);
- pgstat_progress_update_param(PROGRESS_SCAN_BLOCKS_TOTAL,
- hscan->rs_nblocks);
+ prev_indexcursor = &prev_decoded;
- /*
- * Scan all tuples matching the snapshot.
- */
- while ((heapTuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
+ while (!auxtuplesort_empty)
{
- ItemPointer heapcursor = &heapTuple->t_self;
- ItemPointerData rootTuple;
- OffsetNumber root_offnum;
-
CHECK_FOR_INTERRUPTS();
- state->htups += 1;
-
- if ((previous_blkno == InvalidBlockNumber) ||
- (hscan->rs_cblock != previous_blkno))
+ INSTR_TIME_SET_CURRENT(currentTime);
+ elapsed = currentTime;
+ INSTR_TIME_SUBTRACT(elapsed, snapshotTime);
+ if (INSTR_TIME_GET_MILLISEC(elapsed) >= VALIDATE_INDEX_SNAPSHOT_RESET_INTERVAL)
{
- pgstat_progress_update_param(PROGRESS_SCAN_BLOCKS_DONE,
- hscan->rs_cblock);
- previous_blkno = hscan->rs_cblock;
- }
+ PopActiveSnapshot();
+ UnregisterSnapshot(snapshot);
- /*
- * As commented in table_index_build_scan, we should index heap-only
- * tuples under the TIDs of their root tuples; so when we advance onto
- * a new heap page, build a map of root item offsets on the page.
- *
- * This complicates merging against the tuplesort output: we will
- * visit the live tuples in order by their offsets, but the root
- * offsets that we need to compare against the index contents might be
- * ordered differently. So we might have to "look back" within the
- * tuplesort output, but only within the current page. We handle that
- * by keeping a bool array in_index[] showing all the
- * already-passed-over tuplesort output TIDs of the current page. We
- * clear that array here, when advancing onto a new heap page.
- */
- if (hscan->rs_cblock != root_blkno)
- {
- Page page = BufferGetPage(hscan->rs_cbuf);
+ Assert(!TransactionIdIsValid(MyProc->xmin));
- LockBuffer(hscan->rs_cbuf, BUFFER_LOCK_SHARE);
- heap_get_root_tuples(page, root_offsets);
- LockBuffer(hscan->rs_cbuf, BUFFER_LOCK_UNLOCK);
-
- memset(in_index, 0, sizeof(in_index));
-
- root_blkno = hscan->rs_cblock;
+ snapshot = RegisterSnapshot(GetLatestSnapshot());
+ PushActiveSnapshot(snapshot);
+ limitXmin = TransactionIdNewer(limitXmin, snapshot->xmin);
+ INSTR_TIME_SET_CURRENT(snapshotTime);
}
- /* Convert actual tuple TID to root TID */
- rootTuple = *heapcursor;
- root_offnum = ItemPointerGetOffsetNumber(heapcursor);
-
- if (HeapTupleIsHeapOnly(heapTuple))
- {
- root_offnum = root_offsets[root_offnum - 1];
- if (!OffsetNumberIsValid(root_offnum))
- ereport(ERROR,
- (errcode(ERRCODE_DATA_CORRUPTED),
- errmsg_internal("failed to find parent tuple for heap-only tuple at (%u,%u) in table \"%s\"",
- ItemPointerGetBlockNumber(heapcursor),
- ItemPointerGetOffsetNumber(heapcursor),
- RelationGetRelationName(heapRelation))));
- ItemPointerSetOffsetNumber(&rootTuple, root_offnum);
- }
-
- /*
- * "merge" by skipping through the index tuples until we find or pass
- * the current root tuple.
- */
- while (!tuplesort_empty &&
- (!indexcursor ||
- ItemPointerCompare(indexcursor, &rootTuple) < 0))
{
- Datum ts_val;
- bool ts_isnull;
-
- if (indexcursor)
+ Datum ts_val;
+ bool ts_isnull;
+ auxtuplesort_empty = !tuplesort_getdatum(aux_state->tuplesort, true,
+ false, &ts_val, &ts_isnull,
+ NULL);
+ Assert(auxtuplesort_empty || !ts_isnull);
+ if (!auxtuplesort_empty)
+ {
+ itemptr_decode(&auxdecoded, DatumGetInt64(ts_val));
+ auxindexcursor = &auxdecoded;
+ }
+ else
{
- /*
- * Remember index items seen earlier on the current heap page
- */
- if (ItemPointerGetBlockNumber(indexcursor) == root_blkno)
- in_index[ItemPointerGetOffsetNumber(indexcursor) - 1] = true;
+ auxindexcursor = NULL;
}
+ }
- tuplesort_empty = !tuplesort_getdatum(state->tuplesort, true,
- false, &ts_val, &ts_isnull,
- NULL);
- Assert(tuplesort_empty || !ts_isnull);
- if (!tuplesort_empty)
- {
- itemptr_decode(&decoded, DatumGetInt64(ts_val));
- indexcursor = &decoded;
- }
- else
- {
- /* Be tidy */
- indexcursor = NULL;
- }
- }
+ if (!auxtuplesort_empty)
+ {
+ while (!tuplesort_empty && (indexcursor == NULL || /* null on first time here */
+ ItemPointerCompare(indexcursor, auxindexcursor) < 0))
+ {
+ Datum ts_val;
+ bool ts_isnull;
+ prev_decoded = decoded;
+ tuplesort_empty = !tuplesort_getdatum(state->tuplesort, true,
+ false, &ts_val, &ts_isnull,
+ NULL);
+ Assert(tuplesort_empty || !ts_isnull);
+ if (!tuplesort_empty)
+ {
+ itemptr_decode(&decoded, DatumGetInt64(ts_val));
+ indexcursor = &decoded;
+
+ if (ItemPointerCompare(prev_indexcursor, indexcursor) == 0)
+ {
+ elog(DEBUG5, "skipping duplicate tid in target index snapshot: (%u,%u)",
+ ItemPointerGetBlockNumber(indexcursor),
+ ItemPointerGetOffsetNumber(indexcursor));
+ }
+ }
+ else
+ {
+ indexcursor = NULL;
+ }
+
+ CHECK_FOR_INTERRUPTS();
+ }
- /*
- * If the tuplesort has overshot *and* we didn't see a match earlier,
- * then this tuple is missing from the index, so insert it.
- */
- if ((tuplesort_empty ||
- ItemPointerCompare(indexcursor, &rootTuple) > 0) &&
- !in_index[root_offnum - 1])
- {
- MemoryContextReset(econtext->ecxt_per_tuple_memory);
+ if (tuplesort_empty || ItemPointerCompare(indexcursor, auxindexcursor) > 0)
+ {
+ bool call_again = false;
+ bool all_dead = false;
+ ItemPointer tid;
+
+ fetched = *auxindexcursor;
+ tid = &fetched;
+
+ MemoryContextReset(econtext->ecxt_per_tuple_memory);
- /* Set up for predicate or expression evaluation */
- ExecStoreHeapTuple(heapTuple, slot, false);
-
- /*
- * In a partial index, discard tuples that don't satisfy the
- * predicate.
- */
- if (predicate != NULL)
- {
- if (!ExecQual(predicate, econtext))
- continue;
- }
+ if (heapam_index_fetch_tuple(fetch, tid, snapshot, slot, &call_again, &all_dead))
+ {
- /*
- * For the current heap tuple, extract all the attributes we use
- * in this index, and note which are null. This also performs
- * evaluation of any expressions needed.
- */
- FormIndexDatum(indexInfo,
- slot,
- estate,
- values,
- isnull);
+ FormIndexDatum(index_info,
+ slot,
+ estate,
+ values,
+ isnull);
- /*
- * You'd think we should go ahead and build the index tuple here,
- * but some index AMs want to do further processing on the data
- * first. So pass the values[] and isnull[] arrays, instead.
- */
-
- /*
- * If the tuple is already committed dead, you might think we
- * could suppress uniqueness checking, but this is no longer true
- * in the presence of HOT, because the insert is actually a proxy
- * for a uniqueness check on the whole HOT-chain. That is, the
- * tuple we have here could be dead because it was already
- * HOT-updated, and if so the updating transaction will not have
- * thought it should insert index entries. The index AM will
- * check the whole HOT-chain and correctly detect a conflict if
- * there is one.
- */
-
- index_insert(indexRelation,
- values,
- isnull,
- &rootTuple,
- heapRelation,
- indexInfo->ii_Unique ?
- UNIQUE_CHECK_YES : UNIQUE_CHECK_NO,
- false,
- indexInfo);
+ index_insert(index_rel,
+ values,
+ isnull,
+ auxindexcursor, /* insert root tuple */
+ table_rel,
+ index_info->ii_Unique ?
+ UNIQUE_CHECK_YES : UNIQUE_CHECK_NO,
+ false,
+ index_info);
- state->tups_inserted += 1;
+ state->tups_inserted += 1;
+
+ elog(DEBUG5, "inserted tid: (%u,%u), root: (%u, %u)",
+ ItemPointerGetBlockNumber(auxindexcursor),
+ ItemPointerGetOffsetNumber(auxindexcursor),
+ ItemPointerGetBlockNumber(tid),
+ ItemPointerGetOffsetNumber(tid));
+ }
+ else
+ {
+ elog(DEBUG5, "skipping insert to target index because tid not visible: (%u,%u)",
+ ItemPointerGetBlockNumber(auxindexcursor),
+ ItemPointerGetOffsetNumber(auxindexcursor));
+ }
+ }
}
}
-
- table_endscan(scan);
ExecDropSingleTupleTableSlot(slot);
FreeExecutorState(estate);
- /* These may have been pointing to the now-gone estate */
- indexInfo->ii_ExpressionsState = NIL;
- indexInfo->ii_PredicateState = NULL;
+ heapam_index_fetch_end(fetch);
+
+ PopActiveSnapshot();
+ UnregisterSnapshot(snapshot);
+ InvalidateCatalogSnapshot();
+ Assert(MyProc->xmin == InvalidTransactionId);
+#if USE_INJECTION_POINTS
+ if (MyProc->xid == InvalidTransactionId)
+ INJECTION_POINT("heapam_index_validate_scan_no_xid");
+#endif
+
+ return limitXmin;
}
/*
Index: src/backend/catalog/index.c
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c
--- a/src/backend/catalog/index.c (revision 2b5f57977f6d16796121d796835c48e4241b4da1)
+++ b/src/backend/catalog/index.c (revision 3dea72b62adc8806917dc459b82ff44d962bcb12)
@@ -67,6 +67,7 @@
#include "storage/bufmgr.h"
#include "storage/lmgr.h"
#include "storage/predicate.h"
+#include "storage/proc.h"
#include "storage/smgr.h"
#include "utils/builtins.h"
#include "utils/fmgroids.h"
@@ -741,7 +742,8 @@
bits16 constr_flags,
bool allow_system_table_mods,
bool is_internal,
- Oid *constraintId)
+ Oid *constraintId,
+ char relpersistence)
{
Oid heapRelationId = RelationGetRelid(heapRelation);
Relation pg_class;
@@ -752,7 +754,6 @@
bool is_exclusion;
Oid namespaceId;
int i;
- char relpersistence;
bool isprimary = (flags & INDEX_CREATE_IS_PRIMARY) != 0;
bool invalid = (flags & INDEX_CREATE_INVALID) != 0;
bool concurrent = (flags & INDEX_CREATE_CONCURRENT) != 0;
@@ -782,7 +783,6 @@
namespaceId = RelationGetNamespace(heapRelation);
shared_relation = heapRelation->rd_rel->relisshared;
mapped_relation = RelationIsMapped(heapRelation);
- relpersistence = heapRelation->rd_rel->relpersistence;
/*
* check parameters
@@ -1459,13 +1459,151 @@
0,
true, /* allow table to be a system catalog? */
false, /* is_internal? */
- NULL);
+ NULL,
+ heapRelation->rd_rel->relpersistence);
/* Close the relations used and clean up */
index_close(indexRelation, NoLock);
ReleaseSysCache(indexTuple);
ReleaseSysCache(classTuple);
+ return newIndexId;
+}
+
+Oid
+index_concurrently_create_aux(Relation heapRelation, Oid mainIndexId,
+ Oid tablespaceOid, const char *newName)
+{
+ Relation indexRelation;
+ IndexInfo *oldInfo,
+ *newInfo;
+ Oid newIndexId = InvalidOid;
+ HeapTuple indexTuple;
+
+ List *indexColNames = NIL;
+ List *indexExprs = NIL;
+ List *indexPreds = NIL;
+
+ Oid *auxOpclassIds;
+ int16 *auxColoptions;
+
+ indexRelation = index_open(mainIndexId, RowExclusiveLock);
+
+ /* The new index needs some information from the old index */
+ oldInfo = BuildIndexInfo(indexRelation);
+
+ /*
+ * Build of an auxiliary index with exclusion constraints is not
+ * supported.
+ */
+ if (oldInfo->ii_ExclusionOps != NULL)
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("auxiliary index creation for exclusion constraints is not supported")));
+
+ /* Get the array of class and column options IDs from index info */
+ indexTuple = SearchSysCache1(INDEXRELID, ObjectIdGetDatum(mainIndexId));
+ if (!HeapTupleIsValid(indexTuple))
+ elog(ERROR, "cache lookup failed for index %u", mainIndexId);
+
+
+ /*
+ * Fetch the list of expressions and predicates directly from the
+ * catalogs. This cannot rely on the information from IndexInfo of the
+ * old index as these have been flattened for the planner.
+ */
+ if (oldInfo->ii_Expressions != NIL)
+ {
+ Datum exprDatum;
+ char *exprString;
+
+ exprDatum = SysCacheGetAttrNotNull(INDEXRELID, indexTuple,
+ Anum_pg_index_indexprs);
+ exprString = TextDatumGetCString(exprDatum);
+ indexExprs = (List *) stringToNode(exprString);
+ pfree(exprString);
+ }
+ if (oldInfo->ii_Predicate != NIL)
+ {
+ Datum predDatum;
+ char *predString;
+
+ predDatum = SysCacheGetAttrNotNull(INDEXRELID, indexTuple,
+ Anum_pg_index_indpred);
+ predString = TextDatumGetCString(predDatum);
+ indexPreds = (List *) stringToNode(predString);
+
+ /* Also convert to implicit-AND format */
+ indexPreds = make_ands_implicit((Expr *) indexPreds);
+ pfree(predString);
+ }
+
+ /*
+ * Build the index information for the new index. Note that rebuild of
+ * indexes with exclusion constraints is not supported, hence there is no
+ * need to fill all the ii_Exclusion* fields.
+ */
+ newInfo = makeIndexInfo(oldInfo->ii_NumIndexAttrs,
+ oldInfo->ii_NumIndexKeyAttrs,
+ STIR_AM_OID,
+ indexExprs,
+ indexPreds,
+ false, /* aux index are not unique */
+ oldInfo->ii_NullsNotDistinct,
+ false, /* not ready for inserts */
+ true,
+ false); /* aux are not summarizing */
+
+ /*
+ * Extract the list of column names and the column numbers for the new
+ * index information. All this information will be used for the index
+ * creation.
+ */
+ for (int i = 0; i < oldInfo->ii_NumIndexAttrs; i++)
+ {
+ TupleDesc indexTupDesc = RelationGetDescr(indexRelation);
+ Form_pg_attribute att = TupleDescAttr(indexTupDesc, i);
+
+ indexColNames = lappend(indexColNames, NameStr(att->attname));
+ newInfo->ii_IndexAttrNumbers[i] = oldInfo->ii_IndexAttrNumbers[i];
+ }
+
+ auxOpclassIds = palloc0(sizeof(Oid) * newInfo->ii_NumIndexAttrs);
+ auxColoptions = palloc0(sizeof(int16) * newInfo->ii_NumIndexAttrs);
+
+ for (int i = 0; i < newInfo->ii_NumIndexAttrs; i++)
+ {
+ auxOpclassIds[i] = RECORD_STIR_OPS_OID;
+ auxColoptions[i] = 0;
+ }
+
+ newIndexId = index_create(heapRelation,
+ newName,
+ InvalidOid, /* indexRelationId */
+ InvalidOid, /* parentIndexRelid */
+ InvalidOid, /* parentConstraintId */
+ InvalidRelFileNumber, /* relFileNumber */
+ newInfo,
+ indexColNames,
+ STIR_AM_OID,
+ tablespaceOid,
+ indexRelation->rd_indcollation,
+ auxOpclassIds,
+ NULL,
+ auxColoptions,
+ NULL,
+ (Datum) 0,
+ INDEX_CREATE_SKIP_BUILD | INDEX_CREATE_CONCURRENT,
+ 0,
+ true, /* allow table to be a system catalog? */
+ false, /* is_internal? */
+ NULL,
+ RELPERSISTENCE_UNLOGGED);
+
+ /* Close the relations used and clean up */
+ index_close(indexRelation, NoLock);
+ ReleaseSysCache(indexTuple);
+
return newIndexId;
}
@@ -1488,9 +1626,7 @@
int save_nestlevel;
Relation indexRelation;
IndexInfo *indexInfo;
-
- /* This had better make sure that a snapshot is active */
- Assert(ActiveSnapshotSet());
+ Snapshot snapshot = InvalidSnapshot;
/* Open and lock the parent heap relation */
heapRel = table_open(heapRelationId, ShareUpdateExclusiveLock);
@@ -1508,6 +1644,12 @@
indexRelation = index_open(indexRelationId, RowExclusiveLock);
+ Assert(!TransactionIdIsValid(MyProc->xmin));
+ Assert(!TransactionIdIsValid(MyProc->xid));
+
+ /* BuildIndexInfo requires as snapshot for expressions and predicates */
+ snapshot = RegisterSnapshot(GetTransactionSnapshot());
+ PushActiveSnapshot(snapshot);
/*
* We have to re-build the IndexInfo struct, since it was lost in the
* commit of the transaction where this concurrent index was created at
@@ -1518,11 +1660,17 @@
indexInfo->ii_Concurrent = true;
indexInfo->ii_BrokenHotChain = false;
+ PopActiveSnapshot();
+ UnregisterSnapshot(snapshot);
+ snapshot = InvalidSnapshot;
+
/* Now build the index */
- index_build(heapRel, indexRelation, indexInfo, false, true);
+ index_build(heapRel, indexRelation, indexInfo, false, true);
+ Assert(!TransactionIdIsValid(MyProc->xmin));
+
/* Roll back any GUC changes executed by index functions */
- AtEOXact_GUC(false, save_nestlevel);
+ AtEOXact_GUC(false, save_nestlevel);
/* Restore userid and security context */
SetUserIdAndSecContext(save_userid, save_sec_context);
@@ -3177,7 +3325,8 @@
0, /* number of keys */
NULL, /* scan key */
true, /* buffer access strategy OK */
- true); /* syncscan OK */
+ true, /* syncscan OK */
+ false);
while (table_scan_getnextslot(scan, ForwardScanDirection, slot))
{
@@ -3288,34 +3437,59 @@
* making the table append-only by setting use_fsm). However that would
* add yet more locking issues.
*/
-void
-validate_index(Oid heapId, Oid indexId, Snapshot snapshot)
+TransactionId
+validate_index(Oid heapId, Oid indexId, Oid auxIndexId)
{
Relation heapRelation,
- indexRelation;
- IndexInfo *indexInfo;
- IndexVacuumInfo ivinfo;
- ValidateIndexState state;
+ indexRelation,
+ auxIndexRelation;
+ IndexInfo *indexInfo,
+ *auxIndexInfo;
+ Snapshot snapshot;
+ TransactionId limitXmin;
+ IndexVacuumInfo ivinfo, auxivinfo;
+ ValidateIndexState state, auxState;
Oid save_userid;
int save_sec_context;
int save_nestlevel;
+ int main_work_mem_part = (maintenance_work_mem * 8) / 10;
{
const int progress_index[] = {
- PROGRESS_CREATEIDX_PHASE,
- PROGRESS_CREATEIDX_TUPLES_DONE,
- PROGRESS_CREATEIDX_TUPLES_TOTAL,
- PROGRESS_SCAN_BLOCKS_DONE,
- PROGRESS_SCAN_BLOCKS_TOTAL
+ PROGRESS_CREATEIDX_PHASE,
+ PROGRESS_CREATEIDX_TUPLES_DONE,
+ PROGRESS_CREATEIDX_TUPLES_TOTAL,
+ PROGRESS_SCAN_BLOCKS_DONE,
+ PROGRESS_SCAN_BLOCKS_TOTAL
};
const int64 progress_vals[] = {
- PROGRESS_CREATEIDX_PHASE_VALIDATE_IDXSCAN,
- 0, 0, 0, 0
+ PROGRESS_CREATEIDX_PHASE_VALIDATE_IDXSCAN,
+ 0, 0, 0, 0
};
pgstat_progress_update_multi_param(5, progress_index, progress_vals);
}
+ /*
+ * Now take the "reference snapshot" that will be used by validate_index()
+ * to filter candidate tuples. Beware! There might still be snapshots in
+ * use that treat some transaction as in-progress that our reference
+ * snapshot treats as committed. If such a recently-committed transaction
+ * deleted tuples in the table, we will not include them in the index; yet
+ * those transactions which see the deleting one as still-in-progress will
+ * expect such tuples to be there once we mark the index as valid.
+ *
+ * We solve this by waiting for all endangered transactions to exit before
+ * we mark the index as valid.
+ *
+ * We also set ActiveSnapshot to this snap, since functions in indexes may
+ * need a snapshot.
+ */
+ snapshot = RegisterSnapshot(GetTransactionSnapshot());
+ PushActiveSnapshot(snapshot);
+
+ Assert(TransactionIdIsValid(MyProc->xmin));
+
/* Open and lock the parent heap relation */
heapRelation = table_open(heapId, ShareUpdateExclusiveLock);
@@ -3331,6 +3505,7 @@
RestrictSearchPath();
indexRelation = index_open(indexId, RowExclusiveLock);
+ auxIndexRelation = index_open(auxIndexId, RowExclusiveLock);
/*
* Fetch info needed for index_insert. (You might think this should be
@@ -3338,9 +3513,11 @@
* been built in a previous transaction.)
*/
indexInfo = BuildIndexInfo(indexRelation);
+ auxIndexInfo = BuildIndexInfo(auxIndexRelation);
/* mark build is concurrent just for consistency */
indexInfo->ii_Concurrent = true;
+ auxIndexInfo->ii_Concurrent = true;
/*
* Scan the index and gather up all the TIDs into a tuplesort object.
@@ -3353,6 +3530,10 @@
ivinfo.message_level = DEBUG2;
ivinfo.num_heap_tuples = heapRelation->rd_rel->reltuples;
ivinfo.strategy = NULL;
+ ivinfo.validate_index = true;
+
+ auxivinfo = ivinfo;
+ auxivinfo.index = auxIndexRelation;
/*
* Encode TIDs as int8 values for the sort, rather than directly sorting
@@ -3360,9 +3541,27 @@
* is a pass-by-reference type on all platforms, whereas int8 is
* pass-by-value on most platforms.
*/
+ auxState.tuplesort = tuplesort_begin_datum(INT8OID, Int8LessOperator,
+ InvalidOid, false,
+ maintenance_work_mem - main_work_mem_part,
+ NULL, TUPLESORT_NONE);
+ auxState.htups = auxState.itups = auxState.tups_inserted = 0;
+
+ (void) index_bulk_delete(&auxivinfo, NULL,
+ validate_index_callback, (void *) &auxState);
+
+ PopActiveSnapshot();
+ UnregisterSnapshot(snapshot);
+
+ Assert(!TransactionIdIsValid(MyProc->xmin));
+
+ snapshot = RegisterSnapshot(GetLatestSnapshot());
+ PushActiveSnapshot(snapshot);
+
+
state.tuplesort = tuplesort_begin_datum(INT8OID, Int8LessOperator,
InvalidOid, false,
- maintenance_work_mem,
+ main_work_mem_part,
NULL, TUPLESORT_NONE);
state.htups = state.itups = state.tups_inserted = 0;
@@ -3370,38 +3569,63 @@
(void) index_bulk_delete(&ivinfo, NULL,
validate_index_callback, (void *) &state);
+
+
/* Execute the sort */
{
const int progress_index[] = {
- PROGRESS_CREATEIDX_PHASE,
- PROGRESS_SCAN_BLOCKS_DONE,
- PROGRESS_SCAN_BLOCKS_TOTAL
+ PROGRESS_CREATEIDX_PHASE,
+ PROGRESS_SCAN_BLOCKS_DONE,
+ PROGRESS_SCAN_BLOCKS_TOTAL
};
const int64 progress_vals[] = {
- PROGRESS_CREATEIDX_PHASE_VALIDATE_SORT,
- 0, 0
+ PROGRESS_CREATEIDX_PHASE_VALIDATE_SORT,
+ 0, 0
};
pgstat_progress_update_multi_param(3, progress_index, progress_vals);
}
tuplesort_performsort(state.tuplesort);
+ tuplesort_performsort(auxState.tuplesort);
+
+ /*
+ * Drop the reference snapshot. We must do this before waiting out other
+ * snapshot holders, else we will deadlock against other processes also
+ * doing CREATE INDEX CONCURRENTLY, which would see our snapshot as one
+ * they must wait for. But first, save the snapshot's xmin to use as
+ * limitXmin for GetCurrentVirtualXIDs().
+ */
+ limitXmin = snapshot->xmin;
+
+
+ PopActiveSnapshot();
+ UnregisterSnapshot(snapshot);
+ snapshot = InvalidSnapshot;
+
+ Assert(!TransactionIdIsValid(MyProc->xmin));
+
/*
* Now scan the heap and "merge" it with the index
*/
pgstat_progress_update_param(PROGRESS_CREATEIDX_PHASE,
PROGRESS_CREATEIDX_PHASE_VALIDATE_TABLESCAN);
- table_index_validate_scan(heapRelation,
+ limitXmin = TransactionIdNewer(limitXmin, table_index_validate_scan(heapRelation,
indexRelation,
+ auxIndexRelation,
indexInfo,
- snapshot,
- &state);
+ auxIndexInfo,
+ snapshot, /* may be invalid */
+ &state,
+ &auxState));
/* Done with tuplesort object */
tuplesort_end(state.tuplesort);
+ tuplesort_end(auxState.tuplesort);
/* Make sure to release resources cached in indexInfo (if needed). */
index_insert_cleanup(indexRelation, indexInfo);
+ index_insert_cleanup(auxIndexRelation, auxIndexInfo);
elog(DEBUG2,
"validate_index found %.0f heap tuples, %.0f index tuples; inserted %.0f missing tuples",
@@ -3414,8 +3638,13 @@
SetUserIdAndSecContext(save_userid, save_sec_context);
/* Close rels, but keep locks */
+ index_close(auxIndexRelation, NoLock);
index_close(indexRelation, NoLock);
table_close(heapRelation, NoLock);
+
+ Assert(!HaveRegisteredOrActiveSnapshot());
+ Assert(!TransactionIdIsValid(MyProc->xmin));
+ return limitXmin;
}
/*
@@ -3466,6 +3695,12 @@
Assert(!indexForm->indisready);
Assert(!indexForm->indisvalid);
indexForm->indisready = true;
+ break;
+ case INDEX_DROP_CLEAR_READY:
+ Assert(indexForm->indislive);
+ Assert(indexForm->indisready);
+ Assert(!indexForm->indisvalid);
+ indexForm->indisready = false;
break;
case INDEX_CREATE_SET_VALID:
/* Set indisvalid during a CREATE INDEX CONCURRENTLY sequence */
Index: src/backend/catalog/toasting.c
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/catalog/toasting.c b/src/backend/catalog/toasting.c
--- a/src/backend/catalog/toasting.c (revision 2b5f57977f6d16796121d796835c48e4241b4da1)
+++ b/src/backend/catalog/toasting.c (revision 6973360aaf4eb9012a60a5f2d5d46f022ac2d38c)
@@ -324,7 +324,8 @@
BTREE_AM_OID,
rel->rd_rel->reltablespace,
collationIds, opclassIds, NULL, coloptions, NULL, (Datum) 0,
- INDEX_CREATE_IS_PRIMARY, 0, true, true, NULL);
+ INDEX_CREATE_IS_PRIMARY, 0, true, true, NULL,
+ toast_rel->rd_rel->relpersistence);
table_close(toast_rel, NoLock);
Index: src/backend/commands/indexcmds.c
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/commands/indexcmds.c b/src/backend/commands/indexcmds.c
--- a/src/backend/commands/indexcmds.c (revision 2b5f57977f6d16796121d796835c48e4241b4da1)
+++ b/src/backend/commands/indexcmds.c (revision 3dea72b62adc8806917dc459b82ff44d962bcb12)
@@ -69,6 +69,7 @@
#include "utils/regproc.h"
#include "utils/snapmgr.h"
#include "utils/syscache.h"
+#include "utils/injection_point.h"
/* non-export function prototypes */
@@ -112,7 +113,6 @@
Oid relationOid,
const ReindexParams *params);
static void update_relispartition(Oid relationId, bool newval);
-static inline void set_indexsafe_procflags(void);
/*
* callback argument type for RangeVarCallbackForReindexIndex()
@@ -428,8 +428,7 @@
VirtualTransactionId *old_snapshots;
old_snapshots = GetCurrentVirtualXIDs(limitXmin, true, false,
- PROC_IS_AUTOVACUUM | PROC_IN_VACUUM
- | PROC_IN_SAFE_IC,
+ PROC_IS_AUTOVACUUM | PROC_IN_VACUUM,
&n_old_snapshots);
if (progress)
pgstat_progress_update_param(PROGRESS_WAITFOR_TOTAL, n_old_snapshots);
@@ -449,8 +448,7 @@
newer_snapshots = GetCurrentVirtualXIDs(limitXmin,
true, false,
- PROC_IS_AUTOVACUUM | PROC_IN_VACUUM
- | PROC_IN_SAFE_IC,
+ PROC_IS_AUTOVACUUM | PROC_IN_VACUUM,
&n_newer_snapshots);
for (j = i; j < n_old_snapshots; j++)
{
@@ -542,7 +540,9 @@
{
bool concurrent;
char *indexRelationName;
+ char *auxIndexRelationName;
char *accessMethodName;
+ Oid auxIndexRelationId;
Oid *typeIds;
Oid *collationIds;
Oid *opclassIds;
@@ -561,7 +561,6 @@
bool amissummarizing;
amoptions_function amoptions;
bool partitioned;
- bool safe_index;
Datum reloptions;
int16 *coloptions;
IndexInfo *indexInfo;
@@ -571,10 +570,10 @@
int numberOfKeyAttributes;
TransactionId limitXmin;
ObjectAddress address;
+ ObjectAddress auxAddress;
LockRelId heaprelid;
LOCKTAG heaplocktag;
LOCKMODE lockmode;
- Snapshot snapshot;
Oid root_save_userid;
int root_save_sec_context;
int root_save_nestlevel;
@@ -808,6 +807,7 @@
* Select name for index if caller didn't specify
*/
indexRelationName = stmt->idxname;
+ auxIndexRelationName = NULL;
if (indexRelationName == NULL)
indexRelationName = ChooseIndexName(RelationGetRelationName(rel),
namespaceId,
@@ -815,6 +815,12 @@
stmt->excludeOpNames,
stmt->primary,
stmt->isconstraint);
+ if (concurrent)
+ auxIndexRelationName = ChooseRelationName(indexRelationName,
+ NULL,
+ "ccaux",
+ namespaceId,
+ false);
/*
* look up the access method, verify it can handle the requested features
@@ -1116,10 +1122,6 @@
}
}
- /* Is index safe for others to ignore? See set_indexsafe_procflags() */
- safe_index = indexInfo->ii_Expressions == NIL &&
- indexInfo->ii_Predicate == NIL;
-
/*
* Report index creation if appropriate (delay this till after most of the
* error checks)
@@ -1199,7 +1201,8 @@
coloptions, NULL, reloptions,
flags, constr_flags,
allowSystemTableMods, !check_rights,
- &createdConstraintId);
+ &createdConstraintId,
+ rel->rd_rel->relpersistence);
ObjectAddressSet(address, RelationRelationId, indexRelationId);
@@ -1595,6 +1598,28 @@
return address;
}
+ else
+ {
+ Oid save_userid;
+ int save_sec_context;
+ int save_nestlevel;
+
+ GetUserIdAndSecContext(&save_userid, &save_sec_context);
+ SetUserIdAndSecContext(rel->rd_rel->relowner,
+ save_sec_context | SECURITY_RESTRICTED_OPERATION);
+ save_nestlevel = NewGUCNestLevel();
+ RestrictSearchPath();
+
+ auxIndexRelationId = index_concurrently_create_aux(rel, indexRelationId,
+ tablespaceId, auxIndexRelationName);
+ ObjectAddressSet(auxAddress, RelationRelationId, auxIndexRelationId);
+
+ /* Roll back any GUC changes executed by index functions */
+ AtEOXact_GUC(false, save_nestlevel);
+
+ /* Restore userid and security context */
+ SetUserIdAndSecContext(save_userid, save_sec_context);
+ }
/* save lockrelid and locktag for below, then close rel */
heaprelid = rel->rd_lockInfo.lockRelId;
@@ -1626,11 +1651,18 @@
PopActiveSnapshot();
CommitTransactionCommand();
- StartTransactionCommand();
+
+ {
+ StartTransactionCommand();
- /* Tell concurrent index builds to ignore us, if index qualifies */
- if (safe_index)
- set_indexsafe_procflags();
+ WaitForLockers(heaplocktag, ShareLock, true);
+ index_concurrently_build(tableId, auxIndexRelationId);
+
+ CommitTransactionCommand();
+ }
+
+ StartTransactionCommand();
+
/*
* The index is now visible, so we can report the OID. While on it,
@@ -1685,25 +1717,15 @@
* HOT-chain or the extension of the chain is HOT-safe for this index.
*/
- /* Set ActiveSnapshot since functions in the indexes may need it */
- PushActiveSnapshot(GetTransactionSnapshot());
-
/* Perform concurrent build of index */
index_concurrently_build(tableId, indexRelationId);
- /* we can do away with our snapshot */
- PopActiveSnapshot();
-
/*
* Commit this transaction to make the indisready update visible.
*/
CommitTransactionCommand();
StartTransactionCommand();
- /* Tell concurrent index builds to ignore us, if index qualifies */
- if (safe_index)
- set_indexsafe_procflags();
-
/*
* Phase 3 of concurrent index build
*
@@ -1713,41 +1735,17 @@
pgstat_progress_update_param(PROGRESS_CREATEIDX_PHASE,
PROGRESS_CREATEIDX_PHASE_WAIT_2);
WaitForLockers(heaplocktag, ShareLock, true);
+ index_set_state_flags(auxIndexRelationId, INDEX_DROP_CLEAR_READY);
+ CommitTransactionCommand();
- /*
- * Now take the "reference snapshot" that will be used by validate_index()
- * to filter candidate tuples. Beware! There might still be snapshots in
- * use that treat some transaction as in-progress that our reference
- * snapshot treats as committed. If such a recently-committed transaction
- * deleted tuples in the table, we will not include them in the index; yet
- * those transactions which see the deleting one as still-in-progress will
- * expect such tuples to be there once we mark the index as valid.
- *
- * We solve this by waiting for all endangered transactions to exit before
- * we mark the index as valid.
- *
- * We also set ActiveSnapshot to this snap, since functions in indexes may
- * need a snapshot.
- */
- snapshot = RegisterSnapshot(GetTransactionSnapshot());
- PushActiveSnapshot(snapshot);
+ StartTransactionCommand();
/*
* Scan the index and the heap, insert any missing index entries.
*/
- validate_index(tableId, indexRelationId, snapshot);
+ limitXmin = validate_index(tableId, indexRelationId, auxIndexRelationId);
+ Assert(!TransactionIdIsValid(MyProc->xmin));
- /*
- * Drop the reference snapshot. We must do this before waiting out other
- * snapshot holders, else we will deadlock against other processes also
- * doing CREATE INDEX CONCURRENTLY, which would see our snapshot as one
- * they must wait for. But first, save the snapshot's xmin to use as
- * limitXmin for GetCurrentVirtualXIDs().
- */
- limitXmin = snapshot->xmin;
-
- PopActiveSnapshot();
- UnregisterSnapshot(snapshot);
/*
* The snapshot subsystem could still contain registered snapshots that
@@ -1758,14 +1756,32 @@
* transaction, and do our wait before any snapshot has been taken in it.
*/
CommitTransactionCommand();
+
+ {
+ StartTransactionCommand();
+ index_concurrently_set_dead(tableId, auxIndexRelationId);
+ CommitTransactionCommand();
+ }
+
+ WaitForLockers(heaplocktag, ShareLock, true);
+
+
+ {
+ StartTransactionCommand();
+
+ /*
+ * Use PERFORM_DELETION_CONCURRENT_LOCK so that index_drop() uses the
+ * right lock level.
+ */
+ performDeletion(&auxAddress, DROP_RESTRICT,
+ PERFORM_DELETION_CONCURRENT_LOCK | PERFORM_DELETION_INTERNAL);
+ CommitTransactionCommand();
+ }
+
StartTransactionCommand();
- /* Tell concurrent index builds to ignore us, if index qualifies */
- if (safe_index)
- set_indexsafe_procflags();
-
/* We should now definitely not be advertising any xmin. */
- Assert(MyProc->xmin == InvalidTransactionId);
+ Assert(MyProc->xmin == InvalidTransactionId && MyProc->catalogXmin == InvalidTransactionId);
/*
* The index is now valid in the sense that it contains all currently
@@ -3431,9 +3447,9 @@
typedef struct ReindexIndexInfo
{
Oid indexId;
+ Oid auxIndexId;
Oid tableId;
Oid amId;
- bool safe; /* for set_indexsafe_procflags */
} ReindexIndexInfo;
List *heapRelationIds = NIL;
List *indexIds = NIL;
@@ -3558,6 +3574,7 @@
oldcontext = MemoryContextSwitchTo(private_context);
idx = palloc_object(ReindexIndexInfo);
+ idx->auxIndexId = InvalidOid;
idx->indexId = cellOid;
/* other fields set later */
@@ -3608,6 +3625,7 @@
oldcontext = MemoryContextSwitchTo(private_context);
idx = palloc_object(ReindexIndexInfo);
+ idx->auxIndexId = InvalidOid;
idx->indexId = cellOid;
indexIds = lappend(indexIds, idx);
/* other fields set later */
@@ -3689,6 +3707,7 @@
* that invalid indexes are allowed here.
*/
idx = palloc_object(ReindexIndexInfo);
+ idx->auxIndexId = InvalidOid;
idx->indexId = relationOid;
indexIds = lappend(indexIds, idx);
/* other fields set later */
@@ -3754,15 +3773,18 @@
foreach(lc, indexIds)
{
char *concurrentName;
+ char *auxConcurrentName;
ReindexIndexInfo *idx = lfirst(lc);
ReindexIndexInfo *newidx;
Oid newIndexId;
+ Oid auxIndexId;
Relation indexRel;
Relation heapRel;
Oid save_userid;
int save_sec_context;
int save_nestlevel;
Relation newIndexRel;
+ Relation auxIndexRel;
LockRelId *lockrelid;
Oid tablespaceid;
@@ -3781,9 +3803,6 @@
save_nestlevel = NewGUCNestLevel();
RestrictSearchPath();
- /* determine safety of this index for set_indexsafe_procflags */
- idx->safe = (indexRel->rd_indexprs == NIL &&
- indexRel->rd_indpred == NIL);
idx->tableId = RelationGetRelid(heapRel);
idx->amId = indexRel->rd_rel->relam;
@@ -3805,6 +3824,11 @@
"ccnew",
get_rel_namespace(indexRel->rd_index->indrelid),
false);
+ auxConcurrentName = ChooseRelationName(get_rel_name(idx->indexId),
+ NULL,
+ "ccaux",
+ get_rel_namespace(indexRel->rd_index->indrelid),
+ false);
/* Choose the new tablespace, indexes of toast tables are not moved */
if (OidIsValid(params->tablespaceOid) &&
@@ -3819,11 +3843,17 @@
tablespaceid,
concurrentName);
+ auxIndexId = index_concurrently_create_aux(heapRel,
+ idx->indexId,
+ tablespaceid,
+ auxConcurrentName);
+
/*
* Now open the relation of the new index, a session-level lock is
* also needed on it.
*/
newIndexRel = index_open(newIndexId, ShareUpdateExclusiveLock);
+ auxIndexRel = index_open(auxIndexId, ShareUpdateExclusiveLock);
/*
* Save the list of OIDs and locks in private context
@@ -3831,8 +3861,8 @@
oldcontext = MemoryContextSwitchTo(private_context);
newidx = palloc_object(ReindexIndexInfo);
+ newidx->auxIndexId = auxIndexId;
newidx->indexId = newIndexId;
- newidx->safe = idx->safe;
newidx->tableId = idx->tableId;
newidx->amId = idx->amId;
@@ -3850,10 +3880,14 @@
lockrelid = palloc_object(LockRelId);
*lockrelid = newIndexRel->rd_lockInfo.lockRelId;
relationLocks = lappend(relationLocks, lockrelid);
+ lockrelid = palloc_object(LockRelId);
+ *lockrelid = auxIndexRel->rd_lockInfo.lockRelId;
+ relationLocks = lappend(relationLocks, lockrelid);
MemoryContextSwitchTo(oldcontext);
index_close(indexRel, NoLock);
+ index_close(auxIndexRel, NoLock);
index_close(newIndexRel, NoLock);
/* Roll back any GUC changes executed by index functions */
@@ -3919,6 +3953,27 @@
PopActiveSnapshot();
CommitTransactionCommand();
+
+ {
+ StartTransactionCommand();
+ WaitForLockersMultiple(lockTags, ShareLock, true);
+ CommitTransactionCommand();
+ }
+
+ foreach(lc, newIndexIds)
+ {
+ ReindexIndexInfo *newidx = lfirst(lc);
+
+ StartTransactionCommand();
+
+ CHECK_FOR_INTERRUPTS();
+
+ /* Build auxiliary index, it is fast - without any actual heap scan, just an empty index. */
+ index_concurrently_build(newidx->tableId, newidx->auxIndexId);
+
+ CommitTransactionCommand();
+ }
+
StartTransactionCommand();
/*
@@ -3955,13 +4010,6 @@
*/
CHECK_FOR_INTERRUPTS();
- /* Tell concurrent indexing to ignore us, if index qualifies */
- if (newidx->safe)
- set_indexsafe_procflags();
-
- /* Set ActiveSnapshot since functions in the indexes may need it */
- PushActiveSnapshot(GetTransactionSnapshot());
-
/*
* Update progress for the index to build, with the correct parent
* table involved.
@@ -3976,7 +4024,6 @@
/* Perform concurrent build of new index */
index_concurrently_build(newidx->tableId, newidx->indexId);
- PopActiveSnapshot();
CommitTransactionCommand();
}
@@ -3999,12 +4046,21 @@
PROGRESS_CREATEIDX_PHASE_WAIT_2);
WaitForLockersMultiple(lockTags, ShareLock, true);
CommitTransactionCommand();
+
+ StartTransactionCommand();
+ foreach(lc, newIndexIds)
+ {
+ ReindexIndexInfo *newidx = lfirst(lc);
+ CHECK_FOR_INTERRUPTS();
+
+ index_set_state_flags(newidx->auxIndexId, INDEX_DROP_CLEAR_READY);
+ }
+ CommitTransactionCommand();
foreach(lc, newIndexIds)
{
ReindexIndexInfo *newidx = lfirst(lc);
TransactionId limitXmin;
- Snapshot snapshot;
StartTransactionCommand();
@@ -4015,17 +4071,6 @@
*/
CHECK_FOR_INTERRUPTS();
- /* Tell concurrent indexing to ignore us, if index qualifies */
- if (newidx->safe)
- set_indexsafe_procflags();
-
- /*
- * Take the "reference snapshot" that will be used by validate_index()
- * to filter candidate tuples.
- */
- snapshot = RegisterSnapshot(GetTransactionSnapshot());
- PushActiveSnapshot(snapshot);
-
/*
* Update progress for the index to build, with the correct parent
* table involved.
@@ -4037,16 +4082,9 @@
progress_vals[3] = newidx->amId;
pgstat_progress_update_multi_param(4, progress_index, progress_vals);
- validate_index(newidx->tableId, newidx->indexId, snapshot);
+ limitXmin = validate_index(newidx->tableId, newidx->indexId, newidx->auxIndexId);
- /*
- * We can now do away with our active snapshot, we still need to save
- * the xmin limit to wait for older snapshots.
- */
- limitXmin = snapshot->xmin;
-
- PopActiveSnapshot();
- UnregisterSnapshot(snapshot);
+ Assert(!TransactionIdIsValid(MyProc->xmin));
/*
* To ensure no deadlocks, we must commit and start yet another
@@ -4085,13 +4123,6 @@
StartTransactionCommand();
- /*
- * Because this transaction only does catalog manipulations and doesn't do
- * any index operations, we can set the PROC_IN_SAFE_IC flag here
- * unconditionally.
- */
- set_indexsafe_procflags();
-
forboth(lc, indexIds, lc2, newIndexIds)
{
ReindexIndexInfo *oldidx = lfirst(lc);
@@ -4171,6 +4202,16 @@
index_concurrently_set_dead(oldidx->tableId, oldidx->indexId);
}
+ foreach(lc, newIndexIds)
+ {
+ ReindexIndexInfo *newidx = lfirst(lc);
+
+ CHECK_FOR_INTERRUPTS();
+
+ index_concurrently_set_dead(newidx->tableId, newidx->auxIndexId);
+ }
+
+
/* Commit this transaction to make the updates visible. */
CommitTransactionCommand();
StartTransactionCommand();
@@ -4204,6 +4245,18 @@
object.classId = RelationRelationId;
object.objectId = idx->indexId;
object.objectSubId = 0;
+
+ add_exact_object_address(&object, objects);
+ }
+
+ foreach(lc, newIndexIds)
+ {
+ ReindexIndexInfo *idx = lfirst(lc);
+ ObjectAddress object;
+
+ object.classId = RelationRelationId;
+ object.objectId = idx->auxIndexId;
+ object.objectSubId = 0;
add_exact_object_address(&object, objects);
}
@@ -4424,37 +4477,3 @@
heap_freetuple(tup);
table_close(classRel, RowExclusiveLock);
}
-
-/*
- * Set the PROC_IN_SAFE_IC flag in MyProc->statusFlags.
- *
- * When doing concurrent index builds, we can set this flag
- * to tell other processes concurrently running CREATE
- * INDEX CONCURRENTLY or REINDEX CONCURRENTLY to ignore us when
- * doing their waits for concurrent snapshots. On one hand it
- * avoids pointlessly waiting for a process that's not interesting
- * anyway; but more importantly it avoids deadlocks in some cases.
- *
- * This can be done safely only for indexes that don't execute any
- * expressions that could access other tables, so index must not be
- * expressional nor partial. Caller is responsible for only calling
- * this routine when that assumption holds true.
- *
- * (The flag is reset automatically at transaction end, so it must be
- * set for each transaction.)
- */
-static inline void
-set_indexsafe_procflags(void)
-{
- /*
- * This should only be called before installing xid or xmin in MyProc;
- * otherwise, concurrent processes could see an Xmin that moves backwards.
- */
- Assert(MyProc->xid == InvalidTransactionId &&
- MyProc->xmin == InvalidTransactionId);
-
- LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
- MyProc->statusFlags |= PROC_IN_SAFE_IC;
- ProcGlobal->statusFlags[MyProc->pgxactoff] = MyProc->statusFlags;
- LWLockRelease(ProcArrayLock);
-}
Index: src/bin/pg_amcheck/t/006_concurrently.pl
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/bin/pg_amcheck/t/006_concurrently.pl b/src/bin/pg_amcheck/t/006_concurrently.pl
new file mode 100644
--- /dev/null (revision 3dea72b62adc8806917dc459b82ff44d962bcb12)
+++ b/src/bin/pg_amcheck/t/006_concurrently.pl (revision 3dea72b62adc8806917dc459b82ff44d962bcb12)
@@ -0,0 +1,307 @@
+
+# Copyright (c) 2024, PostgreSQL Global Development Group
+
+# Test REINDEX CONCURRENTLY with concurrent modifications and HOT updates
+use strict;
+use warnings;
+
+use Config;
+use Errno;
+
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Time::HiRes qw(usleep);
+use IPC::SysV;
+use threads;
+use Test::More;
+use Test::Builder;
+
+if ($@ || $windows_os)
+{
+ plan skip_all => 'Fork and shared memory are not supported by this platform';
+}
+
+# TODO: refactor to https://metacpan.org/pod/IPC%3A%3AShareable
+my ($pid, $shmem_id, $shmem_key, $shmem_size);
+eval 'sub IPC_CREAT {0001000}' unless defined &IPC_CREAT;
+$shmem_size = 4;
+$shmem_key = rand(1000000);
+$shmem_id = shmget($shmem_key, $shmem_size, &IPC_CREAT | 0777) or die "Can't shmget: $!";
+shmwrite($shmem_id, "wait", 0, $shmem_size) or die "Can't shmwrite: $!";
+
+my $psql_timeout = IPC::Run::timer($PostgreSQL::Test::Utils::timeout_default);
+#
+# Test set-up
+#
+my ($node, $result);
+$node = PostgreSQL::Test::Cluster->new('RC_test');
+$node->init;
+$node->append_conf('postgresql.conf',
+ 'lock_timeout = ' . (1000 * $PostgreSQL::Test::Utils::timeout_default));
+$node->append_conf('postgresql.conf', 'fsync = off');
+$node->start;
+$node->safe_psql('postgres', q(CREATE EXTENSION amcheck));
+$node->safe_psql('postgres', q(CREATE TABLE tbl(i int primary key,
+ c1 money default 0, c2 money default 0,
+ c3 money default 0, updated_at timestamp)));
+$node->safe_psql('postgres', q(CREATE INDEX idx ON tbl(i)));
+
+my $builder = Test::More->builder;
+$builder->use_numbers(0);
+$builder->no_plan();
+
+my $child = $builder->child("pg_bench");
+
+if(!defined($pid = fork())) {
+ # fork returned undef, so unsuccessful
+ die "Cannot fork a child: $!";
+} elsif ($pid == 0) {
+
+ $node->pgbench(
+ '--no-vacuum --client=10 --transactions=10000',
+ 0,
+ [qr{actually processed}],
+ [qr{^$}],
+ 'concurrent INSERTs, UPDATES and RC',
+ {
+ '001_pgbench_concurrent_transaction_inserts' => q(
+ BEGIN;
+ INSERT INTO tbl VALUES(random()*10000,0,0,0,now())
+ on conflict(i) do update set updated_at = now();
+ INSERT INTO tbl VALUES(random()*10000,0,0,0,now())
+ on conflict(i) do update set updated_at = now();
+ INSERT INTO tbl VALUES(random()*10000,0,0,0,now())
+ on conflict(i) do update set updated_at = now();
+ INSERT INTO tbl VALUES(random()*10000,0,0,0,now())
+ on conflict(i) do update set updated_at = now();
+ INSERT INTO tbl VALUES(random()*10000,0,0,0,now())
+ on conflict(i) do update set updated_at = now();
+ COMMIT;
+ ),
+ '002_pgbench_concurrent_transaction_inserts' => q(
+ BEGIN;
+ INSERT INTO tbl VALUES(random()*100000,0,0,0,now())
+ on conflict(i) do update set updated_at = now();
+ INSERT INTO tbl VALUES(random()*100000,0,0,0,now())
+ on conflict(i) do update set updated_at = now();
+ INSERT INTO tbl VALUES(random()*100000,0,0,0,now())
+ on conflict(i) do update set updated_at = now();
+ INSERT INTO tbl VALUES(random()*100000,0,0,0,now())
+ on conflict(i) do update set updated_at = now();
+ INSERT INTO tbl VALUES(random()*100000,0,0,0,now())
+ on conflict(i) do update set updated_at = now();
+ COMMIT;
+ ),
+ # Ensure some HOT updates happen
+ '003_pgbench_concurrent_transaction_updates' => q(
+ BEGIN;
+ INSERT INTO tbl VALUES(random()*1000,0,0,0,now())
+ on conflict(i) do update set updated_at = now();
+ INSERT INTO tbl VALUES(random()*1000,0,0,0,now())
+ on conflict(i) do update set updated_at = now();
+ INSERT INTO tbl VALUES(random()*1000,0,0,0,now())
+ on conflict(i) do update set updated_at = now();
+ INSERT INTO tbl VALUES(random()*1000,0,0,0,now())
+ on conflict(i) do update set updated_at = now();
+ INSERT INTO tbl VALUES(random()*1000,0,0,0,now())
+ on conflict(i) do update set updated_at = now();
+ COMMIT;
+ )
+ });
+
+ if ($child->is_passing()) {
+ shmwrite($shmem_id, "done", 0, $shmem_size) or die "Can't shmwrite: $!";
+ } else {
+ shmwrite($shmem_id, "fail", 0, $shmem_size) or die "Can't shmwrite: $!";
+ }
+
+ my $pg_bench_fork_flag;
+ while (1) {
+ shmread($shmem_id, $pg_bench_fork_flag, 0, $shmem_size) or die "Can't shmread: $!";
+ sleep(0.1);
+ last if $pg_bench_fork_flag eq "stop";
+ }
+} else {
+ my $pg_bench_fork_flag;
+ shmread($shmem_id, $pg_bench_fork_flag, 0, $shmem_size) or die "Can't shmread: $!";
+
+ subtest 'reindex run subtest' => sub {
+ is($pg_bench_fork_flag, "wait", "pg_bench_fork_flag is correct");
+
+ my %psql = (stdin => '', stdout => '', stderr => '');
+ $psql{run} = IPC::Run::start(
+ [ 'psql', '-XA', '-f', '-', '-d', $node->connstr('postgres') ],
+ '<',
+ \$psql{stdin},
+ '>',
+ \$psql{stdout},
+ '2>',
+ \$psql{stderr},
+ $psql_timeout);
+
+ my ($result, $stdout, $stderr, $n, $stderr_saved);
+ $n = 0;
+
+ $node->psql('postgres', q(CREATE FUNCTION predicate_stable() RETURNS bool IMMUTABLE
+ LANGUAGE plpgsql AS $$
+ BEGIN
+ EXECUTE 'SELECT txid_current()';
+ RETURN true;
+ END; $$;));
+
+ $node->psql('postgres', q(CREATE FUNCTION predicate_const(integer) RETURNS bool IMMUTABLE
+ LANGUAGE plpgsql AS $$
+ BEGIN
+ RETURN MOD($1, 2) = 0;
+ END; $$;));
+ while (1)
+ {
+
+ if (int(rand(2)) == 0) {
+ ($result, $stdout, $stderr) = $node->psql('postgres', q(ALTER TABLE tbl SET (parallel_workers=1);));
+ } else {
+ ($result, $stdout, $stderr) = $node->psql('postgres', q(ALTER TABLE tbl SET (parallel_workers=4);));
+ }
+ is($result, '0', 'ALTER TABLE is correct');
+
+ if (1)
+ {
+ ($result, $stdout, $stderr) = $node->psql('postgres', q(REINDEX INDEX CONCURRENTLY idx;));
+ is($result, '0', 'REINDEX is correct');
+
+ if ($result) {
+ diag($stderr);
+ BAIL_OUT($stderr);
+ }
+
+ ($result, $stdout, $stderr) = $node->psql('postgres', q(SELECT bt_index_parent_check('idx', heapallindexed => true, rootdescend => true, checkunique => true);));
+ is($result, '0', 'bt_index_check is correct');
+ if ($result)
+ {
+ diag($stderr);
+ BAIL_OUT($stderr);
+ } else {
+ diag('reindex:)' . $n++);
+ }
+ }
+
+ if (1)
+ {
+ my $variant = int(rand(7));
+ my $sql;
+ if ($variant == 0) {
+ $sql = q(CREATE INDEX CONCURRENTLY idx_2 ON tbl(i, updated_at););
+ } elsif ($variant == 1) {
+ $sql = q(CREATE INDEX CONCURRENTLY idx_2 ON tbl(i, updated_at) WHERE predicate_stable(););
+ } elsif ($variant == 2) {
+ $sql = q(CREATE INDEX CONCURRENTLY idx_2 ON tbl(i, updated_at) WHERE MOD(i, 2) = 0;);
+ } elsif ($variant == 3) {
+ $sql = q(CREATE INDEX CONCURRENTLY idx_2 ON tbl(i, updated_at) WHERE predicate_const(i););
+ } elsif ($variant == 4) {
+ $sql = q(CREATE INDEX CONCURRENTLY idx_2 ON tbl(predicate_const(i)););
+ } elsif ($variant == 5) {
+ $sql = q(CREATE INDEX CONCURRENTLY idx_2 ON tbl(i, predicate_const(i), updated_at) WHERE predicate_const(i););
+ } elsif ($variant == 6) {
+ $sql = q(CREATE UNIQUE INDEX CONCURRENTLY idx_2 ON tbl(i););
+ } else { diag("wrong variant"); }
+
+ diag($sql);
+ ($result, $stdout, $stderr) = $node->psql('postgres', $sql);
+ is($result, '0', 'CREATE INDEX is correct');
+ $stderr_saved = $stderr;
+
+ ($result, $stdout, $stderr) = $node->psql('postgres', q(SELECT bt_index_parent_check('idx_2', heapallindexed => true, rootdescend => true, checkunique => true);));
+ is($result, '0', 'bt_index_check for new index is correct');
+ if ($result)
+ {
+ diag($stderr);
+ diag($stderr_saved);
+ BAIL_OUT($stderr);
+ } else {
+ diag('create:)' . $n++);
+ }
+
+ if (1)
+ {
+ ($result, $stdout, $stderr) = $node->psql('postgres', q(REINDEX INDEX CONCURRENTLY idx_2;));
+ is($result, '0', 'REINDEX 2 is correct');
+ if ($result) {
+ diag($stderr);
+ BAIL_OUT($stderr);
+ }
+
+ ($result, $stdout, $stderr) = $node->psql('postgres', q(SELECT bt_index_parent_check('idx_2', heapallindexed => true, rootdescend => true, checkunique => true);));
+ is($result, '0', 'bt_index_check 2 is correct');
+ if ($result)
+ {
+ diag($stderr);
+ BAIL_OUT($stderr);
+ } else {
+ diag('reindex2:)' . $n++);
+ }
+ }
+
+ ($result, $stdout, $stderr) = $node->psql('postgres', q(DROP INDEX CONCURRENTLY idx_2;));
+ is($result, '0', 'DROP INDEX is correct');
+ }
+ shmread($shmem_id, $pg_bench_fork_flag, 0, $shmem_size) or die "Can't shmread: $!";
+ last if $pg_bench_fork_flag ne "wait";
+ }
+
+ # explicitly shut down psql instances gracefully
+ $psql{stdin} .= "\\q\n";
+ $psql{run}->finish;
+
+ is($pg_bench_fork_flag, "done", "pg_bench_fork_flag is correct");
+ };
+
+ $child->finalize();
+ $child->summary();
+ $node->stop;
+ done_testing();
+
+ shmwrite($shmem_id, "stop", 0, $shmem_size) or die "Can't shmwrite: $!";
+}
+
+# Send query, wait until string matches
+sub send_query_and_wait
+{
+ my ($psql, $query, $untl) = @_;
+ my $ret;
+
+ # For each query we run, we'll restart the timeout. Otherwise the timeout
+ # would apply to the whole test script, and would need to be set very high
+ # to survive when running under Valgrind.
+ $psql_timeout->reset();
+ $psql_timeout->start();
+
+ # send query
+ $$psql{stdin} .= $query;
+ $$psql{stdin} .= "\n";
+
+ # wait for query results
+ $$psql{run}->pump_nb();
+ while (1)
+ {
+ last if $$psql{stdout} =~ /$untl/;
+ if ($psql_timeout->is_expired)
+ {
+ diag("aborting wait: program timed out\n"
+ . "stream contents: >>$$psql{stdout}<<\n"
+ . "pattern searched for: $untl\n");
+ return 0;
+ }
+ if (not $$psql{run}->pumpable())
+ {
+ diag("aborting wait: program died\n"
+ . "stream contents: >>$$psql{stdout}<<\n"
+ . "pattern searched for: $untl\n");
+ return 0;
+ }
+ $$psql{run}->pump();
+ }
+
+ $$psql{stdout} = '';
+
+ return 1;
+}
Index: src/include/access/tableam.h
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/include/access/tableam.h b/src/include/access/tableam.h
--- a/src/include/access/tableam.h (revision 2b5f57977f6d16796121d796835c48e4241b4da1)
+++ b/src/include/access/tableam.h (revision 3dea72b62adc8806917dc459b82ff44d962bcb12)
@@ -24,6 +24,7 @@
#include "storage/read_stream.h"
#include "utils/rel.h"
#include "utils/snapshot.h"
+#include "utils/injection_point.h"
#define DEFAULT_TABLE_ACCESS_METHOD "heap"
@@ -70,6 +71,7 @@
* needed. If table data may be needed, set SO_NEED_TUPLES.
*/
SO_NEED_TUPLES = 1 << 10,
+ SO_RESET_SNAPSHOT = 1 << 11,
} ScanOptions;
/*
@@ -703,11 +705,14 @@
TableScanDesc scan);
/* see table_index_validate_scan for reference about parameters */
- void (*index_validate_scan) (Relation table_rel,
+ TransactionId (*index_validate_scan) (Relation table_rel,
Relation index_rel,
+ Relation aux_index_rel,
struct IndexInfo *index_info,
+ struct IndexInfo *aux_index_info,
Snapshot snapshot,
- struct ValidateIndexState *state);
+ struct ValidateIndexState *state,
+ struct ValidateIndexState *aux_state);
/* ------------------------------------------------------------------------
@@ -931,7 +936,8 @@
static inline TableScanDesc
table_beginscan_strat(Relation rel, Snapshot snapshot,
int nkeys, struct ScanKeyData *key,
- bool allow_strat, bool allow_sync)
+ bool allow_strat, bool allow_sync,
+ bool reset_snapshot)
{
uint32 flags = SO_TYPE_SEQSCAN | SO_ALLOW_PAGEMODE;
@@ -939,6 +945,11 @@
flags |= SO_ALLOW_STRAT;
if (allow_sync)
flags |= SO_ALLOW_SYNC;
+ if (reset_snapshot)
+ {
+ INJECTION_POINT("table_beginscan_strat_reset_snapshots");
+ flags |= (SO_RESET_SNAPSHOT | SO_TEMP_SNAPSHOT);
+ }
return rel->rd_tableam->scan_begin(rel, snapshot, nkeys, key, NULL, flags);
}
@@ -1835,19 +1846,26 @@
*
* See validate_index() for an explanation.
*/
-static inline void
+static inline TransactionId
table_index_validate_scan(Relation table_rel,
- Relation index_rel,
- struct IndexInfo *index_info,
- Snapshot snapshot,
- struct ValidateIndexState *state)
+ Relation index_rel,
+ Relation aux_index_rel,
+ struct IndexInfo *index_info,
+ struct IndexInfo *aux_index_info,
+ Snapshot snapshot,
+ struct ValidateIndexState *state,
+ struct ValidateIndexState *auxstate)
{
- table_rel->rd_tableam->index_validate_scan(table_rel,
- index_rel,
- index_info,
- snapshot,
- state);
+ return table_rel->rd_tableam->index_validate_scan(table_rel,
+ index_rel,
+ aux_index_rel,
+ index_info,
+ aux_index_info,
+ snapshot,
+ state,
+ auxstate);
}
+
/* ----------------------------------------------------------------------------
Index: src/include/catalog/index.h
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/include/catalog/index.h b/src/include/catalog/index.h
--- a/src/include/catalog/index.h (revision 2b5f57977f6d16796121d796835c48e4241b4da1)
+++ b/src/include/catalog/index.h (revision 3dea72b62adc8806917dc459b82ff44d962bcb12)
@@ -26,6 +26,7 @@
INDEX_CREATE_SET_READY,
INDEX_CREATE_SET_VALID,
INDEX_DROP_CLEAR_VALID,
+ INDEX_DROP_CLEAR_READY,
INDEX_DROP_SET_DEAD,
} IndexStateFlagsAction;
@@ -43,6 +44,8 @@
#define REINDEXOPT_MISSING_OK 0x04 /* skip missing relations */
#define REINDEXOPT_CONCURRENTLY 0x08 /* concurrent mode */
+#define VALIDATE_INDEX_SNAPSHOT_RESET_INTERVAL 50 /* 50 ms */
+
/* state info for validate_index bulkdelete callback */
typedef struct ValidateIndexState
{
@@ -86,7 +89,8 @@
bits16 constr_flags,
bool allow_system_table_mods,
bool is_internal,
- Oid *constraintId);
+ Oid *constraintId,
+ char relpersistence);
#define INDEX_CONSTR_CREATE_MARK_AS_PRIMARY (1 << 0)
#define INDEX_CONSTR_CREATE_DEFERRABLE (1 << 1)
@@ -98,6 +102,11 @@
Oid oldIndexId,
Oid tablespaceOid,
const char *newName);
+
+extern Oid index_concurrently_create_aux(Relation heapRelation,
+ Oid mainIndexId,
+ Oid tablespaceOid,
+ const char *newName);
extern void index_concurrently_build(Oid heapRelationId,
Oid indexRelationId);
@@ -144,7 +153,7 @@
bool isreindex,
bool parallel);
-extern void validate_index(Oid heapId, Oid indexId, Snapshot snapshot);
+extern TransactionId validate_index(Oid heapId, Oid indexId, Oid auxIndexId);
extern void index_set_state_flags(Oid indexId, IndexStateFlagsAction action);
Index: src/include/commands/progress.h
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/include/commands/progress.h b/src/include/commands/progress.h
--- a/src/include/commands/progress.h (revision 2b5f57977f6d16796121d796835c48e4241b4da1)
+++ b/src/include/commands/progress.h (revision bbc09a323cc3d6c54f2d26c7c6342d36d7edeb31)
@@ -79,6 +79,7 @@
/* Progress parameters for CREATE INDEX */
/* 3, 4 and 5 reserved for "waitfor" metrics */
+// TODO: new phase names
#define PROGRESS_CREATEIDX_COMMAND 0
#define PROGRESS_CREATEIDX_INDEX_OID 6
#define PROGRESS_CREATEIDX_ACCESS_METHOD_OID 8
@@ -91,6 +92,7 @@
/* 15 and 16 reserved for "block number" metrics */
/* Phases of CREATE INDEX (as advertised via PROGRESS_CREATEIDX_PHASE) */
+// TODO: new phase names
#define PROGRESS_CREATEIDX_PHASE_WAIT_1 1
#define PROGRESS_CREATEIDX_PHASE_BUILD 2
#define PROGRESS_CREATEIDX_PHASE_WAIT_2 3
Index: src/test/regress/expected/create_index.out
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/test/regress/expected/create_index.out b/src/test/regress/expected/create_index.out
--- a/src/test/regress/expected/create_index.out (revision 2b5f57977f6d16796121d796835c48e4241b4da1)
+++ b/src/test/regress/expected/create_index.out (revision d8df9daea76374468c28f8e9d60d83539aad05c8)
@@ -1405,6 +1405,7 @@
CREATE UNIQUE INDEX CONCURRENTLY concur_index3 ON concur_heap(f2);
ERROR: could not create unique index "concur_index3"
DETAIL: Key (f2)=(b) is duplicated.
+DROP INDEX concur_index3_ccaux;
-- test that expression indexes and partial indexes work concurrently
CREATE INDEX CONCURRENTLY concur_index4 on concur_heap(f2) WHERE f1='a';
CREATE INDEX CONCURRENTLY concur_index5 on concur_heap(f2) WHERE f1='x';
@@ -2705,6 +2706,7 @@
CREATE UNIQUE INDEX CONCURRENTLY concur_reindex_ind5 ON concur_reindex_tab4 (c1);
ERROR: could not create unique index "concur_reindex_ind5"
DETAIL: Key (c1)=(1) is duplicated.
+DROP INDEX concur_reindex_ind5_ccaux;
-- Reindexing concurrently this index fails with the same failure.
-- The extra index created is itself invalid, and can be dropped.
REINDEX INDEX CONCURRENTLY concur_reindex_ind5;
@@ -2717,8 +2719,10 @@
c1 | integer | | |
Indexes:
"concur_reindex_ind5" UNIQUE, btree (c1) INVALID
+ "concur_reindex_ind5_ccaux" stir (c1 record_ops) INVALID
"concur_reindex_ind5_ccnew" UNIQUE, btree (c1) INVALID
+DROP INDEX concur_reindex_ind5_ccaux;
DROP INDEX concur_reindex_ind5_ccnew;
-- This makes the previous failure go away, so the index can become valid.
DELETE FROM concur_reindex_tab4 WHERE c1 = 1;
Index: src/test/regress/expected/indexing.out
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/test/regress/expected/indexing.out b/src/test/regress/expected/indexing.out
--- a/src/test/regress/expected/indexing.out (revision 2b5f57977f6d16796121d796835c48e4241b4da1)
+++ b/src/test/regress/expected/indexing.out (revision bbc09a323cc3d6c54f2d26c7c6342d36d7edeb31)
@@ -1571,10 +1571,11 @@
--------------------------------+------------+-----------------------+-------------------------------
parted_isvalid_idx | f | parted_isvalid_tab |
parted_isvalid_idx_11 | f | parted_isvalid_tab_11 | parted_isvalid_tab_1_expr_idx
+ parted_isvalid_idx_11_ccaux | f | parted_isvalid_tab_11 |
parted_isvalid_tab_12_expr_idx | t | parted_isvalid_tab_12 | parted_isvalid_tab_1_expr_idx
parted_isvalid_tab_1_expr_idx | f | parted_isvalid_tab_1 | parted_isvalid_idx
parted_isvalid_tab_2_expr_idx | t | parted_isvalid_tab_2 | parted_isvalid_idx
-(5 rows)
+(6 rows)
drop table parted_isvalid_tab;
-- Check state of replica indexes when attaching a partition.
Index: src/test/regress/sql/create_index.sql
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/test/regress/sql/create_index.sql b/src/test/regress/sql/create_index.sql
--- a/src/test/regress/sql/create_index.sql (revision 2b5f57977f6d16796121d796835c48e4241b4da1)
+++ b/src/test/regress/sql/create_index.sql (revision bbc09a323cc3d6c54f2d26c7c6342d36d7edeb31)
@@ -493,6 +493,7 @@
INSERT INTO concur_heap VALUES ('b','x');
-- check if constraint is enforced properly at build time
CREATE UNIQUE INDEX CONCURRENTLY concur_index3 ON concur_heap(f2);
+DROP INDEX concur_index3_ccaux;
-- test that expression indexes and partial indexes work concurrently
CREATE INDEX CONCURRENTLY concur_index4 on concur_heap(f2) WHERE f1='a';
CREATE INDEX CONCURRENTLY concur_index5 on concur_heap(f2) WHERE f1='x';
@@ -1147,10 +1148,12 @@
INSERT INTO concur_reindex_tab4 VALUES (1), (1), (2);
-- This trick creates an invalid index.
CREATE UNIQUE INDEX CONCURRENTLY concur_reindex_ind5 ON concur_reindex_tab4 (c1);
+DROP INDEX concur_reindex_ind5_ccaux;
-- Reindexing concurrently this index fails with the same failure.
-- The extra index created is itself invalid, and can be dropped.
REINDEX INDEX CONCURRENTLY concur_reindex_ind5;
\d concur_reindex_tab4
+DROP INDEX concur_reindex_ind5_ccaux;
DROP INDEX concur_reindex_ind5_ccnew;
-- This makes the previous failure go away, so the index can become valid.
DELETE FROM concur_reindex_tab4 WHERE c1 = 1;
Index: src/backend/access/transam/twophase.c
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
--- a/src/backend/access/transam/twophase.c (revision bbc09a323cc3d6c54f2d26c7c6342d36d7edeb31)
+++ b/src/backend/access/transam/twophase.c (revision 03c4ff69cbbfa3182e697672d7ea704db293213f)
@@ -459,7 +459,7 @@
proc->vxid.procNumber = INVALID_PROC_NUMBER;
}
proc->xid = xid;
- Assert(proc->xmin == InvalidTransactionId);
+ Assert(proc->xmin == InvalidTransactionId && proc->catalogXmin == InvalidTransactionId);
proc->delayChkptFlags = 0;
proc->statusFlags = 0;
proc->pid = 0;
Index: src/backend/replication/logical/reorderbuffer.c
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
--- a/src/backend/replication/logical/reorderbuffer.c (revision bbc09a323cc3d6c54f2d26c7c6342d36d7edeb31)
+++ b/src/backend/replication/logical/reorderbuffer.c (revision 03c4ff69cbbfa3182e697672d7ea704db293213f)
@@ -1844,6 +1844,7 @@
snap->active_count = 1; /* mark as active so nobody frees it */
snap->regd_count = 0;
snap->xip = (TransactionId *) (snap + 1);
+ snap->catalog = orig_snap->catalog;
memcpy(snap->xip, orig_snap->xip, sizeof(TransactionId) * snap->xcnt);
Index: src/backend/replication/logical/snapbuild.c
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/replication/logical/snapbuild.c b/src/backend/replication/logical/snapbuild.c
--- a/src/backend/replication/logical/snapbuild.c (revision bbc09a323cc3d6c54f2d26c7c6342d36d7edeb31)
+++ b/src/backend/replication/logical/snapbuild.c (revision 03c4ff69cbbfa3182e697672d7ea704db293213f)
@@ -564,6 +564,7 @@
snapshot->active_count = 0;
snapshot->regd_count = 0;
snapshot->snapXactCompletionCount = 0;
+ snapshot->catalog = false; // TODO: or true?
return snapshot;
}
@@ -600,8 +601,8 @@
elog(ERROR, "cannot build an initial slot snapshot, not all transactions are monitored anymore");
/* so we don't overwrite the existing value */
- if (TransactionIdIsValid(MyProc->xmin))
- elog(ERROR, "cannot build an initial slot snapshot when MyProc->xmin already is valid");
+ if (TransactionIdIsValid(MyProc->xmin) || TransactionIdIsValid(MyProc->catalogXmin))
+ elog(ERROR, "cannot build an initial slot snapshot when MyProc->xmin or MyProc->catalogXmin already is valid");
snap = SnapBuildBuildSnapshot(builder);
@@ -622,7 +623,7 @@
elog(ERROR, "cannot build an initial slot snapshot as oldest safe xid %u follows snapshot's xmin %u",
safeXid, snap->xmin);
- MyProc->xmin = snap->xmin;
+ MyProc->xmin = MyProc->catalogXmin = snap->xmin;
/* allocate in transaction context */
newxip = (TransactionId *)
Index: src/backend/replication/walsender.c
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
--- a/src/backend/replication/walsender.c (revision bbc09a323cc3d6c54f2d26c7c6342d36d7edeb31)
+++ b/src/backend/replication/walsender.c (revision 03c4ff69cbbfa3182e697672d7ea704db293213f)
@@ -305,7 +305,7 @@
*/
if (MyDatabaseId == InvalidOid)
{
- Assert(MyProc->xmin == InvalidTransactionId);
+ Assert(MyProc->xmin == InvalidTransactionId && MyProc->catalogXmin == InvalidTransactionId);
LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
MyProc->statusFlags |= PROC_AFFECTS_ALL_HORIZONS;
ProcGlobal->statusFlags[MyProc->pgxactoff] = MyProc->statusFlags;
@@ -2498,7 +2498,7 @@
ReplicationSlot *slot = MyReplicationSlot;
SpinLockAcquire(&slot->mutex);
- MyProc->xmin = InvalidTransactionId;
+ MyProc->xmin = MyProc->catalogXmin = InvalidTransactionId;
/*
* For physical replication we don't need the interlock provided by xmin
@@ -2627,7 +2627,7 @@
if (!TransactionIdIsNormal(feedbackXmin)
&& !TransactionIdIsNormal(feedbackCatalogXmin))
{
- MyProc->xmin = InvalidTransactionId;
+ MyProc->xmin = MyProc->catalogXmin = InvalidTransactionId;
if (MyReplicationSlot != NULL)
PhysicalReplicationSlotNewXmin(feedbackXmin, feedbackCatalogXmin);
return;
@@ -2680,11 +2680,8 @@
PhysicalReplicationSlotNewXmin(feedbackXmin, feedbackCatalogXmin);
else
{
- if (TransactionIdIsNormal(feedbackCatalogXmin)
- && TransactionIdPrecedes(feedbackCatalogXmin, feedbackXmin))
- MyProc->xmin = feedbackCatalogXmin;
- else
- MyProc->xmin = feedbackXmin;
+ MyProc->catalogXmin = feedbackCatalogXmin;
+ MyProc->xmin = feedbackXmin;
}
}
Index: src/backend/storage/ipc/procarray.c
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c
--- a/src/backend/storage/ipc/procarray.c (revision bbc09a323cc3d6c54f2d26c7c6342d36d7edeb31)
+++ b/src/backend/storage/ipc/procarray.c (revision 6c55d9749e2999542d4e6281db733fdd47930796)
@@ -701,7 +701,7 @@
Assert(!proc->subxidStatus.overflowed);
proc->vxid.lxid = InvalidLocalTransactionId;
- proc->xmin = InvalidTransactionId;
+ proc->xmin = proc->catalogXmin = InvalidTransactionId;
/* be sure this is cleared in abort */
proc->delayChkptFlags = 0;
@@ -743,7 +743,7 @@
ProcGlobal->xids[pgxactoff] = InvalidTransactionId;
proc->xid = InvalidTransactionId;
proc->vxid.lxid = InvalidLocalTransactionId;
- proc->xmin = InvalidTransactionId;
+ proc->xmin = proc->catalogXmin = InvalidTransactionId;
/* be sure this is cleared in abort */
proc->delayChkptFlags = 0;
@@ -930,7 +930,7 @@
proc->xid = InvalidTransactionId;
proc->vxid.lxid = InvalidLocalTransactionId;
- proc->xmin = InvalidTransactionId;
+ proc->xmin = proc->catalogXmin = InvalidTransactionId;
proc->recoveryConflictPending = false;
Assert(!(proc->statusFlags & PROC_VACUUM_STATE_MASK));
@@ -1739,8 +1739,6 @@
bool in_recovery = RecoveryInProgress();
TransactionId *other_xids = ProcGlobal->xids;
- /* inferred after ProcArrayLock is released */
- h->catalog_oldest_nonremovable = InvalidTransactionId;
LWLockAcquire(ProcArrayLock, LW_SHARED);
@@ -1761,6 +1759,7 @@
h->oldest_considered_running = initial;
h->shared_oldest_nonremovable = initial;
+ h->catalog_oldest_nonremovable = initial;
h->data_oldest_nonremovable = initial;
/*
@@ -1796,10 +1795,13 @@
int8 statusFlags = ProcGlobal->statusFlags[index];
TransactionId xid;
TransactionId xmin;
+ TransactionId catalogXmin;
+ TransactionId olderXmin;
/* Fetch xid just once - see GetNewTransactionId */
xid = UINT32_ACCESS_ONCE(other_xids[index]);
xmin = UINT32_ACCESS_ONCE(proc->xmin);
+ catalogXmin = UINT32_ACCESS_ONCE(proc->catalogXmin);
/*
* Consider both the transaction's Xmin, and its Xid.
@@ -1809,11 +1811,14 @@
* some not-yet-set Xmin.
*/
xmin = TransactionIdOlder(xmin, xid);
+ catalogXmin = TransactionIdOlder(catalogXmin, xid);
/* if neither is set, this proc doesn't influence the horizon */
- if (!TransactionIdIsValid(xmin))
+ if (!TransactionIdIsValid(xmin) && !TransactionIdIsValid(catalogXmin))
continue;
+ olderXmin = TransactionIdOlder(xmin, catalogXmin);
+
/*
* Don't ignore any procs when determining which transactions might be
* considered running. While slots should ensure logical decoding
@@ -1821,7 +1826,7 @@
* include them here as well..
*/
h->oldest_considered_running =
- TransactionIdOlder(h->oldest_considered_running, xmin);
+ TransactionIdOlder(h->oldest_considered_running, olderXmin);
/*
* Skip over backends either vacuuming (which is ok with rows being
@@ -1833,7 +1838,7 @@
/* shared tables need to take backends in all databases into account */
h->shared_oldest_nonremovable =
- TransactionIdOlder(h->shared_oldest_nonremovable, xmin);
+ TransactionIdOlder(h->shared_oldest_nonremovable, olderXmin);
/*
* Normally sessions in other databases are ignored for anything but
@@ -1859,8 +1864,12 @@
(statusFlags & PROC_AFFECTS_ALL_HORIZONS) ||
in_recovery)
{
- h->data_oldest_nonremovable =
- TransactionIdOlder(h->data_oldest_nonremovable, xmin);
+ if (TransactionIdIsValid(xmin))
+ h->data_oldest_nonremovable =
+ TransactionIdOlder(h->data_oldest_nonremovable, xmin);
+ if (TransactionIdIsValid(olderXmin))
+ h->catalog_oldest_nonremovable =
+ TransactionIdOlder(h->catalog_oldest_nonremovable, olderXmin);
}
}
@@ -1885,6 +1894,8 @@
TransactionIdOlder(h->shared_oldest_nonremovable, kaxmin);
h->data_oldest_nonremovable =
TransactionIdOlder(h->data_oldest_nonremovable, kaxmin);
+ h->catalog_oldest_nonremovable =
+ TransactionIdOlder(h->catalog_oldest_nonremovable, kaxmin);
/* temp relations cannot be accessed in recovery */
}
@@ -1912,7 +1923,6 @@
h->shared_oldest_nonremovable =
TransactionIdOlder(h->shared_oldest_nonremovable,
h->slot_catalog_xmin);
- h->catalog_oldest_nonremovable = h->data_oldest_nonremovable;
h->catalog_oldest_nonremovable =
TransactionIdOlder(h->catalog_oldest_nonremovable,
h->slot_catalog_xmin);
@@ -2092,7 +2102,7 @@
* least in the case we already hold a snapshot), but that's for another day.
*/
static bool
-GetSnapshotDataReuse(Snapshot snapshot)
+GetSnapshotDataReuse(Snapshot snapshot, bool catalog)
{
uint64 curXactCompletionCount;
@@ -2101,6 +2111,9 @@
if (unlikely(snapshot->snapXactCompletionCount == 0))
return false;
+ if (unlikely(snapshot->catalog != catalog))
+ return false;
+
curXactCompletionCount = TransamVariables->xactCompletionCount;
if (curXactCompletionCount != snapshot->snapXactCompletionCount)
return false;
@@ -2125,8 +2138,19 @@
* requirement that concurrent GetSnapshotData() calls yield the same
* xmin.
*/
- if (!TransactionIdIsValid(MyProc->xmin))
- MyProc->xmin = TransactionXmin = snapshot->xmin;
+ if (!catalog)
+ {
+ if (!TransactionIdIsValid(MyProc->xmin))
+ MyProc->xmin = snapshot->xmin;
+ }
+ else
+ {
+ if (!TransactionIdIsValid(MyProc->catalogXmin))
+ MyProc->catalogXmin = snapshot->xmin;
+ }
+
+ if (!TransactionIdIsValid(TransactionXmin))
+ TransactionXmin = snapshot->xmin;
RecentXmin = snapshot->xmin;
Assert(TransactionIdPrecedesOrEquals(TransactionXmin, RecentXmin));
@@ -2173,8 +2197,8 @@
* Note: this function should probably not be called with an argument that's
* not statically allocated (see xip allocation below).
*/
-Snapshot
-GetSnapshotData(Snapshot snapshot)
+static Snapshot
+GetSnapshotDataImpl(Snapshot snapshot, bool catalog)
{
ProcArrayStruct *arrayP = procArray;
TransactionId *other_xids = ProcGlobal->xids;
@@ -2232,7 +2256,7 @@
*/
LWLockAcquire(ProcArrayLock, LW_SHARED);
- if (GetSnapshotDataReuse(snapshot))
+ if (GetSnapshotDataReuse(snapshot, catalog))
{
LWLockRelease(ProcArrayLock);
return snapshot;
@@ -2412,8 +2436,18 @@
replication_slot_xmin = procArray->replication_slot_xmin;
replication_slot_catalog_xmin = procArray->replication_slot_catalog_xmin;
- if (!TransactionIdIsValid(MyProc->xmin))
- MyProc->xmin = TransactionXmin = xmin;
+ if (!catalog)
+ {
+ if (!TransactionIdIsValid(MyProc->xmin))
+ MyProc->xmin = xmin;
+ }
+ else
+ {
+ if (!TransactionIdIsValid(MyProc->catalogXmin))
+ MyProc->catalogXmin = xmin;
+ }
+ if (!TransactionIdIsValid(TransactionXmin))
+ TransactionXmin = xmin;
LWLockRelease(ProcArrayLock);
@@ -2506,6 +2540,7 @@
snapshot->subxcnt = subcount;
snapshot->suboverflowed = suboverflowed;
snapshot->snapXactCompletionCount = curXactCompletionCount;
+ snapshot->catalog = catalog;
snapshot->curcid = GetCurrentCommandId(false);
@@ -2522,6 +2557,19 @@
return snapshot;
}
+Snapshot
+GetSnapshotData(Snapshot snapshot)
+{
+ return GetSnapshotDataImpl(snapshot, false);
+}
+
+
+Snapshot
+GetCatalogSnapshotData(Snapshot snapshot)
+{
+ return GetSnapshotDataImpl(snapshot, true);
+}
+
/*
* ProcArrayInstallImportedXmin -- install imported xmin into MyProc->xmin
*
@@ -2592,7 +2640,7 @@
* GetSnapshotData first, we'll be overwriting a valid xmin here, so
* we don't check that.)
*/
- MyProc->xmin = TransactionXmin = xmin;
+ MyProc->xmin = MyProc->catalogXmin = TransactionXmin = xmin;
result = true;
break;
@@ -2645,7 +2693,7 @@
* Install xmin and propagate the statusFlags that affect how the
* value is interpreted by vacuum.
*/
- MyProc->xmin = TransactionXmin = xmin;
+ MyProc->xmin = MyProc->catalogXmin = TransactionXmin = xmin;
MyProc->statusFlags = (MyProc->statusFlags & ~PROC_XMIN_FLAGS) |
(proc->statusFlags & PROC_XMIN_FLAGS);
ProcGlobal->statusFlags[MyProc->pgxactoff] = MyProc->statusFlags;
@@ -3162,7 +3210,8 @@
*/
void
ProcNumberGetTransactionIds(ProcNumber procNumber, TransactionId *xid,
- TransactionId *xmin, int *nsubxid, bool *overflowed)
+ TransactionId *xmin, TransactionId *catalogXmin,
+ int *nsubxid, bool *overflowed)
{
PGPROC *proc;
@@ -3182,6 +3231,7 @@
{
*xid = proc->xid;
*xmin = proc->xmin;
+ *catalogXmin = proc->catalogXmin;
*nsubxid = proc->subxidStatus.count;
*overflowed = proc->subxidStatus.overflowed;
}
@@ -3356,8 +3406,10 @@
{
/* Fetch xmin just once - might change on us */
TransactionId pxmin = UINT32_ACCESS_ONCE(proc->xmin);
+ TransactionId pcatalogXmin = UINT32_ACCESS_ONCE(proc->catalogXmin);
+ TransactionId olderpXmin = TransactionIdOlder(pxmin, pcatalogXmin);
- if (excludeXmin0 && !TransactionIdIsValid(pxmin))
+ if (excludeXmin0 && !TransactionIdIsValid(olderpXmin))
continue;
/*
@@ -3365,7 +3417,7 @@
* hasn't set xmin yet will not be rejected by this test.
*/
if (!TransactionIdIsValid(limitXmin) ||
- TransactionIdPrecedesOrEquals(pxmin, limitXmin))
+ TransactionIdPrecedesOrEquals(olderpXmin, limitXmin))
{
VirtualTransactionId vxid;
@@ -3456,6 +3508,8 @@
{
/* Fetch xmin just once - can't change on us, but good coding */
TransactionId pxmin = UINT32_ACCESS_ONCE(proc->xmin);
+ TransactionId catalogpXmin = UINT32_ACCESS_ONCE(proc->catalogXmin);
+ TransactionId oldestpXmin = TransactionIdOlder(pxmin, catalogpXmin);
/*
* We ignore an invalid pxmin because this means that backend has
@@ -3466,7 +3520,7 @@
* test here.
*/
if (!TransactionIdIsValid(limitXmin) ||
- (TransactionIdIsValid(pxmin) && !TransactionIdFollows(pxmin, limitXmin)))
+ (TransactionIdIsValid(oldestpXmin) && !TransactionIdFollows(oldestpXmin, limitXmin)))
{
VirtualTransactionId vxid;
Index: src/backend/storage/lmgr/proc.c
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.c
--- a/src/backend/storage/lmgr/proc.c (revision bbc09a323cc3d6c54f2d26c7c6342d36d7edeb31)
+++ b/src/backend/storage/lmgr/proc.c (revision 03c4ff69cbbfa3182e697672d7ea704db293213f)
@@ -382,7 +382,7 @@
MyProc->fpVXIDLock = false;
MyProc->fpLocalTransactionId = InvalidLocalTransactionId;
MyProc->xid = InvalidTransactionId;
- MyProc->xmin = InvalidTransactionId;
+ MyProc->xmin = MyProc->catalogXmin = InvalidTransactionId;
MyProc->pid = MyProcPid;
MyProc->vxid.procNumber = MyProcNumber;
MyProc->vxid.lxid = InvalidLocalTransactionId;
@@ -580,7 +580,7 @@
MyProc->fpVXIDLock = false;
MyProc->fpLocalTransactionId = InvalidLocalTransactionId;
MyProc->xid = InvalidTransactionId;
- MyProc->xmin = InvalidTransactionId;
+ MyProc->xmin = MyProc->catalogXmin = InvalidTransactionId;
MyProc->vxid.procNumber = INVALID_PROC_NUMBER;
MyProc->vxid.lxid = InvalidLocalTransactionId;
MyProc->databaseId = InvalidOid;
Index: src/backend/utils/time/snapmgr.c
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/utils/time/snapmgr.c b/src/backend/utils/time/snapmgr.c
--- a/src/backend/utils/time/snapmgr.c (revision bbc09a323cc3d6c54f2d26c7c6342d36d7edeb31)
+++ b/src/backend/utils/time/snapmgr.c (revision 03c4ff69cbbfa3182e697672d7ea704db293213f)
@@ -290,14 +290,6 @@
Snapshot
GetLatestSnapshot(void)
{
- /*
- * We might be able to relax this, but nothing that could otherwise work
- * needs it.
- */
- if (IsInParallelMode())
- elog(ERROR,
- "cannot update SecondarySnapshot during a parallel operation");
-
/*
* So far there are no cases requiring support for GetLatestSnapshot()
* during logical decoding, but it wouldn't be hard to add if required.
@@ -332,6 +324,16 @@
RegisteredLSN = OldestRegisteredSnapshot->lsn;
}
+ if (CatalogSnapshot != NULL)
+ {
+ if (OldestRegisteredSnapshot == NULL ||
+ TransactionIdPrecedes(CatalogSnapshot->xmin, OldestRegisteredSnapshot->xmin))
+ {
+ OldestRegisteredSnapshot = CatalogSnapshot;
+ RegisteredLSN = CatalogSnapshot->lsn;
+ }
+ }
+
if (OldestActiveSnapshot != NULL)
{
XLogRecPtr ActiveLSN = OldestActiveSnapshot->as_snap->lsn;
@@ -388,7 +390,7 @@
if (CatalogSnapshot == NULL)
{
/* Get new snapshot. */
- CatalogSnapshot = GetSnapshotData(&CatalogSnapshotData);
+ CatalogSnapshot = GetCatalogSnapshotData(&CatalogSnapshotData);
/*
* Make sure the catalog snapshot will be accounted for in decisions
@@ -402,7 +404,7 @@
* NB: it had better be impossible for this to throw error, since the
* CatalogSnapshot pointer is already valid.
*/
- pairingheap_add(&RegisteredSnapshots, &CatalogSnapshot->ph_node);
+ Assert(TransactionIdIsValid(MyProc->catalogXmin));
}
return CatalogSnapshot;
@@ -423,9 +425,8 @@
{
if (CatalogSnapshot)
{
- pairingheap_remove(&RegisteredSnapshots, &CatalogSnapshot->ph_node);
CatalogSnapshot = NULL;
- SnapshotResetXmin();
+ MyProc->catalogXmin = InvalidTransactionId;
}
}
@@ -444,7 +445,7 @@
{
if (CatalogSnapshot &&
ActiveSnapshot == NULL &&
- pairingheap_is_singular(&RegisteredSnapshots))
+ pairingheap_is_empty(&RegisteredSnapshots))
InvalidateCatalogSnapshot();
}
@@ -1081,7 +1082,7 @@
if (resetXmin)
SnapshotResetXmin();
- Assert(resetXmin || MyProc->xmin == 0);
+ Assert(resetXmin || (MyProc->xmin == InvalidTransactionId && MyProc->catalogXmin == InvalidTransactionId));
}
@@ -1626,19 +1627,15 @@
if (ActiveSnapshot != NULL)
return true;
- /*
- * The catalog snapshot is in RegisteredSnapshots when valid, but can be
- * removed at any time due to invalidation processing. If explicitly
- * registered more than one snapshot has to be in RegisteredSnapshots.
- */
- if (CatalogSnapshot != NULL &&
- pairingheap_is_singular(&RegisteredSnapshots))
- return false;
+ return HaveRegisteredSnapshot();
+}
+bool
+HaveRegisteredSnapshot(void)
+{
return !pairingheap_is_empty(&RegisteredSnapshots);
}
-
/*
* Setup a snapshot that replaces normal catalog snapshots that allows catalog
* access to behave just like it did at a certain point in the past.
@@ -1804,6 +1801,7 @@
snapshot->whenTaken = serialized_snapshot.whenTaken;
snapshot->lsn = serialized_snapshot.lsn;
snapshot->snapXactCompletionCount = 0;
+ snapshot->catalog = false;
/* Copy XIDs, if present. */
if (serialized_snapshot.xcnt > 0)
Index: src/include/storage/proc.h
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h
--- a/src/include/storage/proc.h (revision bbc09a323cc3d6c54f2d26c7c6342d36d7edeb31)
+++ b/src/include/storage/proc.h (revision 3dea72b62adc8806917dc459b82ff44d962bcb12)
@@ -56,10 +56,6 @@
*/
#define PROC_IS_AUTOVACUUM 0x01 /* is it an autovac worker? */
#define PROC_IN_VACUUM 0x02 /* currently running lazy vacuum */
-#define PROC_IN_SAFE_IC 0x04 /* currently running CREATE INDEX
- * CONCURRENTLY or REINDEX
- * CONCURRENTLY on non-expressional,
- * non-partial index */
#define PROC_VACUUM_FOR_WRAPAROUND 0x08 /* set by autovac only */
#define PROC_IN_LOGICAL_DECODING 0x10 /* currently doing logical
* decoding outside xact */
@@ -69,13 +65,13 @@
/* flags reset at EOXact */
#define PROC_VACUUM_STATE_MASK \
- (PROC_IN_VACUUM | PROC_IN_SAFE_IC | PROC_VACUUM_FOR_WRAPAROUND)
+ (PROC_IN_VACUUM | PROC_VACUUM_FOR_WRAPAROUND)
/*
* Xmin-related flags. Make sure any flags that affect how the process' Xmin
* value is interpreted by VACUUM are included here.
*/
-#define PROC_XMIN_FLAGS (PROC_IN_VACUUM | PROC_IN_SAFE_IC)
+#define PROC_XMIN_FLAGS (PROC_IN_VACUUM)
/*
* We allow a small number of "weak" relation locks (AccessShareLock,
@@ -179,6 +175,7 @@
* starting our xact, excluding LAZY VACUUM:
* vacuum must not remove tuples deleted by
* xid >= xmin ! */
+ TransactionId catalogXmin;
int pid; /* Backend's process ID; 0 if prepared xact */
Index: src/include/storage/procarray.h
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/include/storage/procarray.h b/src/include/storage/procarray.h
--- a/src/include/storage/procarray.h (revision bbc09a323cc3d6c54f2d26c7c6342d36d7edeb31)
+++ b/src/include/storage/procarray.h (revision 6c55d9749e2999542d4e6281db733fdd47930796)
@@ -45,6 +45,7 @@
extern int GetMaxSnapshotSubxidCount(void);
extern Snapshot GetSnapshotData(Snapshot snapshot);
+extern Snapshot GetCatalogSnapshotData(Snapshot snapshot);
extern bool ProcArrayInstallImportedXmin(TransactionId xmin,
VirtualTransactionId *sourcevxid);
@@ -66,8 +67,8 @@
extern PGPROC *ProcNumberGetProc(int procNumber);
extern void ProcNumberGetTransactionIds(int procNumber, TransactionId *xid,
- TransactionId *xmin, int *nsubxid,
- bool *overflowed);
+ TransactionId *xmin, TransactionId *catalogXmin,
+ int *nsubxid, bool *overflowed);
extern PGPROC *BackendPidGetProc(int pid);
extern PGPROC *BackendPidGetProcWithLock(int pid);
extern int BackendXidGetPid(TransactionId xid);
Index: src/include/utils/snapshot.h
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/include/utils/snapshot.h b/src/include/utils/snapshot.h
--- a/src/include/utils/snapshot.h (revision bbc09a323cc3d6c54f2d26c7c6342d36d7edeb31)
+++ b/src/include/utils/snapshot.h (revision 03c4ff69cbbfa3182e697672d7ea704db293213f)
@@ -183,6 +183,7 @@
bool takenDuringRecovery; /* recovery-shaped snapshot? */
bool copied; /* false if it's a static snapshot */
+ bool catalog; /* snapshot used to access catalog */
CommandId curcid; /* in my xact, CID < curcid are visible */
Index: contrib/amcheck/verify_nbtree.c
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/contrib/amcheck/verify_nbtree.c b/contrib/amcheck/verify_nbtree.c
--- a/contrib/amcheck/verify_nbtree.c (revision 91dd70fc5ddc60cbad5b17c95f17c6a517f36770)
+++ b/contrib/amcheck/verify_nbtree.c (revision 3dea72b62adc8806917dc459b82ff44d962bcb12)
@@ -691,7 +691,8 @@
0, /* number of keys */
NULL, /* scan key */
true, /* buffer access strategy OK */
- true); /* syncscan OK? */
+ true, /* syncscan OK? */
+ false);
/*
* Scan will behave as the first scan of a CREATE INDEX CONCURRENTLY
Index: src/backend/access/brin/brin.c
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/access/brin/brin.c b/src/backend/access/brin/brin.c
--- a/src/backend/access/brin/brin.c (revision 91dd70fc5ddc60cbad5b17c95f17c6a517f36770)
+++ b/src/backend/access/brin/brin.c (revision 3dea72b62adc8806917dc459b82ff44d962bcb12)
@@ -2369,16 +2369,7 @@
leaderparticipates = false;
#endif
- /*
- * Enter parallel mode, and create context for parallel build of brin
- * index
- */
- EnterParallelMode();
- Assert(request > 0);
- pcxt = CreateParallelContext("postgres", "_brin_parallel_build_main",
- request);
-
- scantuplesortstates = leaderparticipates ? request + 1 : request;
+ Assert(!isconcurrent || !TransactionIdIsValid(MyProc->xmin));
/*
* Prepare for scan of the base relation. In a normal index build, we use
@@ -2390,7 +2381,21 @@
if (!isconcurrent)
snapshot = SnapshotAny;
else
+ {
snapshot = RegisterSnapshot(GetTransactionSnapshot());
+ PushActiveSnapshot(snapshot);
+ }
+
+ /*
+ * Enter parallel mode, and create context for parallel build of brin
+ * index
+ */
+ EnterParallelMode();
+ Assert(request > 0);
+ pcxt = CreateParallelContext("postgres", "_brin_parallel_build_main",
+ request);
+
+ scantuplesortstates = leaderparticipates ? request + 1 : request;
/*
* Estimate size for our own PARALLEL_KEY_BRIN_SHARED workspace.
@@ -2429,6 +2434,8 @@
/* Everyone's had a chance to ask for space, so now create the DSM */
InitializeParallelDSM(pcxt);
+ if (IsMVCCSnapshot(snapshot))
+ PopActiveSnapshot();
/* If no DSM segment was available, back out (do serial build) */
if (pcxt->seg == NULL)
@@ -2458,7 +2465,7 @@
table_parallelscan_initialize(heap,
ParallelTableScanFromBrinShared(brinshared),
- snapshot);
+ isconcurrent ? InvalidSnapshot : snapshot);
/*
* Store shared tuplesort-private state, for which we reserved space.
@@ -2504,7 +2511,7 @@
brinleader->nparticipanttuplesorts++;
brinleader->brinshared = brinshared;
brinleader->sharedsort = sharedsort;
- brinleader->snapshot = snapshot;
+ brinleader->snapshot = isconcurrent ? InvalidSnapshot : snapshot;
brinleader->walusage = walusage;
brinleader->bufferusage = bufferusage;
@@ -2518,6 +2525,12 @@
/* Save leader state now that it's clear build will be parallel */
buildstate->bs_leader = brinleader;
+ if (isconcurrent)
+ {
+ WaitForParallelWorkersToAttach(pcxt, true);
+ UnregisterSnapshot(snapshot);
+ }
+
/* Join heap scan ourselves */
if (leaderparticipates)
_brin_leader_participate_as_worker(buildstate, heap, index);
@@ -2526,7 +2539,8 @@
* Caller needs to wait for all launched workers when we return. Make
* sure that the failure-to-start case will not hang forever.
*/
- WaitForParallelWorkersToAttach(pcxt);
+ if (!isconcurrent)
+ WaitForParallelWorkersToAttach(pcxt, false);
}
/*
@@ -2536,6 +2550,7 @@
_brin_end_parallel(BrinLeader *brinleader, BrinBuildState *state)
{
int i;
+ Snapshot snapshot = brinleader->snapshot;
/* Shutdown worker processes */
WaitForParallelWorkersToFinish(brinleader->pcxt);
@@ -2548,8 +2563,10 @@
InstrAccumParallelQuery(&brinleader->bufferusage[i], &brinleader->walusage[i]);
/* Free last reference to MVCC snapshot, if one was used */
- if (IsMVCCSnapshot(brinleader->snapshot))
- UnregisterSnapshot(brinleader->snapshot);
+ Assert(!brinleader->brinshared->isconcurrent || snapshot == InvalidSnapshot);
+ Assert(brinleader->brinshared->isconcurrent || snapshot != InvalidSnapshot);
+ if (snapshot != InvalidSnapshot && IsMVCCSnapshot(snapshot))
+ UnregisterSnapshot(snapshot);
DestroyParallelContext(brinleader->pcxt);
ExitParallelMode();
}
@@ -2800,6 +2817,7 @@
TableScanDesc scan;
double reltuples;
IndexInfo *indexInfo;
+ Snapshot snapshot;
/* Initialize local tuplesort coordination state */
coordinate = palloc0(sizeof(SortCoordinateData));
@@ -2811,8 +2829,21 @@
state->bs_sortstate = tuplesort_begin_index_brin(sortmem, coordinate,
TUPLESORT_NONE);
+ Assert(!brinshared->isconcurrent || !TransactionIdIsValid(MyProc->xmin));
+ /* Join parallel scan */
+ if (brinshared->isconcurrent)
+ {
+ snapshot = RegisterSnapshot(GetTransactionSnapshot());
+ PushActiveSnapshot(snapshot);
+ }
/* Join parallel scan */
indexInfo = BuildIndexInfo(index);
+ if (brinshared->isconcurrent)
+ {
+ PopActiveSnapshot();
+ UnregisterSnapshot(snapshot);
+ }
+ Assert(!brinshared->isconcurrent || !TransactionIdIsValid(MyProc->xmin));
indexInfo->ii_Concurrent = brinshared->isconcurrent;
scan = table_beginscan_parallel(heap,
@@ -2866,8 +2897,7 @@
* The only possible status flag that can be set to the parallel worker is
* PROC_IN_SAFE_IC.
*/
- Assert((MyProc->statusFlags == 0) ||
- (MyProc->statusFlags == PROC_IN_SAFE_IC));
+ Assert(MyProc->statusFlags == 0);
/* Set debug_query_string for individual workers first */
sharedquery = shm_toc_lookup(toc, PARALLEL_KEY_QUERY_TEXT, true);
@@ -2913,8 +2943,12 @@
*/
sortmem = maintenance_work_mem / brinshared->scantuplesortstates;
+ if (brinshared->isconcurrent)
+ PopActiveSnapshot();
_brin_parallel_scan_and_build(buildstate, brinshared, sharedsort,
heapRel, indexRel, sortmem, false);
+ if (brinshared->isconcurrent)
+ PushActiveSnapshot(GetLatestSnapshot());
/* Report WAL/buffer usage during parallel execution */
bufferusage = shm_toc_lookup(toc, PARALLEL_KEY_BUFFER_USAGE, false);
Index: src/backend/access/gin/gininsert.c
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/access/gin/gininsert.c b/src/backend/access/gin/gininsert.c
--- a/src/backend/access/gin/gininsert.c (revision 91dd70fc5ddc60cbad5b17c95f17c6a517f36770)
+++ b/src/backend/access/gin/gininsert.c (revision 3dea72b62adc8806917dc459b82ff44d962bcb12)
@@ -17,6 +17,7 @@
#include "access/gin_private.h"
#include "access/tableam.h"
#include "access/xloginsert.h"
+#include "catalog/index.h"
#include "miscadmin.h"
#include "nodes/execnodes.h"
#include "storage/bufmgr.h"
Index: src/backend/access/gist/gistbuild.c
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/access/gist/gistbuild.c b/src/backend/access/gist/gistbuild.c
--- a/src/backend/access/gist/gistbuild.c (revision 91dd70fc5ddc60cbad5b17c95f17c6a517f36770)
+++ b/src/backend/access/gist/gistbuild.c (revision 3dea72b62adc8806917dc459b82ff44d962bcb12)
@@ -38,6 +38,7 @@
#include "access/gist_private.h"
#include "access/tableam.h"
#include "access/xloginsert.h"
+#include "catalog/index.h"
#include "miscadmin.h"
#include "nodes/execnodes.h"
#include "optimizer/optimizer.h"
Index: src/backend/access/hash/hash.c
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/access/hash/hash.c b/src/backend/access/hash/hash.c
--- a/src/backend/access/hash/hash.c (revision 91dd70fc5ddc60cbad5b17c95f17c6a517f36770)
+++ b/src/backend/access/hash/hash.c (revision 3dea72b62adc8806917dc459b82ff44d962bcb12)
@@ -23,6 +23,7 @@
#include "access/relscan.h"
#include "access/tableam.h"
#include "access/xloginsert.h"
+#include "catalog/index.h"
#include "commands/progress.h"
#include "commands/vacuum.h"
#include "miscadmin.h"
Index: src/backend/access/heap/heapam.c
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
--- a/src/backend/access/heap/heapam.c (revision 91dd70fc5ddc60cbad5b17c95f17c6a517f36770)
+++ b/src/backend/access/heap/heapam.c (revision 3dea72b62adc8806917dc459b82ff44d962bcb12)
@@ -575,6 +575,24 @@
LockBuffer(buffer, BUFFER_LOCK_UNLOCK);
}
+static inline void
+heap_reset_scan_snapshot(TableScanDesc sscan)
+{
+ Assert(ActiveSnapshotSet());
+ PopActiveSnapshot();
+ UnregisterSnapshot(sscan->rs_snapshot);
+ sscan->rs_snapshot = InvalidSnapshot;
+
+ Assert(!TransactionIdIsValid(MyProc->xmin));
+#if USE_INJECTION_POINTS
+ if (!TransactionIdIsValid(MyProc->xid))
+ INJECTION_POINT("heap_reset_scan_snapshot_effective");
+#endif
+
+ sscan->rs_snapshot = RegisterSnapshot(GetLatestSnapshot());
+ PushActiveSnapshot(sscan->rs_snapshot);
+}
+
/*
* heap_fetch_next_buffer - read and pin the next block from MAIN_FORKNUM.
*
@@ -593,6 +611,11 @@
scan->rs_cbuf = InvalidBuffer;
}
+ if (unlikely(scan->rs_base.rs_flags & SO_RESET_SNAPSHOT) & likely(scan->rs_inited))
+ {
+ heap_reset_scan_snapshot((TableScanDesc) scan);
+ }
+
/*
* Be sure to check for interrupts at least once per page. Checks at
* higher code levels won't be able to stop a seqscan that encounters many
@@ -1242,6 +1265,13 @@
if (scan->rs_parallelworkerdata != NULL)
pfree(scan->rs_parallelworkerdata);
+ if (scan->rs_base.rs_flags & SO_RESET_SNAPSHOT)
+ {
+ Assert(scan->rs_base.rs_flags & SO_TEMP_SNAPSHOT);
+ Assert(ActiveSnapshotSet());
+ PopActiveSnapshot();
+ }
+
if (scan->rs_base.rs_flags & SO_TEMP_SNAPSHOT)
UnregisterSnapshot(scan->rs_base.rs_snapshot);
Index: src/backend/access/nbtree/nbtsort.c
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/access/nbtree/nbtsort.c b/src/backend/access/nbtree/nbtsort.c
--- a/src/backend/access/nbtree/nbtsort.c (revision 91dd70fc5ddc60cbad5b17c95f17c6a517f36770)
+++ b/src/backend/access/nbtree/nbtsort.c (revision 3dea72b62adc8806917dc459b82ff44d962bcb12)
@@ -84,6 +84,7 @@
Relation index;
bool isunique;
bool nulls_not_distinct;
+ bool unique_dead_ignored;
} BTSpool;
/*
@@ -377,6 +378,7 @@
btspool->index = index;
btspool->isunique = indexInfo->ii_Unique;
btspool->nulls_not_distinct = indexInfo->ii_NullsNotDistinct;
+ btspool->unique_dead_ignored = indexInfo->ii_Concurrent;
/* Save as primary spool */
buildstate->spool = btspool;
@@ -425,8 +427,9 @@
* the use of parallelism or any other factor.
*/
buildstate->spool->sortstate =
- tuplesort_begin_index_btree(heap, index, buildstate->isunique,
- buildstate->nulls_not_distinct,
+ tuplesort_begin_index_btree(heap, index, btspool->isunique,
+ btspool->nulls_not_distinct,
+ btspool->unique_dead_ignored,
maintenance_work_mem, coordinate,
TUPLESORT_NONE);
@@ -435,7 +438,7 @@
* them out of the uniqueness check. We expect that the second spool (for
* dead tuples) won't get very full, so we give it only work_mem.
*/
- if (indexInfo->ii_Unique)
+ if (indexInfo->ii_Unique && !indexInfo->ii_Concurrent)
{
BTSpool *btspool2 = (BTSpool *) palloc0(sizeof(BTSpool));
SortCoordinate coordinate2 = NULL;
@@ -443,7 +446,7 @@
/* Initialize secondary spool */
btspool2->heap = heap;
btspool2->index = index;
- btspool2->isunique = false;
+ btspool2->isunique = btspool2->unique_dead_ignored = false;
/* Save as secondary spool */
buildstate->spool2 = btspool2;
@@ -466,7 +469,7 @@
* full, so we give it only work_mem
*/
buildstate->spool2->sortstate =
- tuplesort_begin_index_btree(heap, index, false, false, work_mem,
+ tuplesort_begin_index_btree(heap, index, false, false, false, work_mem,
coordinate2, TUPLESORT_NONE);
}
@@ -1145,11 +1148,13 @@
SortSupport sortKeys;
int64 tuples_done = 0;
bool deduplicate;
+ bool fail_on_duplicate;
wstate->bulkstate = smgr_bulk_start_rel(wstate->index, MAIN_FORKNUM);
deduplicate = wstate->inskey->allequalimage && !btspool->isunique &&
BTGetDeduplicateItems(wstate->index);
+ fail_on_duplicate = (btspool->unique_dead_ignored && btspool->isunique && btspool2 == NULL);
if (merge)
{
@@ -1353,6 +1358,80 @@
pfree(dstate);
}
+ else if (fail_on_duplicate)
+ {
+ bool was_valid = false,
+ prev_checked = false,
+ was_null;
+ IndexTuple prev = NULL;
+ TupleTableSlot *slot = MakeSingleTupleTableSlot(RelationGetDescr(wstate->heap),
+ &TTSOpsBufferHeapTuple);
+ IndexFetchTableData *fetch = table_index_fetch_begin(wstate->heap);
+
+ while ((itup = tuplesort_getindextuple(btspool->sortstate, true)) != NULL)
+ {
+ /* When we see first tuple, create first index page */
+ if (state == NULL)
+ state = _bt_pagestate(wstate, 0);
+
+ if (prev != NULL &&
+ ((wstate->inskey->allequalimage &&
+ _bt_keep_natts_fast_wasnull(wstate->index, prev, itup, &was_null) > keysz) ||
+ (_bt_keep_natts_wasnull(wstate->index, prev, itup,wstate->inskey, &was_null) > keysz)
+ ) &&
+ (btspool->nulls_not_distinct && was_null))
+ {
+ bool call_again, ignored, now_valid;
+ ItemPointerData tid;
+ if (!prev_checked)
+ {
+ call_again = false;
+ tid = prev->t_tid;
+ was_valid = table_index_fetch_tuple(fetch, &tid, SnapshotSelf, slot, &call_again, &ignored);
+ prev_checked = true;
+ }
+
+ call_again = false;
+ tid = itup->t_tid;
+ now_valid = table_index_fetch_tuple(fetch, &tid, SnapshotSelf, slot, &call_again, &ignored);
+ if (was_valid && now_valid)
+ {
+ char *key_desc;
+ TupleDesc tupDes = RelationGetDescr(wstate->index);
+ bool isnull[INDEX_MAX_KEYS];
+ Datum values[INDEX_MAX_KEYS];
+
+ index_deform_tuple(itup, tupDes, values, isnull);
+
+ key_desc = BuildIndexValueDescription(wstate->index, values, isnull);
+
+ ereport(ERROR,
+ (errcode(ERRCODE_UNIQUE_VIOLATION),
+ errmsg("could not create unique index \"%s\"",
+ RelationGetRelationName(wstate->index)),
+ key_desc ? errdetail("Key %s is duplicated.", key_desc) :
+ errdetail("Duplicate keys exist."),
+ errtableconstraint(wstate->heap,
+ RelationGetRelationName(wstate->index))));
+ }
+ was_valid |= now_valid;
+ }
+ else
+ {
+ was_valid = false;
+ prev_checked = false;
+ }
+ _bt_buildadd(wstate, state, itup, 0);
+ if (prev) pfree(prev);
+ prev = CopyIndexTuple(itup);
+
+ /* Report progress */
+ pgstat_progress_update_param(PROGRESS_CREATEIDX_TUPLES_DONE,
+ ++tuples_done);
+ }
+ ExecDropSingleTupleTableSlot(slot);
+ table_index_fetch_end(fetch);
+ }
else
{
/* merging and deduplication are both unnecessary */
@@ -1414,17 +1493,7 @@
leaderparticipates = false;
#endif
- /*
- * Enter parallel mode, and create context for parallel build of btree
- * index
- */
- EnterParallelMode();
- Assert(request > 0);
- pcxt = CreateParallelContext("postgres", "_bt_parallel_build_main",
- request);
-
- scantuplesortstates = leaderparticipates ? request + 1 : request;
-
+ Assert(!isconcurrent || !TransactionIdIsValid(MyProc->xmin));
/*
* Prepare for scan of the base relation. In a normal index build, we use
* SnapshotAny because we must retrieve all tuples and do our own time
@@ -1435,7 +1504,20 @@
if (!isconcurrent)
snapshot = SnapshotAny;
else
+ {
snapshot = RegisterSnapshot(GetTransactionSnapshot());
+ PushActiveSnapshot(snapshot);
+ }
+ /*
+ * Enter parallel mode, and create context for parallel build of btree
+ * index
+ */
+ EnterParallelMode();
+ Assert(request > 0);
+ pcxt = CreateParallelContext("postgres", "_bt_parallel_build_main",
+ request);
+
+ scantuplesortstates = leaderparticipates ? request + 1 : request;
/*
* Estimate size for our own PARALLEL_KEY_BTREE_SHARED workspace, and
@@ -1450,7 +1532,7 @@
* Unique case requires a second spool, and so we may have to account for
* another shared workspace for that -- PARALLEL_KEY_TUPLESORT_SPOOL2
*/
- if (!btspool->isunique)
+ if (!btspool->isunique || isconcurrent)
shm_toc_estimate_keys(&pcxt->estimator, 2);
else
{
@@ -1485,6 +1567,8 @@
/* Everyone's had a chance to ask for space, so now create the DSM */
InitializeParallelDSM(pcxt);
+ if (IsMVCCSnapshot(snapshot))
+ PopActiveSnapshot();
/* If no DSM segment was available, back out (do serial build) */
if (pcxt->seg == NULL)
@@ -1515,7 +1599,7 @@
btshared->brokenhotchain = false;
table_parallelscan_initialize(btspool->heap,
ParallelTableScanFromBTShared(btshared),
- snapshot);
+ isconcurrent ? InvalidSnapshot : snapshot);
/*
* Store shared tuplesort-private state, for which we reserved space.
@@ -1529,7 +1613,7 @@
shm_toc_insert(pcxt->toc, PARALLEL_KEY_TUPLESORT, sharedsort);
/* Unique case requires a second spool, and associated shared state */
- if (!btspool->isunique)
+ if (!btspool->isunique || isconcurrent)
sharedsort2 = NULL;
else
{
@@ -1575,7 +1659,7 @@
btleader->btshared = btshared;
btleader->sharedsort = sharedsort;
btleader->sharedsort2 = sharedsort2;
- btleader->snapshot = snapshot;
+ btleader->snapshot = isconcurrent ? InvalidSnapshot : snapshot;
btleader->walusage = walusage;
btleader->bufferusage = bufferusage;
@@ -1589,15 +1673,25 @@
/* Save leader state now that it's clear build will be parallel */
buildstate->btleader = btleader;
+ if (isconcurrent)
+ {
+ WaitForParallelWorkersToAttach(pcxt, true);
+ UnregisterSnapshot(snapshot);
+ }
+
/* Join heap scan ourselves */
if (leaderparticipates)
+ {
+ INJECTION_POINT("_bt_leader_participate_as_worker");
_bt_leader_participate_as_worker(buildstate);
+ }
/*
* Caller needs to wait for all launched workers when we return. Make
* sure that the failure-to-start case will not hang forever.
*/
- WaitForParallelWorkersToAttach(pcxt);
+ if (!isconcurrent)
+ WaitForParallelWorkersToAttach(pcxt, false);
}
/*
@@ -1607,6 +1701,7 @@
_bt_end_parallel(BTLeader *btleader)
{
int i;
+ Snapshot snapshot = btleader->snapshot;
/* Shutdown worker processes */
WaitForParallelWorkersToFinish(btleader->pcxt);
@@ -1619,8 +1714,10 @@
InstrAccumParallelQuery(&btleader->bufferusage[i], &btleader->walusage[i]);
/* Free last reference to MVCC snapshot, if one was used */
- if (IsMVCCSnapshot(btleader->snapshot))
- UnregisterSnapshot(btleader->snapshot);
+ Assert(!btleader->btshared->isconcurrent || snapshot == InvalidSnapshot);
+ Assert(btleader->btshared->isconcurrent || snapshot != InvalidSnapshot);
+ if (snapshot != InvalidSnapshot && IsMVCCSnapshot(snapshot))
+ UnregisterSnapshot(snapshot);
DestroyParallelContext(btleader->pcxt);
ExitParallelMode();
}
@@ -1697,9 +1794,10 @@
leaderworker->index = buildstate->spool->index;
leaderworker->isunique = buildstate->spool->isunique;
leaderworker->nulls_not_distinct = buildstate->spool->nulls_not_distinct;
+ leaderworker->unique_dead_ignored = btleader->btshared->isconcurrent;
/* Initialize second spool, if required */
- if (!btleader->btshared->isunique)
+ if (!btleader->btshared->isunique || btleader->btshared->isconcurrent)
leaderworker2 = NULL;
else
{
@@ -1709,7 +1807,7 @@
/* Initialize worker's own secondary spool */
leaderworker2->heap = leaderworker->heap;
leaderworker2->index = leaderworker->index;
- leaderworker2->isunique = false;
+ leaderworker2->isunique = leaderworker2->unique_dead_ignored = false;
}
/*
@@ -1758,12 +1856,7 @@
ResetUsage();
#endif /* BTREE_BUILD_STATS */
- /*
- * The only possible status flag that can be set to the parallel worker is
- * PROC_IN_SAFE_IC.
- */
- Assert((MyProc->statusFlags == 0) ||
- (MyProc->statusFlags == PROC_IN_SAFE_IC));
+ Assert(MyProc->statusFlags == 0);
/* Set debug_query_string for individual workers first */
sharedquery = shm_toc_lookup(toc, PARALLEL_KEY_QUERY_TEXT, true);
@@ -1796,12 +1889,13 @@
btspool->heap = heapRel;
btspool->index = indexRel;
btspool->isunique = btshared->isunique;
+ btspool->unique_dead_ignored = btshared->isconcurrent;
btspool->nulls_not_distinct = btshared->nulls_not_distinct;
/* Look up shared state private to tuplesort.c */
sharedsort = shm_toc_lookup(toc, PARALLEL_KEY_TUPLESORT, false);
tuplesort_attach_shared(sharedsort, seg);
- if (!btshared->isunique)
+ if (!btshared->isunique || btshared->isconcurrent)
{
btspool2 = NULL;
sharedsort2 = NULL;
@@ -1814,7 +1908,7 @@
/* Initialize worker's own secondary spool */
btspool2->heap = btspool->heap;
btspool2->index = btspool->index;
- btspool2->isunique = false;
+ btspool2->isunique = btspool2->unique_dead_ignored = false;
/* Look up shared state private to tuplesort.c */
sharedsort2 = shm_toc_lookup(toc, PARALLEL_KEY_TUPLESORT_SPOOL2, false);
tuplesort_attach_shared(sharedsort2, seg);
@@ -1825,8 +1919,12 @@
/* Perform sorting of spool, and possibly a spool2 */
sortmem = maintenance_work_mem / btshared->scantuplesortstates;
+ if (btshared->isconcurrent)
+ PopActiveSnapshot();
_bt_parallel_scan_and_sort(btspool, btspool2, btshared, sharedsort,
sharedsort2, sortmem, false);
+ if (btshared->isconcurrent)
+ PushActiveSnapshot(GetLatestSnapshot());
/* Report WAL/buffer usage during parallel execution */
bufferusage = shm_toc_lookup(toc, PARALLEL_KEY_BUFFER_USAGE, false);
@@ -1868,6 +1966,7 @@
TableScanDesc scan;
double reltuples;
IndexInfo *indexInfo;
+ Snapshot snapshot;
/* Initialize local tuplesort coordination state */
coordinate = palloc0(sizeof(SortCoordinateData));
@@ -1880,6 +1979,7 @@
btspool->index,
btspool->isunique,
btspool->nulls_not_distinct,
+ btspool->unique_dead_ignored,
sortmem, coordinate,
TUPLESORT_NONE);
@@ -1902,7 +2002,8 @@
coordinate2->nParticipants = -1;
coordinate2->sharedsort = sharedsort2;
btspool2->sortstate =
- tuplesort_begin_index_btree(btspool->heap, btspool->index, false, false,
+ tuplesort_begin_index_btree(btspool->heap, btspool->index,
+ false, false, false,
Min(sortmem, work_mem), coordinate2,
false);
}
@@ -1917,13 +2018,27 @@
buildstate.indtuples = 0;
buildstate.btleader = NULL;
+ Assert(!btshared->isconcurrent || !TransactionIdIsValid(MyProc->xmin));
/* Join parallel scan */
+ if (btshared->isconcurrent)
+ {
+ snapshot = RegisterSnapshot(GetTransactionSnapshot());
+ PushActiveSnapshot(snapshot);
+ }
indexInfo = BuildIndexInfo(btspool->index);
+ if (btshared->isconcurrent)
+ {
+ PopActiveSnapshot();
+ UnregisterSnapshot(snapshot);
+ }
+ Assert(!btshared->isconcurrent || !TransactionIdIsValid(MyProc->xmin));
+
indexInfo->ii_Concurrent = btshared->isconcurrent;
scan = table_beginscan_parallel(btspool->heap,
ParallelTableScanFromBTShared(btshared));
reltuples = table_index_build_scan(btspool->heap, btspool->index, indexInfo,
- true, progress, _bt_build_callback,
+ true, progress,
+ _bt_build_callback,
(void *) &buildstate, scan);
/* Execute this worker's part of the sort */
Index: src/backend/access/spgist/spginsert.c
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/access/spgist/spginsert.c b/src/backend/access/spgist/spginsert.c
--- a/src/backend/access/spgist/spginsert.c (revision 91dd70fc5ddc60cbad5b17c95f17c6a517f36770)
+++ b/src/backend/access/spgist/spginsert.c (revision 3dea72b62adc8806917dc459b82ff44d962bcb12)
@@ -20,6 +20,7 @@
#include "access/spgist_private.h"
#include "access/tableam.h"
#include "access/xloginsert.h"
+#include "catalog/index.h"
#include "miscadmin.h"
#include "nodes/execnodes.h"
#include "storage/bufmgr.h"
Index: src/backend/optimizer/plan/planner.c
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/optimizer/plan/planner.c b/src/backend/optimizer/plan/planner.c
--- a/src/backend/optimizer/plan/planner.c (revision 91dd70fc5ddc60cbad5b17c95f17c6a517f36770)
+++ b/src/backend/optimizer/plan/planner.c (revision 3dea72b62adc8806917dc459b82ff44d962bcb12)
@@ -61,6 +61,7 @@
#include "utils/lsyscache.h"
#include "utils/rel.h"
#include "utils/selfuncs.h"
+#include "utils/snapmgr.h"
/* GUC parameters */
double cursor_tuple_fraction = DEFAULT_CURSOR_TUPLE_FRACTION;
@@ -6791,6 +6792,7 @@
BlockNumber heap_blocks;
double reltuples;
double allvisfrac;
+ Snapshot snapshot = InvalidSnapshot;
/*
* We don't allow performing parallel operation in standalone backend or
@@ -6842,6 +6844,10 @@
heap = table_open(tableOid, NoLock);
index = index_open(indexOid, NoLock);
+ if (!ActiveSnapshotSet()) {
+ snapshot = RegisterSnapshot(GetTransactionSnapshot());
+ PushActiveSnapshot(snapshot);
+ }
/*
* Determine if it's safe to proceed.
*
@@ -6899,6 +6905,12 @@
parallel_workers--;
done:
+ if (snapshot != InvalidSnapshot)
+ {
+ PopActiveSnapshot();
+ UnregisterSnapshot(snapshot);
+ }
+
index_close(index, NoLock);
table_close(heap, NoLock);
Index: src/backend/access/table/tableam.c
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/access/table/tableam.c b/src/backend/access/table/tableam.c
--- a/src/backend/access/table/tableam.c (revision 103bbb703f974c65be6e238ca2c181f1470ceb25)
+++ b/src/backend/access/table/tableam.c (revision 3dea72b62adc8806917dc459b82ff44d962bcb12)
@@ -29,6 +29,7 @@
#include "storage/bufmgr.h"
#include "storage/shmem.h"
#include "storage/smgr.h"
+#include "storage/proc.h"
/*
* Constants to control the behavior of block allocation to parallel workers
@@ -149,15 +150,23 @@
pscan->phs_snapshot_off = snapshot_off;
- if (IsMVCCSnapshot(snapshot))
+
+ if (snapshot == InvalidSnapshot)
+ {
+ pscan->phs_snapshot_any = false;
+ pscan->phs_snapshot_reset = true;
+ }
+ else if (IsMVCCSnapshot(snapshot))
{
SerializeSnapshot(snapshot, (char *) pscan + pscan->phs_snapshot_off);
pscan->phs_snapshot_any = false;
+ pscan->phs_snapshot_reset = false;
}
else
{
Assert(snapshot == SnapshotAny);
pscan->phs_snapshot_any = true;
+ pscan->phs_snapshot_reset = false;
}
}
@@ -170,7 +179,16 @@
Assert(RelationGetRelid(relation) == pscan->phs_relid);
- if (!pscan->phs_snapshot_any)
+ if (pscan->phs_snapshot_reset)
+ {
+ Assert(!ActiveSnapshotSet());
+ Assert(MyProc->xmin == InvalidTransactionId);
+
+ snapshot = RegisterSnapshot(GetLatestSnapshot());
+ PushActiveSnapshot(snapshot);
+ flags |= (SO_RESET_SNAPSHOT | SO_TEMP_SNAPSHOT);
+ }
+ else if (!pscan->phs_snapshot_any)
{
/* Snapshot was serialized -- restore it */
snapshot = RestoreSnapshot((char *) pscan + pscan->phs_snapshot_off);
Index: src/backend/access/transam/parallel.c
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/access/transam/parallel.c b/src/backend/access/transam/parallel.c
--- a/src/backend/access/transam/parallel.c (revision 103bbb703f974c65be6e238ca2c181f1470ceb25)
+++ b/src/backend/access/transam/parallel.c (revision 9817a8ff254bae0291a320bd306d2ec1280f7592)
@@ -76,6 +76,7 @@
#define PARALLEL_KEY_RELMAPPER_STATE UINT64CONST(0xFFFFFFFFFFFF000D)
#define PARALLEL_KEY_UNCOMMITTEDENUMS UINT64CONST(0xFFFFFFFFFFFF000E)
#define PARALLEL_KEY_CLIENTCONNINFO UINT64CONST(0xFFFFFFFFFFFF000F)
+#define PARALLEL_KEY_SNAPSHOT_SET_FLAG UINT64CONST(0xFFFFFFFFFFFF0010)
/* Fixed-size parallel state. */
typedef struct FixedParallelState
@@ -289,6 +290,9 @@
mul_size(PARALLEL_ERROR_QUEUE_SIZE,
pcxt->nworkers));
shm_toc_estimate_keys(&pcxt->estimator, 1);
+
+ shm_toc_estimate_chunk(&pcxt->estimator, mul_size(sizeof(bool), pcxt->nworkers));
+ shm_toc_estimate_keys(&pcxt->estimator, 1);
/* Estimate how much we'll need for the entrypoint info. */
shm_toc_estimate_chunk(&pcxt->estimator, strlen(pcxt->library_name) +
@@ -359,6 +363,7 @@
char *entrypointstate;
char *uncommittedenumsspace;
char *clientconninfospace;
+ bool *snapshot_set_flag_space;
Size lnamelen;
/* Serialize shared libraries we have loaded. */
@@ -474,6 +479,15 @@
strcpy(entrypointstate, pcxt->library_name);
strcpy(entrypointstate + lnamelen + 1, pcxt->function_name);
shm_toc_insert(pcxt->toc, PARALLEL_KEY_ENTRYPOINT, entrypointstate);
+
+ snapshot_set_flag_space =
+ shm_toc_allocate(pcxt->toc, mul_size(sizeof(bool), pcxt->nworkers));
+ for (i = 0; i < pcxt->nworkers; ++i)
+ {
+ pcxt->worker[i].snapshot_set_flag = snapshot_set_flag_space + i * sizeof(bool);
+ *pcxt->worker[i].snapshot_set_flag = false;
+ }
+ shm_toc_insert(pcxt->toc, PARALLEL_KEY_SNAPSHOT_SET_FLAG, snapshot_set_flag_space);
}
/* Restore previous memory context. */
@@ -511,6 +525,7 @@
if (pcxt->nworkers > 0)
{
char *error_queue_space;
+ bool *snapshot_set_flag_space;
int i;
error_queue_space =
@@ -525,6 +540,11 @@
shm_mq_set_receiver(mq, MyProc);
pcxt->worker[i].error_mqh = shm_mq_attach(mq, pcxt->seg, NULL);
}
+
+ snapshot_set_flag_space =
+ shm_toc_lookup(pcxt->toc, PARALLEL_KEY_SNAPSHOT_SET_FLAG, false);
+ for (i = 0; i < pcxt->nworkers; ++i)
+ snapshot_set_flag_space[i] = false;
}
}
@@ -669,7 +689,7 @@
* call this function at all.
*/
void
-WaitForParallelWorkersToAttach(ParallelContext *pcxt)
+WaitForParallelWorkersToAttach(ParallelContext *pcxt, bool wait_for_snapshot)
{
int i;
@@ -713,9 +733,12 @@
mq = shm_mq_get_queue(pcxt->worker[i].error_mqh);
if (shm_mq_get_sender(mq) != NULL)
{
- /* Yes, so it is known to be attached. */
- pcxt->known_attached_workers[i] = true;
- ++pcxt->nknown_attached_workers;
+ if (!wait_for_snapshot || *(pcxt->worker[i].snapshot_set_flag))
+ {
+ /* Yes, so it is known to be attached. */
+ pcxt->known_attached_workers[i] = true;
+ ++pcxt->nknown_attached_workers;
+ }
}
}
else if (status == BGWH_STOPPED)
@@ -1274,6 +1297,7 @@
shm_toc *toc;
FixedParallelState *fps;
char *error_queue_space;
+ bool *snapshot_flag_set_space;
shm_mq *mq;
shm_mq_handle *mqh;
char *libraryspace;
@@ -1449,6 +1473,9 @@
fps->parallel_leader_pgproc);
PushActiveSnapshot(asnapshot);
+ snapshot_flag_set_space = shm_toc_lookup(toc, PARALLEL_KEY_SNAPSHOT_SET_FLAG, false);
+ snapshot_flag_set_space[ParallelWorkerNumber] = true;
+
/*
* We've changed which tuples we can see, and must therefore invalidate
* system caches.
Index: src/include/access/parallel.h
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/include/access/parallel.h b/src/include/access/parallel.h
--- a/src/include/access/parallel.h (revision 103bbb703f974c65be6e238ca2c181f1470ceb25)
+++ b/src/include/access/parallel.h (revision 9817a8ff254bae0291a320bd306d2ec1280f7592)
@@ -26,6 +26,7 @@
{
BackgroundWorkerHandle *bgwhandle;
shm_mq_handle *error_mqh;
+ bool *snapshot_set_flag;
} ParallelWorkerInfo;
typedef struct ParallelContext
@@ -65,7 +66,7 @@
extern void ReinitializeParallelDSM(ParallelContext *pcxt);
extern void ReinitializeParallelWorkers(ParallelContext *pcxt, int nworkers_to_launch);
extern void LaunchParallelWorkers(ParallelContext *pcxt);
-extern void WaitForParallelWorkersToAttach(ParallelContext *pcxt);
+extern void WaitForParallelWorkersToAttach(ParallelContext *pcxt, bool wait_for_snapshot);
extern void WaitForParallelWorkersToFinish(ParallelContext *pcxt);
extern void DestroyParallelContext(ParallelContext *pcxt);
extern bool ParallelContextActive(void);
Index: src/include/access/relscan.h
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/include/access/relscan.h b/src/include/access/relscan.h
--- a/src/include/access/relscan.h (revision 103bbb703f974c65be6e238ca2c181f1470ceb25)
+++ b/src/include/access/relscan.h (revision ea1fcacc7cead3e2fccf581d20e51244a7107435)
@@ -64,6 +64,7 @@
{
Oid phs_relid; /* OID of relation to scan */
bool phs_syncscan; /* report location to syncscan logic? */
+ bool phs_snapshot_reset;
bool phs_snapshot_any; /* SnapshotAny, not phs_snapshot_data? */
Size phs_snapshot_off; /* data for snapshot */
} ParallelTableScanDescData;
Index: src/include/utils/snapmgr.h
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/include/utils/snapmgr.h b/src/include/utils/snapmgr.h
--- a/src/include/utils/snapmgr.h (revision 103bbb703f974c65be6e238ca2c181f1470ceb25)
+++ b/src/include/utils/snapmgr.h (revision 9817a8ff254bae0291a320bd306d2ec1280f7592)
@@ -96,6 +96,7 @@
extern void WaitForOlderSnapshots(TransactionId limitXmin, bool progress);
extern bool ThereAreNoPriorRegisteredSnapshots(void);
extern bool HaveRegisteredOrActiveSnapshot(void);
+extern bool HaveRegisteredSnapshot(void);
extern char *ExportSnapshot(Snapshot snapshot);
Index: contrib/pgstattuple/pgstattuple.c
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/contrib/pgstattuple/pgstattuple.c b/contrib/pgstattuple/pgstattuple.c
--- a/contrib/pgstattuple/pgstattuple.c (revision 9817a8ff254bae0291a320bd306d2ec1280f7592)
+++ b/contrib/pgstattuple/pgstattuple.c (revision ea1fcacc7cead3e2fccf581d20e51244a7107435)
@@ -286,6 +286,9 @@
case BRIN_AM_OID:
err = "brin index";
break;
+ case STIR_AM_OID:
+ err = "stir index";
+ break;
default:
err = "unknown index";
break;
@@ -329,7 +332,7 @@
errmsg("only heap AM is supported")));
/* Disable syncscan because we assume we scan from block zero upwards */
- scan = table_beginscan_strat(rel, SnapshotAny, 0, NULL, true, false);
+ scan = table_beginscan_strat(rel, SnapshotAny, 0, NULL, true, false, false);
hscan = (HeapScanDesc) scan;
InitDirtySnapshot(SnapshotDirty);
Index: src/backend/access/Makefile
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/access/Makefile b/src/backend/access/Makefile
--- a/src/backend/access/Makefile (revision 9817a8ff254bae0291a320bd306d2ec1280f7592)
+++ b/src/backend/access/Makefile (revision d8df9daea76374468c28f8e9d60d83539aad05c8)
@@ -9,6 +9,6 @@
include $(top_builddir)/src/Makefile.global
SUBDIRS = brin common gin gist hash heap index nbtree rmgrdesc spgist \
- sequence table tablesample transam
+ sequence table tablesample transam stir
include $(top_srcdir)/src/backend/common.mk
Index: src/backend/access/meson.build
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/access/meson.build b/src/backend/access/meson.build
--- a/src/backend/access/meson.build (revision 9817a8ff254bae0291a320bd306d2ec1280f7592)
+++ b/src/backend/access/meson.build (revision d8df9daea76374468c28f8e9d60d83539aad05c8)
@@ -14,3 +14,4 @@
subdir('table')
subdir('tablesample')
subdir('transam')
+subdir('stir')
Index: src/backend/commands/analyze.c
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/commands/analyze.c b/src/backend/commands/analyze.c
--- a/src/backend/commands/analyze.c (revision 9817a8ff254bae0291a320bd306d2ec1280f7592)
+++ b/src/backend/commands/analyze.c (revision 75cd94daf4b0b6147e7f3a386ad1a93fb086653b)
@@ -719,6 +719,7 @@
ivinfo.message_level = elevel;
ivinfo.num_heap_tuples = onerel->rd_rel->reltuples;
ivinfo.strategy = vac_strategy;
+ ivinfo.validate_index = false;
stats = index_vacuum_cleanup(&ivinfo, NULL);
Index: src/backend/commands/vacuumparallel.c
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/commands/vacuumparallel.c b/src/backend/commands/vacuumparallel.c
--- a/src/backend/commands/vacuumparallel.c (revision 9817a8ff254bae0291a320bd306d2ec1280f7592)
+++ b/src/backend/commands/vacuumparallel.c (revision 75cd94daf4b0b6147e7f3a386ad1a93fb086653b)
@@ -883,6 +883,7 @@
ivinfo.estimated_count = pvs->shared->estimated_count;
ivinfo.num_heap_tuples = pvs->shared->reltuples;
ivinfo.strategy = pvs->bstrategy;
+ ivinfo.validate_index = false;
/* Update error traceback information */
pvs->indname = pstrdup(RelationGetRelationName(indrel));
Index: src/include/access/genam.h
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/include/access/genam.h b/src/include/access/genam.h
--- a/src/include/access/genam.h (revision 9817a8ff254bae0291a320bd306d2ec1280f7592)
+++ b/src/include/access/genam.h (revision 75cd94daf4b0b6147e7f3a386ad1a93fb086653b)
@@ -48,6 +48,7 @@
bool analyze_only; /* ANALYZE (without any actual vacuum) */
bool report_progress; /* emit progress.h status reports */
bool estimated_count; /* num_heap_tuples is an estimate */
+ bool validate_index; /* not a vacuum but an index validation */
int message_level; /* ereport level for progress messages */
double num_heap_tuples; /* tuples remaining in heap */
BufferAccessStrategy strategy; /* access strategy for reads */
Index: src/include/access/reloptions.h
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/include/access/reloptions.h b/src/include/access/reloptions.h
--- a/src/include/access/reloptions.h (revision 9817a8ff254bae0291a320bd306d2ec1280f7592)
+++ b/src/include/access/reloptions.h (revision d8df9daea76374468c28f8e9d60d83539aad05c8)
@@ -51,8 +51,9 @@
RELOPT_KIND_VIEW = (1 << 9),
RELOPT_KIND_BRIN = (1 << 10),
RELOPT_KIND_PARTITIONED = (1 << 11),
+ RELOPT_KIND_STIR = (1 << 12),
/* if you add a new kind, make sure you update "last_default" too */
- RELOPT_KIND_LAST_DEFAULT = RELOPT_KIND_PARTITIONED,
+ RELOPT_KIND_LAST_DEFAULT = RELOPT_KIND_STIR,
/* some compilers treat enums as signed ints, so we can't use 1 << 31 */
RELOPT_KIND_MAX = (1 << 30)
} relopt_kind;
Index: src/include/catalog/pg_am.dat
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/include/catalog/pg_am.dat b/src/include/catalog/pg_am.dat
--- a/src/include/catalog/pg_am.dat (revision 9817a8ff254bae0291a320bd306d2ec1280f7592)
+++ b/src/include/catalog/pg_am.dat (revision d8df9daea76374468c28f8e9d60d83539aad05c8)
@@ -33,5 +33,7 @@
{ oid => '3580', oid_symbol => 'BRIN_AM_OID',
descr => 'block range index (BRIN) access method',
amname => 'brin', amhandler => 'brinhandler', amtype => 'i' },
-
+{ oid => '5555', oid_symbol => 'STIR_AM_OID',
+ descr => 'short term index replacement',
+ amname => 'stir', amhandler => 'stirhandler', amtype => 'i' },
]
Index: src/include/catalog/pg_amop.dat
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/include/catalog/pg_amop.dat b/src/include/catalog/pg_amop.dat
--- a/src/include/catalog/pg_amop.dat (revision 9817a8ff254bae0291a320bd306d2ec1280f7592)
+++ b/src/include/catalog/pg_amop.dat (revision d8df9daea76374468c28f8e9d60d83539aad05c8)
@@ -3227,4 +3227,8 @@
amoprighttype => 'point', amopstrategy => '7', amopopr => '@>(box,point)',
amopmethod => 'brin' },
+{ amopfamily => 'stir/record_ops', amoplefttype => 'record',
+ amoprighttype => 'record', amopstrategy => '1', amopopr => '=(record,record)',
+ amopmethod => 'stir' },
+
]
Index: src/include/catalog/pg_opclass.dat
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/include/catalog/pg_opclass.dat b/src/include/catalog/pg_opclass.dat
--- a/src/include/catalog/pg_opclass.dat (revision 9817a8ff254bae0291a320bd306d2ec1280f7592)
+++ b/src/include/catalog/pg_opclass.dat (revision d8df9daea76374468c28f8e9d60d83539aad05c8)
@@ -488,4 +488,8 @@
# no brin opclass for the geometric types except box
+{ oid => '5557', oid_symbol => 'RECORD_STIR_OPS_OID',
+ opcmethod => 'stir', opcname => 'record_ops', opcfamily => 'stir/record_ops',
+ opcintype => 'record' },
+
]
Index: src/include/catalog/pg_opfamily.dat
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/include/catalog/pg_opfamily.dat b/src/include/catalog/pg_opfamily.dat
--- a/src/include/catalog/pg_opfamily.dat (revision 9817a8ff254bae0291a320bd306d2ec1280f7592)
+++ b/src/include/catalog/pg_opfamily.dat (revision d8df9daea76374468c28f8e9d60d83539aad05c8)
@@ -302,6 +302,8 @@
opfmethod => 'btree', opfname => 'multirange_ops' },
{ oid => '4225',
opfmethod => 'hash', opfname => 'multirange_ops' },
+{ oid => '5558',
+ opfmethod => 'stir', opfname => 'record_ops' },
{ oid => '6158',
opfmethod => 'gist', opfname => 'multirange_ops' },
Index: src/include/catalog/pg_proc.dat
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
--- a/src/include/catalog/pg_proc.dat (revision 9817a8ff254bae0291a320bd306d2ec1280f7592)
+++ b/src/include/catalog/pg_proc.dat (revision 6c55d9749e2999542d4e6281db733fdd47930796)
@@ -935,6 +935,10 @@
proname => 'brinhandler', provolatile => 'v',
prorettype => 'index_am_handler', proargtypes => 'internal',
prosrc => 'brinhandler' },
+{ oid => '5556', descr => 'just access method handler',
+ proname => 'stirhandler', provolatile => 'v',
+ prorettype => 'index_am_handler', proargtypes => 'internal',
+ prosrc => 'stirhandler' },
{ oid => '3952', descr => 'brin: standalone scan new table pages',
proname => 'brin_summarize_new_values', provolatile => 'v',
proparallel => 'u', prorettype => 'int4', proargtypes => 'regclass',
@@ -5487,9 +5491,9 @@
proname => 'pg_stat_get_activity', prorows => '100', proisstrict => 'f',
proretset => 't', provolatile => 's', proparallel => 'r',
prorettype => 'record', proargtypes => 'int4',
- proallargtypes => '{int4,oid,int4,oid,text,text,text,text,text,timestamptz,timestamptz,timestamptz,timestamptz,inet,text,int4,xid,xid,text,bool,text,text,int4,text,numeric,text,bool,text,bool,bool,int4,int8}',
- proargmodes => '{i,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
- proargnames => '{pid,datid,pid,usesysid,application_name,state,query,wait_event_type,wait_event,xact_start,query_start,backend_start,state_change,client_addr,client_hostname,client_port,backend_xid,backend_xmin,backend_type,ssl,sslversion,sslcipher,sslbits,ssl_client_dn,ssl_client_serial,ssl_issuer_dn,gss_auth,gss_princ,gss_enc,gss_delegation,leader_pid,query_id}',
+ proallargtypes => '{int4,oid,int4,oid,text,text,text,text,text,timestamptz,timestamptz,timestamptz,timestamptz,inet,text,int4,xid,xid,text,bool,text,text,int4,text,numeric,text,bool,text,bool,bool,int4,int8,xid}',
+ proargmodes => '{i,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
+ proargnames => '{pid,datid,pid,usesysid,application_name,state,query,wait_event_type,wait_event,xact_start,query_start,backend_start,state_change,client_addr,client_hostname,client_port,backend_xid,backend_xmin,backend_type,ssl,sslversion,sslcipher,sslbits,ssl_client_dn,ssl_client_serial,ssl_issuer_dn,gss_auth,gss_princ,gss_enc,gss_delegation,leader_pid,query_id,backend_catalog_xmin}',
prosrc => 'pg_stat_get_activity' },
{ oid => '6318', descr => 'describe wait events',
proname => 'pg_get_wait_events', procost => '10', prorows => '250',
Index: src/include/utils/index_selfuncs.h
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/include/utils/index_selfuncs.h b/src/include/utils/index_selfuncs.h
--- a/src/include/utils/index_selfuncs.h (revision 9817a8ff254bae0291a320bd306d2ec1280f7592)
+++ b/src/include/utils/index_selfuncs.h (revision d8df9daea76374468c28f8e9d60d83539aad05c8)
@@ -70,5 +70,13 @@
Selectivity *indexSelectivity,
double *indexCorrelation,
double *indexPages);
+extern void stircostestimate(struct PlannerInfo *root,
+ struct IndexPath *path,
+ double loop_count,
+ Cost *indexStartupCost,
+ Cost *indexTotalCost,
+ Selectivity *indexSelectivity,
+ double *indexCorrelation,
+ double *indexPages);
#endif /* INDEX_SELFUNCS_H */
Index: src/test/regress/expected/amutils.out
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/test/regress/expected/amutils.out b/src/test/regress/expected/amutils.out
--- a/src/test/regress/expected/amutils.out (revision 9817a8ff254bae0291a320bd306d2ec1280f7592)
+++ b/src/test/regress/expected/amutils.out (revision d8df9daea76374468c28f8e9d60d83539aad05c8)
@@ -173,7 +173,13 @@
spgist | can_exclude | t
spgist | can_include | t
spgist | bogus |
-(36 rows)
+ stir | can_order | f
+ stir | can_unique | f
+ stir | can_multi_col | t
+ stir | can_exclude | f
+ stir | can_include | t
+ stir | bogus |
+(42 rows)
--
-- additional checks for pg_index_column_has_property
Index: src/test/regress/expected/opr_sanity.out
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/test/regress/expected/opr_sanity.out b/src/test/regress/expected/opr_sanity.out
--- a/src/test/regress/expected/opr_sanity.out (revision 9817a8ff254bae0291a320bd306d2ec1280f7592)
+++ b/src/test/regress/expected/opr_sanity.out (revision 75cd94daf4b0b6147e7f3a386ad1a93fb086653b)
@@ -2092,7 +2092,8 @@
4000 | 28 | ^@
4000 | 29 | <^
4000 | 30 | >^
-(124 rows)
+ 5555 | 1 | =
+(125 rows)
-- Check that all opclass search operators have selectivity estimators.
-- This is not absolutely required, but it seems a reasonable thing
Index: src/test/regress/expected/psql.out
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/test/regress/expected/psql.out b/src/test/regress/expected/psql.out
--- a/src/test/regress/expected/psql.out (revision 9817a8ff254bae0291a320bd306d2ec1280f7592)
+++ b/src/test/regress/expected/psql.out (revision d8df9daea76374468c28f8e9d60d83539aad05c8)
@@ -5027,7 +5027,8 @@
heap | Table
heap2 | Table
spgist | Index
-(8 rows)
+ stir | Index
+(9 rows)
\dA *
List of access methods
@@ -5041,7 +5042,8 @@
heap | Table
heap2 | Table
spgist | Index
-(8 rows)
+ stir | Index
+(9 rows)
\dA h*
List of access methods
@@ -5077,7 +5079,8 @@
heap | Table | heap_tableam_handler | heap table access method
heap2 | Table | heap_tableam_handler |
spgist | Index | spghandler | SP-GiST index access method
-(8 rows)
+ stir | Index | stirhandler | short term index replacement
+(9 rows)
\dA+ *
List of access methods
@@ -5091,7 +5094,8 @@
heap | Table | heap_tableam_handler | heap table access method
heap2 | Table | heap_tableam_handler |
spgist | Index | spghandler | SP-GiST index access method
-(8 rows)
+ stir | Index | stirhandler | short term index replacement
+(9 rows)
\dA+ h*
List of access methods
Index: src/backend/catalog/system_views.sql
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
--- a/src/backend/catalog/system_views.sql (revision b24132f98f93d14c64dfe41973337e13d5e7636b)
+++ b/src/backend/catalog/system_views.sql (revision 6c55d9749e2999542d4e6281db733fdd47930796)
@@ -879,6 +879,7 @@
S.state,
S.backend_xid,
s.backend_xmin,
+ s.backend_catalog_xmin,
S.query_id,
S.query,
S.backend_type
Index: src/backend/utils/activity/backend_status.c
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/utils/activity/backend_status.c b/src/backend/utils/activity/backend_status.c
--- a/src/backend/utils/activity/backend_status.c (revision b24132f98f93d14c64dfe41973337e13d5e7636b)
+++ b/src/backend/utils/activity/backend_status.c (revision 6c55d9749e2999542d4e6281db733fdd47930796)
@@ -838,6 +838,7 @@
ProcNumberGetTransactionIds(procNumber,
&localentry->backend_xid,
&localentry->backend_xmin,
+ &localentry->backend_catalog_xmin,
&localentry->backend_subxact_count,
&localentry->backend_subxact_overflowed);
Index: src/backend/utils/adt/pgstatfuncs.c
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c
--- a/src/backend/utils/adt/pgstatfuncs.c (revision b24132f98f93d14c64dfe41973337e13d5e7636b)
+++ b/src/backend/utils/adt/pgstatfuncs.c (revision 6c55d9749e2999542d4e6281db733fdd47930796)
@@ -302,7 +302,7 @@
Datum
pg_stat_get_activity(PG_FUNCTION_ARGS)
{
-#define PG_STAT_GET_ACTIVITY_COLS 31
+#define PG_STAT_GET_ACTIVITY_COLS 32
int num_backends = pgstat_fetch_stat_numbackends();
int curr_backend;
int pid = PG_ARGISNULL(0) ? -1 : PG_GETARG_INT32(0);
@@ -353,6 +353,11 @@
else
nulls[15] = true;
+ if (TransactionIdIsValid(local_beentry->backend_catalog_xmin))
+ values[31] = TransactionIdGetDatum(local_beentry->backend_catalog_xmin);
+ else
+ nulls[31] = true;
+
if (TransactionIdIsValid(local_beentry->backend_xmin))
values[16] = TransactionIdGetDatum(local_beentry->backend_xmin);
else
Index: src/include/utils/backend_status.h
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/include/utils/backend_status.h b/src/include/utils/backend_status.h
--- a/src/include/utils/backend_status.h (revision b24132f98f93d14c64dfe41973337e13d5e7636b)
+++ b/src/include/utils/backend_status.h (revision 6c55d9749e2999542d4e6281db733fdd47930796)
@@ -266,6 +266,8 @@
*/
TransactionId backend_xmin;
+ TransactionId backend_catalog_xmin;
+
/*
* Number of cached subtransactions in the current session.
*/
Index: src/test/regress/expected/rules.out
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
--- a/src/test/regress/expected/rules.out (revision b24132f98f93d14c64dfe41973337e13d5e7636b)
+++ b/src/test/regress/expected/rules.out (revision 6c55d9749e2999542d4e6281db733fdd47930796)
@@ -1759,10 +1759,11 @@
s.state,
s.backend_xid,
s.backend_xmin,
+ s.backend_catalog_xmin,
s.query_id,
s.query,
s.backend_type
- FROM ((pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, gss_delegation, leader_pid, query_id)
+ FROM ((pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, gss_delegation, leader_pid, query_id, backend_catalog_xmin)
LEFT JOIN pg_database d ON ((s.datid = d.oid)))
LEFT JOIN pg_authid u ON ((s.usesysid = u.oid)));
pg_stat_all_indexes| SELECT c.oid AS relid,
@@ -1882,7 +1883,7 @@
gss_princ AS principal,
gss_enc AS encrypted,
gss_delegation AS credentials_delegated
- FROM pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, gss_delegation, leader_pid, query_id)
+ FROM pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, gss_delegation, leader_pid, query_id, backend_catalog_xmin)
WHERE (client_port IS NOT NULL);
pg_stat_io| SELECT backend_type,
object,
@@ -2086,7 +2087,7 @@
w.sync_priority,
w.sync_state,
w.reply_time
- FROM ((pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, gss_delegation, leader_pid, query_id)
+ FROM ((pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, gss_delegation, leader_pid, query_id, backend_catalog_xmin)
JOIN pg_stat_get_wal_senders() w(pid, state, sent_lsn, write_lsn, flush_lsn, replay_lsn, write_lag, flush_lag, replay_lag, sync_priority, sync_state, reply_time) ON ((s.pid = w.pid)))
LEFT JOIN pg_authid u ON ((s.usesysid = u.oid)));
pg_stat_replication_slots| SELECT s.slot_name,
@@ -2120,7 +2121,7 @@
ssl_client_dn AS client_dn,
ssl_client_serial AS client_serial,
ssl_issuer_dn AS issuer_dn
- FROM pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, gss_delegation, leader_pid, query_id)
+ FROM pg_stat_get_activity(NULL::integer) s(datid, pid, usesysid, application_name, state, query, wait_event_type, wait_event, xact_start, query_start, backend_start, state_change, client_addr, client_hostname, client_port, backend_xid, backend_xmin, backend_type, ssl, sslversion, sslcipher, sslbits, ssl_client_dn, ssl_client_serial, ssl_issuer_dn, gss_auth, gss_princ, gss_enc, gss_delegation, leader_pid, query_id, backend_catalog_xmin)
WHERE (client_port IS NOT NULL);
pg_stat_subscription| SELECT su.oid AS subid,
su.subname,
Index: src/backend/access/stir/Makefile
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/access/stir/Makefile b/src/backend/access/stir/Makefile
new file mode 100644
--- /dev/null (revision d8df9daea76374468c28f8e9d60d83539aad05c8)
+++ b/src/backend/access/stir/Makefile (revision d8df9daea76374468c28f8e9d60d83539aad05c8)
@@ -0,0 +1,18 @@
+#-------------------------------------------------------------------------
+#
+# Makefile--
+# Makefile for access/stir
+#
+# IDENTIFICATION
+# src/backend/access/stir/Makefile
+#
+#-------------------------------------------------------------------------
+
+subdir = src/backend/access/stir
+top_builddir = ../../../..
+include $(top_builddir)/src/Makefile.global
+
+OBJS = \
+ stir.o
+
+include $(top_srcdir)/src/backend/common.mk
Index: src/backend/access/stir/meson.build
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/access/stir/meson.build b/src/backend/access/stir/meson.build
new file mode 100644
--- /dev/null (revision d8df9daea76374468c28f8e9d60d83539aad05c8)
+++ b/src/backend/access/stir/meson.build (revision d8df9daea76374468c28f8e9d60d83539aad05c8)
@@ -0,0 +1,5 @@
+# Copyright (c) 2024-2024, PostgreSQL Global Development Group
+
+backend_sources += files(
+ 'stir.c',
+)
Index: src/backend/access/stir/stir.c
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/access/stir/stir.c b/src/backend/access/stir/stir.c
new file mode 100644
--- /dev/null (revision d8df9daea76374468c28f8e9d60d83539aad05c8)
+++ b/src/backend/access/stir/stir.c (revision d8df9daea76374468c28f8e9d60d83539aad05c8)
@@ -0,0 +1,517 @@
+/*-------------------------------------------------------------------------
+ *
+ * stir.c
+ * Implementation of Short-Term Index Replacement.
+ *
+ * Portions Copyright (c) 2024-2024, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ * src/backend/access/stir/stir.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "access/stir.h"
+#include "commands/vacuum.h"
+#include "utils/index_selfuncs.h"
+#include "catalog/pg_opclass.h"
+#include "catalog/pg_opfamily.h"
+#include "utils/catcache.h"
+#include "access/amvalidate.h"
+#include "utils/syscache.h"
+#include "access/htup_details.h"
+#include "catalog/pg_amproc.h"
+#include "catalog/index.h"
+#include "catalog/pg_amop.h"
+#include "utils/regproc.h"
+#include "storage/bufmgr.h"
+#include "access/tableam.h"
+#include "access/reloptions.h"
+#include "utils/memutils.h"
+#include "utils/fmgrprotos.h"
+
+/*
+ * Stir handler function: return IndexAmRoutine with access method parameters
+ * and callbacks.
+ */
+Datum
+stirhandler(PG_FUNCTION_ARGS)
+{
+ IndexAmRoutine *amroutine = makeNode(IndexAmRoutine);
+
+ amroutine->amstrategies = STIR_NSTRATEGIES;
+ amroutine->amsupport = STIR_NPROC;
+ amroutine->amoptsprocnum = STIR_OPTIONS_PROC;
+ amroutine->amcanorder = false;
+ amroutine->amcanorderbyop = false;
+ amroutine->amcanbackward = false;
+ amroutine->amcanunique = false;
+ amroutine->amcanmulticol = true;
+ amroutine->amoptionalkey = true;
+ amroutine->amsearcharray = false;
+ amroutine->amsearchnulls = false;
+ amroutine->amstorage = false;
+ amroutine->amclusterable = false;
+ amroutine->ampredlocks = false;
+ amroutine->amcanparallel = false;
+ amroutine->amcanbuildparallel = false;
+ amroutine->amcaninclude = true;
+ amroutine->amusemaintenanceworkmem = false;
+ amroutine->amparallelvacuumoptions =
+ VACUUM_OPTION_PARALLEL_BULKDEL | VACUUM_OPTION_PARALLEL_CLEANUP;
+ amroutine->amkeytype = InvalidOid;
+
+ amroutine->ambuild = stirbuild;
+ amroutine->ambuildempty = stirbuildempty;
+ amroutine->aminsert = stirinsert;
+ amroutine->aminsertcleanup = NULL;
+ amroutine->ambulkdelete = stirbulkdelete;
+ amroutine->amvacuumcleanup = stirvacuumcleanup;
+ amroutine->amcanreturn = NULL;
+ amroutine->amcostestimate = stircostestimate;
+ amroutine->amoptions = stiroptions;
+ amroutine->amproperty = NULL;
+ amroutine->ambuildphasename = NULL;
+ amroutine->amvalidate = stirvalidate;
+ amroutine->amadjustmembers = NULL;
+ amroutine->ambeginscan = stirbeginscan;
+ amroutine->amrescan = stirrescan;
+ amroutine->amgettuple = NULL;
+ amroutine->amgetbitmap = NULL;
+ amroutine->amendscan = stirendscan;
+ amroutine->ammarkpos = NULL;
+ amroutine->amrestrpos = NULL;
+ amroutine->amestimateparallelscan = NULL;
+ amroutine->aminitparallelscan = NULL;
+ amroutine->amparallelrescan = NULL;
+
+ PG_RETURN_POINTER(amroutine);
+}
+
+bool
+stirvalidate(Oid opclassoid)
+{
+ bool result = true;
+ HeapTuple classtup;
+ Form_pg_opclass classform;
+ Oid opfamilyoid;
+ HeapTuple familytup;
+ Form_pg_opfamily familyform;
+ char *opfamilyname;
+ CatCList *proclist,
+ *oprlist;
+ int i;
+
+ /* Fetch opclass information */
+ classtup = SearchSysCache1(CLAOID, ObjectIdGetDatum(opclassoid));
+ if (!HeapTupleIsValid(classtup))
+ elog(ERROR, "cache lookup failed for operator class %u", opclassoid);
+ classform = (Form_pg_opclass) GETSTRUCT(classtup);
+
+ opfamilyoid = classform->opcfamily;
+
+
+ /* Fetch opfamily information */
+ familytup = SearchSysCache1(OPFAMILYOID, ObjectIdGetDatum(opfamilyoid));
+ if (!HeapTupleIsValid(familytup))
+ elog(ERROR, "cache lookup failed for operator family %u", opfamilyoid);
+ familyform = (Form_pg_opfamily) GETSTRUCT(familytup);
+
+ opfamilyname = NameStr(familyform->opfname);
+
+ /* Fetch all operators and support functions of the opfamily */
+ oprlist = SearchSysCacheList1(AMOPSTRATEGY, ObjectIdGetDatum(opfamilyoid));
+ proclist = SearchSysCacheList1(AMPROCNUM, ObjectIdGetDatum(opfamilyoid));
+
+ /* Check individual operators */
+ for (i = 0; i < oprlist->n_members; i++)
+ {
+ HeapTuple oprtup = &oprlist->members[i]->tuple;
+ Form_pg_amop oprform = (Form_pg_amop) GETSTRUCT(oprtup);
+
+ /* Check it's allowed strategy for stir */
+ if (oprform->amopstrategy < 1 ||
+ oprform->amopstrategy > STIR_NSTRATEGIES)
+ {
+ ereport(INFO,
+ (errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+ errmsg("stir opfamily %s contains operator %s with invalid strategy number %d",
+ opfamilyname,
+ format_operator(oprform->amopopr),
+ oprform->amopstrategy)));
+ result = false;
+ }
+
+ /* stir doesn't support ORDER BY operators */
+ if (oprform->amoppurpose != AMOP_SEARCH ||
+ OidIsValid(oprform->amopsortfamily))
+ {
+ ereport(INFO,
+ (errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+ errmsg("stir opfamily %s contains invalid ORDER BY specification for operator %s",
+ opfamilyname,
+ format_operator(oprform->amopopr))));
+ result = false;
+ }
+
+ /* Check operator signature --- same for all stir strategies */
+ if (!check_amop_signature(oprform->amopopr, BOOLOID,
+ oprform->amoplefttype,
+ oprform->amoprighttype))
+ {
+ ereport(INFO,
+ (errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
+ errmsg("stir opfamily %s contains operator %s with wrong signature",
+ opfamilyname,
+ format_operator(oprform->amopopr))));
+ result = false;
+ }
+ }
+
+
+ ReleaseCatCacheList(proclist);
+ ReleaseCatCacheList(oprlist);
+ ReleaseSysCache(familytup);
+ ReleaseSysCache(classtup);
+
+ return result;
+}
+
+
+void
+StirFillMetapage(Relation index, Page metaPage, bool skipInserts)
+{
+ StirMetaPageData *metadata;
+
+ StirInitPage(metaPage, STIR_META);
+ metadata = StirPageGetMeta(metaPage);
+ memset(metadata, 0, sizeof(StirMetaPageData));
+ metadata->magickNumber = STIR_MAGICK_NUMBER;
+ metadata->skipInserts = skipInserts;
+ ((PageHeader) metaPage)->pd_lower += sizeof(StirMetaPageData);
+}
+
+void
+StirInitMetapage(Relation index, ForkNumber forknum)
+{
+ Buffer metaBuffer;
+ Page metaPage;
+ GenericXLogState *state;
+
+ /*
+ * Make a new page; since it is first page it should be associated with
+ * block number 0 (STIR_METAPAGE_BLKNO). No need to hold the extension
+ * lock because there cannot be concurrent inserters yet.
+ */
+ metaBuffer = ReadBufferExtended(index, forknum, P_NEW, RBM_NORMAL, NULL);
+ LockBuffer(metaBuffer, BUFFER_LOCK_EXCLUSIVE);
+ Assert(BufferGetBlockNumber(metaBuffer) == STIR_METAPAGE_BLKNO);
+
+ /* Initialize contents of meta page */
+ state = GenericXLogStart(index);
+ metaPage = GenericXLogRegisterBuffer(state, metaBuffer,
+ GENERIC_XLOG_FULL_IMAGE);
+ StirFillMetapage(index, metaPage, forknum == INIT_FORKNUM);
+ GenericXLogFinish(state);
+
+ UnlockReleaseBuffer(metaBuffer);
+}
+
+/*
+ * Initialize any page of a stir index.
+ */
+void
+StirInitPage(Page page, uint16 flags)
+{
+ StirPageOpaque opaque;
+
+ PageInit(page, BLCKSZ, sizeof(StirPageOpaqueData));
+
+ opaque = StirPageGetOpaque(page);
+ opaque->flags = flags;
+ opaque->stir_page_id = STIR_PAGE_ID;
+}
+
+static bool
+StirPageAddItem(Page page, StirTuple *tuple)
+{
+ StirTuple *itup;
+ StirPageOpaque opaque;
+ Pointer ptr;
+
+ /* We shouldn't be pointed to an invalid page */
+ Assert(!PageIsNew(page));
+
+ /* Does new tuple fit on the page? */
+ if (StirPageGetFreeSpace(state, page) < sizeof(StirTuple))
+ return false;
+
+ /* Copy new tuple to the end of page */
+ opaque = StirPageGetOpaque(page);
+ itup = StirPageGetTuple(page, opaque->maxoff + 1);
+ memcpy((Pointer) itup, (Pointer) tuple, sizeof(StirTuple));
+
+ /* Adjust maxoff and pd_lower */
+ opaque->maxoff++;
+ ptr = (Pointer) StirPageGetTuple(page, opaque->maxoff + 1);
+ ((PageHeader) page)->pd_lower = ptr - page;
+
+ /* Assert we didn't overrun available space */
+ Assert(((PageHeader) page)->pd_lower <= ((PageHeader) page)->pd_upper);
+ return true;
+}
+
+bool
+stirinsert(Relation index, Datum *values, bool *isnull,
+ ItemPointer ht_ctid, Relation heapRel,
+ IndexUniqueCheck checkUnique,
+ bool indexUnchanged,
+ struct IndexInfo *indexInfo)
+{
+ StirTuple *itup;
+ MemoryContext oldCtx;
+ MemoryContext insertCtx;
+ StirMetaPageData *metaData;
+ Buffer buffer,
+ metaBuffer;
+ Page page;
+ GenericXLogState *state;
+ uint16 blkNo;
+
+ insertCtx = AllocSetContextCreate(CurrentMemoryContext,
+ "Stir insert temporary context",
+ ALLOCSET_DEFAULT_SIZES);
+
+ oldCtx = MemoryContextSwitchTo(insertCtx);
+
+ itup = (StirTuple *) palloc0(sizeof(StirTuple));
+ itup->heapPtr = *ht_ctid;
+
+ metaBuffer = ReadBuffer(index, STIR_METAPAGE_BLKNO);
+
+ for (;;)
+ {
+ LockBuffer(metaBuffer, BUFFER_LOCK_SHARE);
+ metaData = StirPageGetMeta(BufferGetPage(metaBuffer));
+ if (metaData->skipInserts)
+ {
+ UnlockReleaseBuffer(metaBuffer);
+ return false;
+ }
+ blkNo = metaData->lastBlkNo;
+ /* Don't hold metabuffer lock while doing insert */
+ LockBuffer(metaBuffer, BUFFER_LOCK_UNLOCK);
+
+ if (blkNo > 0)
+ {
+ buffer = ReadBuffer(index, blkNo);
+ LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+
+ state = GenericXLogStart(index);
+ page = GenericXLogRegisterBuffer(state, buffer, 0);
+
+ Assert(!PageIsNew(page));
+
+ if (StirPageAddItem(page, itup))
+ {
+ /* Success! Apply the change, clean up, and exit */
+ GenericXLogFinish(state);
+ UnlockReleaseBuffer(buffer);
+ ReleaseBuffer(metaBuffer);
+ MemoryContextSwitchTo(oldCtx);
+ MemoryContextDelete(insertCtx);
+ return false;
+ }
+
+ /* Didn't fit, must try other pages */
+ GenericXLogAbort(state);
+ UnlockReleaseBuffer(buffer);
+ }
+
+ LockBuffer(metaBuffer, BUFFER_LOCK_EXCLUSIVE);
+
+ state = GenericXLogStart(index);
+ metaData = StirPageGetMeta(GenericXLogRegisterBuffer(state, metaBuffer, GENERIC_XLOG_FULL_IMAGE));
+ if (blkNo != metaData->lastBlkNo)
+ {
+ Assert(blkNo < metaData->lastBlkNo);
+ // someone else inserted the new page into the index, lets try again
+ GenericXLogAbort(state);
+ LockBuffer(metaBuffer, BUFFER_LOCK_UNLOCK);
+ continue;
+ }
+ else
+ {
+ /* Must extend the file */
+ buffer = ExtendBufferedRel(BMR_REL(index), MAIN_FORKNUM, NULL,
+ EB_LOCK_FIRST);
+
+ page = GenericXLogRegisterBuffer(state, buffer, GENERIC_XLOG_FULL_IMAGE);
+ StirInitPage(page, 0);
+
+ if (!StirPageAddItem(page, itup))
+ {
+ /* We shouldn't be here since we're inserting to an empty page */
+ elog(ERROR, "could not add new stir tuple to empty page");
+ }
+ metaData->lastBlkNo = BufferGetBlockNumber(buffer);
+ GenericXLogFinish(state);
+
+ UnlockReleaseBuffer(buffer);
+ UnlockReleaseBuffer(metaBuffer);
+
+ MemoryContextSwitchTo(oldCtx);
+ MemoryContextDelete(insertCtx);
+
+ return false;
+ }
+ }
+}
+
+IndexScanDesc stirbeginscan(Relation r, int nkeys, int norderbys)
+{
+ ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("\"%s\" is not a not implemented", __func__)));
+}
+
+void
+stirrescan(IndexScanDesc scan, ScanKey scankey, int nscankeys,
+ ScanKey orderbys, int norderbys)
+{
+ ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("\"%s\" is not a not implemented", __func__)));
+}
+
+void stirendscan(IndexScanDesc scan)
+{
+ ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("\"%s\" is not a not implemented", __func__)));
+}
+
+IndexBuildResult *stirbuild(Relation heap, Relation index,
+ struct IndexInfo *indexInfo)
+{
+ IndexBuildResult *result;
+
+ StirInitMetapage(index, MAIN_FORKNUM);
+
+ result = (IndexBuildResult *) palloc(sizeof(IndexBuildResult));
+ result->heap_tuples = 0;
+ result->index_tuples = 0;
+ return result;
+}
+
+void stirbuildempty(Relation index)
+{
+ StirInitMetapage(index, INIT_FORKNUM);
+}
+
+IndexBulkDeleteResult *stirbulkdelete(IndexVacuumInfo *info,
+ IndexBulkDeleteResult *stats,
+ IndexBulkDeleteCallback callback,
+ void *callback_state)
+{
+ Relation index = info->index;
+ BlockNumber blkno, npages;
+ Buffer buffer;
+ Page page;
+
+ if (!info->validate_index)
+ {
+ StirMarkAsSkipInserts(index);
+
+ ereport(WARNING, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("\"%s\" is not a not implemented, seems like this index need to be dropped", __func__)));
+ return NULL;
+ }
+
+ if (stats == NULL)
+ stats = (IndexBulkDeleteResult *) palloc0(sizeof(IndexBulkDeleteResult));
+
+ /*
+ * Iterate over the pages. We don't care about concurrently added pages,
+ * because TODO
+ */
+ npages = RelationGetNumberOfBlocks(index);
+ for (blkno = STIR_HEAD_BLKNO; blkno < npages; blkno++)
+ {
+ StirTuple *itup, *itupEnd;
+
+ vacuum_delay_point();
+
+ buffer = ReadBufferExtended(index, MAIN_FORKNUM, blkno,
+ RBM_NORMAL, info->strategy);
+
+ LockBuffer(buffer, BUFFER_LOCK_SHARE);
+ page = BufferGetPage(buffer);
+
+ if (PageIsNew(page))
+ {
+ UnlockReleaseBuffer(buffer);
+ continue;
+ }
+
+ itup = StirPageGetTuple(page, FirstOffsetNumber);
+ itupEnd = StirPageGetTuple(page, OffsetNumberNext(StirPageGetMaxOffset(page)));
+ while (itup < itupEnd)
+ {
+ /* Do we have to delete this tuple? */
+ if (callback(&itup->heapPtr, callback_state))
+ {
+ ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("we never delete in stir")));
+ }
+
+ itup = StirPageGetNextTuple(itup);
+ }
+
+ UnlockReleaseBuffer(buffer);
+ }
+
+ return stats;
+}
+
+void StirMarkAsSkipInserts(Relation index)
+{
+ StirMetaPageData *metaData;
+ Buffer metaBuffer;
+ Page metaPage;
+ GenericXLogState *state;
+
+ metaBuffer = ReadBuffer(index, STIR_METAPAGE_BLKNO);
+ LockBuffer(metaBuffer, BUFFER_LOCK_EXCLUSIVE);
+
+ state = GenericXLogStart(index);
+ metaPage = GenericXLogRegisterBuffer(state, metaBuffer,
+ GENERIC_XLOG_FULL_IMAGE);
+ metaData = StirPageGetMeta(metaPage);
+ if (!metaData->skipInserts)
+ {
+ metaData->skipInserts = true;
+ GenericXLogFinish(state);
+ }
+ else
+ {
+ GenericXLogAbort(state);
+ }
+ UnlockReleaseBuffer(metaBuffer);
+}
+
+IndexBulkDeleteResult *stirvacuumcleanup(IndexVacuumInfo *info,
+ IndexBulkDeleteResult *stats)
+{
+ StirMarkAsSkipInserts(info->index);
+ ereport(WARNING, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("\"%s\" is not a not implemented, seems like this index need to be dropped", __func__)));
+ return NULL;
+}
+
+bytea *stiroptions(Datum reloptions, bool validate)
+{
+ return NULL;
+}
+
+void stircostestimate(PlannerInfo *root, IndexPath *path,
+ double loop_count, Cost *indexStartupCost,
+ Cost *indexTotalCost, Selectivity *indexSelectivity,
+ double *indexCorrelation, double *indexPages)
+{
+ ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("\"%s\" is not a not implemented", __func__)));
+}
Index: src/include/access/stir.h
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/include/access/stir.h b/src/include/access/stir.h
new file mode 100644
--- /dev/null (revision d8df9daea76374468c28f8e9d60d83539aad05c8)
+++ b/src/include/access/stir.h (revision d8df9daea76374468c28f8e9d60d83539aad05c8)
@@ -0,0 +1,117 @@
+/*-------------------------------------------------------------------------
+ *
+ * stir.h
+ * header file for postgres stir access method implementation.
+ *
+ *
+ * Portions Copyright (c) 2024-2024, PostgreSQL Global Development Group
+ *
+ * src/include/access/stir.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef _STIR_H_
+#define _STIR_H_
+
+#include "amapi.h"
+#include "xlog.h"
+#include "generic_xlog.h"
+#include "itup.h"
+#include "fmgr.h"
+#include "nodes/pathnodes.h"
+
+/* Support procedures numbers */
+#define STIR_NPROC 0
+
+/* Scan strategies */
+#define STIR_NSTRATEGIES 1
+
+#define STIR_OPTIONS_PROC 0
+
+/* Macros for accessing bloom page structures */
+#define StirPageGetOpaque(page) ((StirPageOpaque) PageGetSpecialPointer(page))
+#define StirPageGetMaxOffset(page) (StirPageGetOpaque(page)->maxoff)
+#define StirPageIsMeta(page) \
+ ((StirPageGetOpaque(page)->flags & BLOOM_META) != 0)
+#define StirPageGetData(page) ((StirTuple *)PageGetContents(page))
+#define StirPageGetTuple(page, offset) \
+ ((StirTuple *)(PageGetContents(page) \
+ + sizeof(StirTuple) * ((offset) - 1)))
+#define StirPageGetNextTuple(tuple) \
+ ((StirTuple *)((Pointer)(tuple) + sizeof(StirTuple)))
+
+
+
+/* Preserved page numbers */
+#define STIR_METAPAGE_BLKNO (0)
+#define STIR_HEAD_BLKNO (1) /* first data page */
+
+
+/* Opaque for stir pages */
+typedef struct StirPageOpaqueData
+{
+ OffsetNumber maxoff; /* number of index tuples on page */
+ uint16 flags; /* see bit definitions below */
+ uint16 unused; /* placeholder to force maxaligning of size of
+ * StirPageOpaqueData and to place
+ * stir_page_id exactly at the end of page */
+ uint16 stir_page_id; /* for identification of STIR indexes */
+} StirPageOpaqueData;
+
+/* Stir page flags */
+#define STIR_META (1<<0)
+
+typedef StirPageOpaqueData *StirPageOpaque;
+
+#define STIR_PAGE_ID 0xFF84
+
+/* Metadata of stir index */
+typedef struct StirMetaPageData
+{
+ uint32 magickNumber;
+ uint16 lastBlkNo;
+ bool skipInserts;
+} StirMetaPageData;
+
+/* Magic number to distinguish stir pages from others */
+#define STIR_MAGICK_NUMBER (0xDBAC0DEF)
+
+#define StirPageGetMeta(page) ((StirMetaPageData *) PageGetContents(page))
+
+typedef struct StirTuple
+{
+ ItemPointerData heapPtr;
+} StirTuple;
+
+#define StirPageGetFreeSpace(state, page) \
+ (BLCKSZ - MAXALIGN(SizeOfPageHeaderData) \
+ - StirPageGetMaxOffset(page) * (sizeof(StirTuple)) \
+ - MAXALIGN(sizeof(StirPageOpaqueData)))
+
+extern void StirFillMetapage(Relation index, Page metaPage, bool skipInserts);
+extern void StirInitMetapage(Relation index, ForkNumber forknum);
+extern void StirInitPage(Page page, uint16 flags);
+extern void StirMarkAsSkipInserts(Relation index);
+
+/* index access method interface functions */
+extern bool stirvalidate(Oid opclassoid);
+extern bool stirinsert(Relation index, Datum *values, bool *isnull,
+ ItemPointer ht_ctid, Relation heapRel,
+ IndexUniqueCheck checkUnique,
+ bool indexUnchanged,
+ struct IndexInfo *indexInfo);
+extern IndexScanDesc stirbeginscan(Relation r, int nkeys, int norderbys);
+extern void stirrescan(IndexScanDesc scan, ScanKey scankey, int nscankeys,
+ ScanKey orderbys, int norderbys);
+extern void stirendscan(IndexScanDesc scan);
+extern IndexBuildResult *stirbuild(Relation heap, Relation index,
+ struct IndexInfo *indexInfo);
+extern void stirbuildempty(Relation index);
+extern IndexBulkDeleteResult *stirbulkdelete(IndexVacuumInfo *info,
+ IndexBulkDeleteResult *stats, IndexBulkDeleteCallback callback,
+ void *callback_state);
+extern IndexBulkDeleteResult *stirvacuumcleanup(IndexVacuumInfo *info,
+ IndexBulkDeleteResult *stats);
+extern bytea *stiroptions(Datum reloptions, bool validate);
+
+#endif
Index: src/backend/utils/sort/tuplesortvariants.c
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/utils/sort/tuplesortvariants.c b/src/backend/utils/sort/tuplesortvariants.c
--- a/src/backend/utils/sort/tuplesortvariants.c (revision 35f233300cd190b0a17e66f2b4bffa2481e62af9)
+++ b/src/backend/utils/sort/tuplesortvariants.c (revision bc1fe05f38fbdda049075b9b1dc238bf0d9c240e)
@@ -123,6 +123,7 @@
bool enforceUnique; /* complain if we find duplicate tuples */
bool uniqueNullsNotDistinct; /* unique constraint null treatment */
+ bool uniqueDeadIgnored;
} TuplesortIndexBTreeArg;
/*
@@ -349,6 +350,7 @@
Relation indexRel,
bool enforceUnique,
bool uniqueNullsNotDistinct,
+ bool uniqueDeadIgnored,
int workMem,
SortCoordinate coordinate,
int sortopt)
@@ -391,6 +393,7 @@
arg->index.indexRel = indexRel;
arg->enforceUnique = enforceUnique;
arg->uniqueNullsNotDistinct = uniqueNullsNotDistinct;
+ arg->uniqueDeadIgnored = uniqueDeadIgnored;
indexScanKey = _bt_mkscankey(indexRel, NULL);
@@ -514,6 +517,7 @@
arg->index.indexRel = indexRel;
arg->enforceUnique = false;
arg->uniqueNullsNotDistinct = false;
+ arg->uniqueDeadIgnored = false;
/* Prepare SortSupport data for each column */
base->sortKeys = (SortSupport) palloc0(base->nKeys *
@@ -1520,6 +1524,7 @@
Datum values[INDEX_MAX_KEYS];
bool isnull[INDEX_MAX_KEYS];
char *key_desc;
+ bool uniqueCheckFail = true;
/*
* Some rather brain-dead implementations of qsort (such as the one in
@@ -1529,18 +1534,56 @@
*/
Assert(tuple1 != tuple2);
- index_deform_tuple(tuple1, tupDes, values, isnull);
+ /* This is fail-fast check, see _bt_load for details. */
+ if (arg->uniqueDeadIgnored)
+ {
+ bool any_tuple_dead,
+ call_again = false,
+ ignored;
+
+ TupleTableSlot *slot = MakeSingleTupleTableSlot(RelationGetDescr(arg->index.heapRel),
+ &TTSOpsBufferHeapTuple);
+ ItemPointerData tid = tuple1->t_tid;
+
+ IndexFetchTableData *fetch = table_index_fetch_begin(arg->index.heapRel);
+ any_tuple_dead = !table_index_fetch_tuple(fetch, &tid, SnapshotSelf, slot, &call_again, &ignored);
+
+ if (!any_tuple_dead)
+ {
+ call_again = false;
+ tid = tuple2->t_tid;
+ any_tuple_dead = !table_index_fetch_tuple(fetch, &tuple2->t_tid, SnapshotSelf, slot, &call_again,
+ &ignored);
+ }
+
+ if (any_tuple_dead)
+ {
+ elog(DEBUG5, "skipping duplicate values because some of them are dead: (%u,%u) vs (%u,%u)",
+ ItemPointerGetBlockNumber(&tuple1->t_tid),
+ ItemPointerGetOffsetNumber(&tuple1->t_tid),
+ ItemPointerGetBlockNumber(&tuple2->t_tid),
+ ItemPointerGetOffsetNumber(&tuple2->t_tid));
+
+ uniqueCheckFail = false;
+ }
+ ExecDropSingleTupleTableSlot(slot);
+ table_index_fetch_end(fetch);
+ }
+ if (uniqueCheckFail)
+ {
+ index_deform_tuple(tuple1, tupDes, values, isnull);
- key_desc = BuildIndexValueDescription(arg->index.indexRel, values, isnull);
+ key_desc = BuildIndexValueDescription(arg->index.indexRel, values, isnull);
- ereport(ERROR,
- (errcode(ERRCODE_UNIQUE_VIOLATION),
- errmsg("could not create unique index \"%s\"",
- RelationGetRelationName(arg->index.indexRel)),
- key_desc ? errdetail("Key %s is duplicated.", key_desc) :
- errdetail("Duplicate keys exist."),
- errtableconstraint(arg->index.heapRel,
- RelationGetRelationName(arg->index.indexRel))));
+ ereport(ERROR,
+ (errcode(ERRCODE_UNIQUE_VIOLATION),
+ errmsg("could not create unique index \"%s\"",
+ RelationGetRelationName(arg->index.indexRel)),
+ key_desc ? errdetail("Key %s is duplicated.", key_desc) :
+ errdetail("Duplicate keys exist."),
+ errtableconstraint(arg->index.heapRel,
+ RelationGetRelationName(arg->index.indexRel))));
+ }
}
/*
Index: src/bin/pg_amcheck/t/007_concurrently_unique.pl
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/bin/pg_amcheck/t/007_concurrently_unique.pl b/src/bin/pg_amcheck/t/007_concurrently_unique.pl
new file mode 100644
--- /dev/null (revision ea1fcacc7cead3e2fccf581d20e51244a7107435)
+++ b/src/bin/pg_amcheck/t/007_concurrently_unique.pl (revision ea1fcacc7cead3e2fccf581d20e51244a7107435)
@@ -0,0 +1,235 @@
+
+# Copyright (c) 2024, PostgreSQL Global Development Group
+
+# Test REINDEX CONCURRENTLY with concurrent modifications and HOT updates
+use strict;
+use warnings;
+
+use Config;
+use Errno;
+
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Time::HiRes qw(usleep);
+use IPC::SysV;
+use threads;
+use Test::More;
+use Test::Builder;
+
+if ($@ || $windows_os)
+{
+ plan skip_all => 'Fork and shared memory are not supported by this platform';
+}
+
+# TODO: refactor to https://metacpan.org/pod/IPC%3A%3AShareable
+my ($pid, $shmem_id, $shmem_key, $shmem_size);
+eval 'sub IPC_CREAT {0001000}' unless defined &IPC_CREAT;
+$shmem_size = 4;
+$shmem_key = rand(1000000);
+$shmem_id = shmget($shmem_key, $shmem_size, &IPC_CREAT | 0777) or die "Can't shmget: $!";
+shmwrite($shmem_id, "wait", 0, $shmem_size) or die "Can't shmwrite: $!";
+
+my $psql_timeout = IPC::Run::timer($PostgreSQL::Test::Utils::timeout_default);
+#
+# Test set-up
+#
+my ($node, $result);
+$node = PostgreSQL::Test::Cluster->new('RC_test');
+$node->init;
+$node->append_conf('postgresql.conf',
+ 'lock_timeout = ' . (1000 * $PostgreSQL::Test::Utils::timeout_default));
+$node->append_conf('postgresql.conf', 'fsync = off');
+$node->append_conf('postgresql.conf', 'autovacuum = off');
+$node->append_conf('postgresql.conf', 'maintenance_work_mem = 128MB');
+$node->append_conf('postgresql.conf', 'shared_buffers = 256MB');
+$node->start;
+$node->safe_psql('postgres', q(CREATE EXTENSION amcheck));
+$node->safe_psql('postgres', q(CREATE UNLOGGED TABLE tbl(i int primary key,
+ c1 money default 0, c2 money default 0,
+ c3 money default 0, updated_at timestamp)));
+$node->safe_psql('postgres', q(CREATE INDEX idx ON tbl(i, updated_at)));
+
+my $builder = Test::More->builder;
+$builder->use_numbers(0);
+$builder->no_plan();
+
+my $child = $builder->child("pg_bench");
+
+if(!defined($pid = fork())) {
+ # fork returned undef, so unsuccessful
+ die "Cannot fork a child: $!";
+} elsif ($pid == 0) {
+
+ # $node->psql('postgres', q(INSERT INTO tbl SELECT i,0,0,0,now() FROM generate_series(1, 1000) s(i);));
+ # while [ $? -eq 0 ]; do make -C src/bin/pg_amcheck/ check PROVE_TESTS='t/007_*' ; done
+
+ $node->pgbench(
+ '--no-vacuum --client=40 --exit-on-abort --transactions=10000',
+ 0,
+ [qr{actually processed}],
+ [qr{^$}],
+ 'concurrent INSERTs, UPDATES and RC',
+ {
+ # Ensure some HOT updates happen
+ '001_pgbench_concurrent_transaction_updates' => q(
+ INSERT INTO tbl VALUES(random()*1000,0,0,0,now()) on conflict(i) do update set updated_at = date_trunc('seconds', now());
+ ),
+ '002_pgbench_concurrent_transaction_updates' => q(
+ INSERT INTO tbl VALUES(random()*100,0,0,0,now()) on conflict(i) do update set updated_at = date_trunc('seconds', now());
+ ),
+ '003_pgbench_concurrent_transaction_updates' => q(
+ INSERT INTO tbl VALUES(random()*10000,0,0,0,now()) on conflict(i) do update set updated_at = date_trunc('seconds', now());
+ ),
+ '004_pgbench_concurrent_transaction_updates' => q(
+ INSERT INTO tbl VALUES(random()*100000,0,0,0,now()) on conflict(i) do update set updated_at = date_trunc('seconds', now());
+ ),
+ });
+
+ if ($child->is_passing()) {
+ shmwrite($shmem_id, "done", 0, $shmem_size) or die "Can't shmwrite: $!";
+ } else {
+ shmwrite($shmem_id, "fail", 0, $shmem_size) or die "Can't shmwrite: $!";
+ }
+
+ my $pg_bench_fork_flag;
+ while (1) {
+ shmread($shmem_id, $pg_bench_fork_flag, 0, $shmem_size) or die "Can't shmread: $!";
+ sleep(0.1);
+ last if $pg_bench_fork_flag eq "stop";
+ }
+} else {
+ my $pg_bench_fork_flag;
+ shmread($shmem_id, $pg_bench_fork_flag, 0, $shmem_size) or die "Can't shmread: $!";
+
+ subtest 'reindex run subtest' => sub {
+ is($pg_bench_fork_flag, "wait", "pg_bench_fork_flag is correct");
+
+ my %psql = (stdin => '', stdout => '', stderr => '');
+ $psql{run} = IPC::Run::start(
+ [ 'psql', '-XA', '-f', '-', '-d', $node->connstr('postgres') ],
+ '<',
+ \$psql{stdin},
+ '>',
+ \$psql{stdout},
+ '2>',
+ \$psql{stderr},
+ $psql_timeout);
+
+ my ($result, $stdout, $stderr, $n, $stderr_saved);
+
+# ok(send_query_and_wait(\%psql, q[SELECT pg_sleep(10);], qr/^.*$/m), 'SELECT');
+
+ while (1)
+ {
+
+ if (int(rand(2)) == 0) {
+ ($result, $stdout, $stderr) = $node->psql('postgres', q(ALTER TABLE tbl SET (parallel_workers=4);));
+ } else {
+ ($result, $stdout, $stderr) = $node->psql('postgres', q(ALTER TABLE tbl SET (parallel_workers=0);));
+ }
+ is($result, '0', 'ALTER TABLE is correct');
+
+
+ if (1)
+ {
+ my $sql = q(select pg_sleep(0); CREATE UNIQUE INDEX CONCURRENTLY idx_2 ON tbl(i););
+
+ ($result, $stdout, $stderr) = $node->psql('postgres', $sql);
+ is($result, '0', 'CREATE INDEX is correct');
+ $stderr_saved = $stderr;
+
+ ($result, $stdout, $stderr) = $node->psql('postgres', q(SELECT bt_index_parent_check('idx_2', heapallindexed => true, rootdescend => true, checkunique => true);));
+ is($result, '0', 'bt_index_check for new index is correct');
+ if ($result)
+ {
+ diag($stderr);
+ diag($stderr_saved);
+ BAIL_OUT($stderr);
+ } else {
+ diag('create:)' . $n++);
+ }
+
+ if (1)
+ {
+ ($result, $stdout, $stderr) = $node->psql('postgres', q(REINDEX INDEX CONCURRENTLY idx_2;));
+ is($result, '0', 'REINDEX 2 is correct');
+ if ($result) {
+ diag($stderr);
+ BAIL_OUT($stderr);
+ }
+
+ ($result, $stdout, $stderr) = $node->psql('postgres', q(SELECT bt_index_parent_check('idx_2', heapallindexed => true, rootdescend => true, checkunique => true);));
+ is($result, '0', 'bt_index_check 2 is correct');
+ if ($result)
+ {
+ diag($stderr);
+ BAIL_OUT($stderr);
+ } else {
+ diag('reindex2:)' . $n++);
+ }
+ }
+
+ ($result, $stdout, $stderr) = $node->psql('postgres', q(DROP INDEX CONCURRENTLY idx_2;));
+ is($result, '0', 'DROP INDEX is correct');
+ }
+ shmread($shmem_id, $pg_bench_fork_flag, 0, $shmem_size) or die "Can't shmread: $!";
+ last if $pg_bench_fork_flag ne "wait";
+ }
+
+ # explicitly shut down psql instances gracefully
+ $psql{stdin} .= "\\q\n";
+ $psql{run}->finish;
+
+ is($pg_bench_fork_flag, "done", "pg_bench_fork_flag is correct");
+ };
+
+ $child->finalize();
+ $child->summary();
+ $node->stop;
+ done_testing();
+
+ shmwrite($shmem_id, "stop", 0, $shmem_size) or die "Can't shmwrite: $!";
+}
+
+# Send query, wait until string matches
+sub send_query_and_wait
+{
+ my ($psql, $query, $untl) = @_;
+ my $ret;
+
+ # For each query we run, we'll restart the timeout. Otherwise the timeout
+ # would apply to the whole test script, and would need to be set very high
+ # to survive when running under Valgrind.
+ $psql_timeout->reset();
+ $psql_timeout->start();
+
+ # send query
+ $$psql{stdin} .= $query;
+ $$psql{stdin} .= "\n";
+
+ # wait for query results
+ $$psql{run}->pump_nb();
+ while (1)
+ {
+ last if $$psql{stdout} =~ /$untl/;
+ if ($psql_timeout->is_expired)
+ {
+ diag("aborting wait: program timed out\n"
+ . "stream contents: >>$$psql{stdout}<<\n"
+ . "pattern searched for: $untl\n");
+ return 0;
+ }
+ if (not $$psql{run}->pumpable())
+ {
+ diag("aborting wait: program died\n"
+ . "stream contents: >>$$psql{stdout}<<\n"
+ . "pattern searched for: $untl\n");
+ return 0;
+ }
+ $$psql{run}->pump();
+ }
+
+ $$psql{stdout} = '';
+
+ return 1;
+}
Index: src/include/access/transam.h
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/include/access/transam.h b/src/include/access/transam.h
--- a/src/include/access/transam.h (revision 35f233300cd190b0a17e66f2b4bffa2481e62af9)
+++ b/src/include/access/transam.h (revision 3a0fa65e328d51b6c97b44a72778b6ee21fe4478)
@@ -344,6 +344,21 @@
return b;
}
+/* return the newer of the two IDs */
+static inline TransactionId
+TransactionIdNewer(TransactionId a, TransactionId b)
+{
+ if (!TransactionIdIsValid(a))
+ return b;
+
+ if (!TransactionIdIsValid(b))
+ return a;
+
+ if (TransactionIdFollows(a, b))
+ return a;
+ return b;
+}
+
/* return the older of the two IDs, assuming they're both normal */
static inline TransactionId
NormalTransactionIdOlder(TransactionId a, TransactionId b)
Index: src/include/utils/tuplesort.h
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/include/utils/tuplesort.h b/src/include/utils/tuplesort.h
--- a/src/include/utils/tuplesort.h (revision 35f233300cd190b0a17e66f2b4bffa2481e62af9)
+++ b/src/include/utils/tuplesort.h (revision 3a0fa65e328d51b6c97b44a72778b6ee21fe4478)
@@ -428,6 +428,7 @@
Relation indexRel,
bool enforceUnique,
bool uniqueNullsNotDistinct,
+ bool uniqueDeadIgnored,
int workMem, SortCoordinate coordinate,
int sortopt);
extern Tuplesortstate *tuplesort_begin_index_hash(Relation heapRel,
Index: src/backend/access/nbtree/nbtutils.c
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/access/nbtree/nbtutils.c b/src/backend/access/nbtree/nbtutils.c
--- a/src/backend/access/nbtree/nbtutils.c (revision 3a0fa65e328d51b6c97b44a72778b6ee21fe4478)
+++ b/src/backend/access/nbtree/nbtutils.c (revision bc1fe05f38fbdda049075b9b1dc238bf0d9c240e)
@@ -100,8 +100,6 @@
ScanDirection dir, bool *continuescan);
static void _bt_checkkeys_look_ahead(IndexScanDesc scan, BTReadPageState *pstate,
int tupnatts, TupleDesc tupdesc);
-static int _bt_keep_natts(Relation rel, IndexTuple lastleft,
- IndexTuple firstright, BTScanInsert itup_key);
/*
@@ -4775,6 +4773,14 @@
return tidpivot;
}
+int
+_bt_keep_natts(Relation rel, IndexTuple lastleft, IndexTuple firstright,
+ BTScanInsert itup_key) {
+ bool ignored;
+ return _bt_keep_natts_wasnull(rel, lastleft, firstright, itup_key, &ignored);
+}
+
+
/*
* _bt_keep_natts - how many key attributes to keep when truncating.
*
@@ -4786,9 +4792,10 @@
* number of key attributes for the index relation. This indicates that the
* caller must use a heap TID as a unique-ifier in new pivot tuple.
*/
-static int
-_bt_keep_natts(Relation rel, IndexTuple lastleft, IndexTuple firstright,
- BTScanInsert itup_key)
+int
+_bt_keep_natts_wasnull(Relation rel, IndexTuple lastleft, IndexTuple firstright,
+ BTScanInsert itup_key,
+ bool *wasnull)
{
int nkeyatts = IndexRelationGetNumberOfKeyAttributes(rel);
TupleDesc itupdesc = RelationGetDescr(rel);
@@ -4814,6 +4821,7 @@
datum1 = index_getattr(lastleft, attnum, itupdesc, &isNull1);
datum2 = index_getattr(firstright, attnum, itupdesc, &isNull2);
+ (*wasnull) |= (isNull1 || isNull2);
if (isNull1 != isNull2)
break;
@@ -4838,6 +4846,13 @@
return keepnatts;
}
+int
+_bt_keep_natts_fast(Relation rel, IndexTuple lastleft, IndexTuple firstright)
+{
+ bool ignored;
+ return _bt_keep_natts_fast_wasnull(rel, lastleft, firstright, &ignored);
+}
+
/*
* _bt_keep_natts_fast - fast bitwise variant of _bt_keep_natts.
*
@@ -4861,7 +4876,8 @@
* more balanced split point.
*/
int
-_bt_keep_natts_fast(Relation rel, IndexTuple lastleft, IndexTuple firstright)
+_bt_keep_natts_fast_wasnull(Relation rel, IndexTuple lastleft, IndexTuple firstright,
+ bool *wasnull)
{
TupleDesc itupdesc = RelationGetDescr(rel);
int keysz = IndexRelationGetNumberOfKeyAttributes(rel);
@@ -4878,6 +4894,7 @@
datum1 = index_getattr(lastleft, attnum, itupdesc, &isNull1);
datum2 = index_getattr(firstright, attnum, itupdesc, &isNull2);
+ *wasnull |= (isNull1 | isNull2);
att = TupleDescAttr(itupdesc, attnum - 1);
if (isNull1 != isNull2)
Index: src/include/access/nbtree.h
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/include/access/nbtree.h b/src/include/access/nbtree.h
--- a/src/include/access/nbtree.h (revision 3a0fa65e328d51b6c97b44a72778b6ee21fe4478)
+++ b/src/include/access/nbtree.h (revision bc1fe05f38fbdda049075b9b1dc238bf0d9c240e)
@@ -1302,8 +1302,15 @@
extern char *btbuildphasename(int64 phasenum);
extern IndexTuple _bt_truncate(Relation rel, IndexTuple lastleft,
IndexTuple firstright, BTScanInsert itup_key);
+extern int _bt_keep_natts(Relation rel, IndexTuple lastleft,
+ IndexTuple firstright, BTScanInsert itup_key);
+extern int _bt_keep_natts_wasnull(Relation rel, IndexTuple lastleft,
+ IndexTuple firstright, BTScanInsert itup_key,
+ bool *wasnull);
extern int _bt_keep_natts_fast(Relation rel, IndexTuple lastleft,
IndexTuple firstright);
+extern int _bt_keep_natts_fast_wasnull(Relation rel, IndexTuple lastleft,
+ IndexTuple firstright, bool *wasnull);
extern bool _bt_check_natts(Relation rel, bool heapkeyspace, Page page,
OffsetNumber offnum);
extern void _bt_check_third_page(Relation rel, Relation heap,
Index: src/backend/optimizer/util/plancat.c
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/optimizer/util/plancat.c b/src/backend/optimizer/util/plancat.c
--- a/src/backend/optimizer/util/plancat.c (revision bc1fe05f38fbdda049075b9b1dc238bf0d9c240e)
+++ b/src/backend/optimizer/util/plancat.c (revision 94aa5d7dab7e8ebd77004b50ba96b1f82a04c249)
@@ -720,6 +720,7 @@
/* Results */
List *results = NIL;
+ bool foundValid = false;
/*
* Quickly return NIL for ON CONFLICT DO NOTHING without an inference
@@ -813,7 +814,13 @@
idxRel = index_open(indexoid, rte->rellockmode);
idxForm = idxRel->rd_index;
- if (!idxForm->indisvalid)
+ /*
+ * We need to consider both indisvalid and indisready indexes because
+ * them may become indisvalid before execution phase. It is required
+ * to keep set of indexes used as arbiter to be the same for all
+ * concurrent transactions.
+ */
+ if (!idxForm->indisready)
goto next;
/*
@@ -835,10 +842,9 @@
errmsg("ON CONFLICT DO UPDATE not supported with exclusion constraints")));
results = lappend_oid(results, idxForm->indexrelid);
- list_free(indexList);
+ foundValid |= idxForm->indisvalid;
index_close(idxRel, NoLock);
- table_close(relation, NoLock);
- return results;
+ break;
}
else if (indexOidFromConstraint != InvalidOid)
{
@@ -932,6 +938,7 @@
goto next;
results = lappend_oid(results, idxForm->indexrelid);
+ foundValid |= idxForm->indisvalid;
next:
index_close(idxRel, NoLock);
}
@@ -939,7 +946,8 @@
list_free(indexList);
table_close(relation, NoLock);
- if (results == NIL)
+ /* It is required to have at least one indisvalid index during the planning. */
+ if (results == NIL || !foundValid)
ereport(ERROR,
(errcode(ERRCODE_INVALID_COLUMN_REFERENCE),
errmsg("there is no unique or exclusion constraint matching the ON CONFLICT specification")));
Index: src/backend/access/index/genam.c
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/backend/access/index/genam.c b/src/backend/access/index/genam.c
--- a/src/backend/access/index/genam.c (revision 94aa5d7dab7e8ebd77004b50ba96b1f82a04c249)
+++ b/src/backend/access/index/genam.c (revision ea1fcacc7cead3e2fccf581d20e51244a7107435)
@@ -454,7 +454,7 @@
*/
sysscan->scan = table_beginscan_strat(heapRelation, snapshot,
nkeys, key,
- true, false);
+ true, false, false);
sysscan->iscan = NULL;
}
Index: src/test/modules/injection_points/Makefile
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/test/modules/injection_points/Makefile b/src/test/modules/injection_points/Makefile
--- a/src/test/modules/injection_points/Makefile (revision 56c9d3f4842baa53d7ab13d0764eae7f305aba0f)
+++ b/src/test/modules/injection_points/Makefile (revision 3dea72b62adc8806917dc459b82ff44d962bcb12)
@@ -13,7 +13,8 @@
REGRESS = injection_points
REGRESS_OPTS = --dlpath=$(top_builddir)/src/test/regress
-ISOLATION = inplace
+ISOLATION = inplace \
+ reset_snapshots
TAP_TESTS = 1
Index: src/test/modules/injection_points/expected/reset_snapshots.out
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/test/modules/injection_points/expected/reset_snapshots.out b/src/test/modules/injection_points/expected/reset_snapshots.out
new file mode 100644
--- /dev/null (revision 3dea72b62adc8806917dc459b82ff44d962bcb12)
+++ b/src/test/modules/injection_points/expected/reset_snapshots.out (revision 3dea72b62adc8806917dc459b82ff44d962bcb12)
@@ -0,0 +1,318 @@
+unused step name: sleep
+Parsed test spec with 2 sessions
+
+starting permutation: set_parallel_workers_1 create_index_concurrently_simple reindex_index_concurrently drop_index detach
+injection_points_attach
+-----------------------
+
+(1 row)
+
+step set_parallel_workers_1: ALTER TABLE test.tbl SET (parallel_workers=0);
+test: NOTICE: notice triggered for injection point heap_reset_scan_snapshot_effective
+test: NOTICE: notice triggered for injection point heapam_index_validate_scan_no_xid
+step create_index_concurrently_simple: CREATE INDEX CONCURRENTLY idx ON test.tbl(i, j);
+test: NOTICE: notice triggered for injection point heap_reset_scan_snapshot_effective
+test: NOTICE: notice triggered for injection point heapam_index_validate_scan_no_xid
+step reindex_index_concurrently: REINDEX INDEX CONCURRENTLY test.idx;
+step drop_index: DROP INDEX CONCURRENTLY test.idx;
+step detach:
+ SELECT injection_points_detach('heapam_index_validate_scan_no_xid');
+ SELECT injection_points_detach('heap_reset_scan_snapshot_effective');
+ SELECT injection_points_detach('_bt_leader_participate_as_worker');
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+
+starting permutation: set_parallel_workers_1 create_unique_index_concurrently_simple reindex_index_concurrently drop_index detach
+injection_points_attach
+-----------------------
+
+(1 row)
+
+step set_parallel_workers_1: ALTER TABLE test.tbl SET (parallel_workers=0);
+test: NOTICE: notice triggered for injection point heap_reset_scan_snapshot_effective
+test: NOTICE: notice triggered for injection point heapam_index_validate_scan_no_xid
+step create_unique_index_concurrently_simple: CREATE UNIQUE INDEX CONCURRENTLY idx ON test.tbl(i);
+test: NOTICE: notice triggered for injection point heap_reset_scan_snapshot_effective
+test: NOTICE: notice triggered for injection point heapam_index_validate_scan_no_xid
+step reindex_index_concurrently: REINDEX INDEX CONCURRENTLY test.idx;
+step drop_index: DROP INDEX CONCURRENTLY test.idx;
+step detach:
+ SELECT injection_points_detach('heapam_index_validate_scan_no_xid');
+ SELECT injection_points_detach('heap_reset_scan_snapshot_effective');
+ SELECT injection_points_detach('_bt_leader_participate_as_worker');
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+
+starting permutation: set_parallel_workers_1 create_index_concurrently_predicate_expression_mod reindex_index_concurrently drop_index detach
+injection_points_attach
+-----------------------
+
+(1 row)
+
+step set_parallel_workers_1: ALTER TABLE test.tbl SET (parallel_workers=0);
+test: NOTICE: notice triggered for injection point heap_reset_scan_snapshot_effective
+test: NOTICE: notice triggered for injection point heapam_index_validate_scan_no_xid
+step create_index_concurrently_predicate_expression_mod: CREATE INDEX CONCURRENTLY idx ON test.tbl(MOD(i, 2), j) WHERE MOD(i, 2) = 0;
+test: NOTICE: notice triggered for injection point heap_reset_scan_snapshot_effective
+test: NOTICE: notice triggered for injection point heapam_index_validate_scan_no_xid
+step reindex_index_concurrently: REINDEX INDEX CONCURRENTLY test.idx;
+step drop_index: DROP INDEX CONCURRENTLY test.idx;
+step detach:
+ SELECT injection_points_detach('heapam_index_validate_scan_no_xid');
+ SELECT injection_points_detach('heap_reset_scan_snapshot_effective');
+ SELECT injection_points_detach('_bt_leader_participate_as_worker');
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+
+starting permutation: set_parallel_workers_1 create_index_concurrently_predicate_set_xid_no_param reindex_index_concurrently drop_index detach
+injection_points_attach
+-----------------------
+
+(1 row)
+
+step set_parallel_workers_1: ALTER TABLE test.tbl SET (parallel_workers=0);
+step create_index_concurrently_predicate_set_xid_no_param: CREATE INDEX CONCURRENTLY idx ON test.tbl(i, j) WHERE test.predicate_stable_no_param();
+step reindex_index_concurrently: REINDEX INDEX CONCURRENTLY test.idx;
+step drop_index: DROP INDEX CONCURRENTLY test.idx;
+step detach:
+ SELECT injection_points_detach('heapam_index_validate_scan_no_xid');
+ SELECT injection_points_detach('heap_reset_scan_snapshot_effective');
+ SELECT injection_points_detach('_bt_leader_participate_as_worker');
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+
+starting permutation: set_parallel_workers_1 create_index_concurrently_predicate_set_xid reindex_index_concurrently drop_index detach
+injection_points_attach
+-----------------------
+
+(1 row)
+
+step set_parallel_workers_1: ALTER TABLE test.tbl SET (parallel_workers=0);
+test: NOTICE: notice triggered for injection point heapam_index_validate_scan_no_xid
+step create_index_concurrently_predicate_set_xid: CREATE INDEX CONCURRENTLY idx ON test.tbl(i, j) WHERE test.predicate_stable(i);
+test: NOTICE: notice triggered for injection point heapam_index_validate_scan_no_xid
+step reindex_index_concurrently: REINDEX INDEX CONCURRENTLY test.idx;
+step drop_index: DROP INDEX CONCURRENTLY test.idx;
+step detach:
+ SELECT injection_points_detach('heapam_index_validate_scan_no_xid');
+ SELECT injection_points_detach('heap_reset_scan_snapshot_effective');
+ SELECT injection_points_detach('_bt_leader_participate_as_worker');
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+
+starting permutation: set_parallel_workers_2 create_index_concurrently_simple wakeup reindex_index_concurrently wakeup drop_index detach
+injection_points_attach
+-----------------------
+
+(1 row)
+
+step set_parallel_workers_2: ALTER TABLE test.tbl SET (parallel_workers=2);
+test: NOTICE: notice triggered for injection point heap_reset_scan_snapshot_effective
+step create_index_concurrently_simple: CREATE INDEX CONCURRENTLY idx ON test.tbl(i, j); <waiting ...>
+step wakeup: SELECT injection_points_wakeup('_bt_leader_participate_as_worker');
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+test: NOTICE: notice triggered for injection point heapam_index_validate_scan_no_xid
+step create_index_concurrently_simple: <... completed>
+test: NOTICE: notice triggered for injection point heap_reset_scan_snapshot_effective
+step reindex_index_concurrently: REINDEX INDEX CONCURRENTLY test.idx; <waiting ...>
+step wakeup: SELECT injection_points_wakeup('_bt_leader_participate_as_worker');
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+test: NOTICE: notice triggered for injection point heapam_index_validate_scan_no_xid
+step reindex_index_concurrently: <... completed>
+step drop_index: DROP INDEX CONCURRENTLY test.idx;
+step detach:
+ SELECT injection_points_detach('heapam_index_validate_scan_no_xid');
+ SELECT injection_points_detach('heap_reset_scan_snapshot_effective');
+ SELECT injection_points_detach('_bt_leader_participate_as_worker');
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+
+starting permutation: set_parallel_workers_2 create_unique_index_concurrently_simple wakeup reindex_index_concurrently wakeup drop_index detach
+injection_points_attach
+-----------------------
+
+(1 row)
+
+step set_parallel_workers_2: ALTER TABLE test.tbl SET (parallel_workers=2);
+test: NOTICE: notice triggered for injection point heap_reset_scan_snapshot_effective
+step create_unique_index_concurrently_simple: CREATE UNIQUE INDEX CONCURRENTLY idx ON test.tbl(i); <waiting ...>
+step wakeup: SELECT injection_points_wakeup('_bt_leader_participate_as_worker');
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+test: NOTICE: notice triggered for injection point heapam_index_validate_scan_no_xid
+step create_unique_index_concurrently_simple: <... completed>
+test: NOTICE: notice triggered for injection point heap_reset_scan_snapshot_effective
+step reindex_index_concurrently: REINDEX INDEX CONCURRENTLY test.idx; <waiting ...>
+step wakeup: SELECT injection_points_wakeup('_bt_leader_participate_as_worker');
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+test: NOTICE: notice triggered for injection point heapam_index_validate_scan_no_xid
+step reindex_index_concurrently: <... completed>
+step drop_index: DROP INDEX CONCURRENTLY test.idx;
+step detach:
+ SELECT injection_points_detach('heapam_index_validate_scan_no_xid');
+ SELECT injection_points_detach('heap_reset_scan_snapshot_effective');
+ SELECT injection_points_detach('_bt_leader_participate_as_worker');
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+
+starting permutation: set_parallel_workers_2 create_index_concurrently_predicate_expression_mod wakeup reindex_index_concurrently wakeup drop_index detach
+injection_points_attach
+-----------------------
+
+(1 row)
+
+step set_parallel_workers_2: ALTER TABLE test.tbl SET (parallel_workers=2);
+test: NOTICE: notice triggered for injection point heap_reset_scan_snapshot_effective
+step create_index_concurrently_predicate_expression_mod: CREATE INDEX CONCURRENTLY idx ON test.tbl(MOD(i, 2), j) WHERE MOD(i, 2) = 0; <waiting ...>
+step wakeup: SELECT injection_points_wakeup('_bt_leader_participate_as_worker');
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+test: NOTICE: notice triggered for injection point heapam_index_validate_scan_no_xid
+step create_index_concurrently_predicate_expression_mod: <... completed>
+test: NOTICE: notice triggered for injection point heap_reset_scan_snapshot_effective
+step reindex_index_concurrently: REINDEX INDEX CONCURRENTLY test.idx; <waiting ...>
+step wakeup: SELECT injection_points_wakeup('_bt_leader_participate_as_worker');
+injection_points_wakeup
+-----------------------
+
+(1 row)
+
+test: NOTICE: notice triggered for injection point heapam_index_validate_scan_no_xid
+step reindex_index_concurrently: <... completed>
+step drop_index: DROP INDEX CONCURRENTLY test.idx;
+step detach:
+ SELECT injection_points_detach('heapam_index_validate_scan_no_xid');
+ SELECT injection_points_detach('heap_reset_scan_snapshot_effective');
+ SELECT injection_points_detach('_bt_leader_participate_as_worker');
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
+injection_points_detach
+-----------------------
+
+(1 row)
+
Index: src/test/modules/injection_points/meson.build
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/test/modules/injection_points/meson.build b/src/test/modules/injection_points/meson.build
--- a/src/test/modules/injection_points/meson.build (revision 56c9d3f4842baa53d7ab13d0764eae7f305aba0f)
+++ b/src/test/modules/injection_points/meson.build (revision 3dea72b62adc8806917dc459b82ff44d962bcb12)
@@ -42,6 +42,7 @@
'isolation': {
'specs': [
'inplace',
+ 'reset_snapshots',
],
},
'tap': {
Index: src/test/modules/injection_points/specs/reset_snapshots.spec
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/src/test/modules/injection_points/specs/reset_snapshots.spec b/src/test/modules/injection_points/specs/reset_snapshots.spec
new file mode 100644
--- /dev/null (revision 3dea72b62adc8806917dc459b82ff44d962bcb12)
+++ b/src/test/modules/injection_points/specs/reset_snapshots.spec (revision 3dea72b62adc8806917dc459b82ff44d962bcb12)
@@ -0,0 +1,114 @@
+setup
+{
+ CREATE EXTENSION injection_points;
+ CREATE SCHEMA test;
+ CREATE TABLE test.tbl(i int primary key, j int);
+ INSERT INTO test.tbl SELECT i, i * I FROM generate_series(1, 200) s(i);
+
+ CREATE FUNCTION test.predicate_stable(integer) RETURNS bool IMMUTABLE
+ LANGUAGE plpgsql AS $$
+ BEGIN
+ EXECUTE 'SELECT txid_current()';
+ RETURN MOD($1, 2) = 0;
+ END; $$;
+
+ CREATE FUNCTION test.predicate_stable_no_param() RETURNS bool IMMUTABLE
+ LANGUAGE plpgsql AS $$
+ BEGIN
+ EXECUTE 'SELECT txid_current()';
+ RETURN false;
+ END; $$;
+}
+
+teardown
+{
+ DROP SCHEMA test CASCADE;
+ DROP EXTENSION injection_points;
+}
+
+session test
+setup {
+ SELECT injection_points_attach('heapam_index_validate_scan_no_xid', 'notice');
+ SELECT injection_points_attach('heap_reset_scan_snapshot_effective', 'notice');
+ SELECT injection_points_attach('_bt_leader_participate_as_worker', 'wait');
+}
+step sleep { SELECT pg_sleep(10); }
+step drop_index { DROP INDEX CONCURRENTLY test.idx; }
+step create_index_concurrently_simple { CREATE INDEX CONCURRENTLY idx ON test.tbl(i, j); }
+step create_unique_index_concurrently_simple { CREATE UNIQUE INDEX CONCURRENTLY idx ON test.tbl(i); }
+step create_index_concurrently_predicate_expression_mod { CREATE INDEX CONCURRENTLY idx ON test.tbl(MOD(i, 2), j) WHERE MOD(i, 2) = 0; }
+step create_index_concurrently_predicate_set_xid { CREATE INDEX CONCURRENTLY idx ON test.tbl(i, j) WHERE test.predicate_stable(i); }
+step create_index_concurrently_predicate_set_xid_no_param { CREATE INDEX CONCURRENTLY idx ON test.tbl(i, j) WHERE test.predicate_stable_no_param(); }
+step reindex_index_concurrently { REINDEX INDEX CONCURRENTLY test.idx; }
+step set_parallel_workers_1 { ALTER TABLE test.tbl SET (parallel_workers=0); }
+step set_parallel_workers_2 { ALTER TABLE test.tbl SET (parallel_workers=2); }
+step detach {
+ SELECT injection_points_detach('heapam_index_validate_scan_no_xid');
+ SELECT injection_points_detach('heap_reset_scan_snapshot_effective');
+ SELECT injection_points_detach('_bt_leader_participate_as_worker');
+}
+
+session wakeup_session
+step wakeup { SELECT injection_points_wakeup('_bt_leader_participate_as_worker'); }
+
+permutation
+ set_parallel_workers_1
+ create_index_concurrently_simple
+ reindex_index_concurrently
+ drop_index
+ detach
+
+permutation
+ set_parallel_workers_1
+ create_unique_index_concurrently_simple
+ reindex_index_concurrently
+ drop_index
+ detach
+
+permutation
+ set_parallel_workers_1
+ create_index_concurrently_predicate_expression_mod
+ reindex_index_concurrently
+ drop_index
+ detach
+
+permutation
+ set_parallel_workers_1
+ create_index_concurrently_predicate_set_xid_no_param
+ reindex_index_concurrently
+ drop_index
+ detach
+
+permutation
+ set_parallel_workers_1
+ create_index_concurrently_predicate_set_xid
+ reindex_index_concurrently
+ drop_index
+ detach
+
+permutation
+ set_parallel_workers_2
+ create_index_concurrently_simple
+ wakeup
+ reindex_index_concurrently
+ wakeup
+ drop_index
+ detach
+
+permutation
+ set_parallel_workers_2
+ create_unique_index_concurrently_simple
+ wakeup
+ reindex_index_concurrently
+ wakeup
+ drop_index
+ detach
+
+permutation
+ set_parallel_workers_2
+ create_index_concurrently_predicate_expression_mod
+ wakeup
+ reindex_index_concurrently
+ wakeup
+ drop_index
+ detach
\ No newline at end of file
view thread (33+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected], [email protected]
Subject: Re: Revisiting {CREATE INDEX, REINDEX} CONCURRENTLY improvements
In-Reply-To: <CANtu0ojHEVU9U_bxgViRmtqNTJ92LnF+76-yzn4axYjGsK2kqQ@mail.gmail.com>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox