bitmap技術解析：redis與roaringBitmap

setbit key offset 1|0
getbit key
bitcount key

// bitops.c
// 操作命令: setbit key offset 0|1
/* SETBIT key offset bitvalue */
void setbitCommand(client *c) {
    robj *o;
    char *err = "bit is not an integer or out of range";
    uint64_t bitoffset;
    ssize_t byte, bit;
    int byteval, bitval;
    long on;
  // 解析 offset 值
    if (getBitOffsetFromArgument(c,c->argv[2],&bitoffset,0,0) != C_OK)
        return;
  // 解析 0|1 值
    if (getLongFromObjectOrReply(c,c->argv[3],&on,err) != C_OK)
        return;
  // 只接受0|1的輸入，其他一律報錯
    /* Bits can only be set or cleared... */
    if (on & ~1) {
        addReplyError(c,err);
        return;
    }
  // 獲取key對應的string物件，方便後續操作
    int dirty;
    if ((o = lookupStringForBitCommand(c,bitoffset,&dirty)) == NULL) return;

  // 計算偏移量: 1byte=8bit, 所以真正的位所在就等於 byte大定位 + 小移位
  // 從高到低計數, 即類似於 big-endian
    /* Get current values */
    byte = bitoffset >> 3;
    byteval = ((uint8_t*)o->ptr)[byte];
    bit = 7 - (bitoffset & 0x7);
    bitval = byteval & (1 << bit);

    /* Either it is newly created, changed length, or the bit changes before and after.
     * Note that the bitval here is actually a decimal number.
     * So we need to use `!!` to convert it to 0 or 1 for comparison. */
    if (dirty || (!!bitval != on)) {
    // 先取反保留當前值, 再重新設定on 進去
        /* Update byte with new bit value. */
        byteval &= ~(1 << bit);
        byteval |= ((on & 0x1) << bit);
        ((uint8_t*)o->ptr)[byte] = byteval;
    // 叢集擴散
        signalModifiedKey(c,c->db,c->argv[1]);
        notifyKeyspaceEvent(NOTIFY_STRING,"setbit",c->argv[1],c->db->id);
        server.dirty++;
    }
  // 返回舊的值給使用者端
    /* Return original value. */
    addReply(c, bitval ? shared.cone : shared.czero);
}
// 查詢key對應的 string 物件
/* This is a helper function for commands implementations that need to write
 * bits to a string object. The command creates or pad with zeroes the string
 * so that the 'maxbit' bit can be addressed. The object is finally
 * returned. Otherwise if the key holds a wrong type NULL is returned and
 * an error is sent to the client. */
robj *lookupStringForBitCommand(client *c, uint64_t maxbit, int *dirty) {
    size_t byte = maxbit >> 3;
    robj *o = lookupKeyWrite(c->db,c->argv[1]);
    if (checkType(c,o,OBJ_STRING)) return NULL;
    if (dirty) *dirty = 0;

    if (o == NULL) {
        o = createObject(OBJ_STRING,sdsnewlen(NULL, byte+1));
        dbAdd(c->db,c->argv[1],o);
        if (dirty) *dirty = 1;
    } else {
        o = dbUnshareStringValue(c->db,c->argv[1],o);
        size_t oldlen = sdslen(o->ptr);
        o->ptr = sdsgrowzero(o->ptr,byte+1);
        if (dirty && oldlen != sdslen(o->ptr)) *dirty = 1;
    }
    return o;
}

// 1. 引入依賴包
        <dependency>
            <groupId>org.roaringbitmap</groupId>
            <artifactId>RoaringBitmap</artifactId>
            <version>0.9.28</version>
        </dependency>
// 2. 建立單元測試
    @Test
    public void testRoaringBitmap() {
        Roaring64NavigableMap bitmapObj = new Roaring64NavigableMap();
        bitmapObj.add(11122233366L);
        boolean exists = bitmapObj.contains(1);
        long eleSize = bitmapObj.getLongCardinality();
        System.out.println("exits:" + exists + ", eleSize:" + eleSize);
    }
