|
|
@@ -30,6 +30,8 @@ import org.elasticsearch.core.Releasable;
|
|
|
import org.elasticsearch.core.ReleasableIterator;
|
|
|
import org.elasticsearch.core.Releasables;
|
|
|
|
|
|
+import java.util.Objects;
|
|
|
+
|
|
|
/**
|
|
|
* An optimized block hash that receives two blocks: tsid and timestamp, which are sorted.
|
|
|
* Since the incoming data is sorted, this block hash appends the incoming data to the internal arrays without lookup.
|
|
|
@@ -39,7 +41,7 @@ public final class TimeSeriesBlockHash extends BlockHash {
|
|
|
private final int tsHashChannel;
|
|
|
private final int timestampIntervalChannel;
|
|
|
|
|
|
- private int lastTsidPosition = 0;
|
|
|
+ private final BytesRef lastTsid = new BytesRef();
|
|
|
private final BytesRefArrayWithSize tsidArray;
|
|
|
|
|
|
private long lastTimestamp;
|
|
|
@@ -62,70 +64,19 @@ public final class TimeSeriesBlockHash extends BlockHash {
|
|
|
Releasables.close(tsidArray, timestampArray, perTsidCountArray);
|
|
|
}
|
|
|
|
|
|
- private OrdinalBytesRefVector getTsidVector(Page page) {
|
|
|
- BytesRefBlock block = page.getBlock(tsHashChannel);
|
|
|
- var ordinalBlock = block.asOrdinals();
|
|
|
- if (ordinalBlock == null) {
|
|
|
- throw new IllegalStateException("expected ordinal block for tsid");
|
|
|
- }
|
|
|
- var ordinalVector = ordinalBlock.asVector();
|
|
|
- if (ordinalVector == null) {
|
|
|
- throw new IllegalStateException("expected ordinal vector for tsid");
|
|
|
- }
|
|
|
- return ordinalVector;
|
|
|
- }
|
|
|
-
|
|
|
- private LongVector getTimestampVector(Page page) {
|
|
|
- final LongBlock timestampsBlock = page.getBlock(timestampIntervalChannel);
|
|
|
- LongVector timestampsVector = timestampsBlock.asVector();
|
|
|
- if (timestampsVector == null) {
|
|
|
- throw new IllegalStateException("expected long vector for timestamp");
|
|
|
- }
|
|
|
- return timestampsVector;
|
|
|
- }
|
|
|
-
|
|
|
@Override
|
|
|
public void add(Page page, GroupingAggregatorFunction.AddInput addInput) {
|
|
|
- final BytesRefVector tsidDict;
|
|
|
- final IntVector tsidOrdinals;
|
|
|
- {
|
|
|
- final var tsidVector = getTsidVector(page);
|
|
|
- tsidDict = tsidVector.getDictionaryVector();
|
|
|
- tsidOrdinals = tsidVector.getOrdinalsVector();
|
|
|
- }
|
|
|
- try (var ordsBuilder = blockFactory.newIntVectorBuilder(tsidOrdinals.getPositionCount())) {
|
|
|
+ final BytesRefBlock tsidBlock = page.getBlock(tsHashChannel);
|
|
|
+ final BytesRefVector tsidVector = Objects.requireNonNull(tsidBlock.asVector(), "tsid input must be a vector");
|
|
|
+ final LongBlock timestampBlock = page.getBlock(timestampIntervalChannel);
|
|
|
+ final LongVector timestampVector = Objects.requireNonNull(timestampBlock.asVector(), "timestamp input must be a vector");
|
|
|
+ try (var ordsBuilder = blockFactory.newIntVectorBuilder(tsidVector.getPositionCount())) {
|
|
|
final BytesRef spare = new BytesRef();
|
|
|
- final BytesRef lastTsid = new BytesRef();
|
|
|
- final LongVector timestampVector = getTimestampVector(page);
|
|
|
- int lastOrd = -1;
|
|
|
- for (int i = 0; i < tsidOrdinals.getPositionCount(); i++) {
|
|
|
- final int newOrd = tsidOrdinals.getInt(i);
|
|
|
- boolean newGroup = false;
|
|
|
- if (lastOrd != newOrd) {
|
|
|
- final var newTsid = tsidDict.getBytesRef(newOrd, spare);
|
|
|
- if (positionCount() == 0) {
|
|
|
- newGroup = true;
|
|
|
- } else if (lastOrd == -1) {
|
|
|
- tsidArray.get(lastTsidPosition, lastTsid);
|
|
|
- newGroup = lastTsid.equals(newTsid) == false;
|
|
|
- } else {
|
|
|
- newGroup = true;
|
|
|
- }
|
|
|
- if (newGroup) {
|
|
|
- endTsidGroup();
|
|
|
- lastTsidPosition = tsidArray.count;
|
|
|
- tsidArray.append(newTsid);
|
|
|
- }
|
|
|
- lastOrd = newOrd;
|
|
|
- }
|
|
|
+ // TODO: optimize incoming ordinal block
|
|
|
+ for (int i = 0; i < tsidVector.getPositionCount(); i++) {
|
|
|
+ final BytesRef tsid = tsidVector.getBytesRef(i, spare);
|
|
|
final long timestamp = timestampVector.getLong(i);
|
|
|
- if (newGroup || timestamp != lastTimestamp) {
|
|
|
- assert newGroup || lastTimestamp >= timestamp : "@timestamp goes backward " + lastTimestamp + " < " + timestamp;
|
|
|
- timestampArray.append(timestamp);
|
|
|
- lastTimestamp = timestamp;
|
|
|
- currentTimestampCount++;
|
|
|
- }
|
|
|
- ordsBuilder.appendInt(timestampArray.count - 1);
|
|
|
+ ordsBuilder.appendInt(addOnePosition(tsid, timestamp));
|
|
|
}
|
|
|
try (var ords = ordsBuilder.build()) {
|
|
|
addInput.add(0, ords);
|
|
|
@@ -133,6 +84,24 @@ public final class TimeSeriesBlockHash extends BlockHash {
|
|
|
}
|
|
|
}
|
|
|
|
|
|
+ private int addOnePosition(BytesRef tsid, long timestamp) {
|
|
|
+ boolean newGroup = false;
|
|
|
+ if (positionCount() == 0 || lastTsid.equals(tsid) == false) {
|
|
|
+ assert positionCount() == 0 || lastTsid.compareTo(tsid) < 0 : "tsid goes backward ";
|
|
|
+ endTsidGroup();
|
|
|
+ tsidArray.append(tsid);
|
|
|
+ tsidArray.get(tsidArray.count - 1, lastTsid);
|
|
|
+ newGroup = true;
|
|
|
+ }
|
|
|
+ if (newGroup || timestamp != lastTimestamp) {
|
|
|
+ assert newGroup || lastTimestamp >= timestamp : "@timestamp goes backward " + lastTimestamp + " < " + timestamp;
|
|
|
+ timestampArray.append(timestamp);
|
|
|
+ lastTimestamp = timestamp;
|
|
|
+ currentTimestampCount++;
|
|
|
+ }
|
|
|
+ return positionCount() - 1;
|
|
|
+ }
|
|
|
+
|
|
|
private void endTsidGroup() {
|
|
|
if (currentTimestampCount > 0) {
|
|
|
perTsidCountArray.append(currentTimestampCount);
|
Nhat Nguyen