stevenzwu commented on code in PR #18544:
URL: https://github.com/apache/iceberg/pull/18544#discussion_r4299216621
##########
core/src/main/java/org/apache/iceberg/mumbling/MumblingBitmap.java:
##########
@@ -19,8 +19,13 @@
package org.apache.iceberg.mumbling;
import java.nio.ByteBuffer;
+import java.util.Map;
+import java.util.NavigableMap;
+import org.apache.curator.shaded.com.google.common.collect.Maps;
Review Comment:
Import `org.apache.iceberg.relocated.com.google.common.collect.Maps` instead
of the Curator-shaded class.
##########
core/src/main/java/org/apache/iceberg/mumbling/MumblingBitmap.java:
##########
Review Comment:
nit: these two lines can use the small helper methods.
##########
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.
The round-trip tests still pass because 128 is above 32. 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.
##########
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);
Review Comment:
do we need to validate firstOverlayIndex equals to baseDescriptors.length?
##########
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) {
Review Comment:
the magic number 31 is a bit difficult to get. should we use
`DENSE_CONTAINER_SIZE`. like
```
if (cardinality < DENSE_CONTAINER_SIZE - 1) {
// sparse container still have room to grow
return sparseSet(sparse, cardinality, posInContainer);
} else if (cardinality >= DENSE_CONTAINER_SIZE) {
// already a dense container
return denseSet(dense, posInContainer);
} else {
// cardinality == 31: convert to dense, reuse converted buffer if it is
present
}
```
##########
core/src/main/java/org/apache/iceberg/mumbling/MumblingBitmap.java:
##########
@@ -51,23 +64,23 @@ class MumblingBitmap implements ManifestBitmap {
private volatile int[] offsets = null;
MumblingBitmap(ByteBuffer data) {
- int version = data.get(data.position()) & 0xFF;
+ int version = ByteBuffers.readByte(data, 0);
if (version != VERSION) {
throw new UnsupportedOperationException("Unsupported Mumbling bitmap
version: " + version);
}
this.data = data;
- this.cardinality =
- (data.get(data.position() + 1) & 0xFF)
- | ((data.get(data.position() + 2) & 0xFF) << 8)
- | ((data.get(data.position() + 3) & 0xFF) << 16);
- this.containerCount =
- (data.get(data.position() + 4) & 0xFF) | ((data.get(data.position() +
5) & 0xFF) << 8);
+ this.cardinality = ByteBuffers.readLittleEndianUnsigned(data, 1, 3);
+ this.containerCount = ByteBuffers.readLittleEndianUnsigned(data, 4, 2);
Preconditions.checkState(
- containerCount <= 8192, "Invalid container count: %s > 8,192 (max)",
containerCount);
+ containerCount < MAX_CONTAINER_COUNT,
Review Comment:
This PR changes the existing `<=` checks to `<`. Container count 8,192 and
cardinality 2,097,152 are valid. Keep `<=`.
##########
core/src/main/java/org/apache/iceberg/mumbling/MutableBitmap.java:
##########
@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.mumbling;
+
+import org.apache.iceberg.ManifestBitmap;
+
+/** Mutation methods for ManifestBitmap. */
+public interface MutableBitmap extends ManifestBitmap {
Review Comment:
I found the naming a bit off. `ManifestBitmap` is a very specific bitmap for
manifest DV. The `MutableBitmap` sounds like a generic bitmap interface. How
about `MutableManifestBitmap`?
##########
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);
Review Comment:
This writer emits `0b0010_0000`, but accoording to spec readers must ignore
the low five bits of a dense descriptor. Should `isDense` check only the top
three bits: `(descriptor & 0b1110_0000) == DENSE_CONTAINER_BIT`?
Dense descriptors may be greater than `DENSE_CONTAINER_SIZE`, but their
payload is always 32 bytes. Should this use `isDense(length) ?
DENSE_CONTAINER_SIZE : length` as the read size?
Is `descriptor` a better arg name than `length`? Below is the only the
caller code of this constructor.
```
if (containerIndex < baseContainerCount) {
overlay =
new Container(
base.buffer(), base.offset(containerIndex),
base.descriptor(containerIndex));
```
--
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]
