stevenzwu commented on code in PR #18544:
URL: https://github.com/apache/iceberg/pull/18544#discussion_r4300034701
##########
core/src/main/java/org/apache/iceberg/mumbling/MumblingBitmap.java:
##########
@@ -138,6 +160,282 @@ private int offset(int containerIndex) {
return offsets[containerIndex];
}
+ public MutableBitmap asMutable() {
+ return new Mutable(this);
+ }
+
+ /**
+ * A copy-on-write {@link MutableBitmap} implementation for Mumbling.
+ *
+ * <p>When a new bit position is set, an overlay is created for the
position's container.
+ *
+ * <p>The initial buffer is not modified.
+ */
+ private static class Mutable implements MutableBitmap {
+ private final NavigableMap<Integer, Container> overlays =
Maps.newTreeMap();
+ private final MumblingBitmap base;
+ private final int baseContainerCount;
+ private int cardinality = 0;
+
+ public Mutable(MumblingBitmap base) {
+ this.base = base;
+ this.baseContainerCount = base.descriptors().length;
+ this.cardinality = base.cardinality();
+ }
+
+ @Override
+ public int cardinality() {
+ return cardinality;
+ }
+
+ @Override
+ public boolean isSet(int pos) {
+ Container overlay = overlays.get(containerIndex(pos));
+ if (overlay != null) {
+ return overlay.isSet(posInContainer(pos));
+ }
+
+ return base.isSet(pos);
+ }
+
+ @Override
+ public boolean set(int pos) {
+ Preconditions.checkArgument(pos >= 0, "Cannot set invalid bit position:
%s < 0", pos);
+ Preconditions.checkArgument(
+ pos < MAX_CARDINALITY, "Cannot set invalid bit position: %s (must be
< 2,097,152)", pos);
+
+ int containerIndex = containerIndex(pos);
+ Container overlay = overlays.get(containerIndex);
+ if (overlay == null) {
+ if (containerIndex < baseContainerCount) {
+ overlay =
+ new Container(
+ base.buffer(), base.offset(containerIndex),
base.descriptor(containerIndex));
+ } else {
+ overlay = new Container();
+ }
+ overlays.put(containerIndex, overlay);
+ }
+
+ if (overlay.set(posInContainer(pos))) {
+ this.cardinality += 1;
+ return true;
+ }
+
+ return false;
+ }
+
+ @Override
+ public ByteBuffer buffer() {
+ int[] descriptorArray = descriptorArray();
+ int containersSize = sizeOfContainers(descriptorArray, 0,
descriptorArray.length);
+ ByteBuffer descBuffer = PFOREncoding.encode(descriptorArray,
descriptorArray.length);
+ int descriptorsSize = descBuffer.remaining();
+
+ ByteBuffer buffer = ByteBuffer.allocate(HEADER_SIZE + descriptorsSize +
containersSize);
+ int currentPos = writeHeader(buffer, cardinality,
descriptorArray.length);
+
+ buffer.put(HEADER_SIZE, descBuffer, descBuffer.position(),
descriptorsSize);
+ currentPos += descriptorsSize;
+
+ int nextContainerIndex = 0;
+ for (Map.Entry<Integer, Container> entry : overlays.entrySet()) {
+ int containerIndex = entry.getKey();
+ currentPos +=
+ copyOriginalContainers(
+ descriptorArray, nextContainerIndex, containerIndex, buffer,
currentPos);
+ currentPos += entry.getValue().writeTo(buffer, currentPos);
+ nextContainerIndex = containerIndex + 1;
+ }
+
+ copyOriginalContainers(
+ descriptorArray, nextContainerIndex, descriptorArray.length, buffer,
currentPos);
+
+ return buffer;
+ }
+
+ private int[] descriptorArray() {
+ int[] baseDescriptors = base.descriptors();
+ int lastOverlayIndex = overlays.isEmpty() ? -1 : overlays.lastKey();
+ int numContainers = Math.max(lastOverlayIndex + 1,
baseDescriptors.length);
+ int[] descriptors = new int[numContainers]; // filled with 0
+ System.arraycopy(baseDescriptors, 0, descriptors, 0,
baseDescriptors.length);
+
+ for (Map.Entry<Integer, Container> entry : overlays.entrySet()) {
+ int containerIndex = entry.getKey();
+ int cardinality = entry.getValue().cardinality();
+ descriptors[containerIndex] = Math.min(cardinality,
DENSE_CONTAINER_SIZE);
+ }
+
+ return descriptors;
+ }
+
+ private int sizeOfContainers(int[] descriptorArray, int startIndex, int
endIndexExclusive) {
+ int size = 0;
+ for (int i = startIndex; i < endIndexExclusive; i += 1) {
+ size += descriptorArray[i];
+ }
+
+ return size;
+ }
+
+ private static int writeHeader(ByteBuffer buffer, int cardinality, int
descriptorCount) {
+ ByteBuffers.writeByte(buffer, VERSION, 0);
+ ByteBuffers.writeLittleEndianUnsigned(buffer, cardinality, 1, 3);
+ ByteBuffers.writeLittleEndianUnsigned(buffer, descriptorCount, 4, 2);
+
+ return HEADER_SIZE;
+ }
+
+ private int copyOriginalContainers(
+ int[] descriptorArray,
+ int startIndex,
+ int endIndexExclusive,
+ ByteBuffer buffer,
+ int currentPos) {
+ if (startIndex < baseContainerCount && startIndex < endIndexExclusive) {
+ int offset = base.offset(startIndex);
+ int sizeToCopy = sizeOfContainers(descriptorArray, startIndex,
endIndexExclusive);
+ if (sizeToCopy > 0) {
+ buffer.put(currentPos, base.buffer(), offset, sizeToCopy);
+ }
+
+ return sizeToCopy;
+ }
+
+ return 0;
+ }
+ }
+
+ private static class Container {
+ private byte[] dense = null;
+ private byte[] sparse = null;
+ private int cardinality = 0;
+
+ private Container() {
+ this.sparse = new byte[DENSE_CONTAINER_SIZE];
+ }
+
+ private Container(ByteBuffer buffer, int offset, int length) {
+ byte[] bytes = new byte[DENSE_CONTAINER_SIZE];
+ buffer.get(offset, bytes, 0, length);
+
+ if (isDense(length)) {
+ this.dense = bytes;
+ this.cardinality = denseCardinality(dense);
+ } else {
+ this.sparse = bytes;
+ this.cardinality = length;
+ }
+ }
+
+ int cardinality() {
+ return cardinality;
+ }
+
+ private boolean isSet(int posInContainer) {
+ if (cardinality < 32) {
+ // sparse
+ for (int i = 0; i < cardinality; i += 1) {
+ if ((sparse[i] & 0xFF) == posInContainer) {
+ return true;
+ }
+ }
+
+ return false;
+ } else {
+ // dense
+ int bitShift = denseBitShift(posInContainer);
+ return ((dense[denseByteIndex(posInContainer)] >>> bitShift) & 0b1) ==
0b1;
+ }
+ }
+
+ private boolean set(int posInContainer) {
+ if (setInternal(posInContainer)) {
+ this.cardinality += 1;
+ return true;
+ }
+
+ return false;
+ }
+
+ int writeTo(ByteBuffer buffer, int offset) {
+ if (cardinality < DENSE_CONTAINER_SIZE) {
+ buffer.put(offset, sparse, 0, cardinality);
+ return cardinality;
+ } else {
+ buffer.put(offset, dense, 0, dense.length);
+ return dense.length;
+ }
+ }
+
+ private boolean setInternal(int posInContainer) {
+ if (cardinality < 31) {
+ return sparseSet(sparse, cardinality, posInContainer);
+ } else if (cardinality > 31) {
+ return denseSet(dense, posInContainer);
+ } else {
+ // convert to dense, reuse converted buffer if it is present
+ if (dense == null) {
+ this.dense = convertToDense(sparse, cardinality);
+ }
+
+ if (denseSet(dense, posInContainer)) {
+ this.sparse = null;
+ return true;
+ }
+
+ // leave the converted dense array for next set attempt
+ return false;
+ }
+ }
+
+ private static boolean sparseSet(byte[] bytes, int length, int
posInContainer) {
+ byte nextInsert = (byte) posInContainer;
+ for (int i = 0; i < length; i += 1) {
+ if (nextInsert == bytes[i]) {
+ return false; // already set
+ } else if ((nextInsert & 0xFF) < (bytes[i] & 0xFF)) {
+ byte replaced = bytes[i];
+ bytes[i] = nextInsert;
+ nextInsert = replaced;
+ }
+ }
+
+ bytes[length] = nextInsert;
+ return true;
+ }
+
+ private static byte[] convertToDense(byte[] sparse, int cardinality) {
+ byte[] dense = new byte[32];
+ for (int i = 0; i < cardinality; i += 1) {
+ denseSet(dense, (sparse[i] & 0xFF));
+ }
+
+ return dense;
+ }
+
+ private static boolean denseSet(byte[] bytes, int posInContainer) {
+ int byteIndex = denseByteIndex(posInContainer);
+ byte bit = (byte) (0b1 << denseBitShift(posInContainer));
+ if ((bytes[byteIndex] & bit) != bit) {
+ bytes[byteIndex] |= bit;
+ return true;
+ }
+
+ return false;
+ }
+
+ private static int denseCardinality(byte[] bytes) {
+ int cardinality = 0;
+ for (int i = 0; i < bytes.length; i += 1) {
+ cardinality += Integer.bitCount(bytes[i]);
Review Comment:
`Integer.bitCount` takes an `int`, and the `byte` is sign-extended first. A
byte `>= 0x80` gains 24 extra one-bits. `0xFF` widens to `0xFFFFFFFF` and
counts as 32 instead of 8. We should mask before counting:
```java
cardinality += Integer.bitCount(bytes[i] & 0xFF);
```
`DENSE_CONTAINER` sets bytes 1, 3, 5, and 7 to `0xFF`, so this container has
32 bits set. This method returns 4 × 32 = 128.
In `descriptorArray()`, `Math.min(cardinality, DENSE_CONTAINER_SIZE)` still
writes descriptor 32, and the `>= 32` branches still read and write the dense
bytes. `testMutableOverlayDenseUpdate` and
`testMutableDenseUpdateSerialization` check the bitmap's own cardinality
counter, the descriptor, and those bytes. They never check this bitcount.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