// 具體實現
// Roaring64NavigableMap
  /**
   * Set all the specified values to true. This can be expected to be slightly faster than calling
   * "add" repeatedly. The provided integers values don't have to be in sorted order, but it may be
   * preferable to sort them from a performance point of view.
   *
   * @param dat set values
   */
  public void add(long... dat) {
    for (long oneLong : dat) {
      addLong(oneLong);
    }
  }
  
  /**
   * Add the value to the container (set the value to "true"), whether it already appears or not.
   *
   * Java lacks native unsigned longs but the x argument is considered to be unsigned. Within
   * bitmaps, numbers are ordered according to {@link Long#compareUnsigned}. We order the numbers
   * like 0, 1, ..., 9223372036854775807, -9223372036854775808, -9223372036854775807,..., -1.
   *
   * @param x long value
   */
  @Override
  public void addLong(long x) {
  // 高低位32位元拆分 (int) (id >> 32)
    int high = high(x);
    int low = low(x);

    // Copy the reference to prevent race-condition
    Map.Entry<Integer, BitmapDataProvider> local = latestAddedHigh;

    BitmapDataProvider bitmap;
    if (local != null && local.getKey().intValue() == high) {
      bitmap = local.getValue();
    } else {
      bitmap = highToBitmap.get(high);
      if (bitmap == null) {
      // 使用 RoaringBitmap 來儲存低層資料, 一級儲存
    // 使用 treemap 儲存整個結構，保證查詢快速
        bitmap = newRoaringBitmap();
        pushBitmapForHigh(high, bitmap);
      }
    // 使用臨時儲存當前高位範例的方式，避免經常查詢map帶來的效能消耗
    // 但實際上這要求使用者端的操作是按序操作的，這樣才能很好利用這個特性，如果只是隨機值的話，效果就大打折扣了
      latestAddedHigh = new AbstractMap.SimpleImmutableEntry<>(high, bitmap);
    }
  // 儲存低位資訊
    bitmap.add(low);
  // 擴容處理
    invalidateAboveHigh(high);
  }

  private void invalidateAboveHigh(int high) {
    // The cardinalities after this bucket may not be valid anymore
    if (compare(firstHighNotValid, high) > 0) {
      // High was valid up to now
      firstHighNotValid = high;

      int indexNotValid = binarySearch(sortedHighs, firstHighNotValid);

      final int indexAfterWhichToReset;
      if (indexNotValid >= 0) {
        indexAfterWhichToReset = indexNotValid;
      } else {
        // We have invalidate a high not already present: added a value for a brand new high
        indexAfterWhichToReset = -indexNotValid - 1;
      }

      // This way, sortedHighs remains sorted, without making a new/shorter array
      Arrays.fill(sortedHighs, indexAfterWhichToReset, sortedHighs.length, highestHigh());
    }
    allValid = false;
  }

// 低位儲存實現
// roaringbitmap
  /**
   * Add the value to the container (set the value to "true"), whether it already appears or not.
   *
   * Java lacks native unsigned integers but the x argument is considered to be unsigned.
   * Within bitmaps, numbers are ordered according to {@link Integer#compareUnsigned}.
   * We order the numbers like 0, 1, ..., 2147483647, -2147483648, -2147483647,..., -1.
   *
   * @param x integer value
   */
  @Override
  public void add(final int x) {
  // 再分高低位儲存, 即32位元拆分為2個16位元, (char) (x >>> 16)
    final char hb = Util.highbits(x);
  // 已經儲存過了，則直接更新值即可
  // highLowContainer = new RoaringArray();
    final int i = highLowContainer.getIndex(hb);
    if (i >= 0) {
    // 此處查詢成功，只是代表高位已經被某些值儲存過了，但低位仍然在變化
      highLowContainer.setContainerAtIndex(i,
          highLowContainer.getContainerAtIndex(i).add(Util.lowbits(x)));
    } else {
    // 否則新插入一個你們數, 預設以陣列形式儲存, 預設初始化大小為4
      final ArrayContainer newac = new ArrayContainer();
      highLowContainer.insertNewKeyValueAt(-i - 1, hb, newac.add(Util.lowbits(x)));
    }
  }
  
  // involves a binary search
  int getIndex(char x) {
    // before the binary search, we optimize for frequent cases
    if ((size == 0) || (keys[size - 1] == x)) {
      return size - 1;
    }
  // 使用二分查詢法查詢值的存在性，實際上內部還有其他優化
    // no luck we have to go through the list
    return this.binarySearch(0, size, x);
  }
  // insert a new key, it is assumed that it does not exist
  void insertNewKeyValueAt(int i, char key, Container value) {
    extendArray(1);
    System.arraycopy(keys, i, keys, i + 1, size - i);
    keys[i] = key;
    System.arraycopy(values, i, values, i + 1, size - i);
    values[i] = value;
    size++;
  }
  // RoaringArray
  protected Container getContainerAtIndex(int i) {
    return this.values[i];
  }
