REQ: allocate the first compactor's storage on first use; add ensureCapacity() A new ReqSketch, an empty heapified one, and one after reset() no longer allocate the storage of the first compactor. It is allocated on the first update, or the first merge of a non-empty sketch, before anything else changes, with the same size as before: twice the compactor's nominal capacity. ensureCapacity() allocates it up front. Results and the serialized form are unchanged. merge() now grows its levels before it changes totalN, min and max. ReqSketchIterator.next() now skips any level without items, instead of assuming every level above level 0 has some. This follows the lazy allocation convention agreed on dev@. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
diff --git a/src/main/java/org/apache/datasketches/req/ReqCompactor.java b/src/main/java/org/apache/datasketches/req/ReqCompactor.java index 151cd8f..71aa4df 100644 --- a/src/main/java/org/apache/datasketches/req/ReqCompactor.java +++ b/src/main/java/org/apache/datasketches/req/ReqCompactor.java
@@ -62,6 +62,15 @@ final boolean hra, final int sectionSize, final ReqDebug reqDebug) { + this(lgWeight, hra, sectionSize, reqDebug, true); + } + + private ReqCompactor( + final byte lgWeight, + final boolean hra, + final int sectionSize, + final ReqDebug reqDebug, + final boolean allocate) { this.lgWeight = lgWeight; this.hra = hra; this.sectionSize = sectionSize; @@ -70,7 +79,29 @@ coin = false; numSections = INIT_NUMBER_OF_SECTIONS; final int nomCap = getNomCapacity(); - buf = new FloatBuffer(2 * nomCap, nomCap, hra); + buf = new FloatBuffer(allocate ? 2 * nomCap : 0, nomCap, hra); + } + + /** + * Constructs a compactor whose buffer has no storage yet. The first compactor of a new sketch uses this, + * and the sketch allocates the storage on first use with {@link #ensureCapacity()}. + * @param lgWeight the lgWeight of this compactor + * @param hra High Rank Accuracy + * @param sectionSize initially the size of k + * @param reqDebug The debug signaling interface + * @return a new compactor with an empty, unallocated buffer + */ + static ReqCompactor newUnallocated(final byte lgWeight, final boolean hra, final int sectionSize, + final ReqDebug reqDebug) { + return new ReqCompactor(lgWeight, hra, sectionSize, reqDebug, false); + } + + /** + * Allocates the buffer storage, twice the nominal capacity, if the buffer has none yet. + * It has no effect otherwise. + */ + void ensureCapacity() { + if (buf.getCapacity() == 0) { buf.ensureCapacity(2 * getNomCapacity()); } } /**
diff --git a/src/main/java/org/apache/datasketches/req/ReqSketch.java b/src/main/java/org/apache/datasketches/req/ReqSketch.java index eda33ac..0bf58d7 100644 --- a/src/main/java/org/apache/datasketches/req/ReqSketch.java +++ b/src/main/java/org/apache/datasketches/req/ReqSketch.java
@@ -74,6 +74,12 @@ * help users understand how the sketch works.</li> * </ul> * + * <p><b>Memory allocation.</b> A new sketch does not allocate storage for items until it is first used: + * the first update, or the first merge of a non-empty sketch. After that, the storage grows as the sketch + * needs it. An {@link OutOfMemoryError} from that first allocation therefore happens in the update or merge, + * not when the sketch is built. Call {@link #ensureCapacity()} right after building the sketch to allocate + * the storage up front instead. Results and the serialized form are the same either way.</p> + * * @see QuantilesAPI * * @author Edo Liberty @@ -141,7 +147,27 @@ hra = highRankAccuracy; this.reqDebug = reqDebug; rand = (reqDebug == null) ? new Random() : new Random(1); - grow(); + growUnallocated(); + } + + /** + * Allocates the storage of the first compactor now, rather than on the first update or merge. + * The storage is the same as a sketch allocated when it was built before: twice the nominal capacity + * of the first compactor. Call this right after building the sketch to move an {@link OutOfMemoryError} + * from the first update or merge to this call. It has no effect if the storage already exists. + * + * <p>{@link #reset()} releases the storage, and it is allocated again on first use.</p> + */ + public void ensureCapacity() { + compactors.get(0).ensureCapacity(); + } + + /** + * Returns true if the first compactor has storage. For tests. + * @return true if the first compactor has storage + */ + boolean hasItemStorage() { + return compactors.get(0).getBuffer().getCapacity() > 0; } /** @@ -384,12 +410,14 @@ throw new SketchesArgumentException( "Both sketches must have the same HighRankAccuracy setting."); } + //allocate first: the first compactor's storage, then any new levels, before the counters change + ensureCapacity(); + //Grow until self has at least as many compactors as other + while (getNumLevels() < other.getNumLevels()) { grow(); } totalN += other.totalN; //update min, max items, n if (Float.isNaN(minItem) || (other.minItem < minItem)) { minItem = other.minItem; } if (Float.isNaN(maxItem) || (other.maxItem > maxItem)) { maxItem = other.maxItem; } - //Grow until self has at least as many compactors as other - while (getNumLevels() < other.getNumLevels()) { grow(); } //Merge the items in all height compactors for (int i = 0; i < other.getNumLevels(); i++) { compactors.get(i).merge(other.compactors.get(i)); @@ -413,7 +441,7 @@ maxItem = Float.NaN; reqSV = null; compactors = new ArrayList<>(); - grow(); + growUnallocated(); //the storage is allocated again on first use } @Override @@ -440,6 +468,7 @@ @Override public void update(final float item) { if (Float.isNaN(item)) { return; } + ensureCapacity(); //first use allocates, before anything changes if (isEmpty()) { minItem = item; maxItem = item; @@ -549,9 +578,18 @@ } private void grow() { + addCompactor(new ReqCompactor((byte)getNumLevels(), hra, k, reqDebug)); + } + + //adds the first compactor of a new or reset sketch, with no storage until first use + private void growUnallocated() { + addCompactor(ReqCompactor.newUnallocated((byte)getNumLevels(), hra, k, reqDebug)); + } + + private void addCompactor(final ReqCompactor compactor) { final byte lgWeight = (byte)getNumLevels(); if ((lgWeight == 0) && (reqDebug != null)) { reqDebug.emitStart(this); } - compactors.add(new ReqCompactor(lgWeight, hra, k, reqDebug)); + compactors.add(compactor); maxNomSize = computeMaxNomSize(); if (reqDebug != null) { reqDebug.emitNewCompactor(lgWeight); } }
diff --git a/src/main/java/org/apache/datasketches/req/ReqSketchBuilder.java b/src/main/java/org/apache/datasketches/req/ReqSketchBuilder.java index 6ed9ce2..c37e926 100644 --- a/src/main/java/org/apache/datasketches/req/ReqSketchBuilder.java +++ b/src/main/java/org/apache/datasketches/req/ReqSketchBuilder.java
@@ -44,6 +44,7 @@ /** * Returns a new ReqSketch with the current configuration of the builder. + * The sketch allocates its item storage on first use; see {@link ReqSketch#ensureCapacity()}. * @return a new ReqSketch */ public ReqSketch build() {
diff --git a/src/main/java/org/apache/datasketches/req/ReqSketchIterator.java b/src/main/java/org/apache/datasketches/req/ReqSketchIterator.java index 8fc69d3..c6c3827 100644 --- a/src/main/java/org/apache/datasketches/req/ReqSketchIterator.java +++ b/src/main/java/org/apache/datasketches/req/ReqSketchIterator.java
@@ -55,16 +55,14 @@ @Override public boolean next() { - if ((numRetainedItems == 0) - || ((cIndex == (compactors.size() - 1)) && (bIndex == (currentBuf.getCount() - 1)))) { - return false; - } - if (bIndex == (currentBuf.getCount() - 1)) { + if (numRetainedItems == 0) { return false; } + if (bIndex < currentBuf.getCount()) { bIndex++; } + //skip any level without items, so the iterator never stops on an empty one + while (bIndex >= currentBuf.getCount()) { + if (cIndex >= (compactors.size() - 1)) { return false; } cIndex++; currentBuf = compactors.get(cIndex).getBuffer(); bIndex = 0; - } else { - bIndex++; } return true; }
diff --git a/src/test/java/org/apache/datasketches/req/ReqLazyAllocationTest.java b/src/test/java/org/apache/datasketches/req/ReqLazyAllocationTest.java new file mode 100644 index 0000000..739a2b2 --- /dev/null +++ b/src/test/java/org/apache/datasketches/req/ReqLazyAllocationTest.java
@@ -0,0 +1,181 @@ +/* + * 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.datasketches.req; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertTrue; + +import java.lang.foreign.MemorySegment; +import java.lang.reflect.Proxy; +import java.util.ArrayList; +import java.util.List; + +import org.apache.datasketches.quantilescommon.QuantilesFloatsSketchIterator; +import org.testng.annotations.Test; + +/** + * Checks that a ReqSketch allocates the storage of its first compactor on first use, or when + * ensureCapacity() is called, with the same results either way. + */ +public class ReqLazyAllocationTest { + private static final int K = 12; + + // A silent ReqDebug: with one, the sketch uses a fixed random seed, so two sketches can be compared byte for byte. + private static final ReqDebug QUIET = (ReqDebug) Proxy.newProxyInstance( + ReqDebug.class.getClassLoader(), new Class<?>[] {ReqDebug.class}, (proxy, method, args) -> null); + + private static ReqSketch newSketch(final boolean hra) { + return ReqSketch.builder().setK(K).setHighRankAccuracy(hra).setReqDebug(QUIET).build(); + } + + private static float value(final int i) { return (i * 7919) % 100_003; } + + private static void checkIterator(final ReqSketch sk) { + final QuantilesFloatsSketchIterator it = sk.iterator(); + long weight = 0; + int count = 0; + while (it.next()) { + it.getQuantile(); + weight += it.getWeight(); + count++; + } + assertFalse(it.next()); //stays exhausted + assertEquals(count, sk.getNumRetained()); + assertEquals(weight, sk.getN()); + } + + @Test + public void emptySketchHoldsNoStorage() { + for (final boolean hra : new boolean[] {true, false}) { + final ReqSketch sk = newSketch(hra); + assertFalse(sk.hasItemStorage()); + sk.update(Float.NaN); //ignored, so not a first use + sk.merge(newSketch(hra)); //merging an empty sketch is not a first use + sk.merge(null); + assertTrue(sk.isEmpty()); + checkIterator(sk); + sk.toString(); + sk.viewCompactorDetail("%4.0f", true); + final ReqSketch reserved = newSketch(hra); + reserved.ensureCapacity(); + assertEquals(sk.toByteArray(), reserved.toByteArray()); + assertFalse(sk.hasItemStorage()); + final ReqSketch copy = ReqSketch.heapify(MemorySegment.ofArray(sk.toByteArray())); + assertFalse(copy.hasItemStorage()); + assertTrue(copy.isEmpty()); + } + } + + @Test + public void ensureCapacity() { + for (final boolean hra : new boolean[] {true, false}) { + final ReqSketch sk = newSketch(hra); + sk.ensureCapacity(); + assertTrue(sk.hasItemStorage()); + final ReqCompactor c0 = sk.getCompactors().get(0); + final FloatBuffer storage = c0.getBuffer(); + assertEquals(storage.getCapacity(), 2 * c0.getNomCapacity()); //the size a new sketch used to allocate + sk.ensureCapacity(); //no effect once allocated + assertEquals(storage.getCapacity(), 2 * c0.getNomCapacity()); + + final ReqSketch lazy = newSketch(hra); + for (int i = 0; i < 100_000; i++) { sk.update(value(i)); lazy.update(value(i)); } + assertEquals(sk.toByteArray(), lazy.toByteArray()); + + final byte[] image = sk.toByteArray(); + sk.ensureCapacity(); //no effect after updates + assertEquals(sk.toByteArray(), image); + } + } + + @Test + public void firstUseAllocates() { + for (final boolean hra : new boolean[] {true, false}) { + final ReqSketch updated = newSketch(hra); + updated.update(1); + assertTrue(updated.hasItemStorage()); + assertEquals(updated.getMinItem(), 1f); + + final ReqSketch other = newSketch(hra); + for (int i = 0; i < 10_000; i++) { other.update(value(i)); } + final ReqSketch lazy = newSketch(hra); + final ReqSketch reserved = newSketch(hra); + reserved.ensureCapacity(); + lazy.merge(other); + reserved.merge(other); + assertTrue(lazy.hasItemStorage()); + assertEquals(lazy.toByteArray(), reserved.toByteArray()); + assertEquals(lazy.getN(), other.getN()); + assertEquals(lazy.getMinItem(), other.getMinItem()); + assertEquals(lazy.getMaxItem(), other.getMaxItem()); + checkIterator(lazy); + } + } + + @Test + public void resetReleasesStorage() { + final ReqSketch sk = newSketch(true); + for (int i = 0; i < 10_000; i++) { sk.update(value(i)); } + sk.reset(); + assertFalse(sk.hasItemStorage()); + assertTrue(sk.isEmpty()); + checkIterator(sk); + sk.update(5); + assertTrue(sk.hasItemStorage()); + assertEquals(sk.getN(), 1); + assertEquals(sk.getMaxItem(), 5f); + } + + @Test + public void iteratorCoversRetainedItems() { + for (final boolean hra : new boolean[] {true, false}) { + for (final int n : new int[] {0, 1, 23, 24, 25, 100, 1000, 10_000}) { + final ReqSketch sk = newSketch(hra); + for (int i = 0; i < n; i++) { sk.update(value(i)); } + checkIterator(sk); + final ReqSketch merged = newSketch(hra); + merged.merge(sk); + merged.merge(sk); + checkIterator(merged); + } + } + } + + @Test + public void iteratorSkipsLevelsWithoutItems() { + //levels 0 and 2 hold items, levels 1 and 3 hold none + final boolean hra = true; + final List<ReqCompactor> compactors = new ArrayList<>(); + final float[][] items = { {1, 2, 3}, {}, {4, 5}, {} }; + long n = 0; + for (int h = 0; h < items.length; h++) { + final FloatBuffer buf = FloatBuffer.reconstruct(items[h], items[h].length, 64, 32, true, hra); + compactors.add(new ReqCompactor((byte) h, hra, 0, K, (byte) 3, buf)); + n += (long) items[h].length << h; + } + final ReqSketch sk = new ReqSketch(K, hra, n, 1, 5, compactors); + sk.setMaxNomSize(sk.computeMaxNomSize()); + sk.setRetainedItems(sk.computeTotalRetainedItems()); + checkIterator(sk); + assertEquals(sk.getN(), 3 + (2 << 2)); + } +}