amogh-jahagirdar commented on code in PR #18544:
URL: https://github.com/apache/iceberg/pull/18544#discussion_r4303459251
##########
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)) {
Review Comment:
Minor: Should we just track if the container is dense or sparse from the
descriptor rather than looking at the cardinality? Right now the read path uses
the descriptor to decide, and this uses the cardinality, which is two separate
sources of truth for the same thing.
--
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]
