draft support for (1) allocating a new index page and (2) relocating

index entries from the metapage to an index page.

this isn't wired into ConveyorBeltGetNewPage yet. it's just stuff
that function will need to call to do this stuff. or so I think.

diff --git a/src/backend/access/conveyor/cbindexpage.c b/src/backend/access/conveyor/cbindexpage.c
index c6a7c0c67c72895082ca624057786e18e89d47e5..81b6050fecfe7b783c2e6a122a70ea7f85b39f7a 100644 (file)
--- a/src/backend/access/conveyor/cbindexpage.c
+++ b/src/backend/access/conveyor/cbindexpage.c
@@ -75,26 +75,34 @@ cb_indexpage_find_logical_page(Page page, CBPageNo pageno,
 }
 
 /*
- * Add an index entry covering 'pageno' and pointing to 'segno' to 'page'.
+ * Add index entries for logical pages beginning at 'pageno'.
  *
- * It is the caller's responsibility to supply the correct index page.
+ * It is the caller's responsibility to supply the correct index page, and
+ * to make sure that there is enough room for the entries to be added.
  */
 void
-cb_indexpage_add_index_entry(Page page, CBPageNo pageno, CBSegNo segno,
-                                                        uint16 pages_per_segment)
+cb_indexpage_add_index_entries(Page page,
+                                                          CBPageNo pageno,
+                                                          unsigned num_index_entries,
+                                                          CBSegNo *index_entries,
+                                                          uint16 pages_per_segment)
 {
        CBIndexPageData *ipd = cb_indexpage_get_special(page);
        unsigned        offset;
 
+       if (num_index_entries < 1 || num_index_entries > CB_INDEXPAGE_INDEX_ENTRIES)
+               elog(ERROR, "can't add %u index entries to an index page",
+                        num_index_entries);
        if (pageno < ipd->cbidx_first_page)
-               elog(ERROR, "can't add index entry for page " UINT64_FORMAT " on an index page that starts at page " UINT64_FORMAT,
+               elog(ERROR, "can't add index entries starting with page " UINT64_FORMAT " to an index page that starts at page " UINT64_FORMAT,
                         pageno, ipd->cbidx_first_page);
        offset = (pageno - ipd->cbidx_first_page) / pages_per_segment;
-       if (offset > CB_INDEXPAGE_INDEX_ENTRIES)
-               elog(ERROR, "can't find index entry for page " UINT64_FORMAT " on an index page that starts at page " UINT64_FORMAT,
-                        pageno, ipd->cbidx_first_page);
+       if (offset + num_index_entries >= CB_INDEXPAGE_INDEX_ENTRIES)
+               elog(ERROR, "can't place %u index entries starting with page " UINT64_FORMAT " on an index page that starts at page " UINT64_FORMAT,
+                        num_index_entries, pageno, ipd->cbidx_first_page);
 
-       ipd->cbidx_entry[offset] = segno;
+       memcpy(&ipd->cbidx_entry[offset], index_entries,
+                  num_index_entries * sizeof(CBSegNo));
 }
 
 /*
@@ -139,7 +147,7 @@ cb_indexpage_get_next_segment(Page page)
  * that's where the count is stored.
  */
 void
-cb_indexpage_increment_pages_initalized(Page page)
+cb_indexpage_increment_pages_initialized(Page page)
 {
        CBIndexPageData *ipd = cb_indexpage_get_special(page);
 
@@ -157,7 +165,7 @@ cb_indexpage_increment_pages_initalized(Page page)
  * that's where the count is stored.
  */
 void
-cb_indexpage_decrement_pages_initalized(Page page)
+cb_indexpage_decrement_pages_initialized(Page page)
 {
        CBIndexPageData *ipd = cb_indexpage_get_special(page);
 

diff --git a/src/backend/access/conveyor/cbmetapage.c b/src/backend/access/conveyor/cbmetapage.c
index 24af6eda26a1e5b85328dec7b820950a1fa26ea6..b58c93d938c618b0b5407820bfa283ed46a7d0f8 100644 (file)
--- a/src/backend/access/conveyor/cbmetapage.c
+++ b/src/backend/access/conveyor/cbmetapage.c
@@ -333,7 +333,7 @@ cb_metapage_add_index_entry(CBMetapageData *meta, CBSegNo segno)
  * make more space in the metapage. In that case, pass relocating = true.
  */
 void
-cb_metapage_remove_index_entries(CBMetapageData *meta, int count,
+cb_metapage_remove_index_entries(CBMetapageData *meta, unsigned count,
                                                                 bool relocating)
 {
        int                     used = cb_metapage_get_index_entries_used(meta);

diff --git a/src/backend/access/conveyor/cbmodify.c b/src/backend/access/conveyor/cbmodify.c
index 87c2c080b00f341f6b648a872665688b69e743b5..9ba50c395ae985409f781094adeadd2f9bc9f842 100644 (file)
--- a/src/backend/access/conveyor/cbmodify.c
+++ b/src/backend/access/conveyor/cbmodify.c
@@ -183,7 +183,7 @@ cb_allocate_payload_segment(RelFileNode *rnode,
 
        if (needs_xlog)
        {
-               xl_cb_allocate_payload_segment  xlrec;
+               xl_cb_allocate_payload_segment xlrec;
                XLogRecPtr      lsn;
 
                xlrec.segno = segno;
@@ -195,7 +195,7 @@ cb_allocate_payload_segment(RelFileNode *rnode,
                if (fsmblock != InvalidBlockNumber)
                        XLogRegisterBlock(1, rnode, fork, fsmblock,
                                                          BufferGetPage(fsmbuffer), REGBUF_STANDARD);
-               XLogRegisterData((char *) &xlrec, sizeof(xlrec));
+               XLogRegisterData((char *) &xlrec, SizeOfCBAllocatePayloadSegment);
                lsn = XLogInsert(RM_CONVEYOR_ID,
                                                 XLOG_CONVEYOR_ALLOCATE_PAYLOAD_SEGMENT);
 
@@ -285,7 +285,7 @@ cb_allocate_index_segment(RelFileNode *rnode,
 
        if (needs_xlog)
        {
-               xl_cb_allocate_index_segment    xlrec;
+               xl_cb_allocate_index_segment xlrec;
                XLogRecPtr      lsn;
 
                xlrec.segno = segno;
@@ -303,7 +303,7 @@ cb_allocate_index_segment(RelFileNode *rnode,
                if (fsmblock != InvalidBlockNumber)
                        XLogRegisterBlock(3, rnode, fork, fsmblock,
                                                          BufferGetPage(fsmbuffer), REGBUF_STANDARD);
-               XLogRegisterData((char *) &xlrec, sizeof(xlrec));
+               XLogRegisterData((char *) &xlrec, SizeOfCBAllocateIndexSegment);
                lsn = XLogInsert(RM_CONVEYOR_ID,
                                                 XLOG_CONVEYOR_ALLOCATE_INDEX_SEGMENT);
 
@@ -317,3 +317,124 @@ cb_allocate_index_segment(RelFileNode *rnode,
 
        END_CRIT_SECTION();
 }
+
+/*
+ * Allocate a new index page in an existing index segment, and optionally
+ * write XLOG for the change.
+ *
+ * 'indexblock' and 'indexbuffer' should be the block number and buffer for
+ * the new page. 'firstindexblock' and 'firstindexbuffer' are the block
+ * number and buffer for the first page of the index segment.
+ *
+ * 'pageno' is the first logical page for which the new index page will
+ * store index information.
+ *
+ * See cb_xlog_allocate_index_page for the corresponding REDO routine.
+ */
+void
+cb_allocate_index_page(RelFileNode *rnode,
+                                          ForkNumber fork,
+                                          BlockNumber indexblock,
+                                          Buffer indexbuffer,
+                                          BlockNumber firstindexblock,
+                                          Buffer firstindexbuffer,
+                                          CBPageNo pageno,
+                                          bool needs_xlog)
+{
+       Page            indexpage;
+       Page            firstindexpage;
+
+       indexpage = BufferGetPage(indexbuffer);
+       firstindexpage = BufferGetPage(firstindexbuffer);
+
+       START_CRIT_SECTION();
+
+       cb_indexpage_initialize(indexpage, pageno, false);
+       cb_indexpage_increment_pages_initialized(firstindexpage);
+
+       if (needs_xlog)
+       {
+               xl_cb_allocate_index_page xlrec;
+               XLogRecPtr      lsn;
+
+               xlrec.pageno = pageno;
+
+               XLogBeginInsert();
+               XLogRegisterBlock(0, rnode, fork, indexblock, indexpage,
+                                                 REGBUF_STANDARD | REGBUF_WILL_INIT);
+               XLogRegisterBlock(1, rnode, fork, firstindexblock,
+                                                 firstindexpage, REGBUF_STANDARD);
+               XLogRegisterData((char *) &xlrec, SizeOfCBAllocateIndexPage);
+               lsn = XLogInsert(RM_CONVEYOR_ID,
+                                                XLOG_CONVEYOR_ALLOCATE_INDEX_PAGE);
+
+               PageSetLSN(indexpage, lsn);
+               PageSetLSN(firstindexpage, lsn);
+       }
+
+       END_CRIT_SECTION();
+}
+
+/*
+ * Relocate index entries from the metapage to a page in an index segment,
+ * and optionally write XLOG for the change.
+ *
+ * 'pageno' is the logical page number for the first index entry that we're
+ * relocating. It is needed to figure out where to place the index entries
+ * on the index page.
+ *
+ * See cb_xlog_allocate_index_segment for the corresponding REDO routine.
+ */
+void
+cb_relocate_index_entries(RelFileNode *rnode,
+                                                 ForkNumber fork,
+                                                 Buffer metabuffer,
+                                                 BlockNumber indexblock,
+                                                 Buffer indexbuffer,
+                                                 CBPageNo pageno,
+                                                 unsigned num_index_entries,
+                                                 CBSegNo *index_entries,
+                                                 uint16 pages_per_segment,
+                                                 bool needs_xlog)
+{
+       Page            metapage;
+       Page            indexpage;
+       CBMetapageData *meta;
+
+       metapage = BufferGetPage(metabuffer);
+       indexpage = BufferGetPage(indexbuffer);
+
+       meta = cb_metapage_get_special(metapage);
+
+       START_CRIT_SECTION();
+
+       cb_indexpage_add_index_entries(indexpage, pageno, num_index_entries,
+                                                                  index_entries, pages_per_segment);
+       cb_metapage_remove_index_entries(meta, num_index_entries, true);
+
+       if (needs_xlog)
+       {
+               xl_cb_relocate_index_entries xlrec;
+               XLogRecPtr      lsn;
+
+               xlrec.pageno = pageno;
+               xlrec.num_index_entries = num_index_entries;
+               xlrec.pages_per_segment = pages_per_segment;
+
+               XLogBeginInsert();
+               XLogRegisterBlock(0, rnode, fork, CONVEYOR_METAPAGE, metapage,
+                                                 REGBUF_STANDARD);
+               XLogRegisterBlock(1, rnode, fork, indexblock, indexpage,
+                                                 REGBUF_STANDARD);
+               XLogRegisterData((char *) &xlrec, SizeOfCBRelocateIndexEntries);
+               XLogRegisterData((char *) index_entries,
+                                                num_index_entries * sizeof(CBSegNo));
+               lsn = XLogInsert(RM_CONVEYOR_ID,
+                                                XLOG_CONVEYOR_RELOCATE_INDEX_ENTRIES);
+
+               PageSetLSN(metapage, lsn);
+               PageSetLSN(indexpage, lsn);
+       }
+
+       END_CRIT_SECTION();
+}

diff --git a/src/backend/access/conveyor/cbxlog.c b/src/backend/access/conveyor/cbxlog.c
index 7c029a76427c057261a76de688bd791abe9d3b49..cc32e40225a7e1003ffe43fc0941450b0af1e177 100644 (file)
--- a/src/backend/access/conveyor/cbxlog.c
+++ b/src/backend/access/conveyor/cbxlog.c
@@ -179,6 +179,89 @@ cb_xlog_allocate_index_segment(XLogReaderState *record)
                UnlockReleaseBuffer(fsmbuffer);
 }
 
+/*
+ * REDO function for cb_allocate_index_page.
+ */
+static void
+cb_xlog_allocate_index_page(XLogReaderState *record)
+{
+       XLogRecPtr      lsn = record->EndRecPtr;
+       xl_cb_allocate_index_page *xlrec;
+       Buffer          indexbuffer;
+       Buffer          firstindexbuffer;
+
+       xlrec = (xl_cb_allocate_index_page *) XLogRecGetData(record);
+
+       /* NB: new index buffer should come first due to lock ordering rules */
+       if (XLogReadBufferForRedo(record, 0, &indexbuffer) == BLK_NEEDS_REDO)
+       {
+               Page    indexpage = BufferGetPage(indexbuffer);
+
+               cb_indexpage_initialize(indexpage, xlrec->pageno, false);
+               PageSetLSN(indexpage, lsn);
+               MarkBufferDirty(indexbuffer);
+       }
+
+       if (XLogReadBufferForRedo(record, 1, &firstindexbuffer) == BLK_NEEDS_REDO)
+       {
+               Page    firstindexpage = BufferGetPage(firstindexbuffer);
+
+               cb_indexpage_increment_pages_initialized(firstindexpage);
+               PageSetLSN(firstindexpage, lsn);
+               MarkBufferDirty(firstindexbuffer);
+       }
+
+       if (BufferIsValid(indexbuffer))
+               UnlockReleaseBuffer(indexbuffer);
+       if (BufferIsValid(firstindexbuffer))
+               UnlockReleaseBuffer(firstindexbuffer);
+}
+
+/*
+ * REDO function for cb_relocate_index_entries.
+ */
+static void
+cb_xlog_relocate_index_entries(XLogReaderState *record)
+{
+       XLogRecPtr      lsn = record->EndRecPtr;
+       xl_cb_relocate_index_entries *xlrec;
+       Buffer          metabuffer;
+       Buffer          indexbuffer;
+
+       xlrec = (xl_cb_relocate_index_entries *) XLogRecGetData(record);
+
+       /* NB: metapage must be last due to lock ordering rules */
+       if (XLogReadBufferForRedo(record, 1, &indexbuffer) == BLK_NEEDS_REDO)
+       {
+               Page    indexpage = BufferGetPage(indexbuffer);
+
+               cb_indexpage_add_index_entries(indexpage, xlrec->pageno,
+                                                                          xlrec->num_index_entries,
+                                                                          xlrec->index_entries,
+                                                                          xlrec->pages_per_segment);
+               cb_indexpage_initialize(indexpage, xlrec->pageno, false);
+               PageSetLSN(indexpage, lsn);
+               MarkBufferDirty(indexbuffer);
+       }
+
+       /* NB: metapage must be last due to lock ordering rules */
+       if (XLogReadBufferForRedo(record, 0, &metabuffer) == BLK_NEEDS_REDO)
+       {
+               Page    metapage = BufferGetPage(metabuffer);
+               CBMetapageData *meta;
+
+               meta = cb_metapage_get_special(metapage);
+               cb_metapage_remove_index_entries(meta, xlrec->num_index_entries, true);
+               PageSetLSN(metapage, lsn);
+               MarkBufferDirty(metabuffer);
+       }
+
+       if (BufferIsValid(metabuffer))
+               UnlockReleaseBuffer(metabuffer);
+       if (BufferIsValid(indexbuffer))
+               UnlockReleaseBuffer(indexbuffer);
+}
+
 /*
  * Main entrypoint for conveyor belt REDO.
  */
@@ -198,6 +281,12 @@ conveyor_redo(XLogReaderState *record)
                case XLOG_CONVEYOR_ALLOCATE_INDEX_SEGMENT:
                        cb_xlog_allocate_index_segment(record);
                        break;
+               case XLOG_CONVEYOR_ALLOCATE_INDEX_PAGE:
+                       cb_xlog_allocate_index_page(record);
+                       break;
+               case XLOG_CONVEYOR_RELOCATE_INDEX_ENTRIES:
+                       cb_xlog_relocate_index_entries(record);
+                       break;
                default:
                        elog(PANIC, "conveyor_redo: unknown op code %u", info);
        }

diff --git a/src/include/access/cbindexpage.h b/src/include/access/cbindexpage.h
index 9e514e6cd415b57f3216ab5085d91135214c16a7..430d258018bcbaf577aed582edc87d57a8fd3710 100644 (file)
--- a/src/include/access/cbindexpage.h
+++ b/src/include/access/cbindexpage.h
@@ -41,13 +41,15 @@ extern void cb_indexpage_initialize(Page page, CBPageNo pageno,
 extern BlockNumber cb_indexpage_find_logical_page(Page page,
                                                                                                  CBPageNo pageno,
                                                                                                  uint16 pages_per_segment);
-extern void cb_indexpage_add_index_entry(Page page, CBPageNo pageno,
-                                                                                CBSegNo segno,
-                                                                                uint16 pages_per_segment);
+extern void cb_indexpage_add_index_entries(Page page,
+                                                                                  CBPageNo pageno,
+                                                                                  unsigned num_index_entries,
+                                                                                  CBSegNo *index_entries,
+                                                                                  uint16 pages_per_segment);
 extern void cb_indexpage_set_next_segment(Page page, CBSegNo segno);
 extern CBSegNo cb_indexpage_get_next_segment(Page page);
-extern void cb_indexpage_increment_pages_initalized(Page page);
-extern void cb_indexpage_decrement_pages_initalized(Page page);
+extern void cb_indexpage_increment_pages_initialized(Page page);
+extern void cb_indexpage_decrement_pages_initialized(Page page);
 
 /*
  * How many index entries will fit into an index segment?

diff --git a/src/include/access/cbmetapage.h b/src/include/access/cbmetapage.h
index 0e1be305cfca2afdd8b16871574e385607732bee..d910afde50d86c057784881bb6ba2860e17b2183 100644 (file)
--- a/src/include/access/cbmetapage.h
+++ b/src/include/access/cbmetapage.h
@@ -111,7 +111,8 @@ extern void cb_metapage_get_bounds(CBMetapageData *meta,
                                                                   CBPageNo *next_logical_page);
 extern int     cb_metapage_get_index_entries_used(CBMetapageData *meta);
 extern void cb_metapage_add_index_entry(CBMetapageData *meta, CBSegNo segno);
-extern void cb_metapage_remove_index_entries(CBMetapageData *meta, int count,
+extern void cb_metapage_remove_index_entries(CBMetapageData *meta,
+                                                                                        unsigned count,
                                                                                         bool relocating);
 extern CBSegNo *cb_metapage_get_index_entry_pointer(CBMetapageData *meta);
 extern void cb_metapage_get_critical_info(CBMetapageData *meta,

diff --git a/src/include/access/cbmodify.h b/src/include/access/cbmodify.h
index fb0152809b5fb5e247f538d9903647b6d4220b8a..5141062cdeeb6259842012f31f59255e7d33c700 100644 (file)
--- a/src/include/access/cbmodify.h
+++ b/src/include/access/cbmodify.h
@@ -70,4 +70,24 @@ extern void cb_allocate_index_segment(RelFileNode *rnode,
                                                                          bool is_extend,
                                                                          bool needs_xlog);
 
+extern void cb_allocate_index_page(RelFileNode *rnode,
+                                                                  ForkNumber fork,
+                                                                  BlockNumber indexblock,
+                                                                  Buffer indexbuffer,
+                                                                  BlockNumber firstindexblock,
+                                                                  Buffer firstindexbuffer,
+                                                                  CBPageNo pageno,
+                                                                  bool needs_xlog);
+
+extern void cb_relocate_index_entries(RelFileNode *rnode,
+                                                                         ForkNumber fork,
+                                                                         Buffer metabuffer,
+                                                                         BlockNumber indexblock,
+                                                                         Buffer indexbuffer,
+                                                                         CBPageNo pageno,
+                                                                         unsigned num_index_entries,
+                                                                         CBSegNo *index_entries,
+                                                                         uint16 pages_per_segment,
+                                                                         bool needs_xlog);
+
 #endif                                                 /* CBMODIFY_H */

diff --git a/src/include/access/cbxlog.h b/src/include/access/cbxlog.h
index c7c2660d5f1043edbaab9802d4c13b31aa6de477..98c25da0c5c8918eddcb7958243eb7b4c00d72d2 100644 (file)
--- a/src/include/access/cbxlog.h
+++ b/src/include/access/cbxlog.h
@@ -22,6 +22,8 @@
 #define XLOG_CONVEYOR_INSERT_PAYLOAD_PAGE                      0x10
 #define        XLOG_CONVEYOR_ALLOCATE_PAYLOAD_SEGMENT          0x20
 #define        XLOG_CONVEYOR_ALLOCATE_INDEX_SEGMENT            0x30
+#define        XLOG_CONVEYOR_ALLOCATE_INDEX_PAGE                       0x40
+#define XLOG_CONVEYOR_RELOCATE_INDEX_ENTRIES           0x50
 
 typedef struct xl_cb_allocate_payload_segment
 {
@@ -29,6 +31,9 @@ typedef struct xl_cb_allocate_payload_segment
        bool            is_extend;
 } xl_cb_allocate_payload_segment;
 
+#define SizeOfCBAllocatePayloadSegment \
+       (offsetof(xl_cb_allocate_payload_segment, is_extend) + sizeof(bool))
+
 typedef struct xl_cb_allocate_index_segment
 {
        CBSegNo         segno;
@@ -36,6 +41,28 @@ typedef struct xl_cb_allocate_index_segment
        bool            is_extend;
 } xl_cb_allocate_index_segment;
 
+#define SizeOfCBAllocateIndexSegment \
+       (offsetof(xl_cb_allocate_index_segment, is_extend) + sizeof(bool))
+
+typedef struct xl_cb_allocate_index_page
+{
+       CBPageNo        pageno;
+} xl_cb_allocate_index_page;
+
+#define SizeOfCBAllocateIndexPage \
+       (offsetof(xl_cb_allocate_index_page, pageno) + sizeof(CBPageNo))
+
+typedef struct xl_cb_relocate_index_entries
+{
+       CBPageNo        pageno;
+       unsigned        num_index_entries;
+       uint16          pages_per_segment;
+       CBSegNo         index_entries[FLEXIBLE_ARRAY_MEMBER];
+} xl_cb_relocate_index_entries;
+
+#define SizeOfCBRelocateIndexEntries \
+       (offsetof(xl_cb_relocate_index_entries, index_entries))
+
 extern void conveyor_desc(StringInfo buf, XLogReaderState *record);
 extern void conveyor_redo(XLogReaderState *record);
 extern const char *conveyor_identify(uint8 info);