blob: 8cd4552a32731642defb9c367f518756dd8929cd [file]
/*
* 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.quantiles;
import static org.apache.datasketches.common.Util.LS;
import static org.apache.datasketches.quantiles.ClassicUtil.checkIsMemorySegmentCompact;
import static org.apache.datasketches.quantiles.DoublesUtil.copyToHeap;
import java.lang.foreign.MemorySegment;
import java.util.Objects;
import org.apache.datasketches.common.MemorySegmentRequest;
import org.apache.datasketches.common.SketchesArgumentException;
import org.apache.datasketches.common.SketchesReadOnlyException;
/**
* Union operation.
*
* @author Lee Rhodes
* @author Kevin Lang
*/
final class QuantilesDoublesUnionImpl extends QuantilesDoublesUnion {
int maxK_;
UpdatableQuantilesDoublesSketch gadget_ = null;
private QuantilesDoublesUnionImpl(final int maxK) {
maxK_ = maxK;
}
/**
* Returns a empty QuantilesDoublesUnion object on the heap.
* @param maxK determines the accuracy and size of the union and is a maximum.
* The effective <i>k</i> can be smaller due to unions with smaller <i>k</i> sketches.
* It is recommended that <i>maxK</i> be a power of 2 to enable unioning of sketches with
* different <i>k</i>.
* @return a new QuantilesDoublesUnionImpl on the Java heap
*/
static QuantilesDoublesUnionImpl heapInstance(final int maxK) {
return new QuantilesDoublesUnionImpl(maxK);
}
/**
* Returns a empty QuantilesDoublesUnion object that uses the given MemorySegment for its internal sketch gadget
* and will be initialized to the empty state.
*
* @param maxK determines the accuracy and size of the union and is a maximum.
* The effective <i>k</i> can be smaller due to unions with smaller <i>k</i> sketches.
* It is recommended that <i>maxK</i> be a power of 2 to enable unioning of sketches with
* different <i>k</i>.
* @param dstSeg the MemorySegment to be used by the internal sketch and must not be null.
* @param mSegReq the MemorySegmentRequest used if the given MemorySegment needs to expand.
* Otherwise, it can be null and the default MemorySegmentRequest will be used.
* @return a QuantilesDoublesUnion object
*/
static QuantilesDoublesUnionImpl directInstance(final int maxK, final MemorySegment dstSeg, final MemorySegmentRequest mSegReq) {
Objects.requireNonNull(dstSeg);
final DirectUpdateDoublesSketch sketch = DirectUpdateDoublesSketch.newInstance(maxK, dstSeg, mSegReq);
final QuantilesDoublesUnionImpl union = new QuantilesDoublesUnionImpl(maxK);
union.maxK_ = maxK;
union.gadget_ = sketch;
return union;
}
/**
* Returns a Heap QuantilesDoublesUnion object that has been initialized with the data from the given
* sketch.
*
* @param sketch A QuantilesDoublesSketch to be used as a source of data only and will not be modified.
* @return a QuantilesDoublesUnion object
*/
static QuantilesDoublesUnionImpl heapifyInstance(final QuantilesDoublesSketch sketch) {
Objects.requireNonNull(sketch);
final int k = sketch.getK();
final QuantilesDoublesUnionImpl union = new QuantilesDoublesUnionImpl(k);
union.maxK_ = k;
union.gadget_ = copyToHeap(sketch);
return union;
}
/**
* Returns a Heap QuantilesDoublesUnion object that has been initialized with the data from the given
* MemorySegment image of a QuantilesDoublesSketch. The srcSeg object will not be modified and a reference to
* it is not retained. The <i>maxK</i> of the resulting union will be that obtained from
* the sketch MemorySegment image.
*
* @param srcSeg an optionally read-only MemorySegment image of a QuantilesDoublesSketch
* @return a QuantilesDoublesUnion object
*/
static QuantilesDoublesUnionImpl heapifyInstance(final MemorySegment srcSeg) {
Objects.requireNonNull(srcSeg);
final HeapUpdateDoublesSketch sketch = HeapUpdateDoublesSketch.heapifyInstance(srcSeg);
final QuantilesDoublesUnionImpl union = new QuantilesDoublesUnionImpl(sketch.getK());
union.gadget_ = sketch;
return union;
}
/**
* Returns an Union object that wraps the data of the given MemorySegment image of a UpdatableQuantilesDoublesSketch.
* The data of the Union will remain in the MemorySegment.
*
* @param srcSeg A MemorySegment image of an updatable QuantilesDoublesSketch to be used as the data structure for the union
* and will be modified.
* @param mSegReq the MemorySegmentRequest used if the given MemorySegment needs to expand.
* Otherwise, it can be null and the default MemorySegmentRequest will be used.
* @return a Union object
*/
static QuantilesDoublesUnionImpl wrapInstance(final MemorySegment srcSeg, final MemorySegmentRequest mSegReq) {
Objects.requireNonNull(srcSeg);
if (srcSeg.isReadOnly()) { throw new SketchesReadOnlyException("Cannot create a Union with a Read Only MemorySegment."); }
final DirectUpdateDoublesSketch sketch = DirectUpdateDoublesSketch.wrapInstance(srcSeg, mSegReq);
final QuantilesDoublesUnionImpl union = new QuantilesDoublesUnionImpl(sketch.getK());
union.gadget_ = sketch;
return union;
}
@Override
public void union(final QuantilesDoublesSketch sketchIn) {
Objects.requireNonNull(sketchIn);
gadget_ = updateLogic(maxK_, gadget_, sketchIn);
gadget_.doublesSV = null;
}
@Override
public void union(final MemorySegment seg) {
Objects.requireNonNull(seg);
if (checkIsMemorySegmentCompact(seg)) {
gadget_ = updateLogic(maxK_, gadget_, QuantilesDoublesSketch.wrap(seg));
} else {
gadget_ = updateLogic(maxK_, gadget_, QuantilesDoublesSketch.writableWrap(seg, null));
}
gadget_.doublesSV = null;
}
@Override
public void update(final double dataItem) {
if (gadget_ == null) {
gadget_ = HeapUpdateDoublesSketch.newInstance(maxK_);
}
gadget_.update(dataItem);
gadget_.doublesSV = null;
}
@Override
public byte[] toByteArray() {
if (gadget_ == null) {
return QuantilesDoublesSketch.builder().setK(maxK_).build().toByteArray();
}
return gadget_.toByteArray();
}
@Override
public UpdatableQuantilesDoublesSketch getResult() {
if (gadget_ == null) {
return HeapUpdateDoublesSketch.newInstance(maxK_);
}
return DoublesUtil.copyToHeap(gadget_);
}
@Override
public UpdatableQuantilesDoublesSketch getResult(final MemorySegment dstSeg, final MemorySegmentRequest mSegReq) {
final long segCapBytes = dstSeg.byteSize();
if (gadget_ == null) {
if (segCapBytes < QuantilesDoublesSketch.getUpdatableStorageBytes(0, 0)) {
throw new SketchesArgumentException("Insufficient capacity for result: " + segCapBytes);
}
return DirectUpdateDoublesSketch.newInstance(maxK_, dstSeg, mSegReq);
}
gadget_.putIntoMemorySegment(dstSeg, false);
return DirectUpdateDoublesSketch.wrapInstance(dstSeg, mSegReq);
}
@Override
public UpdatableQuantilesDoublesSketch getResultAndReset() {
if (gadget_ == null) { return null; } //Intentionally return null here for speed.
final UpdatableQuantilesDoublesSketch ds = gadget_.getSketchAndReset();
gadget_ = null;
return ds;
}
@Override
public void reset() {
gadget_.reset();
}
@Override
public boolean hasMemorySegment() {
return (gadget_ != null) && gadget_.hasMemorySegment();
}
@Override
public boolean isOffHeap() {
return (gadget_ != null) && gadget_.isOffHeap();
}
@Override
public boolean isEmpty() {
return (gadget_ == null) || gadget_.isEmpty();
}
@Override
public boolean isSameResource(final MemorySegment that) {
return (gadget_ == null) ? false : gadget_.isSameResource(that);
}
@Override
public int getMaxK() {
return maxK_;
}
@Override
public int getEffectiveK() {
return (gadget_ != null) ? gadget_.getK() : maxK_;
}
@Override
public String toString() {
return toString(true, false);
}
@Override
public String toString(final boolean sketchSummary, final boolean dataDetail) {
final StringBuilder sb = new StringBuilder();
final String thisSimpleName = this.getClass().getSimpleName();
final int maxK = getMaxK();
final String kStr = String.format("%,d", maxK);
sb.append(LS).append("### Quantiles ").append(thisSimpleName).append(LS);
sb.append(" maxK : ").append(kStr);
if (gadget_ == null) {
sb.append(HeapUpdateDoublesSketch.newInstance(maxK_).toString());
return sb.toString();
}
sb.append(gadget_.toString(sketchSummary, dataDetail));
return sb.toString();
}
//@formatter:off
static UpdatableQuantilesDoublesSketch updateLogic(final int myMaxK, final UpdatableQuantilesDoublesSketch myQS, final QuantilesDoublesSketch other) {
int sw1 = ((myQS == null) ? 0 : myQS.isEmpty() ? 4 : 8);
sw1 |= ((other == null) ? 0 : other.isEmpty() ? 1 : 2);
int outCase = 0; //0=null, 1=NOOP, 2=copy, 3=merge
switch (sw1) {
case 0: outCase = 0; break; //myQS = null, other = null ; return null
case 1: outCase = 4; break; //myQS = null, other = empty; create empty-heap(myMaxK)
case 2: outCase = 2; break; //myQS = null, other = valid; stream or downsample to myMaxK
case 4: outCase = 1; break; //myQS = empty, other = null ; no-op
case 5: outCase = 1; break; //myQS = empty, other = empty; no-op
case 6: outCase = 3; break; //myQS = empty, other = valid; merge
case 8: outCase = 1; break; //myQS = valid, other = null ; no-op
case 9: outCase = 1; break; //myQS = valid, other = empty: no-op
case 10: outCase = 3; break; //myQS = valid, other = valid; merge
default: break; //This cannot happen
}
UpdatableQuantilesDoublesSketch ret = null;
switch (outCase) {
case 0: break; //return null
case 1: ret = myQS; break; //no-op
case 2: { //myQS = null, other = valid; stream or downsample to myMaxK
assert other != null;
if (!other.isEstimationMode()) { //other is exact, stream items in
ret = HeapUpdateDoublesSketch.newInstance(myMaxK);
// exact mode, only need copy base buffer
final DoublesSketchAccessor otherAccessor = DoublesSketchAccessor.wrap(other, false);
for (int i = 0; i < otherAccessor.numItems(); ++i) {
ret.update(otherAccessor.get(i));
}
}
else { //myQS = null, other is est mode
ret = (myMaxK < other.getK())
? other.downSampleInternal(other, myMaxK, null, null) //null seg, null mSegReq
: DoublesUtil.copyToHeap(other); //copy required because caller has handle
}
break;
}
case 3: { //myQS = empty/valid, other = valid; merge
assert other != null;
assert myQS != null;
if (!other.isEstimationMode()) { //other is exact, stream items in
ret = myQS;
// exact mode, only need copy base buffer
final DoublesSketchAccessor otherAccessor = DoublesSketchAccessor.wrap(other, false);
for (int i = 0; i < otherAccessor.numItems(); ++i) {
ret.update(otherAccessor.get(i));
}
} else if (myQS.getK() <= other.getK()) { //I am smaller or equal, thus the target
DoublesMergeImpl.mergeInto(other, myQS);
ret = myQS;
} else if (myQS.isEmpty()) {
if (myQS.hasMemorySegment()) {
final MemorySegment seg = myQS.getMemorySegment(); //myQS is empty, ok to reconfigure
other.putIntoMemorySegment(seg, false); // not compact, but BaseBuf ordered
ret = DirectUpdateDoublesSketch.wrapInstance(seg, null);
} else { //myQS is empty and on heap
ret = DoublesUtil.copyToHeap(other);
}
} else { //Not Empty: myQS has data, downsample to tmp
final UpdatableQuantilesDoublesSketch tmp = QuantilesDoublesSketch.builder().setK(other.getK()).build();
DoublesMergeImpl.downSamplingMergeInto(myQS, tmp); //myData -> tmp
ret = (myQS.hasMemorySegment())
? QuantilesDoublesSketch.builder().setK(other.getK()).build(myQS.getMemorySegment())
: QuantilesDoublesSketch.builder().setK(other.getK()).build();
DoublesMergeImpl.mergeInto(tmp, ret);
DoublesMergeImpl.mergeInto(other, ret);
}
break;
}
case 4: { //myQS = null, other = empty; create empty-heap(myMaxK)
ret = HeapUpdateDoublesSketch.newInstance(myMaxK);
break;
}
default: break; //This cannot happen
}
return ret;
}
//@formatter:on
}