// 陣列的低位儲存實現
// ArrayContainer
  /**
   * running time is in O(n) time if insert is not in order.
   */
  @Override
  public Container add(final char x) {
  // 要插入的值大於當前容量/未大於當前容量分別處理
    if (cardinality == 0 || (cardinality > 0
            && (x) > (content[cardinality - 1]))) {
    // 大於 4096 後，擴充套件為 bitmap儲存結構
      if (cardinality >= DEFAULT_MAX_SIZE) {
        return toBitmapContainer().add(x);
      }
    // 擴容，策略分多種情況處理
      if (cardinality >= this.content.length) {
        increaseCapacity();
      }
    // 直接陣列儲存具體值即可
    // 也就是說，表面看起來這裡可能會被插入重複的值，但是實際這裡插入的是比最大值還大的值
    // 更小的值則會先查詢存在性，再進行找位插入
      content[cardinality++] = x;
    } else {
      int loc = Util.unsignedBinarySearch(content, 0, cardinality, x);
    // 小的值被插入到中間，如果找到相同的值，則本次add將被忽略
    // 也就是說，這種實現的是資料的有序插入
      if (loc < 0) {
        // Transform the ArrayContainer to a BitmapContainer
        // when cardinality = DEFAULT_MAX_SIZE
        if (cardinality >= DEFAULT_MAX_SIZE) {
          return toBitmapContainer().add(x);
        }
        if (cardinality >= this.content.length) {
          increaseCapacity();
        }
        // insertion : shift the elements > x by one position to
        // the right
        // and put x in it's appropriate place
        System.arraycopy(content, -loc - 1, content, -loc, cardinality + loc + 1);
        content[-loc - 1] = x;
        ++cardinality;
      }
    }
    return this;
  }
  // temporarily allow an illegally large size, as long as the operation creating
  // the illegal container does not return it.
  private void increaseCapacity(boolean allowIllegalSize) {
    int newCapacity = (this.content.length == 0) ? DEFAULT_INIT_SIZE
        : this.content.length < 64 ? this.content.length * 2
            : this.content.length < 1067 ? this.content.length * 3 / 2
                : this.content.length * 5 / 4;
    // never allocate more than we will ever need
    if (newCapacity > ArrayContainer.DEFAULT_MAX_SIZE && !allowIllegalSize) {
      newCapacity = ArrayContainer.DEFAULT_MAX_SIZE;
    }
    // if we are within 1/16th of the max, go to max
    if (newCapacity > ArrayContainer.DEFAULT_MAX_SIZE - ArrayContainer.DEFAULT_MAX_SIZE / 16
        && !allowIllegalSize) {
      newCapacity = ArrayContainer.DEFAULT_MAX_SIZE;
    }
    this.content = Arrays.copyOf(this.content, newCapacity);
  }

// bitmap的低位儲存實現
// BitmapContainer
  // 轉移老資料到bitmap的低位儲存中
  /**
   * Copies the data in a bitmap container.
   *
   * @return the bitmap container
   */
  @Override
  public BitmapContainer toBitmapContainer() {
    BitmapContainer bc = new BitmapContainer();
    bc.loadData(this);
    return bc;
  }

  void loadData(final ArrayContainer arrayContainer) {
    this.cardinality = arrayContainer.cardinality;
    for (int k = 0; k < arrayContainer.cardinality; ++k) {
      final char x = arrayContainer.content[k];
    // 取整64, 這個移位是真沒看懂, 反正超過31之後的
      bitmap[(x) / 64] |= (1L << x);
    }
  }
  
  @Override
  public Container add(final char i) {
    final long previous = bitmap[i >>> 6];
    long newval = previous | (1L << i);
    bitmap[i >>> 6] = newval;
    if (USE_BRANCHLESS) {
      cardinality += (int)((previous ^ newval) >>> i);
    } else if (previous != newval) {
      ++cardinality;
    }
    return this;
  }