minmax-9.patch
text/x-diff
Patch
Format: context
| File | + | − |
|---|---|---|
| contrib/pageinspect/Makefile | 1 | 1 |
| contrib/pageinspect/mmfuncs.c | 420 | 0 |
| contrib/pageinspect/pageinspect--1.2.sql | 46 | 0 |
| contrib/pg_xlogdump/rmgrdesc.c | 1 | 0 |
| src/backend/access/common/reloptions.c | 7 | 0 |
| src/backend/access/heap/heapam.c | 18 | 0 |
| src/backend/access/Makefile | 1 | 0 |
| src/backend/access/minmax/Makefile | 17 | 0 |
| src/backend/access/minmax/minmax.c | 1553 | 0 |
| src/backend/access/minmax/mmrevmap.c | 683 | 0 |
| src/backend/access/minmax/mmsortable.c | 225 | 0 |
| src/backend/access/minmax/mmtuple.c | 449 | 0 |
| src/backend/access/minmax/mmxlog.c | 304 | 0 |
| src/backend/access/rmgrdesc/Makefile | 2 | 0 |
| src/backend/access/rmgrdesc/minmaxdesc.c | 95 | 0 |
| src/backend/access/transam/rmgr.c | 1 | 0 |
| src/backend/catalog/index.c | 24 | 0 |
| src/backend/replication/logical/decode.c | 1 | 0 |
| src/backend/storage/page/bufpage.c | 213 | 0 |
| src/backend/utils/adt/selfuncs.c | 24 | 0 |
| src/include/access/heapam.h | 2 | 0 |
| src/include/access/minmax.h | 52 | 0 |
| src/include/access/minmax_internal.h | 104 | 0 |
| src/include/access/minmax_page.h | 88 | 0 |
| src/include/access/minmax_revmap.h | 34 | 0 |
| src/include/access/minmax_tuple.h | 84 | 0 |
| src/include/access/minmax_xlog.h | 134 | 0 |
| src/include/access/reloptions.h | 2 | 0 |
| src/include/access/relscan.h | 3 | 0 |
| src/include/access/rmgrlist.h | 1 | 0 |
| src/include/catalog/index.h | 8 | 0 |
| src/include/catalog/pg_am.h | 2 | 0 |
| src/include/catalog/pg_amop.h | 81 | 0 |
| src/include/catalog/pg_amproc.h | 38 | 0 |
| src/include/catalog/pg_opclass.h | 9 | 0 |
| src/include/catalog/pg_opfamily.h | 10 | 0 |
| src/include/catalog/pg_proc.h | 36 | 0 |
| src/include/storage/bufpage.h | 4 | 0 |
| src/include/utils/selfuncs.h | 1 | 0 |
| src/test/regress/expected/opr_sanity.out | 11 | 0 |
| src/test/regress/sql/opr_sanity.sql | 5 | 0 |
*** a/contrib/pageinspect/Makefile
--- b/contrib/pageinspect/Makefile
***************
*** 1,7 ****
# contrib/pageinspect/Makefile
MODULE_big = pageinspect
! OBJS = rawpage.o heapfuncs.o btreefuncs.o fsmfuncs.o
EXTENSION = pageinspect
DATA = pageinspect--1.2.sql pageinspect--1.0--1.1.sql \
--- 1,7 ----
# contrib/pageinspect/Makefile
MODULE_big = pageinspect
! OBJS = rawpage.o heapfuncs.o btreefuncs.o fsmfuncs.o mmfuncs.o
EXTENSION = pageinspect
DATA = pageinspect--1.2.sql pageinspect--1.0--1.1.sql \
*** /dev/null
--- b/contrib/pageinspect/mmfuncs.c
***************
*** 0 ****
--- 1,420 ----
+ /*
+ * mmfuncs.c
+ * Functions to investigate MinMax indexes
+ *
+ * Copyright (c) 2013, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ * contrib/pageinspect/mmfuncs.c
+ */
+ #include "postgres.h"
+
+ #include "access/htup_details.h"
+ #include "access/minmax.h"
+ #include "access/minmax_internal.h"
+ #include "access/minmax_page.h"
+ #include "access/minmax_revmap.h"
+ #include "access/minmax_tuple.h"
+ #include "catalog/index.h"
+ #include "catalog/pg_type.h"
+ #include "funcapi.h"
+ #include "utils/array.h"
+ #include "utils/builtins.h"
+ #include "utils/lsyscache.h"
+ #include "utils/rel.h"
+ #include "miscadmin.h"
+
+ Datum minmax_page_type(PG_FUNCTION_ARGS);
+ Datum minmax_page_items(PG_FUNCTION_ARGS);
+ Datum minmax_metapage_info(PG_FUNCTION_ARGS);
+ Datum minmax_revmap_array_data(PG_FUNCTION_ARGS);
+ Datum minmax_revmap_data(PG_FUNCTION_ARGS);
+
+ PG_FUNCTION_INFO_V1(minmax_page_type);
+ PG_FUNCTION_INFO_V1(minmax_page_items);
+ PG_FUNCTION_INFO_V1(minmax_metapage_info);
+ PG_FUNCTION_INFO_V1(minmax_revmap_array_data);
+ PG_FUNCTION_INFO_V1(minmax_revmap_data);
+
+ typedef struct mm_page_state
+ {
+ TupleDesc tupdesc;
+ Page page;
+ OffsetNumber offset;
+ bool unusedItem;
+ bool done;
+ AttrNumber attno;
+ DeformedMMTuple *dtup;
+ FmgrInfo outputfn[FLEXIBLE_ARRAY_MEMBER];
+ } mm_page_state;
+
+
+ static Page verify_minmax_page(bytea *raw_page, uint16 type,
+ const char *strtype);
+
+ Datum
+ minmax_page_type(PG_FUNCTION_ARGS)
+ {
+ bytea *raw_page = PG_GETARG_BYTEA_P(0);
+ Page page = VARDATA(raw_page);
+ MinmaxSpecialSpace *special;
+ char *type;
+
+ special = (MinmaxSpecialSpace *) PageGetSpecialPointer(page);
+
+ switch (special->type)
+ {
+ case MINMAX_PAGETYPE_META:
+ type = "meta";
+ break;
+ case MINMAX_PAGETYPE_REVMAP_ARRAY:
+ type = "revmap array";
+ break;
+ case MINMAX_PAGETYPE_REVMAP:
+ type = "revmap";
+ break;
+ case MINMAX_PAGETYPE_REGULAR:
+ type = "regular";
+ break;
+ default:
+ type = psprintf("unknown (%02x)", special->type);
+ break;
+ }
+
+ PG_RETURN_TEXT_P(cstring_to_text(type));
+ }
+
+ /*
+ * Verify that the given bytea contains a minmax page of the indicated page
+ * type, or die in the attempt. A pointer to the page is returned.
+ */
+ static Page
+ verify_minmax_page(bytea *raw_page, uint16 type, const char *strtype)
+ {
+ Page page;
+ int raw_page_size;
+ MinmaxSpecialSpace *special;
+
+ raw_page_size = VARSIZE(raw_page) - VARHDRSZ;
+
+ if (raw_page_size < SizeOfPageHeaderData)
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("input page too small"),
+ errdetail("Expected size %d, got %d", raw_page_size, BLCKSZ)));
+
+ page = VARDATA(raw_page);
+
+ /* verify the special space says this page is what we want */
+ special = (MinmaxSpecialSpace *) PageGetSpecialPointer(page);
+ if (special->type != type)
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("page is not a Minmax page of type \"%s\"", strtype),
+ errdetail("Expected special type %08x, got %08x.",
+ type, special->type)));
+
+ return page;
+ }
+
+
+ #ifdef NOT_YET
+ /*
+ * Extract all item values from a minmax index page
+ *
+ * Usage: SELECT * FROM minmax_page_items(get_raw_page('idx', 1), 'idx'::regclass);
+ */
+ Datum
+ minmax_page_items(PG_FUNCTION_ARGS)
+ {
+ mm_page_state *state;
+ FuncCallContext *fctx;
+
+ if (!superuser())
+ ereport(ERROR,
+ (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+ (errmsg("must be superuser to use raw page functions"))));
+
+ if (SRF_IS_FIRSTCALL())
+ {
+ bytea *raw_page = PG_GETARG_BYTEA_P(0);
+ Oid indexRelid = PG_GETARG_OID(1);
+ Page page;
+ TupleDesc tupdesc;
+ MemoryContext mctx;
+ Relation indexRel;
+ AttrNumber attno;
+
+ /* minimally verify the page we got */
+ page = verify_minmax_page(raw_page, MINMAX_PAGETYPE_REGULAR, "regular");
+
+ /* create a function context for cross-call persistence */
+ fctx = SRF_FIRSTCALL_INIT();
+
+ /* switch to memory context appropriate for multiple function calls */
+ mctx = MemoryContextSwitchTo(fctx->multi_call_memory_ctx);
+
+ /* Build a tuple descriptor for our result type */
+ if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
+ elog(ERROR, "return type must be a row type");
+
+ indexRel = index_open(indexRelid, AccessShareLock);
+
+ state = palloc(offsetof(mm_page_state, outputfn) +
+ sizeof(FmgrInfo) * RelationGetDescr(indexRel)->natts);
+
+ state->tupdesc = CreateTupleDescCopy(RelationGetDescr(indexRel));
+ state->page = page;
+ state->offset = FirstOffsetNumber;
+ state->unusedItem = false;
+ state->done = false;
+ state->dtup = NULL;
+
+ index_close(indexRel, AccessShareLock);
+
+ for (attno = 1; attno <= state->tupdesc->natts; attno++)
+ {
+ Oid output;
+ bool isVarlena;
+
+ getTypeOutputInfo(state->tupdesc->attrs[attno - 1]->atttypid,
+ &output, &isVarlena);
+ fmgr_info(output, &state->outputfn[attno - 1]);
+ }
+
+ fctx->user_fctx = state;
+ fctx->tuple_desc = BlessTupleDesc(tupdesc);
+
+ MemoryContextSwitchTo(mctx);
+ }
+
+ fctx = SRF_PERCALL_SETUP();
+ state = fctx->user_fctx;
+
+ if (!state->done)
+ {
+ HeapTuple result;
+ Datum values[6];
+ bool nulls[6];
+
+ /*
+ * This loop is called once for every attribute of every tuple in the
+ * page. At the start of a tuple, we get a NULL dtup; that's our
+ * signal for obtaining and decoding the next one. If that's not the
+ * case, we output the next attribute.
+ */
+ if (state->dtup == NULL)
+ {
+ MMTuple *tup;
+ MemoryContext mctx;
+ ItemId itemId;
+
+ /* deformed tuple must live across calls */
+ mctx = MemoryContextSwitchTo(fctx->multi_call_memory_ctx);
+
+ /* verify item status: if there's no data, we can't decode */
+ itemId = PageGetItemId(state->page, state->offset);
+ if (ItemIdIsUsed(itemId))
+ {
+ tup = (MMTuple *) PageGetItem(state->page,
+ PageGetItemId(state->page,
+ state->offset));
+ state->dtup = minmax_deform_tuple(state->tupdesc, tup);
+ state->attno = 1;
+ state->unusedItem = false;
+ }
+ else
+ state->unusedItem = true;
+
+ MemoryContextSwitchTo(mctx);
+ }
+ else
+ state->attno++;
+
+ MemSet(nulls, 0, sizeof(nulls));
+
+ if (state->unusedItem)
+ {
+ values[0] = UInt16GetDatum(state->offset);
+ nulls[1] = true;
+ nulls[2] = true;
+ nulls[3] = true;
+ nulls[4] = true;
+ nulls[5] = true;
+ }
+ else
+ {
+ int att = state->attno - 1;
+
+ values[0] = UInt16GetDatum(state->offset);
+ values[1] = UInt16GetDatum(state->attno);
+ values[2] = BoolGetDatum(state->dtup->values[att].allnulls);
+ values[3] = BoolGetDatum(state->dtup->values[att].hasnulls);
+ if (!state->dtup->values[att].allnulls)
+ {
+ FmgrInfo *outputfn = &state->outputfn[att];
+ MMValues *mmvalues = &state->dtup->values[att];
+
+ values[4] = CStringGetTextDatum(OutputFunctionCall(outputfn,
+ mmvalues->min));
+ values[5] = CStringGetTextDatum(OutputFunctionCall(outputfn,
+ mmvalues->max));
+ }
+ else
+ {
+ nulls[4] = true;
+ nulls[5] = true;
+ }
+ }
+
+ result = heap_form_tuple(fctx->tuple_desc, values, nulls);
+
+ /*
+ * If the item was unused, jump straight to the next one; otherwise,
+ * the only cleanup needed here is to set our signal to go to the next
+ * tuple in the following iteration, by freeing the current one.
+ */
+ if (state->unusedItem)
+ state->offset = OffsetNumberNext(state->offset);
+ else if (state->attno >= state->tupdesc->natts)
+ {
+ pfree(state->dtup);
+ state->dtup = NULL;
+ state->offset = OffsetNumberNext(state->offset);
+ }
+
+ /*
+ * If we're beyond the end of the page, set flag to end the function in
+ * the following iteration.
+ */
+ if (state->offset > PageGetMaxOffsetNumber(state->page))
+ state->done = true;
+
+ SRF_RETURN_NEXT(fctx, HeapTupleGetDatum(result));
+ }
+
+ SRF_RETURN_DONE(fctx);
+ }
+ #endif
+
+ Datum
+ minmax_metapage_info(PG_FUNCTION_ARGS)
+ {
+ bytea *raw_page = PG_GETARG_BYTEA_P(0);
+ Page page;
+ MinmaxMetaPageData *meta;
+ TupleDesc tupdesc;
+ Datum values[2];
+ bool nulls[2];
+ ArrayBuildState *astate = NULL;
+ HeapTuple htup;
+ int i;
+
+ page = verify_minmax_page(raw_page, MINMAX_PAGETYPE_META, "metapage");
+
+ /* Build a tuple descriptor for our result type */
+ if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
+ elog(ERROR, "return type must be a row type");
+ tupdesc = BlessTupleDesc(tupdesc);
+
+ /* Extract values from the metapage */
+ meta = (MinmaxMetaPageData *) PageGetContents(page);
+ MemSet(nulls, 0, sizeof(nulls));
+ values[0] = Int32GetDatum(meta->minmaxVersion);
+
+ /* Extract (possibly empty) list of revmap array page numbers. */
+ for (i = 0; i < MAX_REVMAP_ARRAYPAGES; i++)
+ {
+ BlockNumber blkno;
+
+ blkno = meta->revmapArrayPages[i];
+ if (blkno == InvalidBlockNumber)
+ break; /* XXX or continue? */
+ astate = accumArrayResult(astate, Int64GetDatum((int64) blkno),
+ false, INT8OID, CurrentMemoryContext);
+ }
+ if (astate == NULL)
+ nulls[1] = true;
+ else
+ values[1] = makeArrayResult(astate, CurrentMemoryContext);
+
+ htup = heap_form_tuple(tupdesc, values, nulls);
+
+ PG_RETURN_DATUM(HeapTupleGetDatum(htup));
+ }
+
+ /*
+ * Return the BlockNumber array stored in a revmap array page
+ */
+ Datum
+ minmax_revmap_array_data(PG_FUNCTION_ARGS)
+ {
+ bytea *raw_page = PG_GETARG_BYTEA_P(0);
+ Page page;
+ ArrayBuildState *astate = NULL;
+ RevmapArrayContents *contents;
+ Datum blkarr;
+ int i;
+
+ page = verify_minmax_page(raw_page, MINMAX_PAGETYPE_REVMAP_ARRAY,
+ "revmap array");
+
+ contents = (RevmapArrayContents *) PageGetContents(page);
+
+ for (i = 0; i < contents->rma_nblocks; i++)
+ astate = accumArrayResult(astate,
+ Int64GetDatum((int64) contents->rma_blocks[i]),
+ false, INT8OID, CurrentMemoryContext);
+ Assert(astate != NULL);
+
+ blkarr = makeArrayResult(astate, CurrentMemoryContext);
+ PG_RETURN_DATUM(blkarr);
+ }
+
+ /*
+ * Return the TID array stored in a minmax revmap page
+ */
+ Datum
+ minmax_revmap_data(PG_FUNCTION_ARGS)
+ {
+ bytea *raw_page = PG_GETARG_BYTEA_P(0);
+ Page page;
+ RevmapContents *contents;
+ TupleDesc tupdesc;
+ Datum values[2];
+ bool nulls[2];
+ HeapTuple htup;
+ ArrayBuildState *astate = NULL;
+ int i;
+
+ page = verify_minmax_page(raw_page, MINMAX_PAGETYPE_REVMAP, "revmap");
+
+ /* Build a tuple descriptor for our result type */
+ if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
+ elog(ERROR, "return type must be a row type");
+ tupdesc = BlessTupleDesc(tupdesc);
+
+ /* Extract values from the revmap page */
+ contents = (RevmapContents *) PageGetContents(page);
+ MemSet(nulls, 0, sizeof(nulls));
+ values[0] = Int64GetDatum((uint64) contents->rmr_logblk);
+
+ /* Extract (possibly empty) list of TIDs in this page. */
+ for (i = 0; i < REGULAR_REVMAP_PAGE_MAXITEMS; i++)
+ {
+ ItemPointer tid;
+
+ tid = &contents->rmr_tids[i];
+ astate = accumArrayResult(astate,
+ PointerGetDatum(tid),
+ false, TIDOID, CurrentMemoryContext);
+ }
+ if (astate == NULL)
+ nulls[1] = true;
+ else
+ values[1] = makeArrayResult(astate, CurrentMemoryContext);
+
+ htup = heap_form_tuple(tupdesc, values, nulls);
+
+ PG_RETURN_DATUM(HeapTupleGetDatum(htup));
+ }
*** a/contrib/pageinspect/pageinspect--1.2.sql
--- b/contrib/pageinspect/pageinspect--1.2.sql
***************
*** 99,104 **** AS 'MODULE_PATHNAME', 'bt_page_items'
--- 99,150 ----
LANGUAGE C STRICT;
--
+ -- minmax_page_type()
+ --
+ CREATE FUNCTION minmax_page_type(IN page bytea)
+ RETURNS text
+ AS 'MODULE_PATHNAME', 'minmax_page_type'
+ LANGUAGE C STRICT;
+
+ --
+ -- minmax_metapage_info()
+ --
+ CREATE FUNCTION minmax_metapage_info(IN page bytea,
+ OUT version integer, OUT revmap_array_pages BIGINT[])
+ AS 'MODULE_PATHNAME', 'minmax_metapage_info'
+ LANGUAGE C STRICT;
+
+ --
+ -- minmax_page_items()
+ --
+ /* needs more work
+ CREATE FUNCTION minmax_page_items(IN page bytea, IN index_oid oid,
+ OUT itemoffset int,
+ OUT attnum int,
+ OUT allnulls bool,
+ OUT hasnulls bool,
+ OUT min text,
+ OUT max text)
+ RETURNS SETOF record
+ AS 'MODULE_PATHNAME', 'minmax_page_items'
+ LANGUAGE C STRICT;
+ */
+
+ --
+ -- minmax_revmap_array_data()
+ CREATE FUNCTION minmax_revmap_array_data(IN page bytea,
+ OUT revmap_pages BIGINT[])
+ AS 'MODULE_PATHNAME', 'minmax_revmap_array_data'
+ LANGUAGE C STRICT;
+
+ --
+ -- minmax_revmap_data()
+ CREATE FUNCTION minmax_revmap_data(IN page bytea,
+ OUT logblk BIGINT, OUT pages tid[])
+ AS 'MODULE_PATHNAME', 'minmax_revmap_data'
+ LANGUAGE C STRICT;
+
+ --
-- fsm_page_contents()
--
CREATE FUNCTION fsm_page_contents(IN page bytea)
*** a/contrib/pg_xlogdump/rmgrdesc.c
--- b/contrib/pg_xlogdump/rmgrdesc.c
***************
*** 13,18 ****
--- 13,19 ----
#include "access/gist_private.h"
#include "access/hash.h"
#include "access/heapam_xlog.h"
+ #include "access/minmax_xlog.h"
#include "access/multixact.h"
#include "access/nbtree.h"
#include "access/rmgr.h"
*** a/src/backend/access/Makefile
--- b/src/backend/access/Makefile
***************
*** 8,13 **** subdir = src/backend/access
top_builddir = ../../..
include $(top_builddir)/src/Makefile.global
! SUBDIRS = common gin gist hash heap index nbtree rmgrdesc spgist transam
include $(top_srcdir)/src/backend/common.mk
--- 8,13 ----
top_builddir = ../../..
include $(top_builddir)/src/Makefile.global
! SUBDIRS = common gin gist hash heap index minmax nbtree rmgrdesc spgist transam
include $(top_srcdir)/src/backend/common.mk
*** a/src/backend/access/common/reloptions.c
--- b/src/backend/access/common/reloptions.c
***************
*** 209,214 **** static relopt_int intRelOpts[] =
--- 209,221 ----
RELOPT_KIND_HEAP | RELOPT_KIND_TOAST
}, -1, 0, 2000000000
},
+ {
+ {
+ "pages_per_range",
+ "Number of pages that each page range covers in a Minmax index",
+ RELOPT_KIND_MINMAX
+ }, 128, 1, 131072
+ },
/* list terminator */
{{NULL}}
*** a/src/backend/access/heap/heapam.c
--- b/src/backend/access/heap/heapam.c
***************
*** 271,276 **** initscan(HeapScanDesc scan, ScanKey key, bool is_rescan)
--- 271,278 ----
scan->rs_startblock = 0;
}
+ scan->rs_initblock = 0;
+ scan->rs_numblocks = InvalidBlockNumber;
scan->rs_inited = false;
scan->rs_ctup.t_data = NULL;
ItemPointerSetInvalid(&scan->rs_ctup.t_self);
***************
*** 296,301 **** initscan(HeapScanDesc scan, ScanKey key, bool is_rescan)
--- 298,311 ----
pgstat_count_heap_scan(scan->rs_rd);
}
+ void
+ heap_setscanlimits(HeapScanDesc scan, BlockNumber startBlk, BlockNumber numBlks)
+ {
+ scan->rs_startblock = startBlk;
+ scan->rs_initblock = startBlk;
+ scan->rs_numblocks = numBlks;
+ }
+
/*
* heapgetpage - subroutine for heapgettup()
*
***************
*** 636,642 **** heapgettup(HeapScanDesc scan,
*/
if (backward)
{
! finished = (page == scan->rs_startblock);
if (page == 0)
page = scan->rs_nblocks;
page--;
--- 646,653 ----
*/
if (backward)
{
! finished = (page == scan->rs_startblock) ||
! (scan->rs_numblocks != InvalidBlockNumber ? --scan->rs_numblocks <= 0 : false);
if (page == 0)
page = scan->rs_nblocks;
page--;
***************
*** 646,652 **** heapgettup(HeapScanDesc scan,
page++;
if (page >= scan->rs_nblocks)
page = 0;
! finished = (page == scan->rs_startblock);
/*
* Report our new scan position for synchronization purposes. We
--- 657,664 ----
page++;
if (page >= scan->rs_nblocks)
page = 0;
! finished = (page == scan->rs_startblock) ||
! (scan->rs_numblocks != InvalidBlockNumber ? --scan->rs_numblocks <= 0 : false);
/*
* Report our new scan position for synchronization purposes. We
***************
*** 897,903 **** heapgettup_pagemode(HeapScanDesc scan,
*/
if (backward)
{
! finished = (page == scan->rs_startblock);
if (page == 0)
page = scan->rs_nblocks;
page--;
--- 909,916 ----
*/
if (backward)
{
! finished = (page == scan->rs_startblock) ||
! (scan->rs_numblocks != InvalidBlockNumber ? --scan->rs_numblocks <= 0 : false);
if (page == 0)
page = scan->rs_nblocks;
page--;
***************
*** 907,913 **** heapgettup_pagemode(HeapScanDesc scan,
page++;
if (page >= scan->rs_nblocks)
page = 0;
! finished = (page == scan->rs_startblock);
/*
* Report our new scan position for synchronization purposes. We
--- 920,927 ----
page++;
if (page >= scan->rs_nblocks)
page = 0;
! finished = (page == scan->rs_startblock) ||
! (scan->rs_numblocks != InvalidBlockNumber ? --scan->rs_numblocks <= 0 : false);
/*
* Report our new scan position for synchronization purposes. We
*** /dev/null
--- b/src/backend/access/minmax/Makefile
***************
*** 0 ****
--- 1,17 ----
+ #-------------------------------------------------------------------------
+ #
+ # Makefile--
+ # Makefile for access/minmax
+ #
+ # IDENTIFICATION
+ # src/backend/access/minmax/Makefile
+ #
+ #-------------------------------------------------------------------------
+
+ subdir = src/backend/access/minmax
+ top_builddir = ../../../..
+ include $(top_builddir)/src/Makefile.global
+
+ OBJS = minmax.o mmrevmap.o mmtuple.o mmxlog.o mmsortable.o
+
+ include $(top_srcdir)/src/backend/common.mk
*** /dev/null
--- b/src/backend/access/minmax/minmax.c
***************
*** 0 ****
--- 1,1553 ----
+ /*
+ * minmax.c
+ * Implementation of Minmax indexes for Postgres
+ *
+ * Portions Copyright (c) 1996-2013, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * IDENTIFICATION
+ * src/backend/access/minmax/minmax.c
+ *
+ * TODO
+ * * support collatable datatypes
+ * * ScalarArrayOpExpr
+ * * Make use of the stored NULL bits
+ * * we can support unlogged indexes now
+ */
+ #include "postgres.h"
+
+ #include "access/htup_details.h"
+ #include "access/minmax.h"
+ #include "access/minmax_internal.h"
+ #include "access/minmax_page.h"
+ #include "access/minmax_revmap.h"
+ #include "access/minmax_tuple.h"
+ #include "access/minmax_xlog.h"
+ #include "access/reloptions.h"
+ #include "access/relscan.h"
+ #include "access/xlogutils.h"
+ #include "catalog/index.h"
+ #include "catalog/pg_operator.h"
+ #include "commands/vacuum.h"
+ #include "miscadmin.h"
+ #include "pgstat.h"
+ #include "storage/bufmgr.h"
+ #include "storage/freespace.h"
+ #include "storage/lmgr.h"
+ #include "utils/datum.h"
+ #include "utils/lsyscache.h"
+ #include "utils/memutils.h"
+ #include "utils/syscache.h"
+
+
+ /*
+ * We use a MMBuildState during initial construction of a Minmax index.
+ * The running state is kept in a DeformedMMTuple.
+ */
+ typedef struct MMBuildState
+ {
+ Relation irel;
+ int numtuples;
+ Buffer currentInsertBuf;
+ BlockNumber pagesPerRange;
+ BlockNumber currRangeStart;
+ mmRevmapAccess *rmAccess;
+ MinmaxDesc *mmDesc;
+ DeformedMMTuple *dtuple;
+ } MMBuildState;
+
+ /*
+ * Struct used as "opaque" during index scans
+ */
+ typedef struct MinmaxOpaque
+ {
+ BlockNumber pagesPerRange;
+ mmRevmapAccess *rmAccess;
+ MinmaxDesc *mmDesc;
+ } MinmaxOpaque;
+
+ static MinmaxDesc *minmax_build_mmdesc(Relation rel);
+ static void mmbuildCallback(Relation index,
+ HeapTuple htup, Datum *values, bool *isnull,
+ bool tupleIsAlive, void *state);
+ static void mm_doinsert(Relation idxrel, mmRevmapAccess *rmAccess,
+ Buffer *buffer, BlockNumber heapblkno, MMTuple *tup, Size itemsz);
+ static bool mm_getinsertbuffer(Relation irel, Buffer *buffer, Size itemsz);
+
+
+
+ /*
+ * A tuple in the heap is being inserted. To keep a minmax index up to date,
+ * we need to obtain the relevant index tuple, compare its min()/max() stored
+ * values with those of the new tuple; if the tuple values are in range,
+ * there's nothing to do; otherwise we need to update the index (either by
+ * a new index tuple and repointing the revmap, or by overwriting the existing
+ * index tuple).
+ *
+ * If the range is not currently summarized (i.e. the revmap returns InvalidTid
+ * for it), there's nothing to do either.
+ */
+ Datum
+ mminsert(PG_FUNCTION_ARGS)
+ {
+ Relation idxRel = (Relation) PG_GETARG_POINTER(0);
+ Datum *values = (Datum *) PG_GETARG_POINTER(1);
+ bool *nulls = (bool *) PG_GETARG_POINTER(2);
+ ItemPointer heaptid = (ItemPointer) PG_GETARG_POINTER(3);
+
+ /* we ignore the rest of our arguments */
+ MinmaxDesc *mmdesc;
+ mmRevmapAccess *rmAccess;
+ ItemId origlp;
+ MMTuple *mmtup;
+ DeformedMMTuple *dtup;
+ ItemPointerData idxtid;
+ BlockNumber heapBlk;
+ BlockNumber iblk;
+ OffsetNumber ioff;
+ Buffer buf;
+ IndexInfo *indexInfo;
+ Page page;
+ int keyno;
+ bool need_insert = false;
+
+ rmAccess = mmRevmapAccessInit(idxRel, NULL);
+
+ heapBlk = ItemPointerGetBlockNumber(heaptid);
+ mmGetHeapBlockItemptr(rmAccess, heapBlk, &idxtid);
+ /* tuple lock on idxtid is grabbed by mmGetHeapBlockItemptr */
+
+ if (!ItemPointerIsValid(&idxtid))
+ {
+ /* nothing to do, range is unsummarized */
+ mmRevmapAccessTerminate(rmAccess);
+ return BoolGetDatum(false);
+ }
+
+ indexInfo = BuildIndexInfo(idxRel);
+ mmdesc = minmax_build_mmdesc(idxRel);
+
+ iblk = ItemPointerGetBlockNumber(&idxtid);
+ ioff = ItemPointerGetOffsetNumber(&idxtid);
+ Assert(iblk != InvalidBlockNumber);
+ buf = ReadBuffer(idxRel, iblk);
+
+ LockBuffer(buf, BUFFER_LOCK_SHARE);
+ UnlockTuple(idxRel, &idxtid, ShareLock);
+ page = BufferGetPage(buf);
+ origlp = PageGetItemId(page, ioff);
+ mmtup = (MMTuple *) PageGetItem(page, origlp);
+
+ dtup = minmax_deform_tuple(mmdesc, mmtup);
+
+ /*
+ * Compare the key values of the new tuple to the stored index values; our
+ * deformed tuple will get updated if the new tuple doesn't fit the
+ * original range (note this means we can't break out of the loop early).
+ * Make a note of whether this happens, so that we know to insert the
+ * modified tuple later, if necessary.
+ */
+ for (keyno = 0; keyno < indexInfo->ii_NumIndexAttrs; keyno++)
+ {
+ Datum result;
+ FmgrInfo *maybeUpdateFn;
+
+ /* FIXME must be cached somewhere */
+ maybeUpdateFn = index_getprocinfo(idxRel, keyno + 1,
+ MINMAX_PROCNUM_MAYBEUPDATE);
+
+ result = FunctionCall5(maybeUpdateFn,
+ PointerGetDatum(mmdesc),
+ PointerGetDatum(dtup),
+ UInt16GetDatum(keyno + 1),
+ values[keyno],
+ nulls[keyno]);
+ /* if that returned true, we need to insert the updated tuple */
+ if (DatumGetBool(result))
+ need_insert = true;
+ }
+
+ LockBuffer(buf, BUFFER_LOCK_UNLOCK);
+
+ if (need_insert)
+ {
+ Size tupsz;
+ MMTuple *tup;
+
+ tup = minmax_form_tuple(mmdesc, dtup, &tupsz);
+
+ /*
+ * If the size of the original tuple is greater or equal to the new
+ * index tuple, we can overwrite. This saves regular page bloat, and
+ * also saves revmap traffic. This might leave some unused space
+ * before the start of the next tuple, but we don't worry about that
+ * here.
+ *
+ * We avoid doing this when the itempointer of the index tuple would
+ * change, because that would require an update to the revmap while
+ * holding exclusive lock on this page, which would reduce concurrency.
+ *
+ * Note that we continue to acccess 'origlp' here, even though there
+ * was an interval during which the page wasn't locked. Since we hold
+ * pin on the page, this is okay -- the buffer cannot go away from
+ * under us, and also tuples cannot be shuffled around.
+ */
+ if (tupsz >= ItemIdGetLength(origlp))
+ {
+ LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
+
+ START_CRIT_SECTION();
+ PageOverwriteItemData(BufferGetPage(buf),
+ ioff,
+ (Item) tup, tupsz);
+ MarkBufferDirty(buf);
+
+ /* XLOG stuff */
+ if (RelationNeedsWAL(idxRel))
+ {
+ xl_minmax_insert xlrec;
+ XLogRecPtr recptr;
+ XLogRecData rdata[2];
+ uint8 info = XLOG_MINMAX_INSERT;
+
+ xlrec.target.node = idxRel->rd_node;
+ xlrec.target.tid = idxtid;
+ xlrec.overwrite = true;
+ rdata[0].data = (char *) &xlrec;
+ rdata[0].len = SizeOfMinmaxInsert;
+ rdata[0].buffer = InvalidBuffer;
+ rdata[0].next = &(rdata[1]);
+
+ rdata[1].data = (char *) tup;
+ rdata[1].len = tupsz;
+ rdata[1].buffer = buf;
+ rdata[1].buffer_std = true;
+ rdata[1].next = NULL;
+
+ recptr = XLogInsert(RM_MINMAX_ID, info, rdata);
+
+ PageSetLSN(page, recptr);
+ }
+
+ END_CRIT_SECTION();
+
+ LockBuffer(buf, BUFFER_LOCK_UNLOCK);
+ }
+ else
+ {
+ /*
+ * The new tuple is larger than the original one, so we must insert
+ * a new one the slow way.
+ */
+ mm_doinsert(idxRel, rmAccess, &buf, heapBlk, tup, tupsz);
+
+ #ifdef NOT_YET
+ /*
+ * Possible optimization: if we can grab an exclusive lock on the
+ * buffer containing the old tuple right away, we can also seize
+ * the opportunity to prune the old tuple and avoid some bloat.
+ * This is not necessary for correctness.
+ */
+ if (ConditionalLockBuffer(buf))
+ {
+ /* prune the old tuple */
+
+ LockBuffer(buf, BUFFER_LOCK_UNLOCK);
+ }
+ #endif
+ }
+ }
+
+ ReleaseBuffer(buf);
+
+ mmRevmapAccessTerminate(rmAccess);
+
+ return BoolGetDatum(false);
+ }
+
+ /*
+ * ambeginscan implementation for minmax.
+ *
+ * We read the metapage here to determine the pages-per-range number that this
+ * index was built with. Note that since this cannot be changed while we're
+ * holding lock on index, it's not necessary to recompute it during mmrescan.
+ */
+ Datum
+ mmbeginscan(PG_FUNCTION_ARGS)
+ {
+ Relation r = (Relation) PG_GETARG_POINTER(0);
+ int nkeys = PG_GETARG_INT32(1);
+ int norderbys = PG_GETARG_INT32(2);
+ IndexScanDesc scan;
+ MinmaxOpaque *opaque;
+
+ scan = RelationGetIndexScan(r, nkeys, norderbys);
+
+ opaque = (MinmaxOpaque *) palloc(sizeof(MinmaxOpaque));
+ opaque->rmAccess = mmRevmapAccessInit(r, &opaque->pagesPerRange);
+ scan->opaque = opaque;
+
+ PG_RETURN_POINTER(scan);
+ }
+
+ /*
+ * Execute the index scan.
+ *
+ * This works by reading index TIDs from the revmap, and obtaining the index
+ * tuples pointed to by them; the min/max values in them are compared to the
+ * scan keys. We return into the TID bitmap all the pages in ranges
+ * corresponding to index tuples that match the scan keys.
+ *
+ * If a TID from the revmap is read as InvalidTID, we know that range is
+ * unsummarized. Pages in those ranges need to be returned regardless of scan
+ * keys.
+ */
+ Datum
+ mmgetbitmap(PG_FUNCTION_ARGS)
+ {
+ IndexScanDesc scan = (IndexScanDesc) PG_GETARG_POINTER(0);
+ TIDBitmap *tbm = (TIDBitmap *) PG_GETARG_POINTER(1);
+ Relation idxRel = scan->indexRelation;
+ Buffer currIdxBuf = InvalidBuffer;
+ MinmaxDesc *mmdesc = minmax_build_mmdesc(idxRel);
+ Oid heapOid;
+ Relation heapRel;
+ MinmaxOpaque *opaque;
+ BlockNumber nblocks;
+ BlockNumber heapBlk;
+ int totalpages = 0;
+ int keyno;
+ FmgrInfo *compareFn;
+
+ opaque = (MinmaxOpaque *) scan->opaque;
+ pgstat_count_index_scan(idxRel);
+
+ /*
+ * XXX We need to know the size of the table so that we know how long to
+ * iterate on the revmap. There's room for improvement here, in that we
+ * could have the revmap tell us when to stop iterating.
+ */
+ heapOid = IndexGetRelation(RelationGetRelid(idxRel), false);
+ heapRel = heap_open(heapOid, AccessShareLock);
+ nblocks = RelationGetNumberOfBlocks(heapRel);
+ heap_close(heapRel, AccessShareLock);
+
+ /*
+ * Obtain compare functions for all indexed column. Maybe it'd be possible
+ * to do this lazily only the first time we see a scan key that involves
+ * each particular attribute.
+ */
+ compareFn = palloc(sizeof(FmgrInfo) * mmdesc->tupdesc->natts);
+ for (keyno = 0; keyno < mmdesc->tupdesc->natts; keyno++)
+ {
+ FmgrInfo *tmp;
+
+ tmp = index_getprocinfo(idxRel, keyno + 1, MINMAX_PROCNUM_COMPARE);
+ fmgr_info_copy(&compareFn[keyno], tmp, CurrentMemoryContext);
+ }
+
+ /*
+ * Now scan the revmap. We start by querying for heap page 0,
+ * incrementing by the number of pages per range; this gives us a full
+ * view of the table.
+ */
+ for (heapBlk = 0; heapBlk < nblocks; heapBlk += opaque->pagesPerRange)
+ {
+ ItemPointerData itupptr;
+ bool addrange;
+
+ mmGetHeapBlockItemptr(opaque->rmAccess, heapBlk, &itupptr);
+
+ /*
+ * For revmap items that return InvalidTID, we must return the whole
+ * range; otherwise, fetch the index item and compare it to the scan
+ * keys.
+ */
+ if (!ItemPointerIsValid(&itupptr))
+ {
+ addrange = true;
+ }
+ else
+ {
+ Page page;
+ OffsetNumber idxoffno;
+ BlockNumber idxblkno;
+ MMTuple *tup;
+ DeformedMMTuple *dtup;
+ int keyno;
+
+ idxoffno = ItemPointerGetOffsetNumber(&itupptr);
+ idxblkno = ItemPointerGetBlockNumber(&itupptr);
+
+ if (currIdxBuf == InvalidBuffer ||
+ idxblkno != BufferGetBlockNumber(currIdxBuf))
+ {
+ if (currIdxBuf != InvalidBuffer)
+ UnlockReleaseBuffer(currIdxBuf);
+
+ Assert(idxblkno != InvalidBlockNumber);
+ currIdxBuf = ReadBuffer(idxRel, idxblkno);
+ LockBuffer(currIdxBuf, BUFFER_LOCK_SHARE);
+ }
+
+ UnlockTuple(idxRel, &itupptr, ShareLock);
+
+ page = BufferGetPage(currIdxBuf);
+ tup = (MMTuple *)
+ PageGetItem(page, PageGetItemId(page, idxoffno));
+ /* XXX probably need copies */
+ dtup = minmax_deform_tuple(mmdesc, tup);
+
+ /*
+ * Compare scan keys with min/max values stored in range. If scan
+ * keys are matched, the page range must be added to the bitmap.
+ */
+ for (keyno = 0, addrange = true;
+ keyno < scan->numberOfKeys;
+ keyno++)
+ {
+ ScanKey key = &scan->keyData[keyno];
+ AttrNumber keyattno = key->sk_attno;
+ Datum add;
+
+ /*
+ * The analysis we need to make to decide whether to include a
+ * page range in the output result is: is it possible for a
+ * tuple contained within the min/max interval specified by
+ * this index tuple to match what's specified by the scan key?
+ * For example, for a query qual such as "WHERE col < 10" we
+ * need to include a range whose minimum value is less than
+ * 10.
+ *
+ * When there are multiple scan keys, failure to meet the
+ * criteria for a single one of them is enough to discard the
+ * range as a whole.
+ */
+ add = FunctionCall5Coll(&compareFn[keyattno - 1],
+ InvalidOid, /* FIXME collation */
+ PointerGetDatum(mmdesc),
+ PointerGetDatum(dtup),
+ Int16GetDatum(keyattno),
+ UInt16GetDatum(key->sk_strategy),
+ key->sk_argument);
+ addrange = DatumGetBool(add);
+
+ /*
+ * If the current scan key doesn't match the range values,
+ * don't look at further ones.
+ */
+ if (!addrange)
+ break;
+ }
+
+ /* XXX anything to free here? */
+ pfree(dtup);
+ }
+
+ if (addrange)
+ {
+ BlockNumber pageno;
+
+ for (pageno = heapBlk;
+ pageno <= heapBlk + opaque->pagesPerRange - 1;
+ pageno++)
+ {
+ tbm_add_page(tbm, pageno);
+ totalpages++;
+ }
+ }
+ }
+
+ if (currIdxBuf != InvalidBuffer)
+ UnlockReleaseBuffer(currIdxBuf);
+
+ /*
+ * XXX We have an approximation of the number of *pages* that our scan
+ * returns, but we don't have a precise idea of the number of heap tuples
+ * involved.
+ */
+ PG_RETURN_INT64(totalpages * 10);
+ }
+
+ Datum
+ mmrescan(PG_FUNCTION_ARGS)
+ {
+ IndexScanDesc scan = (IndexScanDesc) PG_GETARG_POINTER(0);
+ ScanKey scankey = (ScanKey) PG_GETARG_POINTER(1);
+ /* other arguments ignored */
+
+ if (scankey && scan->numberOfKeys > 0)
+ memmove(scan->keyData, scankey,
+ scan->numberOfKeys * sizeof(ScanKeyData));
+
+ PG_RETURN_VOID();
+ }
+
+ Datum
+ mmendscan(PG_FUNCTION_ARGS)
+ {
+ IndexScanDesc scan = (IndexScanDesc) PG_GETARG_POINTER(0);
+ MinmaxOpaque *opaque = (MinmaxOpaque *) scan->opaque;
+
+ mmRevmapAccessTerminate(opaque->rmAccess);
+ pfree(opaque);
+
+ PG_RETURN_VOID();
+ }
+
+ Datum
+ mmmarkpos(PG_FUNCTION_ARGS)
+ {
+ elog(ERROR, "MinMax does not support mark/restore");
+ PG_RETURN_VOID();
+ }
+
+ Datum
+ mmrestrpos(PG_FUNCTION_ARGS)
+ {
+ elog(ERROR, "MinMax does not support mark/restore");
+ PG_RETURN_VOID();
+ }
+
+ /*
+ * Per-heap-tuple callback for IndexBuildHeapScan.
+ *
+ * Note we don't worry about the page range at the end of the table here; it is
+ * present in the build state struct after we're called the last time, but not
+ * inserted into the index. Caller must ensure to do so, if appropriate.
+ */
+ static void
+ mmbuildCallback(Relation index,
+ HeapTuple htup,
+ Datum *values,
+ bool *isnull,
+ bool tupleIsAlive,
+ void *state)
+ {
+ MMBuildState *mmstate = (MMBuildState *) state;
+ BlockNumber thisblock;
+ int i;
+
+ thisblock = ItemPointerGetBlockNumber(&htup->t_self);
+
+ /*
+ * If we're in a new block which belongs to the next range, summarize what
+ * we've got and start afresh. We're careful to avoid being fooled by
+ * wraparound.
+ */
+ if (thisblock > (mmstate->currRangeStart + mmstate->pagesPerRange - 1) ||
+ thisblock < mmstate->currRangeStart)
+ {
+ MMTuple *tup;
+ Size size;
+
+ MINMAX_elog(DEBUG2, "mmbuildCallback: completed a range: %u--%u",
+ mmstate->currRangeStart,
+ mmstate->currRangeStart + mmstate->pagesPerRange);
+ #if 0
+ for (i = 0; i < mmstate->indexDesc->natts; i++)
+ {
+ elog(DEBUG2, "completed a range for column %d, range: %u .. %u",
+ i,
+ DatumGetUInt32(mmstate->dtuple->values[i].min),
+ DatumGetUInt32(mmstate->dtuple->values[i].max));
+ }
+ #endif
+
+ /*
+ * Create the index tuple and insert it.
+ */
+ tup = minmax_form_tuple(mmstate->mmDesc, mmstate->dtuple, &size);
+ mm_doinsert(mmstate->irel, mmstate->rmAccess,
+ &mmstate->currentInsertBuf, mmstate->currRangeStart, tup,
+ size);
+ mmstate->numtuples++;
+ pfree(tup);
+
+ /* and set state to correspond to the new current range */
+ mmstate->currRangeStart += mmstate->pagesPerRange;
+
+ /* re-initialize state for the new range */
+ minmax_dtuple_initialize(mmstate->dtuple, mmstate->mmDesc);
+ }
+
+ /* Accumulate the current tuple into the running state */
+ for (i = 0; i < mmstate->mmDesc->tupdesc->natts; i++)
+ {
+ FmgrInfo *maybeUpdateFn;
+
+ /* FIXME must be cached somewhere */
+ maybeUpdateFn = index_getprocinfo(index, i + 1,
+ MINMAX_PROCNUM_MAYBEUPDATE);
+
+ /*
+ * Update dtuple state, if and as necessary.
+ */
+ FunctionCall5(maybeUpdateFn,
+ PointerGetDatum(mmstate->mmDesc),
+ PointerGetDatum(mmstate->dtuple),
+ UInt16GetDatum(i + 1), values[i], isnull[i]);
+ }
+ }
+
+ /*
+ * Initialize a MMBuildState appropriate to create tuples on the given index.
+ */
+ static MMBuildState *
+ initialize_mm_buildstate(Relation heapRel, Relation idxRel,
+ mmRevmapAccess *rmAccess, BlockNumber pagesPerRange,
+ IndexInfo *indexInfo)
+ {
+ MMBuildState *mmstate;
+
+ mmstate = palloc(sizeof(MMBuildState));
+
+ mmstate->irel = idxRel;
+ mmstate->numtuples = 0;
+ mmstate->currentInsertBuf = InvalidBuffer;
+ mmstate->pagesPerRange = pagesPerRange;
+ mmstate->currRangeStart = 0;
+ mmstate->rmAccess = rmAccess;
+ mmstate->mmDesc = minmax_build_mmdesc(idxRel);
+ mmstate->dtuple = minmax_new_dtuple(mmstate->mmDesc);
+ minmax_dtuple_initialize(mmstate->dtuple, mmstate->mmDesc);
+
+ return mmstate;
+ }
+
+ /*
+ * Initialize a page with the given type.
+ *
+ * Caller is responsible for marking it dirty, as appropriate.
+ */
+ void
+ mm_page_init(Page page, uint16 type)
+ {
+ MinmaxSpecialSpace *special;
+
+ PageInit(page, BLCKSZ, sizeof(MinmaxSpecialSpace));
+
+ special = (MinmaxSpecialSpace *) PageGetSpecialPointer(page);
+ special->type = type;
+ }
+
+ /*
+ * Initialize a new minmax index' metapage.
+ */
+ void
+ mm_metapage_init(Page page, BlockNumber pagesPerRange, uint16 version)
+ {
+ MinmaxMetaPageData *metadata;
+ int i;
+
+ mm_page_init(page, MINMAX_PAGETYPE_META);
+
+ metadata = (MinmaxMetaPageData *) PageGetContents(page);
+
+ metadata->pagesPerRange = pagesPerRange;
+ metadata->minmaxVersion = version;
+ for (i = 0; i < MAX_REVMAP_ARRAYPAGES; i++)
+ metadata->revmapArrayPages[i] = InvalidBlockNumber;
+ }
+
+ /*
+ * mmbuild() -- build a new minmax index.
+ */
+ Datum
+ mmbuild(PG_FUNCTION_ARGS)
+ {
+ Relation heap = (Relation) PG_GETARG_POINTER(0);
+ Relation index = (Relation) PG_GETARG_POINTER(1);
+ IndexInfo *indexInfo = (IndexInfo *) PG_GETARG_POINTER(2);
+ IndexBuildResult *result;
+ double reltuples;
+ mmRevmapAccess *rmAccess;
+ MMBuildState *mmstate;
+ Buffer meta;
+ BlockNumber pagesPerRange;
+
+ /*
+ * We expect to be called exactly once for any index relation.
+ */
+ if (RelationGetNumberOfBlocks(index) != 0)
+ elog(ERROR, "index \"%s\" already contains data",
+ RelationGetRelationName(index));
+
+ /* partial indexes not supported */
+ if (indexInfo->ii_Predicate != NIL)
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("partial indexes not supported")));
+ /* expressions not supported (yet?) */
+ if (indexInfo->ii_Expressions != NIL)
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("expression indexes not supported")));
+
+ meta = mm_getnewbuffer(index);
+ START_CRIT_SECTION();
+ mm_metapage_init(BufferGetPage(meta), MinmaxGetPagesPerRange(index),
+ MINMAX_CURRENT_VERSION);
+ MarkBufferDirty(meta);
+
+ if (RelationNeedsWAL(index))
+ {
+ xl_minmax_createidx xlrec;
+ XLogRecPtr recptr;
+ XLogRecData rdata;
+ Page page;
+
+ xlrec.node = index->rd_node;
+ xlrec.version = MINMAX_CURRENT_VERSION;
+ xlrec.pagesPerRange = MinmaxGetPagesPerRange(index);
+
+ rdata.buffer = InvalidBuffer;
+ rdata.data = (char *) &xlrec;
+ rdata.len = SizeOfMinmaxCreateIdx;
+ rdata.next = NULL;
+
+ recptr = XLogInsert(RM_MINMAX_ID, XLOG_MINMAX_CREATE_INDEX, &rdata);
+
+ page = BufferGetPage(meta);
+ PageSetLSN(page, recptr);
+ }
+
+ UnlockReleaseBuffer(meta);
+ END_CRIT_SECTION();
+
+ /*
+ * Set up an empty revmap, and get access to it
+ */
+ mmRevmapCreate(index);
+ rmAccess = mmRevmapAccessInit(index, &pagesPerRange);
+
+ /*
+ * Initialize our state, including the deformed tuple state.
+ */
+ mmstate = initialize_mm_buildstate(heap, index, rmAccess, pagesPerRange,
+ indexInfo);
+
+ /*
+ * Now scan the relation. No syncscan allowed here because we want the
+ * heap blocks in physical order.
+ */
+ reltuples = IndexBuildHeapScan(heap, index, indexInfo, false,
+ mmbuildCallback, (void *) mmstate);
+
+ /* FIXME process the final batch */
+
+
+ /* release the last index buffer used */
+ if (!BufferIsInvalid(mmstate->currentInsertBuf))
+ {
+ ReleaseBuffer(mmstate->currentInsertBuf);
+ mmstate->currentInsertBuf = InvalidBuffer;
+ }
+
+ mmRevmapAccessTerminate(mmstate->rmAccess);
+
+ /*
+ * Return statistics
+ */
+ result = (IndexBuildResult *) palloc(sizeof(IndexBuildResult));
+
+ result->heap_tuples = reltuples;
+ result->index_tuples = mmstate->numtuples;
+
+ PG_RETURN_POINTER(result);
+ }
+
+ Datum
+ mmbuildempty(PG_FUNCTION_ARGS)
+ {
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("unlogged MinMax indexes are not supported")));
+
+ PG_RETURN_VOID();
+ }
+
+ static MinmaxDesc *
+ minmax_build_mmdesc(Relation rel)
+ {
+ MinmaxOpers **opers;
+ MinmaxDesc *mmdesc;
+ TupleDesc tupdesc;
+ int totalopers = 0;
+ int totalstored = 0;
+ int keyno;
+ long totalsize;
+ int curroffset;
+ Datum indclassDatum;
+ oidvector *indclass;
+ bool isnull;
+
+ tupdesc = RelationGetDescr(rel);
+
+ indclassDatum = SysCacheGetAttr(INDEXRELID, rel->rd_indextuple,
+ Anum_pg_index_indclass, &isnull);
+ Assert(!isnull);
+ indclass = (oidvector *) DatumGetPointer(indclassDatum);
+ opers = (MinmaxOpers **) palloc(sizeof(MinmaxOpers *) * tupdesc->natts);
+ for (keyno = 0; keyno < tupdesc->natts; keyno++)
+ {
+ Oid opfam = get_opclass_family(indclass->values[keyno]);
+ Oid idxtypid = tupdesc->attrs[keyno]->atttypid;
+ FmgrInfo getOpers;
+
+ fmgr_info_copy(&getOpers,
+ index_getprocinfo(rel, keyno + 1, MINMAX_PROCNUM_GETOPERS),
+ CurrentMemoryContext);
+
+ opers[keyno] = (MinmaxOpers *)
+ DatumGetPointer(FunctionCall2(&getOpers,
+ ObjectIdGetDatum(opfam),
+ ObjectIdGetDatum(idxtypid)));
+ totalopers += opers[keyno]->nopers;
+ totalstored += opers[keyno]->nstored;
+ }
+
+ totalsize = offsetof(MinmaxDesc, perCol) +
+ sizeof(MinmaxDescPerCol) * tupdesc->natts +
+ sizeof(FmgrInfo) * totalopers +
+ sizeof(Oid) * totalopers;
+
+ mmdesc = palloc(totalsize);
+ mmdesc->tupdesc = CreateTupleDescCopy(tupdesc);
+ mmdesc->disktdesc = NULL; /* generated lazily */
+ mmdesc->totalstored = totalstored;
+ mmdesc->operoids = (Oid *) ((char *) mmdesc +
+ offsetof(MinmaxDesc, perCol) +
+ sizeof(MinmaxDescPerCol) * tupdesc->natts);
+ mmdesc->opers = (FmgrInfo *) ((char *) mmdesc->operoids +
+ sizeof(Oid) * totalopers);
+
+ curroffset = 0;
+ for (keyno = 0; keyno < tupdesc->natts; keyno++)
+ {
+ int opno;
+
+ mmdesc->perCol[keyno].numopers = opers[keyno]->nopers;
+ mmdesc->perCol[keyno].numstored = opers[keyno]->nstored;
+
+ /* Copy the operator OIDs from what the opclass told us */
+ mmdesc->perCol[keyno].operoids = mmdesc->operoids + curroffset;
+ memcpy(mmdesc->perCol[keyno].operoids, opers[keyno]->opers,
+ sizeof(Oid) * opers[keyno]->nopers);
+
+ /*
+ * Fill in the operator's FmgrInfos. XXX is it possible to do this
+ * lazily to avoid initializing operators that go unused much of the
+ * time? We'd need a tweak to minmax_get_operfn plus a flag array
+ * to indicate which ones have been initialized ...
+ */
+ mmdesc->perCol[keyno].opers = mmdesc->opers + curroffset;
+ for (opno = 0; opno < mmdesc->perCol[keyno].numopers; opno++)
+ fmgr_info(mmdesc->perCol[keyno].operoids[opno],
+ &mmdesc->perCol[keyno].opers[opno]);
+
+ curroffset += opers[keyno]->nopers;
+ }
+
+ for (keyno = 0; keyno < tupdesc->natts; keyno++)
+ pfree(opers[keyno]);
+ pfree(opers);
+
+ return mmdesc;
+ }
+
+ /*
+ * qsort comparator for ItemPointerData items
+ */
+ static int
+ qsortCompareItemPointers(const void *a, const void *b)
+ {
+ return ItemPointerCompare((ItemPointer) a, (ItemPointer) b);
+ }
+
+ /*
+ * Remove index tuples that are no longer useful.
+ *
+ * While at it, return in nonsummed the array (and in numnonsummed its size) of
+ * block numbers for which the revmap returns InvalidTid; this is used in a
+ * later stage to execute re-summarization. (Each block number returned
+ * corresponds to the heap page number with which each unsummarized range
+ * starts.) Space for the array is palloc'ed, and must be freed by caller.
+ *
+ * idxRel is the index relation; heapNumBlocks is the size of the heap
+ * relation; strategy is appropriate for bulk scanning.
+ */
+ static void
+ remove_deletable_tuples(Relation idxRel, BlockNumber heapNumBlocks,
+ BufferAccessStrategy strategy,
+ BlockNumber **nonsummed, int *numnonsummed)
+ {
+ HASHCTL hctl;
+ HTAB *tuples;
+ HASH_SEQ_STATUS status;
+ BlockNumber nblocks;
+ BlockNumber blk;
+ mmRevmapAccess *rmAccess;
+ BlockNumber heapBlk;
+ BlockNumber pagesPerRange;
+ int numitems = 0;
+ int numdeletable = 0;
+ ItemPointerData *deletable;
+ int start;
+ int i;
+ BlockNumber *nonsumm = NULL;
+ int maxnonsumm = 0;
+ int numnonsumm = 0;
+
+ typedef struct DeletableTuple
+ {
+ ItemPointerData tid;
+ bool referenced;
+ } DeletableTuple;
+
+ nblocks = RelationGetNumberOfBlocks(idxRel);
+
+ /* Initialize hash used to track deletable tuples */
+ memset(&hctl, 0, sizeof(hctl));
+ hctl.keysize = sizeof(ItemPointerData);
+ hctl.entrysize = sizeof(DeletableTuple);
+ hctl.hcxt = CurrentMemoryContext;
+ hctl.hash = tag_hash;
+
+ /* assume ten entries per page. No harm in getting this wrong */
+ tuples = hash_create("mmvacuumcleanup", nblocks * 10, &hctl,
+ HASH_CONTEXT | HASH_FUNCTION | HASH_ELEM);
+
+ /*
+ * Scan the index sequentially, entering each item into a hash table.
+ * Initially, the items are marked as not referenced.
+ */
+ for (blk = 0; blk < nblocks; blk++)
+ {
+ Buffer buf;
+ Page page;
+ OffsetNumber offno;
+ MinmaxSpecialSpace *special;
+
+ vacuum_delay_point();
+
+ buf = ReadBufferExtended(idxRel, MAIN_FORKNUM, blk, RBM_NORMAL,
+ strategy);
+ page = BufferGetPage(buf);
+
+ /*
+ * Verify the type of the page we got; if it's not a regular page,
+ * ignore it.
+ */
+ special = (MinmaxSpecialSpace *) PageGetSpecialPointer(page);
+ if (special->type != MINMAX_PAGETYPE_REGULAR)
+ {
+ ReleaseBuffer(buf);
+ continue;
+ }
+
+ /*
+ * Enter each live tuple into the hash table
+ */
+ LockBuffer(buf, BUFFER_LOCK_SHARE);
+ for (offno = 1; offno <= PageGetMaxOffsetNumber(page); offno++)
+ {
+ ItemPointerData tid;
+ ItemId itemid;
+ bool found;
+ DeletableTuple *hitem;
+
+ itemid = PageGetItemId(page, offno);
+ if (!ItemIdHasStorage(itemid))
+ continue;
+
+ ItemPointerSet(&tid, blk, offno);
+ hitem = (DeletableTuple *)
+ hash_search(tuples, &tid, HASH_ENTER, &found);
+ Assert(!found);
+ hitem->referenced = false;
+ numitems++;
+ }
+ UnlockReleaseBuffer(buf);
+ }
+
+ /*
+ * Now scan the revmap, and determine which of these TIDs are still
+ * referenced
+ */
+ rmAccess = mmRevmapAccessInit(idxRel, &pagesPerRange);
+ for (heapBlk = 0; heapBlk < heapNumBlocks; heapBlk += pagesPerRange)
+ {
+ ItemPointerData itupptr;
+ DeletableTuple *hitem;
+ bool found;
+
+ mmGetHeapBlockItemptr(rmAccess, heapBlk, &itupptr);
+
+ if (!ItemPointerIsValid(&itupptr))
+ {
+ /*
+ * Ignore revmap entries set to invalid. Before doing so, if the
+ * heap page range is complete but not summarized, store its
+ * initial page number in the unsummarized array, for later
+ * summarization.
+ */
+ if (heapBlk + pagesPerRange < heapNumBlocks)
+ {
+ if (maxnonsumm == 0)
+ {
+ Assert(!nonsumm);
+ maxnonsumm = 8;
+ nonsumm = palloc(sizeof(BlockNumber) * maxnonsumm);
+ }
+ else if (numnonsumm >= maxnonsumm)
+ {
+ maxnonsumm *= 2;
+ nonsumm = repalloc(nonsumm, sizeof(BlockNumber) * maxnonsumm);
+ }
+
+ nonsumm[numnonsumm++] = heapBlk;
+ }
+
+ continue;
+ }
+ else
+ UnlockTuple(idxRel, &itupptr, ShareLock);
+
+ hitem = (DeletableTuple *) hash_search(tuples,
+ &itupptr,
+ HASH_FIND,
+ &found);
+ /*
+ * If the item is not in the hash, it must have been inserted after the
+ * index was scanned, and therefore we should leave things well alone.
+ * (There might be a leftover entry, but it's okay because next vacuum
+ * will remove it.)
+ */
+ if (!found)
+ continue;
+
+ hitem->referenced = true;
+
+ /* discount items set as referenced */
+ numitems--;
+ }
+ Assert(numitems >= 0);
+
+ mmRevmapAccessTerminate(rmAccess);
+
+ /*
+ * Now scan the hash, and keep track of the removable (i.e. not referenced,
+ * not locked) tuples.
+ */
+ deletable = palloc(sizeof(ItemPointerData) * numitems);
+
+ hash_freeze(tuples);
+ hash_seq_init(&status, tuples);
+ for (;;)
+ {
+ DeletableTuple *hitem;
+
+ hitem = hash_seq_search(&status);
+ if (!hitem)
+ break;
+ if (hitem->referenced)
+ continue;
+ if (!ConditionalLockTuple(idxRel, &hitem->tid, ExclusiveLock))
+ continue;
+
+ /*
+ * By here, we know this tuple is not referenced from the revmap.
+ * Also, since we hold the tuple lock, we know that if there is a
+ * concurrent scan that had obtained the tuple before the reference
+ * got removed, either that scan is not looking at the tuple (because
+ * that would have prevented us from getting the tuple lock) or it is
+ * holding the containing buffer's lock. If the former, then there's
+ * no problem with removing the tuple immediately; if the latter, we
+ * will block below trying to acquire that lock, so by the time we are
+ * unblocked, the concurrent scan will no longer be interested in the
+ * tuple contents anymore. Therefore, this tuple can be removed from
+ * the block.
+ */
+ UnlockTuple(idxRel, &hitem->tid, ExclusiveLock);
+
+ deletable[numdeletable++] = hitem->tid;
+ }
+
+ /*
+ * Now sort the array of deletable index tuples, and walk this array by
+ * pages doing bulk deletion of items on each page; the free space map is
+ * updated for pages on which we delete item.
+ */
+ qsort(deletable, numdeletable, sizeof(ItemPointerData),
+ qsortCompareItemPointers);
+
+ for (start = 0, i = 0; i < numdeletable; i++)
+ {
+ /*
+ * Are we at the end of the items that together belong in one
+ * particular page? If so, then it's deletion time.
+ */
+ if (i == numdeletable - 1 ||
+ (ItemPointerGetBlockNumber(&deletable[start]) !=
+ ItemPointerGetBlockNumber(&deletable[i + 1])))
+ {
+ OffsetNumber *offnos;
+ int noffs;
+ Buffer buf;
+ Page page;
+ int j;
+ BlockNumber blk;
+ int freespace;
+
+ vacuum_delay_point();
+
+ blk = ItemPointerGetBlockNumber(&deletable[start]);
+ buf = ReadBufferExtended(idxRel, MAIN_FORKNUM, blk,
+ RBM_NORMAL, strategy);
+ LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
+ page = BufferGetPage(buf);
+
+ noffs = i + 1 - start;
+ offnos = palloc(sizeof(OffsetNumber) * noffs);
+
+ for (j = 0; j < noffs; j++)
+ offnos[j] = ItemPointerGetOffsetNumber(&deletable[start + j]);
+
+ /*
+ * Now defragment the page.
+ */
+ START_CRIT_SECTION();
+
+ PageIndexDeleteNoCompact(page, offnos, noffs);
+ MarkBufferDirty(buf);
+
+ /* XLOG stuff */
+ if (RelationNeedsWAL(idxRel))
+ {
+ xl_minmax_bulkremove xlrec;
+ XLogRecPtr recptr;
+ XLogRecData rdata[2];
+
+ xlrec.node = idxRel->rd_node;
+ xlrec.block = blk;
+ rdata[0].data = (char *) &xlrec;
+ rdata[0].len = SizeOfMinmaxBulkRemove;
+ rdata[0].buffer = InvalidBuffer;
+ rdata[0].buffer_std = false;
+ rdata[0].next = &(rdata[1]);
+
+ /*
+ * The OffsetNumber array is not actually in the buffer, but we
+ * pretend that it is. When XLogInsert stores the whole
+ * buffer, the offset array need not be stored too.
+ */
+ rdata[1].data = (char *) offnos;
+ rdata[1].len = sizeof(OffsetNumber) * noffs;
+ rdata[1].buffer = buf;
+ rdata[1].buffer_std = true;
+ rdata[1].next = NULL;
+
+ recptr = XLogInsert(RM_MINMAX_ID, XLOG_MINMAX_BULKREMOVE,
+ rdata);
+
+ PageSetLSN(page, recptr);
+ }
+
+ END_CRIT_SECTION();
+
+ /* next iteration starts where this one ended */
+ start = i + 1;
+
+ /* remember free space while we have the buffer locked */
+ freespace = PageGetFreeSpace(page);
+
+ UnlockReleaseBuffer(buf);
+ pfree(offnos);
+
+ RecordPageWithFreeSpace(idxRel, blk, freespace);
+ }
+ }
+
+ pfree(deletable);
+
+ /* Finally, ensure the index' FSM is consistent */
+ FreeSpaceMapVacuum(idxRel);
+
+ *nonsummed = nonsumm;
+ *numnonsummed = numnonsumm;
+
+ hash_destroy(tuples);
+ }
+
+ /*
+ * Summarize the given page ranges of the given index.
+ */
+ static void
+ rerun_summarization(Relation idxRel, Relation heapRel,
+ mmRevmapAccess *rmAccess, BlockNumber pagesPerRange,
+ BlockNumber *nonsummarized, int numnonsummarized)
+ {
+ int i;
+ IndexInfo *indexInfo;
+ MMBuildState *mmstate;
+
+ indexInfo = BuildIndexInfo(idxRel);
+
+ mmstate = initialize_mm_buildstate(heapRel, idxRel, rmAccess, pagesPerRange, indexInfo);
+
+ for (i = 0; i < numnonsummarized; i++)
+ {
+ BlockNumber blk = nonsummarized[i];
+ ItemPointerData iptr;
+ MMTuple *tup;
+ Size size;
+
+ mmstate->currRangeStart = blk;
+
+ mmGetHeapBlockItemptr(rmAccess, blk, &iptr);
+ /* it can't have been re-summarized concurrently .. */
+ Assert(!ItemPointerIsValid(&iptr));
+
+ IndexBuildHeapRangeScan(heapRel, idxRel, indexInfo, false,
+ blk, pagesPerRange,
+ mmbuildCallback, (void *) mmstate);
+
+ /*
+ * Create the index tuple containing min/max values, and insert it.
+ * Note mmbuildCallback didn't have the chance to actually insert
+ * anything into the index, because the heapscan should have ended
+ * just as it reached the final tuple in the range.
+ */
+ tup = minmax_form_tuple(mmstate->mmDesc, mmstate->dtuple, &size);
+ mm_doinsert(mmstate->irel, mmstate->rmAccess,
+ &mmstate->currentInsertBuf, mmstate->currRangeStart, tup,
+ size);
+ mmstate->numtuples++;
+ pfree(tup);
+
+ minmax_dtuple_initialize(mmstate->dtuple, mmstate->mmDesc);
+ }
+
+ if (!BufferIsInvalid(mmstate->currentInsertBuf))
+ {
+ ReleaseBuffer(mmstate->currentInsertBuf);
+ mmstate->currentInsertBuf = InvalidBuffer;
+ }
+ }
+
+ /*
+ * ambulkdelete
+ * Since there are no per-heap-tuple index tuples in minmax indexes,
+ * there's not a lot we can do here.
+ *
+ * XXX we could mark item tuples as "dirty" (when a minimum or maximum heap
+ * tuple is deleted), meaning the need to re-run summarization on the affected
+ * range. Need to an extra flag in mmtuples for that.
+ */
+ Datum
+ mmbulkdelete(PG_FUNCTION_ARGS)
+ {
+ /* other arguments are not currently used */
+ IndexBulkDeleteResult *stats = (IndexBulkDeleteResult *) PG_GETARG_POINTER(1);
+
+ /* allocate stats if first time through, else re-use existing struct */
+ if (stats == NULL)
+ stats = (IndexBulkDeleteResult *) palloc0(sizeof(IndexBulkDeleteResult));
+
+ PG_RETURN_POINTER(stats);
+ }
+
+ /*
+ * This routine is in charge of "vacuuming" a minmax index: 1) remove index
+ * tuples that are no longer referenced from the revmap. 2) summarize ranges
+ * that are currently unsummarized.
+ */
+ Datum
+ mmvacuumcleanup(PG_FUNCTION_ARGS)
+ {
+ IndexVacuumInfo *info = (IndexVacuumInfo *) PG_GETARG_POINTER(0);
+ IndexBulkDeleteResult *stats = (IndexBulkDeleteResult *) PG_GETARG_POINTER(1);
+ mmRevmapAccess *rmAccess;
+ BlockNumber *nonsummarized = NULL;
+ int numnonsummarized;
+ Relation heapRel;
+ BlockNumber heapNumBlocks;
+ BlockNumber pagesPerRange;
+
+ /* No-op in ANALYZE ONLY mode */
+ if (info->analyze_only)
+ PG_RETURN_POINTER(stats);
+
+ rmAccess = mmRevmapAccessInit(info->index, &pagesPerRange);
+
+ heapRel = heap_open(IndexGetRelation(RelationGetRelid(info->index), false),
+ AccessShareLock);
+
+ /*
+ * First scan the index, removing index tuples that are no longer
+ * referenced from the revmap. While at it, collect the page numbers of
+ * ranges that are not summarized.
+ */
+ heapNumBlocks = RelationGetNumberOfBlocks(heapRel);
+ remove_deletable_tuples(info->index, heapNumBlocks, info->strategy,
+ &nonsummarized, &numnonsummarized);
+
+ /* and summarize the ranges collected above */
+ if (nonsummarized)
+ {
+ rerun_summarization(info->index, heapRel, rmAccess, pagesPerRange,
+ nonsummarized, numnonsummarized);
+ pfree(nonsummarized);
+ }
+
+ mmRevmapAccessTerminate(rmAccess);
+ heap_close(heapRel, AccessShareLock);
+
+ PG_RETURN_POINTER(stats);
+ }
+
+ Datum
+ mmoptions(PG_FUNCTION_ARGS)
+ {
+ Datum reloptions = PG_GETARG_DATUM(0);
+ bool validate = PG_GETARG_BOOL(1);
+ relopt_value *options;
+ MinmaxOptions *rdopts;
+ int numoptions;
+ static const relopt_parse_elt tab[] = {
+ {"pages_per_range", RELOPT_TYPE_INT, offsetof(MinmaxOptions, pagesPerRange)}
+ };
+
+ options = parseRelOptions(reloptions, validate, RELOPT_KIND_MINMAX,
+ &numoptions);
+
+ /* if none set, we're done */
+ if (numoptions == 0)
+ PG_RETURN_NULL();
+
+ rdopts = allocateReloptStruct(sizeof(MinmaxOptions), options, numoptions);
+
+ fillRelOptions((void *) rdopts, sizeof(MinmaxOptions), options, numoptions,
+ validate, tab, lengthof(tab));
+
+ pfree(options);
+
+ PG_RETURN_BYTEA_P(rdopts);
+ }
+
+ /*
+ * Insert an index tuple into the index relation. The revmap is updated to
+ * mark the range containing the given page as pointing to the inserted entry.
+ *
+ * The buffer, if valid, is checked for free space to insert the new entry;
+ * if there isn't enough, a new buffer is obtained and pinned.
+ *
+ * The buffer is marked dirty.
+ */
+ static void
+ mm_doinsert(Relation idxrel, mmRevmapAccess *rmAccess, Buffer *buffer,
+ BlockNumber heapblkno, MMTuple *tup, Size itemsz)
+ {
+ Page page;
+ BlockNumber blk;
+ OffsetNumber off;
+ bool extended;
+
+ itemsz = MAXALIGN(itemsz);
+
+ extended = mm_getinsertbuffer(idxrel, buffer, itemsz);
+ page = BufferGetPage(*buffer);
+
+ if (PageGetFreeSpace(page) < itemsz)
+ ereport(ERROR,
+ (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
+ errmsg("index row size %lu exceeds maximum for index \"%s\"",
+ itemsz, RelationGetRelationName(idxrel))));
+
+ START_CRIT_SECTION();
+ off = PageAddItem(page, (Item) tup, itemsz, InvalidOffsetNumber,
+ false, false);
+ MarkBufferDirty(*buffer);
+
+ blk = BufferGetBlockNumber(*buffer);
+ MINMAX_elog(DEBUG2, "inserted tuple (%u,%u) for range starting at %u",
+ blk, off, heapblkno);
+
+ /* XLOG stuff */
+ if (RelationNeedsWAL(idxrel))
+ {
+ xl_minmax_insert xlrec;
+ XLogRecPtr recptr;
+ XLogRecData rdata[2];
+ uint8 info = XLOG_MINMAX_INSERT;
+
+ xlrec.target.node = idxrel->rd_node;
+ ItemPointerSet(&xlrec.target.tid, blk, off);
+ xlrec.overwrite = false;
+ rdata[0].data = (char *) &xlrec;
+ rdata[0].len = SizeOfMinmaxInsert;
+ rdata[0].buffer = InvalidBuffer;
+ rdata[0].buffer_std = false;
+ rdata[0].next = &(rdata[1]);
+
+ rdata[1].data = (char *) tup;
+ rdata[1].len = itemsz;
+ rdata[1].buffer = *buffer;
+ rdata[1].buffer_std = true;
+ rdata[1].next = NULL;
+
+ /*
+ * If this is the first tuple in the page, we can reinit the page
+ * instead of restoring the whole thing. Set flag, and hide buffer
+ * references from XLogInsert.
+ */
+ if (extended)
+ {
+ info |= XLOG_MINMAX_INIT_PAGE;
+ rdata[1].buffer = InvalidBuffer;
+ }
+
+ recptr = XLogInsert(RM_MINMAX_ID, info, rdata);
+
+ PageSetLSN(page, recptr);
+ }
+
+ END_CRIT_SECTION();
+
+ /*
+ * Note we need to keep the lock on the buffer until after the revmap
+ * has been updated. Otherwise, a concurrent scanner could try to obtain
+ * the index tuple from the revmap before we're done writing it.
+ */
+ mmSetHeapBlockItemptr(rmAccess, heapblkno, blk, off);
+
+ LockBuffer(*buffer, BUFFER_LOCK_UNLOCK);
+ }
+
+ /*
+ * Return an exclusively-locked buffer resulting from extending the relation.
+ */
+ Buffer
+ mm_getnewbuffer(Relation irel)
+ {
+ Buffer buffer;
+ bool needLock = !RELATION_IS_LOCAL(irel);
+
+ /* FIXME need to request a MaxFSMRequestSize page from the FSM here */
+
+ if (needLock)
+ LockRelationForExtension(irel, ExclusiveLock);
+
+ buffer = ReadBuffer(irel, P_NEW);
+ LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+
+ MINMAX_elog(DEBUG2, "mm_getnewbuffer: extending to page %u",
+ BufferGetBlockNumber(buffer));
+
+ if (needLock)
+ UnlockRelationForExtension(irel, ExclusiveLock);
+
+ return buffer;
+ }
+
+ /*
+ * Return a pinned and locked buffer which can be used to insert an index item
+ * of size itemsz.
+ *
+ * The passed buffer argument is tested for free space; if it has enough, it is
+ * locked and returned. Otherwise, that buffer (if valid) is unpinned, a new
+ * buffer is obtained, and returned pinned and locked.
+ *
+ * If there's no existing page with enough free to accomodate the new item,
+ * the relation is extended. This function returns true if this happens, false
+ * otherwise.
+ */
+ static bool
+ mm_getinsertbuffer(Relation irel, Buffer *buffer, Size itemsz)
+ {
+ Buffer buf;
+ bool extended = false;
+
+ buf = *buffer;
+
+ if (BufferIsInvalid(buf) ||
+ (PageGetFreeSpace(BufferGetPage(buf)) < itemsz))
+ {
+ Page page;
+
+ /*
+ * By the time we break out of this loop, buf is a locked and pinned
+ * buffer which has enough free space to satisfy the requirement.
+ */
+ for (;;)
+ {
+ BlockNumber blk;
+ int freespace;
+
+ blk = GetPageWithFreeSpace(irel, itemsz);
+ if (blk == InvalidBlockNumber)
+ {
+ /*
+ * There's not enough free space in any existing index page,
+ * according to the FSM: extend the relation to obtain a shiny
+ * new page.
+ */
+ buf = mm_getnewbuffer(irel);
+ page = BufferGetPage(buf);
+ mm_page_init(page, MINMAX_PAGETYPE_REGULAR);
+
+ /*
+ * If an entirely new page does not contain enough free space
+ * for the new item, then surely that item is oversized.
+ * Complain loudly; but first make sure we record the page as
+ * free, for next time.
+ */
+ freespace = PageGetFreeSpace(page);
+ RecordPageWithFreeSpace(irel, BufferGetBlockNumber(buf),
+ freespace);
+ if (freespace < itemsz)
+ ereport(ERROR,
+ (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
+ errmsg("index row size %lu exceeds maximum %lu for index \"%s\"",
+ (unsigned long) itemsz,
+ (unsigned long) freespace,
+ RelationGetRelationName(irel))));
+ extended = true;
+ break;
+ }
+
+ Assert(blk != InvalidBlockNumber);
+ buf = ReadBuffer(irel, blk);
+ LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
+ page = BufferGetPage(buf);
+ freespace = PageGetFreeSpace(page);
+ if (freespace >= itemsz)
+ break;
+
+ /* Not really enough space: register reality and start over */
+ UnlockReleaseBuffer(buf);
+ RecordPageWithFreeSpace(irel, blk, freespace);
+ }
+
+ if (!BufferIsInvalid(*buffer))
+ ReleaseBuffer(*buffer);
+ *buffer = buf;
+ }
+ else
+ LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
+
+ /*
+ * XXX we could perhaps avoid this if we used RelationSetTargetBlock ...
+ */
+ if (extended)
+ FreeSpaceMapVacuum(irel);
+
+ return extended;
+ }
+
+ FmgrInfo *
+ minmax_get_operfn(MinmaxDesc *mmdesc, AttrNumber attno, uint16 operno)
+ {
+ return &(mmdesc->perCol[attno - 1].opers[operno]);
+ }
*** /dev/null
--- b/src/backend/access/minmax/mmrevmap.c
***************
*** 0 ****
--- 1,683 ----
+ /*
+ * mmrevmap.c
+ * Reverse range map for MinMax indexes
+ *
+ * The reverse range map (revmap) is a translation structure for minmax
+ * indexes: for each page range, there is one most-up-to-date summary tuple,
+ * and its location is tracked by the revmap. Whenever a new tuple is inserted
+ * into a table that violates the previously recorded min/max values, a new
+ * tuple is inserted into the index and the revmap is updated to point to it.
+ *
+ * The pages of the revmap are interspersed in the index's main fork. The
+ * first revmap page is always the index's page number one (that is,
+ * immediately after the metapage). Subsequent revmap pages are allocated as
+ * they are needed; their locations are tracked by "array pages". The metapage
+ * contains a large BlockNumber array, which correspond to array pages. Thus,
+ * to find the second revmap page, we read the metapage and obtain the block
+ * number of the first array page; we then read that page, and the first
+ * element in it is the revmap page we're looking for.
+ *
+ * Portions Copyright (c) 1996-2013, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * IDENTIFICATION
+ * src/backend/access/minmax/mmrevmap.c
+ */
+ #include "postgres.h"
+
+ #include "access/heapam_xlog.h"
+ #include "access/minmax.h"
+ #include "access/minmax_internal.h"
+ #include "access/minmax_page.h"
+ #include "access/minmax_revmap.h"
+ #include "access/minmax_xlog.h"
+ #include "access/rmgr.h"
+ #include "miscadmin.h"
+ #include "storage/bufmgr.h"
+ #include "storage/lmgr.h"
+ #include "storage/relfilenode.h"
+ #include "storage/smgr.h"
+ #include "utils/memutils.h"
+
+
+
+ /*
+ * In regular revmap pages, each item stores an ItemPointerData. These defines
+ * let one find the logical revmap page number and index number of the revmap
+ * item for the given heap block number.
+ */
+ #define HEAPBLK_TO_REVMAP_BLK(pagesPerRange, heapBlk) \
+ ((heapBlk / pagesPerRange) / REGULAR_REVMAP_PAGE_MAXITEMS)
+ #define HEAPBLK_TO_REVMAP_INDEX(pagesPerRange, heapBlk) \
+ ((heapBlk / pagesPerRange) % REGULAR_REVMAP_PAGE_MAXITEMS)
+
+ /*
+ * In array revmap pages, each item stores a BlockNumber. These defines let
+ * one find the page and index number of a given revmap block number. Note
+ * that the first revmap page (revmap logical page number 0) is always stored
+ * in physical block number 1, so array pages do not store that one.
+ */
+ #define MAPBLK_TO_RMARRAY_BLK(rmBlk) ((rmBlk - 1) / ARRAY_REVMAP_PAGE_MAXITEMS)
+ #define MAPBLK_TO_RMARRAY_INDEX(rmBlk) ((rmBlk - 1) % ARRAY_REVMAP_PAGE_MAXITEMS)
+
+
+ struct mmRevmapAccess
+ {
+ Relation idxrel;
+ BlockNumber pagesPerRange;
+ Buffer metaBuf;
+ Buffer currBuf;
+ Buffer currArrayBuf;
+ BlockNumber *revmapArrayPages;
+ };
+ /* typedef appears in minmax_revmap.h */
+
+
+ /*
+ * Initialize an access object for a reverse range map, which can be used to
+ * read stuff from it. This must be freed by mmRevmapAccessTerminate when caller
+ * is done with it.
+ */
+ mmRevmapAccess *
+ mmRevmapAccessInit(Relation idxrel, BlockNumber *pagesPerRange)
+ {
+ mmRevmapAccess *rmAccess;
+ Buffer meta;
+ MinmaxMetaPageData *metadata;
+
+ meta = ReadBuffer(idxrel, MINMAX_METAPAGE_BLKNO);
+ metadata = (MinmaxMetaPageData *) PageGetContents(BufferGetPage(meta));
+
+ rmAccess = palloc(sizeof(mmRevmapAccess));
+ rmAccess->metaBuf = meta;
+ rmAccess->idxrel = idxrel;
+ rmAccess->pagesPerRange = metadata->pagesPerRange;
+ rmAccess->currBuf = InvalidBuffer;
+ rmAccess->currArrayBuf = InvalidBuffer;
+ rmAccess->revmapArrayPages = NULL;
+
+ if (pagesPerRange)
+ *pagesPerRange = metadata->pagesPerRange;
+
+ return rmAccess;
+ }
+
+ /*
+ * Release resources associated with a revmap access object.
+ */
+ void
+ mmRevmapAccessTerminate(mmRevmapAccess *rmAccess)
+ {
+ if (rmAccess->revmapArrayPages != NULL)
+ pfree(rmAccess->revmapArrayPages);
+ if (rmAccess->metaBuf != InvalidBuffer)
+ ReleaseBuffer(rmAccess->metaBuf);
+ if (rmAccess->currBuf != InvalidBuffer)
+ ReleaseBuffer(rmAccess->currBuf);
+ if (rmAccess->currArrayBuf != InvalidBuffer)
+ ReleaseBuffer(rmAccess->currArrayBuf);
+ pfree(rmAccess);
+ }
+
+ /*
+ * In the given revmap page, which is used in a minmax index of pagesPerRange
+ * pages per range, set the element corresponding to heap block number heapBlk
+ * to the value (blkno, offno).
+ *
+ * Caller must have obtained the correct revmap page.
+ *
+ * This is used both in regular operation and during WAL replay.
+ */
+ void
+ rm_page_set_iptr(Page page, BlockNumber pagesPerRange, BlockNumber heapBlk,
+ BlockNumber blkno, OffsetNumber offno)
+ {
+ RevmapContents *contents;
+ ItemPointerData *iptr;
+
+ contents = (RevmapContents *) PageGetContents(page);
+ iptr = (ItemPointerData *) contents->rmr_tids;
+ iptr += HEAPBLK_TO_REVMAP_INDEX(pagesPerRange, heapBlk);
+
+ ItemPointerSet(iptr, blkno, offno);
+ }
+
+ /*
+ * Initialize a new regular revmap page, which stores the given revmap logical
+ * page number. The newly allocated physical block number is returned.
+ *
+ * Used both by regular code path as well as during xlog replay.
+ */
+ BlockNumber
+ initialize_rmr_page(Buffer newbuf, BlockNumber mapBlk)
+ {
+ BlockNumber blkno;
+ Page page;
+ RevmapContents *contents;
+
+ page = BufferGetPage(newbuf);
+
+ mm_page_init(page, MINMAX_PAGETYPE_REVMAP);
+ contents = (RevmapContents *) PageGetContents(page);
+ contents->rmr_logblk = mapBlk;
+ /* the rmr_tids array is initialized to all invalid by PageInit */
+
+ blkno = BufferGetBlockNumber(newbuf);
+
+ return blkno;
+ }
+
+ /*
+ * Lock the metapage as specified by called, and update the given rmAccess with
+ * the metapage data. The metapage buffer is locked when this function
+ * returns; it's the caller's responsibility to unlock it.
+ */
+ static void
+ rmaccess_get_metapage(mmRevmapAccess *rmAccess, int lockmode)
+ {
+ MinmaxMetaPageData *metadata;
+ MinmaxSpecialSpace *special PG_USED_FOR_ASSERTS_ONLY;
+ Page metapage;
+
+ LockBuffer(rmAccess->metaBuf, lockmode);
+ metapage = BufferGetPage(rmAccess->metaBuf);
+
+ #ifdef USE_ASSERT_CHECKING
+ /* ensure we really got the metapage */
+ special = (MinmaxSpecialSpace *) PageGetSpecialPointer(metapage);
+ Assert(special->type == MINMAX_PAGETYPE_META);
+ #endif
+
+ /* first time through? allocate the array */
+ if (rmAccess->revmapArrayPages == NULL)
+ rmAccess->revmapArrayPages =
+ palloc(sizeof(BlockNumber) * MAX_REVMAP_ARRAYPAGES);
+
+ metadata = (MinmaxMetaPageData *) PageGetContents(metapage);
+ memcpy(rmAccess->revmapArrayPages, metadata->revmapArrayPages,
+ sizeof(BlockNumber) * MAX_REVMAP_ARRAYPAGES);
+ }
+
+ /*
+ * Given a buffer (hopefully containing a blank page), set it up as a revmap
+ * array page.
+ *
+ * Used both by regular code path as well as during xlog replay.
+ */
+ void
+ initialize_rma_page(Buffer buf)
+ {
+ Page arrayPg;
+ RevmapArrayContents *contents;
+
+ arrayPg = BufferGetPage(buf);
+ mm_page_init(arrayPg, MINMAX_PAGETYPE_REVMAP_ARRAY);
+ contents = (RevmapArrayContents *) PageGetContents(arrayPg);
+ contents->rma_nblocks = 0;
+ /* set the whole array to InvalidBlockNumber */
+ memset(contents->rma_blocks, 0xFF,
+ sizeof(BlockNumber) * ARRAY_REVMAP_PAGE_MAXITEMS);
+ }
+
+ /*
+ * Update the metapage, so that item arrayBlkIdx in the array of revmap array
+ * pages points to block number newPgBlkno.
+ */
+ static void
+ update_minmax_metapg(Relation idxrel, Buffer meta, uint32 arrayBlkIdx,
+ BlockNumber newPgBlkno)
+ {
+ MinmaxMetaPageData *metadata;
+
+ metadata = (MinmaxMetaPageData *) PageGetContents(BufferGetPage(meta));
+
+ START_CRIT_SECTION();
+ metadata->revmapArrayPages[arrayBlkIdx] = newPgBlkno;
+ MarkBufferDirty(meta);
+ if (RelationNeedsWAL(idxrel))
+ {
+ xl_minmax_metapg_set xlrec;
+ XLogRecPtr recptr;
+ XLogRecData rdata;
+
+ xlrec.node = idxrel->rd_node;
+ xlrec.blkidx = arrayBlkIdx;
+ xlrec.newpg = newPgBlkno;
+
+ rdata.data = (char *) &xlrec;
+ rdata.len = SizeOfMinmaxMetapgSet;
+ rdata.buffer = InvalidBuffer;
+ rdata.buffer_std = false;
+ rdata.next = NULL;
+
+ recptr = XLogInsert(RM_MINMAX_ID, XLOG_MINMAX_METAPG_SET, &rdata);
+ PageSetLSN(BufferGetPage(meta), recptr);
+ }
+ END_CRIT_SECTION();
+ }
+
+ /*
+ * Given a logical revmap block number, find its physical block number.
+ *
+ * Note this might involve up to two buffer reads, including a possible
+ * update to the metapage.
+ *
+ * If extend is set to true, and the page hasn't been set yet, extend the
+ * array to point to a newly allocated page.
+ */
+ static BlockNumber
+ rm_get_phys_blkno(mmRevmapAccess *rmAccess, BlockNumber mapBlk, bool extend)
+ {
+ int arrayBlkIdx;
+ BlockNumber arrayBlk;
+ RevmapArrayContents *contents;
+ int revmapIdx;
+ BlockNumber targetblk;
+
+ /* the first revmap page is always block number 1 */
+ if (mapBlk == 0)
+ return (BlockNumber) 1;
+
+ /*
+ * For all other cases, take the long route of checking the metapage and
+ * revmap array pages.
+ */
+
+ /*
+ * Copy the revmap array from the metapage into private storage, if not
+ * done already in this scan.
+ */
+ if (rmAccess->revmapArrayPages == NULL)
+ {
+ rmaccess_get_metapage(rmAccess, BUFFER_LOCK_SHARE);
+ LockBuffer(rmAccess->metaBuf, BUFFER_LOCK_UNLOCK);
+ }
+
+ /*
+ * Consult the metapage array; if the array page we need is not set there,
+ * we need to extend the index to allocate the array page, and update the
+ * metapage array.
+ */
+ arrayBlkIdx = MAPBLK_TO_RMARRAY_BLK(mapBlk);
+ if (arrayBlkIdx > MAX_REVMAP_ARRAYPAGES)
+ elog(ERROR, "non-existant revmap array page requested");
+
+ arrayBlk = rmAccess->revmapArrayPages[arrayBlkIdx];
+ if (arrayBlk == InvalidBlockNumber)
+ {
+ /* if not asked to extend, there's no further work to do here */
+ if (!extend)
+ return InvalidBlockNumber;
+
+ /*
+ * If we need to create a new array page, check the metapage again;
+ * someone might have created it after the last time we read the
+ * metapage. This time we acquire an exclusive lock, since we may need
+ * to extend. Lock before doing the physical relation extension, to
+ * avoid leaving an unused page around in case someone does this
+ * concurrently. Note that, unfortunately, we will be keeping the lock
+ * on the metapage alongside the relation extension lock, while doing a
+ * syscall involving disk I/O. Extending to add a new revmap array page
+ * is fairly infrequent, so it shouldn't be too bad.
+ *
+ * XXX it is possible to extend the relation unconditionally before
+ * locking the metapage, and later if we find that someone else had
+ * already added this page, save the page in FSM as MaxFSMRequestSize.
+ * That would be better for concurrency. Explore someday.
+ */
+ rmaccess_get_metapage(rmAccess, BUFFER_LOCK_EXCLUSIVE);
+
+ if (rmAccess->revmapArrayPages[arrayBlkIdx] == InvalidBlockNumber)
+ {
+ BlockNumber newPgBlkno;
+
+ /*
+ * Ok, definitely need to allocate a new revmap array page;
+ * initialize a new page to the initial (empty) array revmap state
+ * and register it in metapage.
+ */
+ START_CRIT_SECTION();
+ rmAccess->currArrayBuf = mm_getnewbuffer(rmAccess->idxrel);
+ initialize_rma_page(rmAccess->currArrayBuf);
+ MarkBufferDirty(rmAccess->currArrayBuf);
+ if (RelationNeedsWAL(rmAccess->idxrel))
+ {
+ xl_minmax_init_rmpg xlrec;
+ XLogRecPtr recptr;
+ XLogRecData rdata;
+
+ xlrec.node = rmAccess->idxrel->rd_node;
+ xlrec.blkno = BufferGetBlockNumber(rmAccess->currArrayBuf);
+ xlrec.array = true;
+ xlrec.logblk = InvalidBlockNumber;
+
+ rdata.data = (char *) &xlrec;
+ rdata.len = SizeOfMinmaxInitRmpg;
+ rdata.buffer = InvalidBuffer; /* FIXME */
+ rdata.buffer_std = false;
+ rdata.next = NULL;
+
+ recptr = XLogInsert(RM_MINMAX_ID, XLOG_MINMAX_INIT_RMPG, &rdata);
+ PageSetLSN(BufferGetPage(rmAccess->currArrayBuf), recptr);
+ }
+ END_CRIT_SECTION();
+ LockBuffer(rmAccess->currArrayBuf, BUFFER_LOCK_UNLOCK);
+ newPgBlkno = BufferGetBlockNumber(rmAccess->currArrayBuf);
+ rmAccess->revmapArrayPages[arrayBlkIdx] = newPgBlkno;
+
+ MINMAX_elog(DEBUG2, "allocated block for revmap array page: %u",
+ BufferGetBlockNumber(rmAccess->currArrayBuf));
+
+ /* Update the metapage to point to the new array page. */
+ update_minmax_metapg(rmAccess->idxrel, rmAccess->metaBuf, arrayBlkIdx,
+ newPgBlkno);
+ }
+
+ LockBuffer(rmAccess->metaBuf, BUFFER_LOCK_UNLOCK);
+ arrayBlk = rmAccess->revmapArrayPages[arrayBlkIdx];
+ }
+
+ /*
+ * By here, we know the array page is set in the metapage array. Read that
+ * page; except that if we just allocated it, or we already hold pin on it,
+ * we don't need to read it again. XXX but we didn't hold lock!
+ */
+ Assert(arrayBlk != InvalidBlockNumber);
+
+ if (rmAccess->currArrayBuf == InvalidBuffer ||
+ BufferGetBlockNumber(rmAccess->currArrayBuf) != arrayBlk)
+ {
+ if (rmAccess->currArrayBuf != InvalidBuffer)
+ ReleaseBuffer(rmAccess->currArrayBuf);
+
+ rmAccess->currArrayBuf =
+ ReadBuffer(rmAccess->idxrel, arrayBlk);
+ }
+
+ LockBuffer(rmAccess->currArrayBuf, BUFFER_LOCK_SHARE);
+
+ /*
+ * And now we can inspect its contents; if the target page is set, we can
+ * just return. Even if not set, we can also return if caller asked us not
+ * to extend the revmap.
+ */
+ contents = (RevmapArrayContents *)
+ PageGetContents(BufferGetPage(rmAccess->currArrayBuf));
+ revmapIdx = MAPBLK_TO_RMARRAY_INDEX(mapBlk);
+ if (!extend || revmapIdx <= contents->rma_nblocks - 1)
+ {
+ LockBuffer(rmAccess->currArrayBuf, BUFFER_LOCK_UNLOCK);
+
+ return contents->rma_blocks[revmapIdx];
+ }
+
+ /*
+ * Trade our shared lock in the array page for exclusive, because we now
+ * need to allocate one more revmap page and modify the array page.
+ */
+ LockBuffer(rmAccess->currArrayBuf, BUFFER_LOCK_UNLOCK);
+ LockBuffer(rmAccess->currArrayBuf, BUFFER_LOCK_EXCLUSIVE);
+
+ contents = (RevmapArrayContents *)
+ PageGetContents(BufferGetPage(rmAccess->currArrayBuf));
+
+ /*
+ * If someone else already set the value while we were waiting for the
+ * exclusive lock, we're done; otherwise, allocate a new block as the
+ * new revmap page, and update the array page to point to it.
+ *
+ * FIXME -- what if we were asked not to extend?
+ */
+ if (contents->rma_blocks[revmapIdx] != InvalidBlockNumber)
+ {
+ targetblk = contents->rma_blocks[revmapIdx];
+ }
+ else
+ {
+ Buffer newbuf;
+
+ START_CRIT_SECTION();
+ newbuf = mm_getnewbuffer(rmAccess->idxrel);
+ targetblk = initialize_rmr_page(newbuf, mapBlk);
+ MarkBufferDirty(newbuf);
+ if (RelationNeedsWAL(rmAccess->idxrel))
+ {
+ xl_minmax_init_rmpg xlrec;
+ XLogRecPtr recptr;
+ XLogRecData rdata;
+
+ xlrec.node = rmAccess->idxrel->rd_node;
+ xlrec.blkno = BufferGetBlockNumber(newbuf);
+ xlrec.array = false;
+ xlrec.logblk = mapBlk;
+
+ rdata.data = (char *) &xlrec;
+ rdata.len = SizeOfMinmaxInitRmpg;
+ rdata.buffer = InvalidBuffer;
+ rdata.buffer_std = false;
+ rdata.next = NULL;
+
+ recptr = XLogInsert(RM_MINMAX_ID, XLOG_MINMAX_INIT_RMPG, &rdata);
+ PageSetLSN(BufferGetPage(newbuf), recptr);
+ }
+ END_CRIT_SECTION();
+
+ UnlockReleaseBuffer(newbuf);
+
+ /*
+ * Modify the revmap array page to point to the newly allocated revmap
+ * page.
+ */
+ START_CRIT_SECTION();
+
+ contents->rma_blocks[revmapIdx] = targetblk;
+ /*
+ * XXX this rma_nblocks assignment should probably be conditional on the
+ * current rma_blocks value.
+ */
+ contents->rma_nblocks = revmapIdx + 1;
+ MarkBufferDirty(rmAccess->currArrayBuf);
+
+ /* XLOG stuff */
+ if (RelationNeedsWAL(rmAccess->idxrel))
+ {
+ xl_minmax_rmarray_set xlrec;
+ XLogRecPtr recptr;
+ XLogRecData rdata[2];
+ uint8 info;
+
+ info = XLOG_MINMAX_RMARRAY_SET;
+
+ xlrec.node = rmAccess->idxrel->rd_node;
+ xlrec.rmarray = BufferGetBlockNumber(rmAccess->currArrayBuf);
+ xlrec.blkidx = revmapIdx;
+ xlrec.newpg = targetblk;
+
+ rdata[0].data = (char *) &xlrec;
+ rdata[0].len = SizeOfMinmaxRmarraySet;
+ rdata[0].buffer = InvalidBuffer;
+ rdata[0].buffer_std = false;
+ rdata[0].next = &rdata[1];
+
+ rdata[1].data = NULL;
+ rdata[1].len = 0;
+ rdata[1].buffer = rmAccess->currArrayBuf;
+ rdata[1].buffer_std = false;
+ rdata[1].next = NULL;
+
+ recptr = XLogInsert(RM_MINMAX_ID, info, rdata);
+ PageSetLSN(BufferGetPage(rmAccess->currArrayBuf), recptr);
+ }
+
+ END_CRIT_SECTION();
+ }
+
+ LockBuffer(rmAccess->currArrayBuf, BUFFER_LOCK_UNLOCK);
+
+ return targetblk;
+ }
+
+ /*
+ * Set the TID of the index entry corresponding to the range that includes
+ * the given heap page to the given item pointer.
+ *
+ * The map is extended, if necessary.
+ */
+ void
+ mmSetHeapBlockItemptr(mmRevmapAccess *rmAccess, BlockNumber heapBlk,
+ BlockNumber blkno, OffsetNumber offno)
+ {
+ BlockNumber mapBlk;
+ bool extend = false;
+
+ mapBlk = HEAPBLK_TO_REVMAP_BLK(rmAccess->pagesPerRange, heapBlk);
+
+ /* Translate the map block number to physical location */
+ mapBlk = rm_get_phys_blkno(rmAccess, mapBlk, true);
+
+ MINMAX_elog(DEBUG2, "setting %u/%u in logical page %lu (physical %u) for heap %u",
+ blkno, offno,
+ HEAPBLK_TO_REVMAP_BLK(rmAccess->pagesPerRange, heapBlk),
+ mapBlk, heapBlk);
+
+ /*
+ * Obtain the buffer from which we need to read. If we already have the
+ * correct buffer in our access struct, use that; otherwise, release that,
+ * (if valid) and read the one we need.
+ */
+ if (rmAccess->currBuf == InvalidBuffer ||
+ mapBlk != BufferGetBlockNumber(rmAccess->currBuf))
+ {
+ if (rmAccess->currBuf != InvalidBuffer)
+ ReleaseBuffer(rmAccess->currBuf);
+
+ Assert(mapBlk != InvalidBlockNumber);
+ rmAccess->currBuf = ReadBuffer(rmAccess->idxrel, mapBlk);
+ }
+
+ LockBuffer(rmAccess->currBuf, BUFFER_LOCK_EXCLUSIVE);
+ START_CRIT_SECTION();
+
+ rm_page_set_iptr(BufferGetPage(rmAccess->currBuf),
+ rmAccess->pagesPerRange,
+ heapBlk,
+ blkno, offno);
+
+ MarkBufferDirty(rmAccess->currBuf);
+
+ /* XLOG stuff */
+ if (RelationNeedsWAL(rmAccess->idxrel))
+ {
+ xl_minmax_rm_set xlrec;
+ XLogRecPtr recptr;
+ XLogRecData rdata[2];
+ uint8 info;
+
+ info = XLOG_MINMAX_REVMAP_SET;
+
+ xlrec.node = rmAccess->idxrel->rd_node;
+ xlrec.mapBlock = mapBlk;
+ xlrec.pagesPerRange = rmAccess->pagesPerRange;
+ xlrec.heapBlock = heapBlk;
+ ItemPointerSet(&(xlrec.newval), blkno, offno);
+
+ rdata[0].data = (char *) &xlrec;
+ rdata[0].len = SizeOfMinmaxRevmapSet;
+ rdata[0].buffer = InvalidBuffer;
+ rdata[0].buffer_std = false;
+ rdata[0].next = &(rdata[1]);
+
+ rdata[1].data = NULL;
+ rdata[1].len = 0;
+ rdata[1].buffer = rmAccess->currBuf;
+ rdata[1].buffer_std = false;
+ rdata[1].next = NULL;
+
+ if (extend)
+ {
+ info |= XLOG_MINMAX_INIT_PAGE;
+ /* If the page is new, there's no need for a full page image */
+ rdata[0].next = NULL;
+ }
+
+ recptr = XLogInsert(RM_MINMAX_ID, info, rdata);
+ PageSetLSN(BufferGetPage(rmAccess->currBuf), recptr);
+ }
+
+ END_CRIT_SECTION();
+
+ LockBuffer(rmAccess->currBuf, BUFFER_LOCK_UNLOCK);
+ }
+
+
+ /*
+ * Return the TID of the index entry corresponding to the range that includes
+ * the given heap page. If the TID is valid, the tuple is locked with
+ * LockTuple. It is the caller's responsibility to release that lock.
+ */
+ void
+ mmGetHeapBlockItemptr(mmRevmapAccess *rmAccess, BlockNumber heapBlk,
+ ItemPointerData *out)
+ {
+ BlockNumber mapBlk;
+ RevmapContents *contents;
+ ItemPointerData *iptr;
+
+ mapBlk = HEAPBLK_TO_REVMAP_BLK(rmAccess->pagesPerRange, heapBlk);
+ /* Translate the map block number to physical location */
+ mapBlk = rm_get_phys_blkno(rmAccess, mapBlk, false);
+ if (mapBlk == InvalidBlockNumber)
+ {
+ ItemPointerSetInvalid(out);
+ return;
+ }
+
+ if (rmAccess->currBuf == InvalidBuffer ||
+ BufferGetBlockNumber(rmAccess->currBuf) != mapBlk)
+ {
+ if (rmAccess->currBuf != InvalidBuffer)
+ ReleaseBuffer(rmAccess->currBuf);
+
+ Assert(mapBlk != InvalidBlockNumber);
+ rmAccess->currBuf = ReadBuffer(rmAccess->idxrel, mapBlk);
+ }
+
+ LockBuffer(rmAccess->currBuf, BUFFER_LOCK_SHARE);
+
+ contents = (RevmapContents *)
+ PageGetContents(BufferGetPage(rmAccess->currBuf));
+ iptr = contents->rmr_tids;
+ iptr += HEAPBLK_TO_REVMAP_INDEX(rmAccess->pagesPerRange, heapBlk);
+
+ ItemPointerCopy(iptr, out);
+
+ if (ItemPointerIsValid(iptr))
+ LockTuple(rmAccess->idxrel, iptr, ShareLock);
+
+ LockBuffer(rmAccess->currBuf, BUFFER_LOCK_UNLOCK);
+ }
+
+ /*
+ * Initialize the revmap of a new minmax index.
+ *
+ * NB -- caller is assumed to WAL-log this operation
+ */
+ void
+ mmRevmapCreate(Relation idxrel)
+ {
+ Buffer buf;
+
+ /*
+ * The first page of the revmap is always stored in block number 1 of the
+ * main fork. Because of this, the only thing we need to do is request
+ * a new page; we assume we are called immediately after the metapage has
+ * been initialized.
+ */
+ buf = mm_getnewbuffer(idxrel);
+ Assert(BufferGetBlockNumber(buf) == 1);
+
+ mm_page_init(BufferGetPage(buf), MINMAX_PAGETYPE_REVMAP);
+ MarkBufferDirty(buf);
+
+ UnlockReleaseBuffer(buf);
+ }
*** /dev/null
--- b/src/backend/access/minmax/mmsortable.c
***************
*** 0 ****
--- 1,225 ----
+ /*
+ * minmax_sortable.c
+ * Implementation of Minmax indexes for sortable datatypes
+ * (that is, anything with a btree opclass)
+ *
+ * Portions Copyright (c) 1996-2014, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * IDENTIFICATION
+ * src/backend/access/minmax/mmsortable.c
+ */
+ #include "postgres.h"
+
+ #include "access/minmax_internal.h"
+ #include "access/minmax_tuple.h"
+ #include "access/skey.h"
+ #include "utils/datum.h"
+ #include "utils/lsyscache.h"
+ #include "utils/syscache.h"
+
+ #define OPER_LT 0
+ #define OPER_LTEQ 1
+ #define OPER_EQ 2
+ #define OPER_GTEQ 3
+ #define OPER_GT 4
+
+
+ PG_FUNCTION_INFO_V1(mmSortableGetOpers);
+ PG_FUNCTION_INFO_V1(mmSortableMaybeUpdateValues);
+ PG_FUNCTION_INFO_V1(mmSortableCompare);
+
+ Datum mmSortableGetOpers(PG_FUNCTION_ARGS);
+ Datum mmSortableMaybeUpdateValues(PG_FUNCTION_ARGS);
+ Datum mmSortableCompare(PG_FUNCTION_ARGS);
+
+ /*
+ * Return the number and OIDs of (the functions that underlie) operators we
+ * need to build a minmax index, as a pointer to a newly palloc'ed MinmaxOpers.
+ */
+ Datum
+ mmSortableGetOpers(PG_FUNCTION_ARGS)
+ {
+ Oid opfamily = PG_GETARG_OID(0);
+ Oid optypid = PG_GETARG_OID(1);
+ MinmaxOpers *mmopers;
+
+ mmopers = palloc(offsetof(MinmaxOpers, opers) + sizeof(Oid) * 5);
+ mmopers->nopers = 5; /* <, <=, =, >=, > */
+ mmopers->nstored = 2; /* min, max */
+
+ mmopers->opers[OPER_LT] =
+ get_opcode(get_opfamily_member(opfamily, optypid, optypid,
+ BTLessStrategyNumber));
+ mmopers->opers[OPER_LTEQ] =
+ get_opcode(get_opfamily_member(opfamily, optypid, optypid,
+ BTLessEqualStrategyNumber));
+ mmopers->opers[OPER_EQ] =
+ get_opcode(get_opfamily_member(opfamily, optypid, optypid,
+ BTEqualStrategyNumber));
+ mmopers->opers[OPER_GTEQ] =
+ get_opcode(get_opfamily_member(opfamily, optypid, optypid,
+ BTGreaterEqualStrategyNumber));
+ mmopers->opers[OPER_GT] =
+ get_opcode(get_opfamily_member(opfamily, optypid, optypid,
+ BTGreaterStrategyNumber));
+
+ PG_RETURN_POINTER(mmopers);
+ }
+
+ /*
+ * Examine the given index tuple (which contains partial status of a certain
+ * page range) by comparing it to the given value that comes from another heap
+ * tuple. If the new value is outside the domain specified by the existing
+ * tuple values, update the index range and return true. Otherwise, return
+ * false and do not modify in this case.
+ */
+ Datum
+ mmSortableMaybeUpdateValues(PG_FUNCTION_ARGS)
+ {
+ MinmaxDesc *mmdesc = (MinmaxDesc *) PG_GETARG_POINTER(0);
+ DeformedMMTuple *dtuple = (DeformedMMTuple *) PG_GETARG_POINTER(1);
+ AttrNumber attno = PG_GETARG_UINT16(2);
+ Datum newval = PG_GETARG_DATUM(3);
+ bool isnull = PG_GETARG_DATUM(4);
+ FmgrInfo *cmpFn;
+ Datum compar;
+ bool updated = false;
+ Oid colloid = InvalidOid; /* FIXME -- figure out collation stuff */
+
+ /*
+ * If the new value is null, we record that we saw it if it's the first
+ * one; otherwise, there's nothing to do.
+ */
+ if (isnull)
+ {
+ if (dtuple->perCol[attno - 1].hasnulls)
+ PG_RETURN_BOOL(false);
+
+ dtuple->perCol[attno - 1].hasnulls = true;
+ PG_RETURN_BOOL(true);
+ }
+
+ /*
+ * If the recorded value is null, store the new value (which we know is not
+ * null) as both minimum and maximum, and we're done.
+ */
+ if (dtuple->perCol[attno - 1].allnulls)
+ {
+ dtuple->perCol[attno - 1].values[0] =
+ datumCopy(newval, mmdesc->tupdesc->attrs[attno - 1]->attbyval,
+ mmdesc->tupdesc->attrs[attno - 1]->attlen);
+ dtuple->perCol[attno - 1].values[1] =
+ datumCopy(newval, mmdesc->tupdesc->attrs[attno - 1]->attbyval,
+ mmdesc->tupdesc->attrs[attno - 1]->attlen);
+ dtuple->perCol[attno - 1].allnulls = false;
+ PG_RETURN_BOOL(true);
+ }
+
+ /*
+ * Otherwise, need to compare the new value with the existing boundaries
+ * and update them accordingly. First check if it's less than the existing
+ * minimum.
+ */
+ cmpFn = minmax_get_operfn(mmdesc, attno, OPER_LT);
+ compar = FunctionCall2Coll(cmpFn, colloid, newval,
+ dtuple->perCol[attno - 1].values[0]);
+ if (DatumGetBool(compar))
+ {
+ dtuple->perCol[attno - 1].values[0] =
+ datumCopy(newval, mmdesc->tupdesc->attrs[attno - 1]->attbyval,
+ mmdesc->tupdesc->attrs[attno - 1]->attlen);
+ updated = true;
+ }
+
+ /*
+ * And now compare it to the existing maximum.
+ */
+ cmpFn = minmax_get_operfn(mmdesc, attno, OPER_GT);
+ compar = FunctionCall2Coll(cmpFn, colloid, newval,
+ dtuple->perCol[attno - 1].values[1]);
+ if (DatumGetBool(compar))
+ {
+ dtuple->perCol[attno - 1].values[1] =
+ datumCopy(newval, mmdesc->tupdesc->attrs[attno - 1]->attbyval,
+ mmdesc->tupdesc->attrs[attno - 1]->attlen);
+ updated = true;
+ }
+
+ PG_RETURN_BOOL(updated);
+ }
+
+ /*
+ * Given an index tuple corresponding to a certain page range, and a scan key
+ * (represented by its index attribute number, the value and an operator
+ * strategy number), return whether the scan key is consistent with the page
+ * range. Return true if so, false otherwise.
+ *
+ * XXX what do we need to do with NULL values here, if anything?
+ */
+ Datum
+ mmSortableCompare(PG_FUNCTION_ARGS)
+ {
+ MinmaxDesc *mmdesc = (MinmaxDesc *) PG_GETARG_POINTER(0);
+ DeformedMMTuple *dtup = (DeformedMMTuple *) PG_GETARG_POINTER(1);
+ AttrNumber attno = PG_GETARG_INT16(2);
+ StrategyNumber strat = PG_GETARG_UINT16(3);
+ Datum value = PG_GETARG_DATUM(4);
+ Datum matches;
+ Oid colloid = InvalidOid; /* figure out collation stuff */
+ FmgrInfo *cmpFn;
+
+ switch (strat)
+ {
+ case BTLessStrategyNumber:
+ cmpFn = minmax_get_operfn(mmdesc, attno, OPER_LT);
+ matches = FunctionCall2Coll(cmpFn, colloid,
+ dtup->perCol[attno - 1].values[0],
+ value);
+ break;
+ case BTLessEqualStrategyNumber:
+ cmpFn = minmax_get_operfn(mmdesc, attno, OPER_LTEQ);
+ matches = FunctionCall2Coll(cmpFn, colloid,
+ dtup->perCol[attno - 1].values[0],
+ value);
+ break;
+ case BTEqualStrategyNumber:
+
+ /*
+ * In the equality case (WHERE col = someval), we want to return
+ * the current page range if the minimum value in the range <= scan
+ * key, and the maximum value >= scan key.
+ */
+ cmpFn = minmax_get_operfn(mmdesc, attno, OPER_LTEQ);
+ matches = FunctionCall2Coll(cmpFn, colloid,
+ dtup->perCol[attno - 1].values[0],
+ value);
+ if (!DatumGetBool(matches))
+ break;
+ /* max() >= scankey */
+ cmpFn = minmax_get_operfn(mmdesc, attno, OPER_GTEQ);
+ matches = FunctionCall2Coll(cmpFn, colloid,
+ dtup->perCol[attno - 1].values[1],
+ value);
+ break;
+ case BTGreaterEqualStrategyNumber:
+ cmpFn = minmax_get_operfn(mmdesc, attno, OPER_GTEQ);
+ matches = FunctionCall2Coll(cmpFn, colloid,
+ dtup->perCol[attno - 1].values[1],
+ value);
+ break;
+ case BTGreaterStrategyNumber:
+ cmpFn = minmax_get_operfn(mmdesc, attno, OPER_GT);
+ matches = FunctionCall2Coll(cmpFn, colloid,
+ dtup->perCol[attno - 1].values[1],
+ value);
+ break;
+ default:
+ /* shouldn't happen */
+ elog(ERROR, "invalid strategy number %d", strat);
+ matches = 0;
+ break;
+ }
+
+ PG_RETURN_DATUM(matches);
+ }
*** /dev/null
--- b/src/backend/access/minmax/mmtuple.c
***************
*** 0 ****
--- 1,449 ----
+ /*
+ * MinMax-specific tuples
+ * Method implementations for tuples in minmax indexes.
+ *
+ * The intended interface is that code outside this file only deals with
+ * DeformedMMTuples, and convert to and from the on-disk representation by
+ * using functions in this file.
+ *
+ * NOTES
+ *
+ * A minmax tuple is similar to a heap tuple, with a few key differences. The
+ * first interesting difference is that the tuple header is much simpler, only
+ * containing its total length and a small area for flags. Also, the stored
+ * data does not match the relation tuple descriptor exactly: for each
+ * attribute in the descriptor, the index tuple carries an arbitrary number
+ * of values, depending on the opclass.
+ *
+ * Also, for each column of the index relation there are two null bits: one
+ * (hasnulls) stores whether any tuple within the page range has that column
+ * set to null; the other one (allnulls) stores whether the column values are
+ * all null. If allnulls is true, then the tuple data area does not contain
+ * values for that column at all; whereas it does if the hasnulls is set.
+ * Note the size of the null bitmask may not be the same as that of the
+ * datum array.
+ *
+ * Portions Copyright (c) 1996-2013, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * IDENTIFICATION
+ * src/backend/access/minmax/mmtuple.c
+ */
+ #include "postgres.h"
+
+ #include "access/htup_details.h"
+ #include "access/minmax_tuple.h"
+ #include "access/tupdesc.h"
+ #include "access/tupmacs.h"
+
+
+ static inline void mm_deconstruct_tuple(MinmaxDesc *mmdesc,
+ char *tp, bits8 *nullbits, bool nulls,
+ Datum *values, bool *allnulls, bool *hasnulls);
+
+
+ /*
+ * Return a tuple descriptor used for on-disk storage of minmax tuples.
+ */
+ static TupleDesc
+ mmtuple_disk_tupdesc(MinmaxDesc *mmdesc)
+ {
+ /* We cache these in the MinmaxDesc */
+ if (mmdesc->disktdesc == NULL)
+ {
+ int i;
+ int j;
+ AttrNumber attno = 1;
+ TupleDesc tupdesc;
+
+ tupdesc = CreateTemplateTupleDesc(mmdesc->totalstored, false);
+
+ for (i = 0; i < mmdesc->tupdesc->natts; i++)
+ {
+ for (j = 0; j < mmdesc->perCol[i].numstored; j++)
+ TupleDescInitEntry(tupdesc, attno++, NULL,
+ mmdesc->tupdesc->attrs[i]->atttypid,
+ mmdesc->tupdesc->attrs[i]->atttypmod,
+ 0);
+ }
+
+ mmdesc->disktdesc = tupdesc;
+ }
+
+ return mmdesc->disktdesc;
+ }
+
+ /*
+ * Generate a new on-disk tuple to be inserted in a minmax index.
+ */
+ MMTuple *
+ minmax_form_tuple(MinmaxDesc *mmdesc, DeformedMMTuple *tuple, Size *size)
+ {
+ Datum *values;
+ bool *nulls;
+ bool anynulls = false;
+ MMTuple *rettuple;
+ int keyno;
+ int idxattno;
+ uint16 phony_infomask;
+ bits8 *phony_nullbitmap;
+ Size len,
+ hoff,
+ data_len;
+
+ Assert(mmdesc->totalstored > 0);
+
+ values = palloc(sizeof(Datum) * mmdesc->totalstored);
+ nulls = palloc0(sizeof(bool) * mmdesc->totalstored);
+ phony_nullbitmap = palloc(sizeof(bits8) * BITMAPLEN(mmdesc->totalstored));
+
+ /*
+ * Set up the values/nulls arrays for heap_fill_tuple
+ */
+ for (idxattno = 0, keyno = 0; keyno < mmdesc->tupdesc->natts; keyno++)
+ {
+ int datumno;
+
+ /*
+ * "allnulls" is set when there's no nonnull value in any row in
+ * the column; when this happens, there is no data to store. Thus
+ * set the nullable bits for all data elements of this column and
+ * we're done.
+ */
+ if (tuple->perCol[keyno].allnulls)
+ {
+ for (datumno = 0;
+ datumno < mmdesc->perCol[keyno].numstored;
+ datumno++)
+ nulls[idxattno++] = true;
+ anynulls = true;
+ continue;
+ }
+
+ /*
+ * The "hasnulls" bit is set when there are some null values in the
+ * data. We still need to store a real value, but the presence of this
+ * means we need a null bitmap.
+ */
+ if (tuple->perCol[keyno].hasnulls)
+ anynulls = true;
+
+ for (datumno = 0;
+ datumno < mmdesc->perCol[keyno].numstored;
+ datumno++)
+ /* XXX datumCopy ?? */
+ values[idxattno++] = tuple->perCol[keyno].values[datumno];
+ }
+
+ /* compute total space needed */
+ len = SizeOfMinMaxTuple;
+ if (anynulls)
+ {
+ /*
+ * We need a double-length bitmap on an on-disk minmax index tuple;
+ * the first half stores the "allnulls" bits, the second stores
+ * "hasnulls".
+ */
+ len += BITMAPLEN(mmdesc->tupdesc->natts * 2);
+ }
+
+ /*
+ * TODO: we can probably do away with alignment here, and save some
+ * precious disk space. When there's no bitmap we can save 6 bytes. Maybe
+ * we can use the first col's type alignment instead of maxalign.
+ */
+ len = hoff = MAXALIGN(len);
+
+ data_len = heap_compute_data_size(mmtuple_disk_tupdesc(mmdesc),
+ values, nulls);
+
+ len += data_len;
+
+ rettuple = palloc0(len);
+ rettuple->mt_info = hoff;
+ Assert((rettuple->mt_info & MMIDX_OFFSET_MASK) == hoff);
+
+ /*
+ * The infomask and null bitmap as computed by heap_fill_tuple are useless
+ * to us. However, that function will not accept a null infomask; and we
+ * need to pass a valid null bitmap so that it will correctly skip
+ * outputting null attributes in the data area.
+ */
+ heap_fill_tuple(mmtuple_disk_tupdesc(mmdesc),
+ values,
+ nulls,
+ (char *) rettuple + hoff,
+ data_len,
+ &phony_infomask,
+ phony_nullbitmap);
+
+ /* done with these */
+ pfree(values);
+ pfree(nulls);
+ pfree(phony_nullbitmap);
+
+ /*
+ * Now fill in the real null bitmasks. allnulls first.
+ */
+ if (anynulls)
+ {
+ bits8 *bitP;
+ int bitmask;
+
+ rettuple->mt_info |= MMIDX_NULLS_MASK;
+
+ bitP = ((bits8 *) (rettuple + SizeOfMinMaxTuple)) - 1;
+ bitmask = HIGHBIT;
+ for (keyno = 0; keyno < mmdesc->tupdesc->natts; keyno++)
+ {
+ if (bitmask != HIGHBIT)
+ bitmask <<= 1;
+ else
+ {
+ bitP += 1;
+ *bitP = 0x0;
+ bitmask = 1;
+ }
+
+ if (tuple->perCol[keyno].allnulls)
+ continue;
+
+ *bitP |= bitmask;
+ }
+ /* hasnulls bits follow */
+ for (keyno = 0; keyno < mmdesc->tupdesc->natts; keyno++)
+ {
+ if (bitmask != HIGHBIT)
+ bitmask <<= 1;
+ else
+ {
+ bitP += 1;
+ *bitP = 0x0;
+ bitmask = 1;
+ }
+
+ if (tuple->perCol[keyno].hasnulls)
+ continue;
+
+ *bitP |= bitmask;
+ }
+ }
+
+ *size = len;
+ return rettuple;
+ }
+
+ /*
+ * Free a tuple created by minmax_form_tuple
+ */
+ void
+ minmax_free_tuple(MMTuple *tuple)
+ {
+ pfree(tuple);
+ }
+
+ DeformedMMTuple *
+ minmax_new_dtuple(MinmaxDesc *mmdesc)
+ {
+ DeformedMMTuple *dtup;
+ char *currdatum;
+ long basesize;
+ int i;
+
+ basesize = MAXALIGN(sizeof(DeformedMMTuple) +
+ sizeof(MMValues) * mmdesc->tupdesc->natts);
+ dtup = palloc0(basesize + sizeof(Datum) * mmdesc->totalstored);
+ currdatum = (char *) dtup + basesize;
+ for (i = 0; i < mmdesc->tupdesc->natts; i++)
+ {
+ dtup->perCol[i].allnulls = true;
+ dtup->perCol[i].hasnulls = false;
+ dtup->perCol[i].values = (Datum *) currdatum;
+ currdatum += sizeof(Datum) * mmdesc->perCol[i].numstored;
+ }
+
+ return dtup;
+ }
+
+ void
+ minmax_dtuple_initialize(DeformedMMTuple *dtuple, MinmaxDesc *mmdesc)
+ {
+ int i;
+
+ for (i = 0; i < mmdesc->tupdesc->natts; i++)
+ {
+ /*
+ * FIXME -- we may need to pfree() some datums here before clobbering
+ * the whole thing
+ */
+ dtuple->perCol[i].allnulls = true;
+ dtuple->perCol[i].hasnulls = false;
+ memset(dtuple->perCol[i].values, 0,
+ sizeof(Datum) * mmdesc->perCol[i].numstored);
+ }
+ }
+
+ /*
+ * Convert a MMTuple back to a DeformedMMTuple. This is the reverse of
+ * minmax_form_tuple.
+ *
+ * Note we don't need the "on disk tupdesc" here; we rely on our own routine to
+ * deconstruct the tuple from the on-disk format.
+ *
+ * XXX some callers might need copies of each datum; if so we need
+ * to apply datumCopy inside the loop. We probably also need a
+ * minmax_free_dtuple() function.
+ */
+ DeformedMMTuple *
+ minmax_deform_tuple(MinmaxDesc *mmdesc, MMTuple *tuple)
+ {
+ DeformedMMTuple *dtup;
+ Datum *values;
+ bool *allnulls;
+ bool *hasnulls;
+ char *tp;
+ bits8 *nullbits;
+ int keyno;
+ int valueno;
+
+ dtup = minmax_new_dtuple(mmdesc);
+
+ values = palloc(sizeof(Datum) * mmdesc->totalstored);
+ allnulls = palloc(sizeof(bool) * mmdesc->tupdesc->natts);
+ hasnulls = palloc(sizeof(bool) * mmdesc->tupdesc->natts);
+
+ tp = (char *) tuple + MMTupleDataOffset(tuple);
+
+ if (MMTupleHasNulls(tuple))
+ nullbits = (bits8 *) ((char *) tuple + SizeOfMinMaxTuple);
+ else
+ nullbits = NULL;
+ mm_deconstruct_tuple(mmdesc,
+ tp, nullbits, MMTupleHasNulls(tuple),
+ values, allnulls, hasnulls);
+
+ /*
+ * Iterate to assign each of the values to the corresponding item
+ * in the values array of each column.
+ */
+ for (valueno = 0, keyno = 0; keyno < mmdesc->tupdesc->natts; keyno++)
+ {
+ int i;
+
+ if (allnulls[keyno])
+ {
+ valueno += mmdesc->perCol[keyno].numstored;
+ continue;
+ }
+
+ dtup->perCol[keyno].values =
+ palloc(sizeof(Datum) * mmdesc->totalstored);
+
+ /* XXX optional datumCopy()? */
+ for (i = 0; i < mmdesc->perCol[keyno].numstored; i++)
+ dtup->perCol[keyno].values[i] = values[valueno++];
+
+ dtup->perCol[keyno].hasnulls = hasnulls[keyno];
+ dtup->perCol[keyno].allnulls = false;
+ }
+
+ pfree(values);
+ pfree(allnulls);
+ pfree(hasnulls);
+
+ return dtup;
+ }
+
+ /*
+ * mm_deconstruct_tuple
+ * Guts of attribute extraction from an on-disk minmax tuple.
+ *
+ * Its arguments are:
+ * mmdesc minmax descriptor for the stored tuple
+ * tp pointer to the tuple data area
+ * nullbits pointer to the tuple nulls bitmask
+ * nulls "has nulls" bit in tuple infomask
+ * values output values, array of size mmdesc->totalstored
+ * allnulls output "allnulls", size mmdesc->tupdesc->natts
+ * hasnulls output "hasnulls", size mmdesc->tupdesc->natts
+ *
+ * Output arrays must have been allocated by caller.
+ */
+ static inline void
+ mm_deconstruct_tuple(MinmaxDesc *mmdesc,
+ char *tp, bits8 *nullbits, bool nulls,
+ Datum *values, bool *allnulls, bool *hasnulls)
+ {
+ int attnum;
+ int stored;
+ TupleDesc diskdsc;
+ long off = 0;
+
+ /*
+ * First iterate to natts to obtain both null flags for each attribute.
+ */
+ for (attnum = 0; attnum < mmdesc->tupdesc->natts; attnum++)
+ {
+ /*
+ * the "all nulls" bit means that all values in the page range for
+ * this column are nulls. Therefore there are no values in the tuple
+ * data area.
+ */
+ if (nulls && att_isnull(attnum, nullbits))
+ {
+ allnulls[attnum] = true;
+ continue;
+ }
+
+ allnulls[attnum] = false;
+
+ /*
+ * the "has nulls" bit means that some tuples have nulls, but others
+ * have not-null values. Therefore we know the tuple contains data for
+ * this column.
+ *
+ * The hasnulls bits follow the allnulls bits in the same bitmask.
+ */
+ hasnulls[attnum] =
+ nulls && att_isnull(mmdesc->tupdesc->natts + attnum, hasnulls);
+ }
+
+ /*
+ * Iterate to obtain each attribute's stored values. Note that since we
+ * may reuse attribute entries for more than one column, we cannot cache
+ * offsets here.
+ */
+ diskdsc = mmtuple_disk_tupdesc(mmdesc);
+ for (stored = 0, attnum = 0; attnum < mmdesc->tupdesc->natts; attnum++)
+ {
+ int datumno;
+
+ if (allnulls[attnum])
+ {
+ stored += mmdesc->perCol[attnum].numstored;
+ continue;
+ }
+
+ for (datumno = 0;
+ datumno < mmdesc->perCol[attnum].numstored;
+ datumno++)
+ {
+ Form_pg_attribute thisatt = diskdsc->attrs[stored];
+
+ if (thisatt->attlen == -1)
+ {
+ off = att_align_pointer(off, thisatt->attalign, -1,
+ tp + off);
+ }
+ else
+ {
+ /* not varlena, so safe to use att_align_nominal */
+ off = att_align_nominal(off, thisatt->attalign);
+ }
+
+ values[stored++] = fetchatt(thisatt, tp + off);
+
+ off = att_addlength_pointer(off, thisatt->attlen, tp + off);
+ }
+ }
+ }
*** /dev/null
--- b/src/backend/access/minmax/mmxlog.c
***************
*** 0 ****
--- 1,304 ----
+ /*
+ * mmxlog.c
+ * XLog replay routines for MinMax indexes
+ *
+ * Portions Copyright (c) 1996-2013, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * IDENTIFICATION
+ * src/backend/access/minmax/mmxlog.c
+ */
+ #include "postgres.h"
+
+ #include "access/minmax.h"
+ #include "access/minmax_internal.h"
+ #include "access/minmax_page.h"
+ #include "access/minmax_tuple.h"
+ #include "access/minmax_xlog.h"
+ #include "access/xlogutils.h"
+ #include "storage/freespace.h"
+
+
+ /*
+ * xlog replay routines
+ */
+ static void
+ minmax_xlog_createidx(XLogRecPtr lsn, XLogRecord *record)
+ {
+ xl_minmax_createidx *xlrec = (xl_minmax_createidx *) XLogRecGetData(record);
+ Buffer buf;
+ Page page;
+
+ /* Backup blocks are not used in create_index records */
+ Assert(!(record->xl_info & XLR_BKP_BLOCK_MASK));
+
+ /* create the index' metapage */
+ buf = XLogReadBuffer(xlrec->node, MINMAX_METAPAGE_BLKNO, true);
+ Assert(BufferIsValid(buf));
+ page = (Page) BufferGetPage(buf);
+ mm_metapage_init(page, xlrec->pagesPerRange, xlrec->version);
+ PageSetLSN(page, lsn);
+ MarkBufferDirty(buf);
+ UnlockReleaseBuffer(buf);
+
+ /* also initialize its first revmap page */
+ buf = XLogReadBuffer(xlrec->node, 1, true);
+ Assert(BufferIsValid(buf));
+ page = (Page) BufferGetPage(buf);
+ mm_page_init(page, MINMAX_PAGETYPE_REVMAP);
+ PageSetLSN(page, lsn);
+ MarkBufferDirty(buf);
+ UnlockReleaseBuffer(buf);
+ }
+
+ static void
+ minmax_xlog_insert(XLogRecPtr lsn, XLogRecord *record)
+ {
+ xl_minmax_insert *xlrec = (xl_minmax_insert *) XLogRecGetData(record);
+ BlockNumber blkno;
+ Buffer buffer;
+ Page page;
+ OffsetNumber offnum;
+ int tuplen;
+ MMTuple *mmtuple;
+
+ /* If we have a full-page image, restore it and we're done */
+ if (record->xl_info & XLR_BKP_BLOCK(0))
+ {
+ (void) RestoreBackupBlock(lsn, record, 0, false, false);
+ return;
+ }
+
+ blkno = ItemPointerGetBlockNumber(&(xlrec->target.tid));
+ if (record->xl_info & XLOG_MINMAX_INIT_PAGE)
+ {
+ buffer = XLogReadBuffer(xlrec->target.node, blkno, true);
+ Assert(BufferIsValid(buffer));
+ page = (Page) BufferGetPage(buffer);
+
+ mm_page_init(page, MINMAX_PAGETYPE_REGULAR);
+ }
+ else
+ {
+ buffer = XLogReadBuffer(xlrec->target.node, blkno, false);
+ if (!BufferIsValid(buffer))
+ return;
+ page = (Page) BufferGetPage(buffer);
+
+ if (lsn <= PageGetLSN(page)) /* changes are applied */
+ {
+ UnlockReleaseBuffer(buffer);
+ return;
+ }
+ }
+ offnum = ItemPointerGetOffsetNumber(&(xlrec->target.tid));
+ if (PageGetMaxOffsetNumber(page) + 1 < offnum)
+ elog(PANIC, "minmax_xlog_insert: invalid max offset number");
+
+ tuplen = record->xl_len - SizeOfMinmaxInsert;
+ mmtuple = (MMTuple *) ((char *) xlrec + SizeOfMinmaxInsert);
+
+ if (xlrec->overwrite)
+ PageOverwriteItemData(page, offnum, (Item) mmtuple, tuplen);
+ else
+ {
+ offnum = PageAddItem(page, (Item) mmtuple, tuplen, offnum, true, false);
+ if (offnum == InvalidOffsetNumber)
+ elog(PANIC, "minmax_xlog_insert: failed to add tuple");
+ }
+
+ PageSetLSN(page, lsn);
+
+ MarkBufferDirty(buffer);
+ UnlockReleaseBuffer(buffer);
+
+ /* XXX no FSM updates here ... */
+ }
+
+ static void
+ minmax_xlog_bulkremove(XLogRecPtr lsn, XLogRecord *record)
+ {
+ xl_minmax_bulkremove *xlrec = (xl_minmax_bulkremove *) XLogRecGetData(record);
+ Buffer buffer;
+ Page page;
+ OffsetNumber *offnos;
+ int noffs;
+ Size freespace;
+
+ /* If we have a full-page image, restore it and we're done */
+ if (record->xl_info & XLR_BKP_BLOCK(0))
+ {
+ (void) RestoreBackupBlock(lsn, record, 0, false, false);
+ return;
+ }
+
+ buffer = XLogReadBuffer(xlrec->node, xlrec->block, false);
+ if (!BufferIsValid(buffer))
+ return;
+ page = (Page) BufferGetPage(buffer);
+
+ if (lsn <= PageGetLSN(page)) /* changes are applied */
+ {
+ UnlockReleaseBuffer(buffer);
+ return;
+ }
+
+ offnos = (OffsetNumber *) ((char *) xlrec + SizeOfMinmaxBulkRemove);
+ noffs = (record->xl_len - SizeOfMinmaxBulkRemove) / sizeof(OffsetNumber);
+
+ PageIndexDeleteNoCompact(page, offnos, noffs);
+ freespace = PageGetFreeSpace(page);
+
+ PageSetLSN(page, lsn);
+
+ MarkBufferDirty(buffer);
+ UnlockReleaseBuffer(buffer);
+
+ /* update FSM as well */
+ XLogRecordPageWithFreeSpace(xlrec->node, xlrec->block, freespace);
+ }
+
+ static void
+ minmax_xlog_revmap_set(XLogRecPtr lsn, XLogRecord *record)
+ {
+ xl_minmax_rm_set *xlrec = (xl_minmax_rm_set *) XLogRecGetData(record);
+ bool init;
+ Buffer buffer;
+ Page page;
+
+ /* If we have a full-page image, restore it and we're done */
+ if (record->xl_info & XLR_BKP_BLOCK(0))
+ {
+ (void) RestoreBackupBlock(lsn, record, 0, false, false);
+ return;
+ }
+
+ init = (record->xl_info & XLOG_MINMAX_INIT_PAGE) != 0;
+ buffer = XLogReadBuffer(xlrec->node, xlrec->mapBlock, init);
+ Assert(BufferIsValid(buffer));
+ page = BufferGetPage(buffer);
+ if (init)
+ mm_page_init(page, MINMAX_PAGETYPE_REVMAP);
+
+ rm_page_set_iptr(page, xlrec->pagesPerRange, xlrec->heapBlock,
+ ItemPointerGetBlockNumber(&(xlrec->newval)),
+ ItemPointerGetOffsetNumber(&(xlrec->newval)));
+
+ PageSetLSN(page, lsn);
+ MarkBufferDirty(buffer);
+ UnlockReleaseBuffer(buffer);
+ }
+
+ static void
+ minmax_xlog_metapg_set(XLogRecPtr lsn, XLogRecord *record)
+ {
+ xl_minmax_metapg_set *xlrec = (xl_minmax_metapg_set *) XLogRecGetData(record);
+ Buffer meta;
+ Page metapg;
+ MinmaxMetaPageData *metadata;
+
+ /* If we have a full-page image, restore it and we're done */
+ if (record->xl_info & XLR_BKP_BLOCK(0))
+ {
+ (void) RestoreBackupBlock(lsn, record, 0, false, false);
+ return;
+ }
+
+ meta = XLogReadBuffer(xlrec->node, MINMAX_METAPAGE_BLKNO, false);
+ Assert(BufferIsValid(meta));
+
+ metapg = BufferGetPage(meta);
+ metadata = (MinmaxMetaPageData *) PageGetContents(metapg);
+ metadata->revmapArrayPages[xlrec->blkidx] = xlrec->newpg;
+
+ PageSetLSN(metapg, lsn);
+ MarkBufferDirty(meta);
+ UnlockReleaseBuffer(meta);
+ }
+
+ static void
+ minmax_xlog_init_rmpg(XLogRecPtr lsn, XLogRecord *record)
+ {
+ xl_minmax_init_rmpg *xlrec = (xl_minmax_init_rmpg *) XLogRecGetData(record);
+ Buffer buffer;
+
+ if (record->xl_info & XLR_BKP_BLOCK(0))
+ {
+ (void) RestoreBackupBlock(lsn, record, 0, false, false);
+ return;
+ }
+
+ buffer = XLogReadBuffer(xlrec->node, xlrec->blkno, true);
+ Assert(BufferIsValid(buffer));
+
+ if (xlrec->array)
+ initialize_rma_page(buffer);
+ else
+ initialize_rmr_page(buffer, xlrec->logblk);
+
+ PageSetLSN(BufferGetPage(buffer), lsn);
+ MarkBufferDirty(buffer);
+ UnlockReleaseBuffer(buffer);
+ }
+
+ static void
+ minmax_xlog_rmarray_set(XLogRecPtr lsn, XLogRecord *record)
+ {
+ xl_minmax_rmarray_set *xlrec = (xl_minmax_rmarray_set *) XLogRecGetData(record);
+ Buffer buffer;
+ Page page;
+ RevmapArrayContents *contents;
+
+ /* If we have a full-page image, restore it and we're done */
+ if (record->xl_info & XLR_BKP_BLOCK(0))
+ {
+ (void) RestoreBackupBlock(lsn, record, 0, false, false);
+ return;
+ }
+
+ buffer = XLogReadBuffer(xlrec->node, xlrec->rmarray, false);
+ Assert(BufferIsValid(buffer));
+
+ page = BufferGetPage(buffer);
+
+ contents = (RevmapArrayContents *) PageGetContents(page);
+ contents->rma_blocks[xlrec->blkidx] = xlrec->newpg;
+ contents->rma_nblocks = xlrec->blkidx + 1; /* XXX is this okay? */
+
+ PageSetLSN(page, lsn);
+ MarkBufferDirty(buffer);
+ UnlockReleaseBuffer(buffer);
+ }
+
+ void
+ minmax_redo(XLogRecPtr lsn, XLogRecord *record)
+ {
+ uint8 info = record->xl_info & ~XLR_INFO_MASK;
+
+ switch (info & XLOG_MINMAX_OPMASK)
+ {
+ case XLOG_MINMAX_CREATE_INDEX:
+ minmax_xlog_createidx(lsn, record);
+ break;
+ case XLOG_MINMAX_INSERT:
+ minmax_xlog_insert(lsn, record);
+ break;
+ case XLOG_MINMAX_BULKREMOVE:
+ minmax_xlog_bulkremove(lsn, record);
+ break;
+ case XLOG_MINMAX_REVMAP_SET:
+ minmax_xlog_revmap_set(lsn, record);
+ break;
+ case XLOG_MINMAX_METAPG_SET:
+ minmax_xlog_metapg_set(lsn, record);
+ break;
+ case XLOG_MINMAX_RMARRAY_SET:
+ minmax_xlog_rmarray_set(lsn, record);
+ break;
+ case XLOG_MINMAX_INIT_RMPG:
+ minmax_xlog_init_rmpg(lsn, record);
+ break;
+ default:
+ elog(PANIC, "minmax_redo: unknown op code %u", info);
+ }
+ }
*** a/src/backend/access/rmgrdesc/Makefile
--- b/src/backend/access/rmgrdesc/Makefile
***************
*** 9,15 **** top_builddir = ../../../..
include $(top_builddir)/src/Makefile.global
OBJS = clogdesc.o dbasedesc.o gindesc.o gistdesc.o hashdesc.o heapdesc.o \
! mxactdesc.o nbtdesc.o relmapdesc.o seqdesc.o smgrdesc.o spgdesc.o \
standbydesc.o tblspcdesc.o xactdesc.o xlogdesc.o
include $(top_srcdir)/src/backend/common.mk
--- 9,16 ----
include $(top_builddir)/src/Makefile.global
OBJS = clogdesc.o dbasedesc.o gindesc.o gistdesc.o hashdesc.o heapdesc.o \
! minmaxdesc.o mxactdesc.o nbtdesc.o relmapdesc.o seqdesc.o \
! smgrdesc.o spgdesc.o \
standbydesc.o tblspcdesc.o xactdesc.o xlogdesc.o
include $(top_srcdir)/src/backend/common.mk
*** /dev/null
--- b/src/backend/access/rmgrdesc/minmaxdesc.c
***************
*** 0 ****
--- 1,95 ----
+ /*-------------------------------------------------------------------------
+ *
+ * minmaxdesc.c
+ * rmgr descriptor routines for MinMax indexes
+ *
+ * Portions Copyright (c) 1996-2013, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ *
+ * IDENTIFICATION
+ * src/backend/access/rmgrdesc/minmaxdesc.c
+ *
+ *-------------------------------------------------------------------------
+ */
+ #include "postgres.h"
+
+ #include "access/minmax_xlog.h"
+
+ static void
+ out_target(StringInfo buf, xl_minmax_tid *target)
+ {
+ appendStringInfo(buf, "rel %u/%u/%u; tid %u/%u",
+ target->node.spcNode, target->node.dbNode, target->node.relNode,
+ ItemPointerGetBlockNumber(&(target->tid)),
+ ItemPointerGetOffsetNumber(&(target->tid)));
+ }
+
+ void
+ minmax_desc(StringInfo buf, XLogRecord *record)
+ {
+ char *rec = XLogRecGetData(record);
+ uint8 info = record->xl_info & ~XLR_INFO_MASK;
+
+ info &= XLOG_MINMAX_OPMASK;
+ if (info == XLOG_MINMAX_CREATE_INDEX)
+ {
+ xl_minmax_createidx *xlrec = (xl_minmax_createidx *) rec;
+
+ appendStringInfo(buf, "create index: %u/%u/%u",
+ xlrec->node.spcNode, xlrec->node.dbNode,
+ xlrec->node.relNode);
+ }
+ else if (info == XLOG_MINMAX_INSERT)
+ {
+ xl_minmax_insert *xlrec = (xl_minmax_insert *) rec;
+
+ if (record->xl_info & XLOG_MINMAX_INIT_PAGE)
+ appendStringInfo(buf, "insert(init): ");
+ else
+ appendStringInfo(buf, "insert: ");
+ out_target(buf, &(xlrec->target));
+ }
+ else if (info == XLOG_MINMAX_BULKREMOVE)
+ {
+ xl_minmax_bulkremove *xlrec = (xl_minmax_bulkremove *) rec;
+
+ appendStringInfo(buf, "bulkremove: rel %u/%u/%u blk %u",
+ xlrec->node.spcNode, xlrec->node.dbNode,
+ xlrec->node.relNode, xlrec->block);
+ }
+ else if (info == XLOG_MINMAX_REVMAP_SET)
+ {
+ xl_minmax_rm_set *xlrec = (xl_minmax_rm_set *) rec;
+
+ appendStringInfo(buf, "revmap set: rel %u/%u/%u mapblk %u pagesPerRange %u item %u value %u/%u",
+ xlrec->node.spcNode, xlrec->node.dbNode,
+ xlrec->node.relNode, xlrec->mapBlock,
+ xlrec->pagesPerRange, xlrec->heapBlock,
+ ItemPointerGetBlockNumber(&(xlrec->newval)),
+ ItemPointerGetOffsetNumber(&(xlrec->newval)));
+ }
+ else if (info == XLOG_MINMAX_METAPG_SET)
+ {
+ xl_minmax_metapg_set *xlrec = (xl_minmax_metapg_set *) rec;
+
+ appendStringInfo(buf, "metapg: rel %u/%u/%u array revmap idx %d block %u",
+ xlrec->node.spcNode, xlrec->node.dbNode,
+ xlrec->node.relNode,
+ xlrec->blkidx, xlrec->newpg);
+ }
+ else if (info == XLOG_MINMAX_RMARRAY_SET)
+ {
+ xl_minmax_rmarray_set *xlrec = (xl_minmax_rmarray_set *) rec;
+
+ appendStringInfoString(buf, "revmap array: ");
+ appendStringInfo(buf, "rel %u/%u/%u array pg %u revmap idx %d block %u",
+ xlrec->node.spcNode, xlrec->node.dbNode,
+ xlrec->node.relNode,
+ xlrec->rmarray,
+ xlrec->blkidx, xlrec->newpg);
+ }
+
+ else
+ appendStringInfo(buf, "UNKNOWN");
+ }
*** a/src/backend/access/transam/rmgr.c
--- b/src/backend/access/transam/rmgr.c
***************
*** 12,17 ****
--- 12,18 ----
#include "access/gist_private.h"
#include "access/hash.h"
#include "access/heapam_xlog.h"
+ #include "access/minmax_xlog.h"
#include "access/multixact.h"
#include "access/nbtree.h"
#include "access/spgist.h"
*** a/src/backend/catalog/index.c
--- b/src/backend/catalog/index.c
***************
*** 2096,2101 **** IndexBuildHeapScan(Relation heapRelation,
--- 2096,2122 ----
IndexBuildCallback callback,
void *callback_state)
{
+ return IndexBuildHeapRangeScan(heapRelation, indexRelation,
+ indexInfo, allow_sync,
+ 0, InvalidBlockNumber,
+ callback, callback_state);
+ }
+
+ /*
+ * As above, except that instead of scanning the complete heap, only the given
+ * number of blocks are scanned. Scan to end-of-rel can be signalled by
+ * passing InvalidBlockNumber as numblocks.
+ */
+ double
+ IndexBuildHeapRangeScan(Relation heapRelation,
+ Relation indexRelation,
+ IndexInfo *indexInfo,
+ bool allow_sync,
+ BlockNumber start_blockno,
+ BlockNumber numblocks,
+ IndexBuildCallback callback,
+ void *callback_state)
+ {
bool is_system_catalog;
bool checking_uniqueness;
HeapScanDesc scan;
***************
*** 2166,2171 **** IndexBuildHeapScan(Relation heapRelation,
--- 2187,2195 ----
true, /* buffer access strategy OK */
allow_sync); /* syncscan OK? */
+ /* set our endpoints */
+ heap_setscanlimits(scan, start_blockno, numblocks);
+
reltuples = 0;
/*
*** a/src/backend/replication/logical/decode.c
--- b/src/backend/replication/logical/decode.c
***************
*** 132,137 **** LogicalDecodingProcessRecord(LogicalDecodingContext *ctx, XLogRecord *record)
--- 132,138 ----
case RM_GIST_ID:
case RM_SEQ_ID:
case RM_SPGIST_ID:
+ case RM_MINMAX_ID:
break;
case RM_NEXT_ID:
elog(ERROR, "unexpected RM_NEXT_ID rmgr_id: %u", (RmgrIds) buf.record.xl_rmid);
*** a/src/backend/storage/page/bufpage.c
--- b/src/backend/storage/page/bufpage.c
***************
*** 324,329 **** PageAddItem(Page page,
--- 324,364 ----
}
/*
+ * PageOverwriteItemData
+ * Overwrite the data for the item at the given offset.
+ *
+ * The new data must fit in the existing data space for the old tuple.
+ */
+ void
+ PageOverwriteItemData(Page page, OffsetNumber offset, Item item, Size size)
+ {
+ PageHeader phdr = (PageHeader) page;
+ ItemId itemId;
+
+ /*
+ * Be wary about corrupted page pointers
+ */
+ if (phdr->pd_lower < SizeOfPageHeaderData ||
+ phdr->pd_lower > phdr->pd_upper ||
+ phdr->pd_upper > phdr->pd_special ||
+ phdr->pd_special > BLCKSZ)
+ ereport(PANIC,
+ (errcode(ERRCODE_DATA_CORRUPTED),
+ errmsg("corrupted page pointers: lower = %u, upper = %u, special = %u",
+ phdr->pd_lower, phdr->pd_upper, phdr->pd_special)));
+
+ itemId = PageGetItemId(phdr, offset);
+ if (!ItemIdIsUsed(itemId) || !ItemIdHasStorage(itemId))
+ elog(ERROR, "existing item to overwrite is not used");
+
+ if (ItemIdGetLength(itemId) < size)
+ elog(ERROR, "existing item is not large enough to be overwritten");
+
+ memcpy((char *) page + ItemIdGetOffset(itemId), item, size);
+ ItemIdSetNormal(itemId, ItemIdGetOffset(itemId), size);
+ }
+
+ /*
* PageGetTempPage
* Get a temporary page in local memory for special processing.
* The returned page is not initialized at all; caller must do that.
***************
*** 399,405 **** PageRestoreTempPage(Page tempPage, Page oldPage)
}
/*
! * sorting support for PageRepairFragmentation and PageIndexMultiDelete
*/
typedef struct itemIdSortData
{
--- 434,441 ----
}
/*
! * sorting support for PageRepairFragmentation, PageIndexMultiDelete,
! * PageIndexDeleteNoCompact
*/
typedef struct itemIdSortData
{
***************
*** 896,901 **** PageIndexMultiDelete(Page page, OffsetNumber *itemnos, int nitems)
--- 932,1113 ----
phdr->pd_upper = upper;
}
+ /*
+ * PageIndexDeleteNoCompact
+ * Delete the given items for an index page, and defragment the resulting
+ * free space, but do not compact the item pointers array.
+ *
+ * itemnos is the array of tuples to delete; nitems is its size. maxIdxTuples
+ * is the maximum number of tuples that can exist in a page.
+ *
+ * Unused items at the end of the array are removed.
+ *
+ * This is used for index AMs that require that existing TIDs of live tuples
+ * remain unchanged.
+ */
+ void
+ PageIndexDeleteNoCompact(Page page, OffsetNumber *itemnos, int nitems)
+ {
+ PageHeader phdr = (PageHeader) page;
+ LocationIndex pd_lower = phdr->pd_lower;
+ LocationIndex pd_upper = phdr->pd_upper;
+ LocationIndex pd_special = phdr->pd_special;
+ int nline;
+ bool empty;
+ OffsetNumber offnum;
+ int nextitm;
+
+ /*
+ * As with PageRepairFragmentation, paranoia seems justified.
+ */
+ if (pd_lower < SizeOfPageHeaderData ||
+ pd_lower > pd_upper ||
+ pd_upper > pd_special ||
+ pd_special > BLCKSZ ||
+ pd_special != MAXALIGN(pd_special))
+ ereport(ERROR,
+ (errcode(ERRCODE_DATA_CORRUPTED),
+ errmsg("corrupted page pointers: lower = %u, upper = %u, special = %u",
+ pd_lower, pd_upper, pd_special)));
+
+ /*
+ * Scan the existing item pointer array and mark as unused those that are
+ * in our kill-list; make sure any non-interesting ones are marked unused
+ * as well.
+ */
+ nline = PageGetMaxOffsetNumber(page);
+ empty = true;
+ nextitm = 0;
+ for (offnum = FirstOffsetNumber; offnum <= nline; offnum = OffsetNumberNext(offnum))
+ {
+ ItemId lp;
+ ItemLength itemlen;
+ ItemOffset offset;
+
+ lp = PageGetItemId(page, offnum);
+
+ itemlen = ItemIdGetLength(lp);
+ offset = ItemIdGetOffset(lp);
+
+ if (ItemIdIsUsed(lp))
+ {
+ if (offset < pd_upper ||
+ (offset + itemlen) > pd_special ||
+ offset != MAXALIGN(offset))
+ ereport(ERROR,
+ (errcode(ERRCODE_DATA_CORRUPTED),
+ errmsg("corrupted item pointer: offset = %u, length = %u",
+ offset, (unsigned int) itemlen)));
+
+ if (nextitm < nitems && offnum == itemnos[nextitm])
+ {
+ /* this one is on our list to delete, so mark it unused */
+ ItemIdSetUnused(lp);
+ nextitm++;
+ }
+ else if (ItemIdHasStorage(lp))
+ {
+ /* This one's live -- must do the compaction dance */
+ empty = false;
+ }
+ else
+ {
+ /* get rid of this one too */
+ ItemIdSetUnused(lp);
+ }
+ }
+ }
+
+ /* this will catch invalid or out-of-order itemnos[] */
+ if (nextitm != nitems)
+ elog(ERROR, "incorrect index offsets supplied");
+
+ if (empty)
+ {
+ /* Page is completely empty, so just reset it quickly */
+ phdr->pd_lower = SizeOfPageHeaderData;
+ phdr->pd_upper = pd_special;
+ }
+ else
+ {
+ /* There are live items: need to compact the page the hard way */
+ itemIdSortData itemidbase[MaxOffsetNumber];
+ itemIdSort itemidptr;
+ int i;
+ Size totallen;
+ Offset upper;
+
+ /*
+ * Scan the page taking note of each item that we need to preserve.
+ * This includes both live items (those that contain data) and
+ * interspersed unused ones. It's critical to preserve these unused
+ * items, because otherwise the offset numbers for later live items
+ * would change, which is not acceptable. Unused items might get used
+ * again later; that is fine.
+ */
+ itemidptr = itemidbase;
+ totallen = 0;
+ for (i = 0; i < nline; i++, itemidptr++)
+ {
+ ItemId lp;
+
+ itemidptr->offsetindex = i;
+
+ lp = PageGetItemId(page, i + 1);
+ if (ItemIdHasStorage(lp))
+ {
+ itemidptr->itemoff = ItemIdGetOffset(lp);
+ itemidptr->alignedlen = MAXALIGN(ItemIdGetLength(lp));
+ totallen += itemidptr->alignedlen;
+ }
+ else
+ {
+ itemidptr->itemoff = 0;
+ itemidptr->alignedlen = 0;
+ }
+ }
+ /* By here, there are exactly nline elements in itemidbase array */
+
+ if (totallen > (Size) (pd_special - pd_lower))
+ ereport(ERROR,
+ (errcode(ERRCODE_DATA_CORRUPTED),
+ errmsg("corrupted item lengths: total %u, available space %u",
+ (unsigned int) totallen, pd_special - pd_lower)));
+
+ /* sort itemIdSortData array into decreasing itemoff order */
+ qsort((char *) itemidbase, nline, sizeof(itemIdSortData),
+ itemoffcompare);
+
+ /*
+ * Defragment the data areas of each tuple, being careful to preserve
+ * each item's position in the linp array.
+ */
+ upper = pd_special;
+ PageClearHasFreeLinePointers(page);
+ for (i = 0, itemidptr = itemidbase; i < nline; i++, itemidptr++)
+ {
+ ItemId lp;
+
+ lp = PageGetItemId(page, itemidptr->offsetindex + 1);
+ if (itemidptr->alignedlen == 0)
+ {
+ PageSetHasFreeLinePointers(page);
+ ItemIdSetUnused(lp);
+ continue;
+ }
+ upper -= itemidptr->alignedlen;
+ memmove((char *) page + upper,
+ (char *) page + itemidptr->itemoff,
+ itemidptr->alignedlen);
+ lp->lp_off = upper;
+ /* lp_flags and lp_len remain the same as originally */
+ }
+
+ /* Set the new page limits */
+ phdr->pd_upper = upper;
+ phdr->pd_lower = SizeOfPageHeaderData + i * sizeof(ItemIdData);
+ }
+ }
/*
* Set checksum for a page in shared buffers.
*** a/src/backend/utils/adt/selfuncs.c
--- b/src/backend/utils/adt/selfuncs.c
***************
*** 7349,7351 **** gincostestimate(PG_FUNCTION_ARGS)
--- 7349,7375 ----
PG_RETURN_VOID();
}
+
+ Datum
+ mmcostestimate(PG_FUNCTION_ARGS)
+ {
+ PlannerInfo *root = (PlannerInfo *) PG_GETARG_POINTER(0);
+ IndexPath *path = (IndexPath *) PG_GETARG_POINTER(1);
+ double loop_count = PG_GETARG_FLOAT8(2);
+ Cost *indexStartupCost = (Cost *) PG_GETARG_POINTER(3);
+ Cost *indexTotalCost = (Cost *) PG_GETARG_POINTER(4);
+ Selectivity *indexSelectivity = (Selectivity *) PG_GETARG_POINTER(5);
+ double *indexCorrelation = (double *) PG_GETARG_POINTER(6);
+ IndexOptInfo *index = path->indexinfo;
+
+ *indexStartupCost = (Cost) seq_page_cost * index->pages * loop_count;
+ *indexTotalCost = *indexStartupCost;
+
+ *indexSelectivity =
+ clauselist_selectivity(root, path->indexquals,
+ path->indexinfo->rel->relid,
+ JOIN_INNER, NULL);
+ *indexCorrelation = 1;
+
+ PG_RETURN_VOID();
+ }
*** a/src/include/access/heapam.h
--- b/src/include/access/heapam.h
***************
*** 112,117 **** extern HeapScanDesc heap_beginscan_strat(Relation relation, Snapshot snapshot,
--- 112,119 ----
bool allow_strat, bool allow_sync);
extern HeapScanDesc heap_beginscan_bm(Relation relation, Snapshot snapshot,
int nkeys, ScanKey key);
+ extern void heap_setscanlimits(HeapScanDesc scan, BlockNumber startBlk,
+ BlockNumber endBlk);
extern void heap_rescan(HeapScanDesc scan, ScanKey key);
extern void heap_endscan(HeapScanDesc scan);
extern HeapTuple heap_getnext(HeapScanDesc scan, ScanDirection direction);
*** /dev/null
--- b/src/include/access/minmax.h
***************
*** 0 ****
--- 1,52 ----
+ /*
+ * AM-callable functions for MinMax indexes
+ *
+ * Portions Copyright (c) 1996-2013, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * IDENTIFICATION
+ * src/include/access/minmax.h
+ */
+ #ifndef MINMAX_H
+ #define MINMAX_H
+
+ #include "fmgr.h"
+ #include "nodes/execnodes.h"
+ #include "utils/relcache.h"
+
+
+ /*
+ * prototypes for functions in minmax.c (external entry points for minmax)
+ */
+ extern Datum mmbuild(PG_FUNCTION_ARGS);
+ extern Datum mmbuildempty(PG_FUNCTION_ARGS);
+ extern Datum mminsert(PG_FUNCTION_ARGS);
+ extern Datum mmbeginscan(PG_FUNCTION_ARGS);
+ extern Datum mmgettuple(PG_FUNCTION_ARGS);
+ extern Datum mmgetbitmap(PG_FUNCTION_ARGS);
+ extern Datum mmrescan(PG_FUNCTION_ARGS);
+ extern Datum mmendscan(PG_FUNCTION_ARGS);
+ extern Datum mmmarkpos(PG_FUNCTION_ARGS);
+ extern Datum mmrestrpos(PG_FUNCTION_ARGS);
+ extern Datum mmbulkdelete(PG_FUNCTION_ARGS);
+ extern Datum mmvacuumcleanup(PG_FUNCTION_ARGS);
+ extern Datum mmcanreturn(PG_FUNCTION_ARGS);
+ extern Datum mmcostestimate(PG_FUNCTION_ARGS);
+ extern Datum mmoptions(PG_FUNCTION_ARGS);
+
+ /*
+ * Storage type for MinMax' reloptions
+ */
+ typedef struct MinmaxOptions
+ {
+ int32 vl_len_; /* varlena header (do not touch directly!) */
+ BlockNumber pagesPerRange;
+ } MinmaxOptions;
+
+ #define MINMAX_DEFAULT_PAGES_PER_RANGE 128
+ #define MinmaxGetPagesPerRange(relation) \
+ ((relation)->rd_options ? \
+ ((MinmaxOptions *) (relation)->rd_options)->pagesPerRange : \
+ MINMAX_DEFAULT_PAGES_PER_RANGE)
+
+ #endif /* MINMAX_H */
*** /dev/null
--- b/src/include/access/minmax_internal.h
***************
*** 0 ****
--- 1,104 ----
+ /*
+ * minmax_internal.h
+ * internal declarations for MinMax indexes
+ *
+ * Portions Copyright (c) 1996-2013, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * IDENTIFICATION
+ * src/include/access/minmax_internal.h
+ */
+ #ifndef MINMAX_INTERNAL_H
+ #define MINMAX_INTERNAL_H
+
+ #include "fmgr.h"
+ #include "storage/buf.h"
+ #include "storage/bufpage.h"
+ #include "storage/off.h"
+ #include "utils/relcache.h"
+
+ /* returned by "GetOpers" amproc */
+ typedef struct MinmaxOpers
+ {
+ /* Number of columns stored in an index column of this opclass */
+ int nstored;
+
+ /* Operators that this opclass uses */
+ int nopers;
+ Oid opers[FLEXIBLE_ARRAY_MEMBER];
+ } MinmaxOpers;
+
+ /*
+ * A MinmaxDesc is a struct designed to enable decoding a MinMax tuple from the
+ * on-disk format to a DeformedMMTuple. We store all the necessary FmgrInfo
+ * structs for all columns in a single array; for each column, the operators
+ * involved start at the "operoffset" item of that array.
+ *
+ * Note: we assume, for now, that the data stored for each column is the same
+ * datatype as the indexed heap column. This restriction can be lifted by
+ * having an Oid array pointer on the PerCol struct, where each member of the
+ * array indicates the typid of the stored data.
+ */
+ typedef struct MinmaxDescPerCol
+ {
+ uint16 numopers; /* number of operators we need */
+ uint16 numstored; /* number of stored columns */
+ #if 0
+ Oid typid; /* OID of indexed datatype */
+ uint32 typmod; /* typmod of indexed datatype */
+ #endif
+ Oid *operoids; /* array of operator OIDs (size numopers) */
+ FmgrInfo *opers; /* array of operators (same size) */
+ } MinmaxDescPerCol;
+
+ typedef struct MinmaxDesc
+ {
+ /* tuple descriptor of the index relation */
+ TupleDesc tupdesc;
+
+ /* cached copy for on-disk tuples; generated at first use */
+ TupleDesc disktdesc;
+
+ /* total number of Datum entries that are stored on-disk for all columns */
+ int totalstored;
+
+ /*
+ * The "getOpers" opclass-specific procedure returns an array of OIDs for
+ * the operators it uses internally. We copy the OIDs into our own array;
+ * we also have an array of FmgrInfos where we initialize the operators,
+ * so that the opclass can call them from there.
+ */
+ Oid *operoids;
+ FmgrInfo *opers;
+
+ /* per-column info */
+ MinmaxDescPerCol perCol[FLEXIBLE_ARRAY_MEMBER]; /* tupdesc->natts entries long */
+ } MinmaxDesc;
+
+ extern void mm_metapage_init(Page page, BlockNumber pagesPerRange,
+ uint16 version);
+ extern Buffer mm_getnewbuffer(Relation irel);
+ extern void rm_page_set_iptr(Page page, BlockNumber pagesPerRange,
+ BlockNumber heapBlk, BlockNumber blkno, OffsetNumber offno);
+ extern BlockNumber initialize_rmr_page(Buffer newbuf, BlockNumber mapBlk);
+ extern void initialize_rma_page(Buffer buf);
+
+ extern FmgrInfo *minmax_get_operfn(MinmaxDesc *mmdesc, AttrNumber attno,
+ uint16 operno);
+
+ /* Procedure strategy numbers */
+ #define MINMAX_PROCNUM_GETOPERS 1
+ #define MINMAX_PROCNUM_MAYBEUPDATE 2
+ #define MINMAX_PROCNUM_COMPARE 3
+
+ #define MINMAX_DEBUG
+
+ /* we allow debug if using GCC; otherwise don't bother */
+ #if defined(MINMAX_DEBUG) && defined(__GNUC__)
+ #define MINMAX_elog(level, ...) elog(level, __VA_ARGS__)
+ #else
+ #define MINMAX_elog(a) void(0)
+ #endif
+
+
+ #endif /* MINMAX_INTERNAL_H */
*** /dev/null
--- b/src/include/access/minmax_page.h
***************
*** 0 ****
--- 1,88 ----
+ /*
+ * prototypes and definitions for minmax page layouts
+ *
+ * Portions Copyright (c) 1996-2013, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * IDENTIFICATION
+ * src/include/access/minmax_page.h
+ *
+ * NOTES
+ *
+ * These structs should really be private to specific minmax files, but it's
+ * useful to have them here so that they can be used by pageinspect and similar
+ * tools.
+ */
+ #ifndef MINMAX_PAGE_H
+ #define MINMAX_PAGE_H
+
+
+ /* special space on all minmax pages stores a "type" identifier */
+ #define MINMAX_PAGETYPE_META 0xF091
+ #define MINMAX_PAGETYPE_REVMAP_ARRAY 0xF092
+ #define MINMAX_PAGETYPE_REVMAP 0xF093
+ #define MINMAX_PAGETYPE_REGULAR 0xF094
+
+ typedef struct MinmaxSpecialSpace
+ {
+ uint16 type;
+ } MinmaxSpecialSpace;
+
+ /* Metapage definitions */
+ typedef struct MinmaxMetaPageData
+ {
+ uint32 minmaxVersion;
+ BlockNumber pagesPerRange;
+ BlockNumber revmapArrayPages[1]; /* actually MAX_REVMAP_ARRAYPAGES */
+ } MinmaxMetaPageData;
+
+ /*
+ * Number of array pages listed in metapage. Need to consider leaving enough
+ * space for the page header, the metapage struct, and the minmax special
+ * space.
+ */
+ #define MAX_REVMAP_ARRAYPAGES \
+ ((BLCKSZ - \
+ MAXALIGN(SizeOfPageHeaderData) - \
+ offsetof(MinmaxMetaPageData, revmapArrayPages) - \
+ MAXALIGN(sizeof(MinmaxSpecialSpace)) ) / \
+ sizeof(BlockNumber))
+
+ #define MINMAX_CURRENT_VERSION 1
+
+ #define MINMAX_METAPAGE_BLKNO 0
+
+ /* Definitions for regular revmap pages */
+ typedef struct RevmapContents
+ {
+ int32 rmr_logblk; /* logical blkno of this revmap page */
+ ItemPointerData rmr_tids[1]; /* really REGULAR_REVMAP_PAGE_MAXITEMS */
+ } RevmapContents;
+
+ #define REGULAR_REVMAP_CONTENT_SIZE \
+ (BLCKSZ - MAXALIGN(SizeOfPageHeaderData) - \
+ offsetof(RevmapContents, rmr_tids) - \
+ MAXALIGN(sizeof(MinmaxSpecialSpace)))
+ /* max num of items in the array */
+ #define REGULAR_REVMAP_PAGE_MAXITEMS \
+ (REGULAR_REVMAP_CONTENT_SIZE / sizeof(ItemPointerData))
+
+ /* Definitions for array revmap pages */
+ typedef struct RevmapArrayContents
+ {
+ int32 rma_nblocks;
+ BlockNumber rma_blocks[1]; /* really ARRAY_REVMAP_PAGE_MAXITEMS */
+ } RevmapArrayContents;
+
+ #define REVMAP_ARRAY_CONTENT_SIZE \
+ (BLCKSZ - MAXALIGN(SizeOfPageHeaderData) - \
+ offsetof(RevmapArrayContents, rma_blocks) - \
+ MAXALIGN(sizeof(MinmaxSpecialSpace)))
+ /* max num of items in the array */
+ #define ARRAY_REVMAP_PAGE_MAXITEMS \
+ (REVMAP_ARRAY_CONTENT_SIZE / sizeof(BlockNumber))
+
+
+ extern void mm_page_init(Page page, uint16 type);
+
+ #endif /* MINMAX_PAGE_H */
*** /dev/null
--- b/src/include/access/minmax_revmap.h
***************
*** 0 ****
--- 1,34 ----
+ /*
+ * prototypes for minmax reverse range maps
+ *
+ * Portions Copyright (c) 1996-2013, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * IDENTIFICATION
+ * src/include/access/minmax_revmap.h
+ */
+
+ #ifndef MINMAX_REVMAP_H
+ #define MINMAX_REVMAP_H
+
+ #include "storage/block.h"
+ #include "storage/itemptr.h"
+ #include "storage/off.h"
+ #include "utils/relcache.h"
+
+ /* struct definition lives in mmrevmap.c */
+ typedef struct mmRevmapAccess mmRevmapAccess;
+
+ extern mmRevmapAccess *mmRevmapAccessInit(Relation idxrel, BlockNumber *pagesPerRange);
+ extern void mmRevmapAccessTerminate(mmRevmapAccess *rmAccess);
+
+ extern void mmRevmapCreate(Relation idxrel);
+ extern void mmSetHeapBlockItemptr(mmRevmapAccess *rmAccess, BlockNumber blk,
+ BlockNumber blkno, OffsetNumber offno);
+ extern void mmGetHeapBlockItemptr(mmRevmapAccess *rmAccess, BlockNumber blk,
+ ItemPointerData *iptr);
+ extern void mmRevmapTruncate(mmRevmapAccess *rmAccess,
+ BlockNumber heapNumBlocks);
+
+
+ #endif /* MINMAX_REVMAP_H */
*** /dev/null
--- b/src/include/access/minmax_tuple.h
***************
*** 0 ****
--- 1,84 ----
+ /*
+ * Declarations for dealing with MinMax-specific tuples.
+ *
+ * Portions Copyright (c) 1996-2013, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * IDENTIFICATION
+ * src/include/access/minmax_tuple.h
+ */
+ #ifndef MINMAX_TUPLE_H
+ #define MINMAX_TUPLE_H
+
+ #include "access/minmax_internal.h"
+ #include "access/tupdesc.h"
+
+
+ /*
+ * A minmax index stores one index tuple per page range. Each index tuple
+ * has one MMValues struct for each indexed column; in turn, each MMValues
+ * has (besides the null flags) an array of Datum whose size is determined by
+ * the opclass.
+ */
+ typedef struct MMValues
+ {
+ bool hasnulls; /* is there any nulls in the page range? */
+ bool allnulls; /* are all values nulls in the page range? */
+ Datum *values; /* current accumulated values */
+ } MMValues;
+
+ /*
+ * This struct represents one index tuple, comprising the minimum and maximum
+ * values for all indexed columns, within one page range. These values can
+ * only be meaningfully decoded with an appropriate MinmaxDesc.
+ */
+ typedef struct DeformedMMTuple
+ {
+ bool unused;
+ MMValues perCol[FLEXIBLE_ARRAY_MEMBER];
+ } DeformedMMTuple;
+
+ /*
+ * An on-disk minmax tuple. This is possibly followed by a nulls bitmask, with
+ * room for 2 null bits (two bits for each value stored); an opclass-defined
+ * number of Datum values for each column follow.
+ */
+ typedef struct MMTuple
+ {
+ /* ---------------
+ * mt_info is laid out in the following fashion:
+ *
+ * 7th (high) bit: has nulls
+ * 6th bit: unused
+ * 5th bit: unused
+ * 4-0 bit: offset of data
+ * ---------------
+ */
+ uint8 mt_info;
+ } MMTuple;
+
+ #define SizeOfMinMaxTuple (offsetof(MMTuple, mt_info) + sizeof(uint8))
+
+ /*
+ * t_info manipulation macros
+ */
+ #define MMIDX_OFFSET_MASK 0x1F
+ /* bit 0x20 is not used at present */
+ /* bit 0x40 is not used at present */
+ #define MMIDX_NULLS_MASK 0x80
+
+ #define MMTupleDataOffset(mmtup) ((Size) (((MMTuple *) (mmtup))->mt_info & MMIDX_OFFSET_MASK))
+ #define MMTupleHasNulls(mmtup) (((((MMTuple *) (mmtup))->mt_info & MMIDX_NULLS_MASK)) != 0)
+
+
+ extern MMTuple *minmax_form_tuple(MinmaxDesc *mmdesc,
+ DeformedMMTuple *tuple, Size *size);
+ extern void minmax_free_tuple(MMTuple *tuple);
+
+ extern DeformedMMTuple *minmax_new_dtuple(MinmaxDesc *mmdesc);
+ extern void minmax_dtuple_initialize(DeformedMMTuple *dtuple,
+ MinmaxDesc *mmdesc);
+ extern DeformedMMTuple *minmax_deform_tuple(MinmaxDesc *mmdesc,
+ MMTuple *tuple);
+
+ #endif /* MINMAX_TUPLE_H */
*** /dev/null
--- b/src/include/access/minmax_xlog.h
***************
*** 0 ****
--- 1,134 ----
+ /*-------------------------------------------------------------------------
+ *
+ * minmax_xlog.h
+ * POSTGRES MinMax access XLOG definitions.
+ *
+ *
+ * Portions Copyright (c) 1996-2013, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/access/minmax_xlog.h
+ *
+ *-------------------------------------------------------------------------
+ */
+ #ifndef MINMAX_XLOG_H
+ #define MINMAX_XLOG_H
+
+ #include "access/xlog.h"
+ #include "storage/bufpage.h"
+ #include "storage/itemptr.h"
+ #include "storage/relfilenode.h"
+ #include "utils/relcache.h"
+
+
+ /*
+ * WAL record definitions for minmax's WAL operations
+ *
+ * XLOG allows to store some information in high 4 bits of log
+ * record xl_info field.
+ */
+ #define XLOG_MINMAX_CREATE_INDEX 0x00
+ #define XLOG_MINMAX_INSERT 0x10
+ #define XLOG_MINMAX_BULKREMOVE 0x20
+ #define XLOG_MINMAX_REVMAP_SET 0x30
+ #define XLOG_MINMAX_METAPG_SET 0x40
+ #define XLOG_MINMAX_RMARRAY_SET 0x50
+ #define XLOG_MINMAX_INIT_RMPG 0x60
+
+ #define XLOG_MINMAX_OPMASK 0x70
+ /*
+ * When we insert the first item on a new page, we restore the entire page in
+ * redo.
+ */
+ #define XLOG_MINMAX_INIT_PAGE 0x80
+
+ /* This is what we need to know about a minmax index create */
+ typedef struct xl_minmax_createidx
+ {
+ BlockNumber pagesPerRange;
+ RelFileNode node;
+ uint16 version;
+ } xl_minmax_createidx;
+ #define SizeOfMinmaxCreateIdx (offsetof(xl_minmax_createidx, version) + sizeof(uint16))
+
+ /* All that we need to find a minmax tuple */
+ typedef struct xl_minmax_tid
+ {
+ RelFileNode node;
+ ItemPointerData tid;
+ } xl_minmax_tid;
+
+ #define SizeOfMinmaxTid (offsetof(xl_minmax_tid, tid) + SizeOfIptrData)
+
+ /* This is what we need to know about a minmax tuple insert */
+ typedef struct xl_minmax_insert
+ {
+ xl_minmax_tid target;
+ bool overwrite;
+ /* tuple data follows at end of struct */
+ } xl_minmax_insert;
+
+ #define SizeOfMinmaxInsert (offsetof(xl_minmax_insert, overwrite) + sizeof(bool))
+
+ /* This is what we need to know about a bulk minmax tuple remove */
+ typedef struct xl_minmax_bulkremove
+ {
+ RelFileNode node;
+ BlockNumber block;
+ /* offset number array follows at end of struct */
+ } xl_minmax_bulkremove;
+
+ #define SizeOfMinmaxBulkRemove (offsetof(xl_minmax_bulkremove, block) + sizeof(BlockNumber))
+
+ /* This is what we need to know about a revmap "set heap ptr" */
+ typedef struct xl_minmax_rm_set
+ {
+ RelFileNode node;
+ BlockNumber mapBlock;
+ int pagesPerRange;
+ BlockNumber heapBlock;
+ ItemPointerData newval;
+ } xl_minmax_rm_set;
+
+ #define SizeOfMinmaxRevmapSet (offsetof(xl_minmax_rm_set, newval) + SizeOfIptrData)
+
+ /* This is what we need to know about a "metapage set" operation */
+ typedef struct xl_minmax_metapg_set
+ {
+ RelFileNode node;
+ uint32 blkidx;
+ BlockNumber newpg;
+ } xl_minmax_metapg_set;
+
+ #define SizeOfMinmaxMetapgSet (offsetof(xl_minmax_metapg_set, newpg) + \
+ sizeof(BlockNumber))
+
+ /* This is what we need to know about a "revmap array set" operation */
+ typedef struct xl_minmax_rmarray_set
+ {
+ RelFileNode node;
+ BlockNumber rmarray;
+ uint32 blkidx;
+ BlockNumber newpg;
+ } xl_minmax_rmarray_set;
+
+ #define SizeOfMinmaxRmarraySet (offsetof(xl_minmax_rmarray_set, newpg) + \
+ sizeof(BlockNumber))
+
+ /* This is what we need to know when we initialize a new revmap page */
+ typedef struct xl_minmax_init_rmpg
+ {
+ RelFileNode node;
+ bool array; /* array revmap page or regular revmap page */
+ BlockNumber blkno;
+ BlockNumber logblk; /* only used by regular revmap pages */
+ } xl_minmax_init_rmpg;
+
+ #define SizeOfMinmaxInitRmpg (offsetof(xl_minmax_init_rmpg, blkno) + \
+ sizeof(BlockNumber))
+
+
+ extern void minmax_desc(StringInfo buf, XLogRecord *record);
+ extern void minmax_redo(XLogRecPtr lsn, XLogRecord *record);
+
+ #endif /* MINMAX_XLOG_H */
*** a/src/include/access/reloptions.h
--- b/src/include/access/reloptions.h
***************
*** 45,52 **** typedef enum relopt_kind
RELOPT_KIND_TABLESPACE = (1 << 7),
RELOPT_KIND_SPGIST = (1 << 8),
RELOPT_KIND_VIEW = (1 << 9),
/* if you add a new kind, make sure you update "last_default" too */
! RELOPT_KIND_LAST_DEFAULT = RELOPT_KIND_VIEW,
/* some compilers treat enums as signed ints, so we can't use 1 << 31 */
RELOPT_KIND_MAX = (1 << 30)
} relopt_kind;
--- 45,53 ----
RELOPT_KIND_TABLESPACE = (1 << 7),
RELOPT_KIND_SPGIST = (1 << 8),
RELOPT_KIND_VIEW = (1 << 9),
+ RELOPT_KIND_MINMAX = (1 << 10),
/* if you add a new kind, make sure you update "last_default" too */
! RELOPT_KIND_LAST_DEFAULT = RELOPT_KIND_MINMAX,
/* some compilers treat enums as signed ints, so we can't use 1 << 31 */
RELOPT_KIND_MAX = (1 << 30)
} relopt_kind;
*** a/src/include/access/relscan.h
--- b/src/include/access/relscan.h
***************
*** 35,42 **** typedef struct HeapScanDescData
bool rs_temp_snap; /* unregister snapshot at scan end? */
/* state set up at initscan time */
! BlockNumber rs_nblocks; /* number of blocks to scan */
BlockNumber rs_startblock; /* block # to start at */
BufferAccessStrategy rs_strategy; /* access strategy for reads */
bool rs_syncscan; /* report location to syncscan logic? */
--- 35,44 ----
bool rs_temp_snap; /* unregister snapshot at scan end? */
/* state set up at initscan time */
! BlockNumber rs_nblocks; /* total number of blocks in rel */
BlockNumber rs_startblock; /* block # to start at */
+ BlockNumber rs_initblock; /* block # to consider initial of rel */
+ BlockNumber rs_numblocks; /* number of blocks to scan */
BufferAccessStrategy rs_strategy; /* access strategy for reads */
bool rs_syncscan; /* report location to syncscan logic? */
*** a/src/include/access/rmgrlist.h
--- b/src/include/access/rmgrlist.h
***************
*** 42,44 **** PG_RMGR(RM_GIN_ID, "Gin", gin_redo, gin_desc, gin_xlog_startup, gin_xlog_cleanup
--- 42,45 ----
PG_RMGR(RM_GIST_ID, "Gist", gist_redo, gist_desc, gist_xlog_startup, gist_xlog_cleanup)
PG_RMGR(RM_SEQ_ID, "Sequence", seq_redo, seq_desc, NULL, NULL)
PG_RMGR(RM_SPGIST_ID, "SPGist", spg_redo, spg_desc, spg_xlog_startup, spg_xlog_cleanup)
+ PG_RMGR(RM_MINMAX_ID, "MinMax", minmax_redo, minmax_desc, NULL, NULL)
*** a/src/include/catalog/index.h
--- b/src/include/catalog/index.h
***************
*** 97,102 **** extern double IndexBuildHeapScan(Relation heapRelation,
--- 97,110 ----
bool allow_sync,
IndexBuildCallback callback,
void *callback_state);
+ extern double IndexBuildHeapRangeScan(Relation heapRelation,
+ Relation indexRelation,
+ IndexInfo *indexInfo,
+ bool allow_sync,
+ BlockNumber start_blockno,
+ BlockNumber end_blockno,
+ IndexBuildCallback callback,
+ void *callback_state);
extern void validate_index(Oid heapId, Oid indexId, Snapshot snapshot);
*** a/src/include/catalog/pg_am.h
--- b/src/include/catalog/pg_am.h
***************
*** 132,136 **** DESCR("GIN index access method");
--- 132,138 ----
DATA(insert OID = 4000 ( spgist 0 5 f f f f f t f t f f f 0 spginsert spgbeginscan spggettuple spggetbitmap spgrescan spgendscan spgmarkpos spgrestrpos spgbuild spgbuildempty spgbulkdelete spgvacuumcleanup spgcanreturn spgcostestimate spgoptions ));
DESCR("SP-GiST index access method");
#define SPGIST_AM_OID 4000
+ DATA(insert OID = 3580 ( minmax 5 3 f f f f t t f t t f f 0 mminsert mmbeginscan - mmgetbitmap mmrescan mmendscan mmmarkpos mmrestrpos mmbuild mmbuildempty mmbulkdelete mmvacuumcleanup - mmcostestimate mmoptions ));
+ #define MINMAX_AM_OID 3580
#endif /* PG_AM_H */
*** a/src/include/catalog/pg_amop.h
--- b/src/include/catalog/pg_amop.h
***************
*** 845,848 **** DATA(insert ( 3550 869 869 25 s 932 783 0 ));
--- 845,929 ----
DATA(insert ( 3550 869 869 26 s 933 783 0 ));
DATA(insert ( 3550 869 869 27 s 934 783 0 ));
+ /*
+ * MinMax int4_ops
+ */
+ DATA(insert ( 4054 23 23 1 s 97 3580 0 ));
+ DATA(insert ( 4054 23 23 2 s 523 3580 0 ));
+ DATA(insert ( 4054 23 23 3 s 96 3580 0 ));
+ DATA(insert ( 4054 23 23 4 s 525 3580 0 ));
+ DATA(insert ( 4054 23 23 5 s 521 3580 0 ));
+
+ /*
+ * MinMax numeric_ops
+ */
+ DATA(insert ( 4055 1700 1700 1 s 1754 3580 0 ));
+ DATA(insert ( 4055 1700 1700 2 s 1755 3580 0 ));
+ DATA(insert ( 4055 1700 1700 3 s 1752 3580 0 ));
+ DATA(insert ( 4055 1700 1700 4 s 1757 3580 0 ));
+ DATA(insert ( 4055 1700 1700 5 s 1756 3580 0 ));
+
+ /*
+ * MinMax text_ops
+ */
+ DATA(insert ( 4056 25 25 1 s 664 3580 0 ));
+ DATA(insert ( 4056 25 25 2 s 665 3580 0 ));
+ DATA(insert ( 4056 25 25 3 s 98 3580 0 ));
+ DATA(insert ( 4056 25 25 4 s 667 3580 0 ));
+ DATA(insert ( 4056 25 25 5 s 666 3580 0 ));
+
+ /*
+ * MinMax time_ops
+ */
+ DATA(insert ( 4057 1083 1083 1 s 1110 3580 0 ));
+ DATA(insert ( 4057 1083 1083 2 s 1111 3580 0 ));
+ DATA(insert ( 4057 1083 1083 3 s 1108 3580 0 ));
+ DATA(insert ( 4057 1083 1083 4 s 1113 3580 0 ));
+ DATA(insert ( 4057 1083 1083 5 s 1112 3580 0 ));
+
+ /*
+ * MinMax timetz_ops
+ */
+ DATA(insert ( 4058 1266 1266 1 s 1552 3580 0 ));
+ DATA(insert ( 4058 1266 1266 2 s 1553 3580 0 ));
+ DATA(insert ( 4058 1266 1266 3 s 1550 3580 0 ));
+ DATA(insert ( 4058 1266 1266 4 s 1555 3580 0 ));
+ DATA(insert ( 4058 1266 1266 5 s 1554 3580 0 ));
+
+ /*
+ * MinMax timestamp_ops
+ */
+ DATA(insert ( 4059 1114 1114 1 s 2062 3580 0 ));
+ DATA(insert ( 4059 1114 1114 2 s 2063 3580 0 ));
+ DATA(insert ( 4059 1114 1114 3 s 2060 3580 0 ));
+ DATA(insert ( 4059 1114 1114 4 s 2065 3580 0 ));
+ DATA(insert ( 4059 1114 1114 5 s 2064 3580 0 ));
+
+ /*
+ * MinMax timestamptz_ops
+ */
+ DATA(insert ( 4060 1184 1184 1 s 1322 3580 0 ));
+ DATA(insert ( 4060 1184 1184 2 s 1323 3580 0 ));
+ DATA(insert ( 4060 1184 1184 3 s 1320 3580 0 ));
+ DATA(insert ( 4060 1184 1184 4 s 1325 3580 0 ));
+ DATA(insert ( 4060 1184 1184 5 s 1324 3580 0 ));
+
+ /*
+ * MinMax date_ops
+ */
+ DATA(insert ( 4061 1082 1082 1 s 1095 3580 0 ));
+ DATA(insert ( 4061 1082 1082 2 s 1096 3580 0 ));
+ DATA(insert ( 4061 1082 1082 3 s 1093 3580 0 ));
+ DATA(insert ( 4061 1082 1082 4 s 1098 3580 0 ));
+ DATA(insert ( 4061 1082 1082 5 s 1097 3580 0 ));
+
+ /*
+ * MinMax char_ops
+ */
+ DATA(insert ( 4062 18 18 1 s 631 3580 0 ));
+ DATA(insert ( 4062 18 18 2 s 632 3580 0 ));
+ DATA(insert ( 4062 18 18 3 s 92 3580 0 ));
+ DATA(insert ( 4062 18 18 4 s 634 3580 0 ));
+ DATA(insert ( 4062 18 18 5 s 633 3580 0 ));
+
#endif /* PG_AMOP_H */
*** a/src/include/catalog/pg_amproc.h
--- b/src/include/catalog/pg_amproc.h
***************
*** 431,434 **** DATA(insert ( 4017 25 25 3 4029 ));
--- 431,472 ----
DATA(insert ( 4017 25 25 4 4030 ));
DATA(insert ( 4017 25 25 5 4031 ));
+ /* minmax */
+ DATA(insert ( 4054 23 23 1 3383 ));
+ DATA(insert ( 4054 23 23 2 3384 ));
+ DATA(insert ( 4054 23 23 3 3385 ));
+
+ DATA(insert ( 4055 1700 1700 1 3383 ));
+ DATA(insert ( 4055 1700 1700 2 3384 ));
+ DATA(insert ( 4055 1700 1700 3 3385 ));
+
+ DATA(insert ( 4056 25 25 1 3383 ));
+ DATA(insert ( 4056 25 25 2 3384 ));
+ DATA(insert ( 4056 25 25 3 3385 ));
+
+ DATA(insert ( 4057 1083 1083 1 3383 ));
+ DATA(insert ( 4057 1083 1083 2 3384 ));
+ DATA(insert ( 4057 1083 1083 3 3385 ));
+
+ DATA(insert ( 4058 1266 1266 1 3383 ));
+ DATA(insert ( 4058 1266 1266 2 3384 ));
+ DATA(insert ( 4058 1266 1266 3 3385 ));
+
+ DATA(insert ( 4059 1114 1114 1 3383 ));
+ DATA(insert ( 4059 1114 1114 2 3384 ));
+ DATA(insert ( 4059 1114 1114 3 3385 ));
+
+ DATA(insert ( 4060 1184 1184 1 3383 ));
+ DATA(insert ( 4060 1184 1184 2 3384 ));
+ DATA(insert ( 4060 1184 1184 3 3385 ));
+
+ DATA(insert ( 4061 1082 1082 1 3383 ));
+ DATA(insert ( 4061 1082 1082 2 3384 ));
+ DATA(insert ( 4061 1082 1082 3 3385 ));
+
+ DATA(insert ( 4062 18 18 1 3383 ));
+ DATA(insert ( 4062 18 18 2 3384 ));
+ DATA(insert ( 4062 18 18 3 3385 ));
+
+
#endif /* PG_AMPROC_H */
*** a/src/include/catalog/pg_opclass.h
--- b/src/include/catalog/pg_opclass.h
***************
*** 235,239 **** DATA(insert ( 403 jsonb_ops PGNSP PGUID 4033 3802 t 0 ));
--- 235,248 ----
DATA(insert ( 405 jsonb_ops PGNSP PGUID 4034 3802 t 0 ));
DATA(insert ( 2742 jsonb_ops PGNSP PGUID 4036 3802 t 25 ));
DATA(insert ( 2742 jsonb_path_ops PGNSP PGUID 4037 3802 f 23 ));
+ DATA(insert ( 3580 int4_ops PGNSP PGUID 4054 23 t 0 ));
+ DATA(insert ( 3580 numeric_ops PGNSP PGUID 4055 1700 t 0 ));
+ DATA(insert ( 3580 text_ops PGNSP PGUID 4056 25 t 0 ));
+ DATA(insert ( 3580 time_ops PGNSP PGUID 4057 1083 t 0 ));
+ DATA(insert ( 3580 timetz_ops PGNSP PGUID 4058 1266 t 0 ));
+ DATA(insert ( 3580 timestamp_ops PGNSP PGUID 4059 1114 t 0 ));
+ DATA(insert ( 3580 timestamptz_ops PGNSP PGUID 4060 1184 t 0 ));
+ DATA(insert ( 3580 date_ops PGNSP PGUID 4061 1082 t 0 ));
+ DATA(insert ( 3580 char_ops PGNSP PGUID 4062 18 t 0 ));
#endif /* PG_OPCLASS_H */
*** a/src/include/catalog/pg_opfamily.h
--- b/src/include/catalog/pg_opfamily.h
***************
*** 157,160 **** DATA(insert OID = 4035 ( 783 jsonb_ops PGNSP PGUID ));
--- 157,170 ----
DATA(insert OID = 4036 ( 2742 jsonb_ops PGNSP PGUID ));
DATA(insert OID = 4037 ( 2742 jsonb_path_ops PGNSP PGUID ));
+ DATA(insert OID = 4054 ( 3580 int4_ops PGNSP PGUID ));
+ DATA(insert OID = 4055 ( 3580 numeric_ops PGNSP PGUID ));
+ DATA(insert OID = 4056 ( 3580 text_ops PGNSP PGUID ));
+ DATA(insert OID = 4057 ( 3580 time_ops PGNSP PGUID ));
+ DATA(insert OID = 4058 ( 3580 timetz_ops PGNSP PGUID ));
+ DATA(insert OID = 4059 ( 3580 timestamp_ops PGNSP PGUID ));
+ DATA(insert OID = 4060 ( 3580 timestamptz_ops PGNSP PGUID ));
+ DATA(insert OID = 4061 ( 3580 date_ops PGNSP PGUID ));
+ DATA(insert OID = 4062 ( 3580 char_ops PGNSP PGUID ));
+
#endif /* PG_OPFAMILY_H */
*** a/src/include/catalog/pg_proc.h
--- b/src/include/catalog/pg_proc.h
***************
*** 565,570 **** DESCR("btree(internal)");
--- 565,598 ----
DATA(insert OID = 2785 ( btoptions PGNSP PGUID 12 1 0 0 0 f f f f t f s 2 0 17 "1009 16" _null_ _null_ _null_ _null_ btoptions _null_ _null_ _null_ ));
DESCR("btree(internal)");
+ DATA(insert OID = 3789 ( mmgetbitmap PGNSP PGUID 12 1 0 0 0 f f f f t f v 2 0 20 "2281 2281" _null_ _null_ _null_ _null_ mmgetbitmap _null_ _null_ _null_ ));
+ DESCR("minmax(internal)");
+ DATA(insert OID = 3790 ( mminsert PGNSP PGUID 12 1 0 0 0 f f f f t f v 6 0 16 "2281 2281 2281 2281 2281 2281" _null_ _null_ _null_ _null_ mminsert _null_ _null_ _null_ ));
+ DESCR("minmax(internal)");
+ DATA(insert OID = 3791 ( mmbeginscan PGNSP PGUID 12 1 0 0 0 f f f f t f v 3 0 2281 "2281 2281 2281" _null_ _null_ _null_ _null_ mmbeginscan _null_ _null_ _null_ ));
+ DESCR("minmax(internal)");
+ DATA(insert OID = 3792 ( mmrescan PGNSP PGUID 12 1 0 0 0 f f f f t f v 5 0 2278 "2281 2281 2281 2281 2281" _null_ _null_ _null_ _null_ mmrescan _null_ _null_ _null_ ));
+ DESCR("minmax(internal)");
+ DATA(insert OID = 3793 ( mmendscan PGNSP PGUID 12 1 0 0 0 f f f f t f v 1 0 2278 "2281" _null_ _null_ _null_ _null_ mmendscan _null_ _null_ _null_ ));
+ DESCR("minmax(internal)");
+ DATA(insert OID = 3794 ( mmmarkpos PGNSP PGUID 12 1 0 0 0 f f f f t f v 1 0 2278 "2281" _null_ _null_ _null_ _null_ mmmarkpos _null_ _null_ _null_ ));
+ DESCR("minmax(internal)");
+ DATA(insert OID = 3795 ( mmrestrpos PGNSP PGUID 12 1 0 0 0 f f f f t f v 1 0 2278 "2281" _null_ _null_ _null_ _null_ mmrestrpos _null_ _null_ _null_ ));
+ DESCR("minmax(internal)");
+ DATA(insert OID = 3796 ( mmbuild PGNSP PGUID 12 1 0 0 0 f f f f t f v 3 0 2281 "2281 2281 2281" _null_ _null_ _null_ _null_ mmbuild _null_ _null_ _null_ ));
+ DESCR("minmax(internal)");
+ DATA(insert OID = 3797 ( mmbuildempty PGNSP PGUID 12 1 0 0 0 f f f f t f v 1 0 2278 "2281" _null_ _null_ _null_ _null_ mmbuildempty _null_ _null_ _null_ ));
+ DESCR("minmax(internal)");
+ DATA(insert OID = 3798 ( mmbulkdelete PGNSP PGUID 12 1 0 0 0 f f f f t f v 4 0 2281 "2281 2281 2281 2281" _null_ _null_ _null_ _null_ mmbulkdelete _null_ _null_ _null_ ));
+ DESCR("minmax(internal)");
+ DATA(insert OID = 3799 ( mmvacuumcleanup PGNSP PGUID 12 1 0 0 0 f f f f t f v 2 0 2281 "2281 2281" _null_ _null_ _null_ _null_ mmvacuumcleanup _null_ _null_ _null_ ));
+ DESCR("minmax(internal)");
+ DATA(insert OID = 3800 ( mmcostestimate PGNSP PGUID 12 1 0 0 0 f f f f t f v 7 0 2278 "2281 2281 2281 2281 2281 2281 2281" _null_ _null_ _null_ _null_ mmcostestimate _null_ _null_ _null_ ));
+ DESCR("minmax(internal)");
+ DATA(insert OID = 3801 ( mmoptions PGNSP PGUID 12 1 0 0 0 f f f f t f s 2 0 17 "1009 16" _null_ _null_ _null_ _null_ mmoptions _null_ _null_ _null_ ));
+ DESCR("minmax(internal)");
+
+
DATA(insert OID = 339 ( poly_same PGNSP PGUID 12 1 0 0 0 f f f f t f i 2 0 16 "604 604" _null_ _null_ _null_ _null_ poly_same _null_ _null_ _null_ ));
DATA(insert OID = 340 ( poly_contain PGNSP PGUID 12 1 0 0 0 f f f f t f i 2 0 16 "604 604" _null_ _null_ _null_ _null_ poly_contain _null_ _null_ _null_ ));
DATA(insert OID = 341 ( poly_left PGNSP PGUID 12 1 0 0 0 f f f f t f i 2 0 16 "604 604" _null_ _null_ _null_ _null_ poly_left _null_ _null_ _null_ ));
***************
*** 4064,4069 **** DATA(insert OID = 2747 ( arrayoverlap PGNSP PGUID 12 1 0 0 0 f f f f t f i
--- 4092,4105 ----
DATA(insert OID = 2748 ( arraycontains PGNSP PGUID 12 1 0 0 0 f f f f t f i 2 0 16 "2277 2277" _null_ _null_ _null_ _null_ arraycontains _null_ _null_ _null_ ));
DATA(insert OID = 2749 ( arraycontained PGNSP PGUID 12 1 0 0 0 f f f f t f i 2 0 16 "2277 2277" _null_ _null_ _null_ _null_ arraycontained _null_ _null_ _null_ ));
+ /* Minmax */
+ DATA(insert OID = 3383 ( minmax_sortable_getopers PGNSP PGUID 12 1 0 0 0 f f f f t f i 2 0 2281 "2281 2281" _null_ _null_ _null_ _null_ mmSortableGetOpers _null_ _null_ _null_ ));
+ DESCR("MinMax sortable datatype support");
+ DATA(insert OID = 3384 ( minmax_sortable_update_values PGNSP PGUID 12 1 0 0 0 f f f f t f i 5 0 16 "2281 2281 2281 2281 2281" _null_ _null_ _null_ _null_ mmSortableMaybeUpdateValues _null_ _null_ _null_ ));
+ DESCR("MinMax sortable datatype support");
+ DATA(insert OID = 3385 ( minmax_sortable_compare PGNSP PGUID 12 1 0 0 0 f f f f t f i 5 0 16 "2281 2281 2281 2281 2281" _null_ _null_ _null_ _null_ mmSortableCompare _null_ _null_ _null_ ));
+ DESCR("MinMax sortable datatype support");
+
/* userlock replacements */
DATA(insert OID = 2880 ( pg_advisory_lock PGNSP PGUID 12 1 0 0 0 f f f f t f v 1 0 2278 "20" _null_ _null_ _null_ _null_ pg_advisory_lock_int8 _null_ _null_ _null_ ));
DESCR("obtain exclusive advisory lock");
*** a/src/include/storage/bufpage.h
--- b/src/include/storage/bufpage.h
***************
*** 393,398 **** extern void PageInit(Page page, Size pageSize, Size specialSize);
--- 393,400 ----
extern bool PageIsVerified(Page page, BlockNumber blkno);
extern OffsetNumber PageAddItem(Page page, Item item, Size size,
OffsetNumber offsetNumber, bool overwrite, bool is_heap);
+ extern void PageOverwriteItemData(Page page, OffsetNumber offset, Item item,
+ Size size);
extern Page PageGetTempPage(Page page);
extern Page PageGetTempPageCopy(Page page);
extern Page PageGetTempPageCopySpecial(Page page);
***************
*** 403,408 **** extern Size PageGetExactFreeSpace(Page page);
--- 405,412 ----
extern Size PageGetHeapFreeSpace(Page page);
extern void PageIndexTupleDelete(Page page, OffsetNumber offset);
extern void PageIndexMultiDelete(Page page, OffsetNumber *itemnos, int nitems);
+ extern void PageIndexDeleteNoCompact(Page page, OffsetNumber *itemnos,
+ int nitems);
extern char *PageSetChecksumCopy(Page page, BlockNumber blkno);
extern void PageSetChecksumInplace(Page page, BlockNumber blkno);
*** a/src/include/utils/selfuncs.h
--- b/src/include/utils/selfuncs.h
***************
*** 195,200 **** extern Datum hashcostestimate(PG_FUNCTION_ARGS);
--- 195,201 ----
extern Datum gistcostestimate(PG_FUNCTION_ARGS);
extern Datum spgcostestimate(PG_FUNCTION_ARGS);
extern Datum gincostestimate(PG_FUNCTION_ARGS);
+ extern Datum mmcostestimate(PG_FUNCTION_ARGS);
/* Functions in array_selfuncs.c */
*** a/src/test/regress/expected/opr_sanity.out
--- b/src/test/regress/expected/opr_sanity.out
***************
*** 1591,1596 **** ORDER BY 1, 2, 3;
--- 1591,1601 ----
2742 | 9 | ?
2742 | 10 | ?|
2742 | 11 | ?&
+ 3580 | 1 | <
+ 3580 | 2 | <=
+ 3580 | 3 | =
+ 3580 | 4 | >=
+ 3580 | 5 | >
4000 | 1 | <<
4000 | 1 | ~<~
4000 | 2 | &<
***************
*** 1613,1619 **** ORDER BY 1, 2, 3;
4000 | 15 | >
4000 | 16 | @>
4000 | 18 | =
! (80 rows)
-- Check that all opclass search operators have selectivity estimators.
-- This is not absolutely required, but it seems a reasonable thing
--- 1618,1624 ----
4000 | 15 | >
4000 | 16 | @>
4000 | 18 | =
! (85 rows)
-- Check that all opclass search operators have selectivity estimators.
-- This is not absolutely required, but it seems a reasonable thing
***************
*** 1775,1785 **** WHERE NOT (
-- GIN has six support functions. 1-3 are mandatory, 5 is optional, and
-- at least one of 4 and 6 must be given.
-- SP-GiST has five support functions, all mandatory
amname = 'btree' AND procnums @> '{1}' OR
amname = 'hash' AND procnums = '{1}' OR
amname = 'gist' AND procnums @> '{1, 2, 3, 4, 5, 6, 7}' OR
amname = 'gin' AND (procnums @> '{1, 2, 3}' AND (procnums && '{4, 6}')) OR
! amname = 'spgist' AND procnums = '{1, 2, 3, 4, 5}'
);
amname | opfname | amproclefttype | amprocrighttype | procnums
--------+---------+----------------+-----------------+----------
--- 1780,1792 ----
-- GIN has six support functions. 1-3 are mandatory, 5 is optional, and
-- at least one of 4 and 6 must be given.
-- SP-GiST has five support functions, all mandatory
+ -- MinMax has three support functions, all mandatory
amname = 'btree' AND procnums @> '{1}' OR
amname = 'hash' AND procnums = '{1}' OR
amname = 'gist' AND procnums @> '{1, 2, 3, 4, 5, 6, 7}' OR
amname = 'gin' AND (procnums @> '{1, 2, 3}' AND (procnums && '{4, 6}')) OR
! amname = 'spgist' AND procnums = '{1, 2, 3, 4, 5}' OR
! amname = 'minmax' AND procnums = '{1, 2, 3}'
);
amname | opfname | amproclefttype | amprocrighttype | procnums
--------+---------+----------------+-----------------+----------
***************
*** 1800,1806 **** WHERE NOT (
amname = 'hash' AND procnums = '{1}' OR
amname = 'gist' AND procnums @> '{1, 2, 3, 4, 5, 6, 7}' OR
amname = 'gin' AND (procnums @> '{1, 2, 3}' AND (procnums && '{4, 6}')) OR
! amname = 'spgist' AND procnums = '{1, 2, 3, 4, 5}'
);
amname | opcname | procnums
--------+---------+----------
--- 1807,1814 ----
amname = 'hash' AND procnums = '{1}' OR
amname = 'gist' AND procnums @> '{1, 2, 3, 4, 5, 6, 7}' OR
amname = 'gin' AND (procnums @> '{1, 2, 3}' AND (procnums && '{4, 6}')) OR
! amname = 'spgist' AND procnums = '{1, 2, 3, 4, 5}' OR
! amname = 'minmax' AND procnums = '{1, 2, 3}'
);
amname | opcname | procnums
--------+---------+----------
*** a/src/test/regress/sql/opr_sanity.sql
--- b/src/test/regress/sql/opr_sanity.sql
***************
*** 1178,1188 **** WHERE NOT (
-- GIN has six support functions. 1-3 are mandatory, 5 is optional, and
-- at least one of 4 and 6 must be given.
-- SP-GiST has five support functions, all mandatory
amname = 'btree' AND procnums @> '{1}' OR
amname = 'hash' AND procnums = '{1}' OR
amname = 'gist' AND procnums @> '{1, 2, 3, 4, 5, 6, 7}' OR
amname = 'gin' AND (procnums @> '{1, 2, 3}' AND (procnums && '{4, 6}')) OR
! amname = 'spgist' AND procnums = '{1, 2, 3, 4, 5}'
);
-- Also, check if there are any pg_opclass entries that don't seem to have
--- 1178,1190 ----
-- GIN has six support functions. 1-3 are mandatory, 5 is optional, and
-- at least one of 4 and 6 must be given.
-- SP-GiST has five support functions, all mandatory
+ -- MinMax has three support functions, all mandatory
amname = 'btree' AND procnums @> '{1}' OR
amname = 'hash' AND procnums = '{1}' OR
amname = 'gist' AND procnums @> '{1, 2, 3, 4, 5, 6, 7}' OR
amname = 'gin' AND (procnums @> '{1, 2, 3}' AND (procnums && '{4, 6}')) OR
! amname = 'spgist' AND procnums = '{1, 2, 3, 4, 5}' OR
! amname = 'minmax' AND procnums = '{1, 2, 3}'
);
-- Also, check if there are any pg_opclass entries that don't seem to have
***************
*** 1201,1207 **** WHERE NOT (
amname = 'hash' AND procnums = '{1}' OR
amname = 'gist' AND procnums @> '{1, 2, 3, 4, 5, 6, 7}' OR
amname = 'gin' AND (procnums @> '{1, 2, 3}' AND (procnums && '{4, 6}')) OR
! amname = 'spgist' AND procnums = '{1, 2, 3, 4, 5}'
);
-- Unfortunately, we can't check the amproc link very well because the
--- 1203,1210 ----
amname = 'hash' AND procnums = '{1}' OR
amname = 'gist' AND procnums @> '{1, 2, 3, 4, 5, 6, 7}' OR
amname = 'gin' AND (procnums @> '{1, 2, 3}' AND (procnums && '{4, 6}')) OR
! amname = 'spgist' AND procnums = '{1, 2, 3, 4, 5}' OR
! amname = 'minmax' AND procnums = '{1, 2, 3}'
);
-- Unfortunately, we can't check the amproc link very well because the