Fix interaction of partitioned tables with BulkInsertState.

When copying into a partitioned table, the target heap may change from
one tuple to next.  We must ask ReadBufferBI() to get a new buffer
every time such change occurs.  To do that, use new function
ReleaseBulkInsertStatePin().  This fixes the bug that tuples ended up
being inserted into the wrong partition, which occurred exactly
because the wrong buffer was used.

Amit Langote, per a suggestion from Robert Haas.  Some cosmetic
adjustments by me.

Reports by 高增琦 (Gao Zengqi), Venkata B Nagothi, and
Ragnar Ouchterlony.

Discussion: http://postgr.es/m/CAFmBtr32FDOqofo8yG-4mjzL1HnYHxXK5S9OGFJ%3D%3DcJpgEW4vA%40mail.gmail.com
Discussion: http://postgr.es/m/CAEyp7J9WiX0L3DoiNcRrY-9iyw%3DqP%2Bj%3DDLsAnNFF1xT2J1ggfQ%40mail.gmail.com
Discussion: http://postgr.es/m/16d73804-c9cd-14c5-463e-5caad563ff77%40agama.tv
Discussion: http://postgr.es/m/CA+TgmoaiZpDVUUN8LZ4jv1qFE_QyR+H9ec+79f5vNczYarg5Zg@mail.gmail.com

diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index 1ce42ea9702f98db6feffeb245f0d803209a39aa..5fd7f1e1a200da7a4646e4de7c971fad80bd5b41 100644 (file)
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -2324,6 +2324,17 @@ FreeBulkInsertState(BulkInsertState bistate)
        pfree(bistate);
 }
 
+/*
+ * ReleaseBulkInsertStatePin - release a buffer currently held in bistate
+ */
+void
+ReleaseBulkInsertStatePin(BulkInsertState bistate)
+{
+       if (bistate->current_buf != InvalidBuffer)
+               ReleaseBuffer(bistate->current_buf);
+       bistate->current_buf = InvalidBuffer;
+}
+
 
 /*
  *     heap_insert             - insert tuple into a heap

diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c
index c05e14e26f76f3eebfde5d1ca16d496dc4e628cb..6b199c122f1dc597218c150c78c4063e0e62aebc 100644 (file)
--- a/src/backend/commands/copy.c
+++ b/src/backend/commands/copy.c
@@ -2307,6 +2307,7 @@ CopyFrom(CopyState cstate)
        uint64          processed = 0;
        bool            useHeapMultiInsert;
        int                     nBufferedTuples = 0;
+       int                     prev_leaf_part_index = -1;
 
 #define MAX_BUFFERED_TUPLES 1000
        HeapTuple  *bufferedTuples = NULL;      /* initialize to silence warning */
@@ -2561,6 +2562,17 @@ CopyFrom(CopyState cstate)
                        Assert(leaf_part_index >= 0 &&
                                   leaf_part_index < cstate->num_partitions);
 
+                       /*
+                        * If this tuple is mapped to a partition that is not same as the
+                        * previous one, we'd better make the bulk insert mechanism gets a
+                        * new buffer.
+                        */
+                       if (prev_leaf_part_index != leaf_part_index)
+                       {
+                               ReleaseBulkInsertStatePin(bistate);
+                               prev_leaf_part_index = leaf_part_index;
+                       }
+
                        /*
                         * Save the old ResultRelInfo and switch to the one corresponding
                         * to the selected partition.

diff --git a/src/include/access/heapam.h b/src/include/access/heapam.h
index ee7e05a3ec72a4e25859963b6ed8d9ed5716c006..a864f7860d8b6ac16f91d77a341b4931048f0f65 100644 (file)
--- a/src/include/access/heapam.h
+++ b/src/include/access/heapam.h
@@ -147,6 +147,7 @@ extern void setLastTid(const ItemPointer tid);
 
 extern BulkInsertState GetBulkInsertState(void);
 extern void FreeBulkInsertState(BulkInsertState);
+extern void ReleaseBulkInsertStatePin(BulkInsertState bistate);
 
 extern Oid heap_insert(Relation relation, HeapTuple tup, CommandId cid,
                        int options, BulkInsertState bistate);