granicus.if.org Git - postgresql/blob - contrib/bloom/blscan.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * blscan.c
   4  *              Bloom index scan functions.
   5  *
   6  * Copyright (c) 2016-2019, PostgreSQL Global Development Group
   7  *
   8  * IDENTIFICATION
   9  *        contrib/bloom/blscan.c
  10  *
  11  *-------------------------------------------------------------------------
  12  */
  13 #include "postgres.h"
  14
  15 #include "access/relscan.h"
  16 #include "bloom.h"
  17 #include "miscadmin.h"
  18 #include "pgstat.h"
  19 #include "storage/bufmgr.h"
  20 #include "storage/lmgr.h"
  21 #include "utils/memutils.h"
  22 #include "utils/rel.h"
  23
  24 /*
  25  * Begin scan of bloom index.
  26  */
  27 IndexScanDesc
  28 blbeginscan(Relation r, int nkeys, int norderbys)
  29 {
  30         IndexScanDesc scan;
  31         BloomScanOpaque so;
  32
  33         scan = RelationGetIndexScan(r, nkeys, norderbys);
  34
  35         so = (BloomScanOpaque) palloc(sizeof(BloomScanOpaqueData));
  36         initBloomState(&so->state, scan->indexRelation);
  37         so->sign = NULL;
  38
  39         scan->opaque = so;
  40
  41         return scan;
  42 }
  43
  44 /*
  45  * Rescan a bloom index.
  46  */
  47 void
  48 blrescan(IndexScanDesc scan, ScanKey scankey, int nscankeys,
  49                  ScanKey orderbys, int norderbys)
  50 {
  51         BloomScanOpaque so = (BloomScanOpaque) scan->opaque;
  52
  53         if (so->sign)
  54                 pfree(so->sign);
  55         so->sign = NULL;
  56
  57         if (scankey && scan->numberOfKeys > 0)
  58         {
  59                 memmove(scan->keyData, scankey,
  60                                 scan->numberOfKeys * sizeof(ScanKeyData));
  61         }
  62 }
  63
  64 /*
  65  * End scan of bloom index.
  66  */
  67 void
  68 blendscan(IndexScanDesc scan)
  69 {
  70         BloomScanOpaque so = (BloomScanOpaque) scan->opaque;
  71
  72         if (so->sign)
  73                 pfree(so->sign);
  74         so->sign = NULL;
  75 }
  76
  77 /*
  78  * Insert all matching tuples into a bitmap.
  79  */
  80 int64
  81 blgetbitmap(IndexScanDesc scan, TIDBitmap *tbm)
  82 {
  83         int64           ntids = 0;
  84         BlockNumber blkno = BLOOM_HEAD_BLKNO,
  85                                 npages;
  86         int                     i;
  87         BufferAccessStrategy bas;
  88         BloomScanOpaque so = (BloomScanOpaque) scan->opaque;
  89
  90         if (so->sign == NULL)
  91         {
  92                 /* New search: have to calculate search signature */
  93                 ScanKey         skey = scan->keyData;
  94
  95                 so->sign = palloc0(sizeof(BloomSignatureWord) * so->state.opts.bloomLength);
  96
  97                 for (i = 0; i < scan->numberOfKeys; i++)
  98                 {
  99                         /*
 100                          * Assume bloom-indexable operators to be strict, so nothing could
 101                          * be found for NULL key.
 102                          */
 103                         if (skey->sk_flags & SK_ISNULL)
 104                         {
 105                                 pfree(so->sign);
 106                                 so->sign = NULL;
 107                                 return 0;
 108                         }
 109
 110                         /* Add next value to the signature */
 111                         signValue(&so->state, so->sign, skey->sk_argument,
 112                                           skey->sk_attno - 1);
 113
 114                         skey++;
 115                 }
 116         }
 117
 118         /*
 119          * We're going to read the whole index. This is why we use appropriate
 120          * buffer access strategy.
 121          */
 122         bas = GetAccessStrategy(BAS_BULKREAD);
 123         npages = RelationGetNumberOfBlocks(scan->indexRelation);
 124
 125         for (blkno = BLOOM_HEAD_BLKNO; blkno < npages; blkno++)
 126         {
 127                 Buffer          buffer;
 128                 Page            page;
 129
 130                 buffer = ReadBufferExtended(scan->indexRelation, MAIN_FORKNUM,
 131                                                                         blkno, RBM_NORMAL, bas);
 132
 133                 LockBuffer(buffer, BUFFER_LOCK_SHARE);
 134                 page = BufferGetPage(buffer);
 135                 TestForOldSnapshot(scan->xs_snapshot, scan->indexRelation, page);
 136
 137                 if (!PageIsNew(page) && !BloomPageIsDeleted(page))
 138                 {
 139                         OffsetNumber offset,
 140                                                 maxOffset = BloomPageGetMaxOffset(page);
 141
 142                         for (offset = 1; offset <= maxOffset; offset++)
 143                         {
 144                                 BloomTuple *itup = BloomPageGetTuple(&so->state, page, offset);
 145                                 bool            res = true;
 146
 147                                 /* Check index signature with scan signature */
 148                                 for (i = 0; i < so->state.opts.bloomLength; i++)
 149                                 {
 150                                         if ((itup->sign[i] & so->sign[i]) != so->sign[i])
 151                                         {
 152                                                 res = false;
 153                                                 break;
 154                                         }
 155                                 }
 156
 157                                 /* Add matching tuples to bitmap */
 158                                 if (res)
 159                                 {
 160                                         tbm_add_tuples(tbm, &itup->heapPtr, 1, true);
 161                                         ntids++;
 162                                 }
 163                         }
 164                 }
 165
 166                 UnlockReleaseBuffer(buffer);
 167                 CHECK_FOR_INTERRUPTS();
 168         }
 169         FreeAccessStrategy(bas);
 170
 171         return ntids;
 172 }