You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by GitBox <gi...@apache.org> on 2018/09/02 07:19:00 UTC
[GitHub] sanastas closed pull request #5732: Oak v0.1
sanastas closed pull request #5732: Oak v0.1
URL: https://github.com/apache/incubator-druid/pull/5732
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/processing/pom.xml b/processing/pom.xml
index 220abc4192f..1b45c0632c3 100644
--- a/processing/pom.xml
+++ b/processing/pom.xml
@@ -30,6 +30,11 @@
</parent>
<dependencies>
+ <dependency>
+ <groupId>oak</groupId>
+ <artifactId>oak</artifactId>
+ <version>1.0-SNAPSHOT</version>
+ </dependency>
<dependency>
<groupId>io.druid</groupId>
<artifactId>druid-common</artifactId>
diff --git a/processing/src/main/java/io/druid/segment/incremental/ExternalDataIncrementalIndex.java b/processing/src/main/java/io/druid/segment/incremental/ExternalDataIncrementalIndex.java
new file mode 100644
index 00000000000..93d3ac90f82
--- /dev/null
+++ b/processing/src/main/java/io/druid/segment/incremental/ExternalDataIncrementalIndex.java
@@ -0,0 +1,396 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets 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 io.druid.segment.incremental;
+
+import com.google.common.base.Supplier;
+import com.google.common.collect.Iterators;
+import com.google.common.collect.Maps;
+
+import io.druid.data.input.InputRow;
+import io.druid.data.input.MapBasedRow;
+import io.druid.data.input.Row;
+import io.druid.query.aggregation.AggregatorFactory;
+import io.druid.query.aggregation.PostAggregator;
+import io.druid.segment.DimensionHandler;
+import io.druid.segment.DimensionIndexer;
+
+import java.util.Comparator;
+import java.util.Deque;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ConcurrentNavigableMap;
+import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ */
+public abstract class ExternalDataIncrementalIndex<AggregatorType> extends IncrementalIndex<AggregatorType>
+{
+
+ /* basic constractor */
+ ExternalDataIncrementalIndex(
+ IncrementalIndexSchema incrementalIndexSchema,
+ boolean deserializeComplexMetrics,
+ boolean reportParseExceptions,
+ boolean concurrentEventAdd
+ )
+ {
+ super(incrementalIndexSchema, deserializeComplexMetrics, reportParseExceptions, concurrentEventAdd);
+ }
+
+ protected abstract FactsHolder getFacts();
+
+ // Note: This method needs to be thread safe.
+ protected abstract Integer addToFacts(
+ AggregatorFactory[] metrics,
+ boolean deserializeComplexMetrics,
+ boolean reportParseExceptions,
+ InputRow row,
+ AtomicInteger numEntries,
+ TimeAndDims key,
+ ThreadLocal<InputRow> rowContainer,
+ Supplier<InputRow> rowSupplier,
+ boolean skipMaxRowsInMemoryCheck
+ ) throws IndexSizeExceededException;
+
+ @Override
+ public int add(InputRow row, boolean skipMaxRowsInMemoryCheck) throws IndexSizeExceededException
+ {
+ TimeAndDims key = toTimeAndDims(row);
+ final int rv = addToFacts(
+ metrics,
+ deserializeComplexMetrics,
+ reportParseExceptions,
+ row,
+ numEntries,
+ key,
+ in,
+ rowSupplier,
+ skipMaxRowsInMemoryCheck
+ );
+ updateMaxIngestedTime(row.getTimestamp());
+ return rv;
+ }
+
+ protected long getMinTimeMillis()
+ {
+ return getFacts().getMinTimeMillis();
+ }
+
+ protected long getMaxTimeMillis()
+ {
+ return getFacts().getMaxTimeMillis();
+ }
+
+ public Iterable<io.druid.segment.incremental.IncrementalIndex.TimeAndDims> timeRangeIterable(
+ boolean descending, long timeStart, long timeEnd)
+ {
+ return getFacts().timeRangeIterable(descending, timeStart, timeEnd);
+ }
+
+ public Iterable<io.druid.segment.incremental.IncrementalIndex.TimeAndDims> keySet()
+ {
+ return getFacts().keySet();
+ }
+
+ public Iterable<Row> iterableWithPostAggregations(final List<PostAggregator> postAggs, final boolean descending)
+ {
+ return new Iterable<Row>()
+ {
+ @Override
+ public Iterator<Row> iterator()
+ {
+ final List<DimensionDesc> dimensions = getDimensions();
+
+ return Iterators.transform(
+ getFacts().iterator(descending),
+ timeAndDims -> {
+ final int rowOffset = timeAndDims.getRowIndex();
+
+ Object[] theDims = timeAndDims.getDims();
+
+ Map<String, Object> theVals = Maps.newLinkedHashMap();
+ for (int i = 0; i < theDims.length; ++i) {
+ Object dim = theDims[i];
+ DimensionDesc dimensionDesc = dimensions.get(i);
+ if (dimensionDesc == null) {
+ continue;
+ }
+ String dimensionName = dimensionDesc.getName();
+ DimensionHandler handler = dimensionDesc.getHandler();
+ if (dim == null || handler.getLengthOfEncodedKeyComponent(dim) == 0) {
+ theVals.put(dimensionName, null);
+ continue;
+ }
+ final DimensionIndexer indexer = dimensionDesc.getIndexer();
+ Object rowVals = indexer.convertUnsortedEncodedKeyComponentToActualArrayOrList(dim, DimensionIndexer.LIST);
+ theVals.put(dimensionName, rowVals);
+ }
+
+ AggregatorType[] aggs = getAggsForRow(timeAndDims);
+ for (int i = 0; i < aggs.length; ++i) {
+ theVals.put(metrics[i].getName(), getAggVal(aggs[i], timeAndDims, i));
+ }
+
+ if (postAggs != null) {
+ for (PostAggregator postAgg : postAggs) {
+ theVals.put(postAgg.getName(), postAgg.compute(theVals));
+ }
+ }
+
+ return new MapBasedRow(timeAndDims.getTimestamp(), theVals);
+ }
+ );
+ }
+ };
+ }
+
+
+ interface FactsHolder
+ {
+ /**
+ * @return the previous rowIndex associated with the specified key, or
+ * {@code TimeAndDims#EMPTY_ROW_INDEX} if there was no mapping for the key.
+ */
+ int getPriorIndex(io.druid.segment.incremental.IncrementalIndex.TimeAndDims key);
+
+ long getMinTimeMillis();
+
+ long getMaxTimeMillis();
+
+ Iterator<io.druid.segment.incremental.IncrementalIndex.TimeAndDims> iterator(boolean descending);
+
+ Iterable<io.druid.segment.incremental.IncrementalIndex.TimeAndDims> timeRangeIterable(boolean descending, long timeStart, long timeEnd);
+
+ Iterable<TimeAndDims> keySet();
+
+ /**
+ * @return the previous rowIndex associated with the specified key, or
+ * {@code TimeAndDims#EMPTY_ROW_INDEX} if there was no mapping for the key.
+ */
+ int putIfAbsent(io.druid.segment.incremental.IncrementalIndex.TimeAndDims key, int rowIndex);
+
+ void clear();
+ }
+
+ static class RollupFactsHolder implements FactsHolder
+ {
+ private final boolean sortFacts;
+ // Can't use Set because we need to be able to get from collection
+ private final ConcurrentMap<io.druid.segment.incremental.IncrementalIndex.TimeAndDims, io.druid.segment.incremental.IncrementalIndex.TimeAndDims>
+ facts;
+ private final List<io.druid.segment.incremental.IncrementalIndex.DimensionDesc> dimensionDescsList;
+
+ public RollupFactsHolder(boolean sortFacts, Comparator<io.druid.segment.incremental.IncrementalIndex.TimeAndDims> timeAndDimsComparator, List<io.druid.segment.incremental.IncrementalIndex.DimensionDesc> dimensionDescsList)
+ {
+ this.sortFacts = sortFacts;
+ if (sortFacts) {
+ this.facts = new ConcurrentSkipListMap<>(timeAndDimsComparator);
+ } else {
+ this.facts = new ConcurrentHashMap<>();
+ }
+ this.dimensionDescsList = dimensionDescsList;
+ }
+
+ @Override
+ public int getPriorIndex(io.druid.segment.incremental.IncrementalIndex.TimeAndDims key)
+ {
+ io.druid.segment.incremental.IncrementalIndex.TimeAndDims timeAndDims = facts.get(key);
+ return timeAndDims == null ?
+ io.druid.segment.incremental.IncrementalIndex.TimeAndDims.EMPTY_ROW_INDEX :
+ timeAndDims.getRowIndex();
+ }
+
+ @Override
+ public long getMinTimeMillis()
+ {
+ if (sortFacts) {
+ return ((ConcurrentNavigableMap<io.druid.segment.incremental.IncrementalIndex.TimeAndDims, io.druid.segment.incremental.IncrementalIndex.TimeAndDims>) facts).firstKey().getTimestamp();
+ } else {
+ throw new UnsupportedOperationException("can't get minTime from unsorted facts data.");
+ }
+ }
+
+ @Override
+ public long getMaxTimeMillis()
+ {
+ if (sortFacts) {
+ return ((ConcurrentNavigableMap<io.druid.segment.incremental.IncrementalIndex.TimeAndDims, io.druid.segment.incremental.IncrementalIndex.TimeAndDims>) facts).lastKey().getTimestamp();
+ } else {
+ throw new UnsupportedOperationException("can't get maxTime from unsorted facts data.");
+ }
+ }
+
+ @Override
+ public Iterator<io.druid.segment.incremental.IncrementalIndex.TimeAndDims> iterator(boolean descending)
+ {
+ if (descending && sortFacts) {
+ return ((ConcurrentNavigableMap<io.druid.segment.incremental.IncrementalIndex.TimeAndDims, io.druid.segment.incremental.IncrementalIndex.TimeAndDims>) facts).descendingMap().keySet().iterator();
+ }
+ return keySet().iterator();
+ }
+
+ @Override
+ public Iterable<io.druid.segment.incremental.IncrementalIndex.TimeAndDims> timeRangeIterable(boolean descending, long timeStart, long timeEnd)
+ {
+ if (!sortFacts) {
+ throw new UnsupportedOperationException("can't get timeRange from unsorted facts data.");
+ }
+ io.druid.segment.incremental.IncrementalIndex.TimeAndDims
+ start = new io.druid.segment.incremental.IncrementalIndex.TimeAndDims(timeStart, new Object[]{}, dimensionDescsList);
+ io.druid.segment.incremental.IncrementalIndex.TimeAndDims
+ end = new io.druid.segment.incremental.IncrementalIndex.TimeAndDims(timeEnd, new Object[]{}, dimensionDescsList);
+ ConcurrentNavigableMap<io.druid.segment.incremental.IncrementalIndex.TimeAndDims, io.druid.segment.incremental.IncrementalIndex.TimeAndDims> subMap =
+ ((ConcurrentNavigableMap<io.druid.segment.incremental.IncrementalIndex.TimeAndDims, io.druid.segment.incremental.IncrementalIndex.TimeAndDims>) facts).subMap(start, end);
+ final Map<io.druid.segment.incremental.IncrementalIndex.TimeAndDims, io.druid.segment.incremental.IncrementalIndex.TimeAndDims> rangeMap = descending ? subMap.descendingMap() : subMap;
+ return rangeMap.keySet();
+ }
+
+ @Override
+ public Iterable<io.druid.segment.incremental.IncrementalIndex.TimeAndDims> keySet()
+ {
+ return facts.keySet();
+ }
+
+ @Override
+ public int putIfAbsent(io.druid.segment.incremental.IncrementalIndex.TimeAndDims key, int rowIndex)
+ {
+ // setRowIndex() must be called before facts.putIfAbsent() for visibility of rowIndex from concurrent readers.
+ key.setRowIndex(rowIndex);
+ io.druid.segment.incremental.IncrementalIndex.TimeAndDims prev = facts.putIfAbsent(key, key);
+ return prev == null ? io.druid.segment.incremental.IncrementalIndex.TimeAndDims.EMPTY_ROW_INDEX : prev.getRowIndex();
+ }
+
+ @Override
+ public void clear()
+ {
+ facts.clear();
+ }
+ }
+
+ static class PlainFactsHolder implements FactsHolder
+ {
+ private final boolean sortFacts;
+ private final ConcurrentMap<Long, Deque<io.druid.segment.incremental.IncrementalIndex.TimeAndDims>> facts;
+
+ public PlainFactsHolder(boolean sortFacts)
+ {
+ this.sortFacts = sortFacts;
+ if (sortFacts) {
+ this.facts = new ConcurrentSkipListMap<>();
+ } else {
+ this.facts = new ConcurrentHashMap<>();
+ }
+ }
+
+ @Override
+ public int getPriorIndex(io.druid.segment.incremental.IncrementalIndex.TimeAndDims key)
+ {
+ // always return EMPTY_ROW_INDEX to indicate that no prior key cause we always add new row
+ return io.druid.segment.incremental.IncrementalIndex.TimeAndDims.EMPTY_ROW_INDEX;
+ }
+
+ @Override
+ public long getMinTimeMillis()
+ {
+ if (sortFacts) {
+ return ((ConcurrentNavigableMap<Long, Deque<io.druid.segment.incremental.IncrementalIndex.TimeAndDims>>) facts).firstKey();
+ } else {
+ throw new UnsupportedOperationException("can't get minTime from unsorted facts data.");
+ }
+ }
+
+ @Override
+ public long getMaxTimeMillis()
+ {
+ if (sortFacts) {
+ return ((ConcurrentNavigableMap<Long, Deque<io.druid.segment.incremental.IncrementalIndex.TimeAndDims>>) facts).lastKey();
+ } else {
+ throw new UnsupportedOperationException("can't get maxTime from unsorted facts data.");
+ }
+ }
+
+ @Override
+ public Iterator<io.druid.segment.incremental.IncrementalIndex.TimeAndDims> iterator(boolean descending)
+ {
+ if (descending && sortFacts) {
+ return concat(((ConcurrentNavigableMap<Long, Deque<io.druid.segment.incremental.IncrementalIndex.TimeAndDims>>) facts)
+ .descendingMap().values(), true).iterator();
+ }
+ return concat(facts.values(), false).iterator();
+ }
+
+ @Override
+ public Iterable<io.druid.segment.incremental.IncrementalIndex.TimeAndDims> timeRangeIterable(boolean descending, long timeStart, long timeEnd)
+ {
+ ConcurrentNavigableMap<Long, Deque<io.druid.segment.incremental.IncrementalIndex.TimeAndDims>> subMap =
+ ((ConcurrentNavigableMap<Long, Deque<io.druid.segment.incremental.IncrementalIndex.TimeAndDims>>) facts).subMap(timeStart, timeEnd);
+ final Map<Long, Deque<io.druid.segment.incremental.IncrementalIndex.TimeAndDims>> rangeMap = descending ? subMap.descendingMap() : subMap;
+ return concat(rangeMap.values(), descending);
+ }
+
+ private Iterable<io.druid.segment.incremental.IncrementalIndex.TimeAndDims> concat(
+ final Iterable<Deque<io.druid.segment.incremental.IncrementalIndex.TimeAndDims>> iterable,
+ final boolean descending
+ )
+ {
+ return () -> Iterators.concat(
+ Iterators.transform(
+ iterable.iterator(),
+ input -> descending ? input.descendingIterator() : input.iterator()
+ )
+ );
+ }
+
+ @Override
+ public Iterable<io.druid.segment.incremental.IncrementalIndex.TimeAndDims> keySet()
+ {
+ return concat(facts.values(), false);
+ }
+
+ @Override
+ public int putIfAbsent(io.druid.segment.incremental.IncrementalIndex.TimeAndDims key, int rowIndex)
+ {
+ Long time = key.getTimestamp();
+ Deque<io.druid.segment.incremental.IncrementalIndex.TimeAndDims> rows = facts.get(time);
+ if (rows == null) {
+ facts.putIfAbsent(time, new ConcurrentLinkedDeque<>());
+ // in race condition, rows may be put by other thread, so always get latest status from facts
+ rows = facts.get(time);
+ }
+ // setRowIndex() must be called before rows.add() for visibility of rowIndex from concurrent readers.
+ key.setRowIndex(rowIndex);
+ rows.add(key);
+ // always return EMPTY_ROW_INDEX to indicate that we always add new row
+ return io.druid.segment.incremental.IncrementalIndex.TimeAndDims.EMPTY_ROW_INDEX;
+ }
+
+ @Override
+ public void clear()
+ {
+ facts.clear();
+ }
+ }
+
+}
diff --git a/processing/src/main/java/io/druid/segment/incremental/IncrementalIndex.java b/processing/src/main/java/io/druid/segment/incremental/IncrementalIndex.java
index 332781b9bee..9932e30f6b4 100644
--- a/processing/src/main/java/io/druid/segment/incremental/IncrementalIndex.java
+++ b/processing/src/main/java/io/druid/segment/incremental/IncrementalIndex.java
@@ -25,7 +25,6 @@
import com.google.common.base.Supplier;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.Iterators;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.primitives.Ints;
@@ -34,7 +33,6 @@
import io.druid.common.config.NullHandling;
import io.druid.common.guava.GuavaUtils;
import io.druid.data.input.InputRow;
-import io.druid.data.input.MapBasedRow;
import io.druid.data.input.Row;
import io.druid.data.input.impl.DimensionSchema;
import io.druid.data.input.impl.DimensionsSpec;
@@ -83,16 +81,10 @@
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
-import java.util.Deque;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentLinkedDeque;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.ConcurrentNavigableMap;
-import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicInteger;
@@ -100,7 +92,7 @@
*/
public abstract class IncrementalIndex<AggregatorType> extends AbstractIndex implements Iterable<Row>, Closeable
{
- private volatile DateTime maxIngestedEventTime;
+ protected volatile DateTime maxIngestedEventTime;
// Used to discover ValueType based on the class of values in a row
// Also used to convert between the duplicate ValueType enums in DimensionSchema (druid-api) and main druid.
@@ -222,27 +214,27 @@ public ColumnCapabilities getColumnCapabilities(String columnName)
return virtualColumns.wrap(new IncrementalIndexInputRowColumnSelectorFactory());
}
- private final long minTimestamp;
+ protected final long minTimestamp;
private final Granularity gran;
private final boolean rollup;
private final List<Function<InputRow, InputRow>> rowTransformers;
private final VirtualColumns virtualColumns;
- private final AggregatorFactory[] metrics;
+ protected final AggregatorFactory[] metrics;
private final AggregatorType[] aggs;
- private final boolean deserializeComplexMetrics;
- private final boolean reportParseExceptions;
+ protected final boolean deserializeComplexMetrics;
+ protected final boolean reportParseExceptions;
private final Metadata metadata;
private final Map<String, MetricDesc> metricDescs;
private final Map<String, DimensionDesc> dimensionDescs;
- private final List<DimensionDesc> dimensionDescsList;
+ protected final List<DimensionDesc> dimensionDescsList;
private final Map<String, ColumnCapabilitiesImpl> columnCapabilities;
- private final AtomicInteger numEntries = new AtomicInteger();
+ protected final AtomicInteger numEntries = new AtomicInteger();
// This is modified on add() in a critical section.
- private final ThreadLocal<InputRow> in = new ThreadLocal<>();
- private final Supplier<InputRow> rowSupplier = in::get;
+ protected final ThreadLocal<InputRow> in = new ThreadLocal<>();
+ protected final Supplier<InputRow> rowSupplier = in::get;
/**
* Setting deserializeComplexMetrics to false is necessary for intermediate aggregation such as groupBy that
@@ -256,7 +248,7 @@ public ColumnCapabilities getColumnCapabilities(String columnName)
* value for aggregators that return metrics other than float.
* @param reportParseExceptions flag whether or not to report ParseExceptions that occur while extracting values
* from input rows
- * @param concurrentEventAdd flag whether ot not adding of input rows should be thread-safe
+ * @param concurrentEventAdd flag whether or not adding of input rows should be thread-safe
*/
protected IncrementalIndex(
final IncrementalIndexSchema incrementalIndexSchema,
@@ -430,6 +422,21 @@ public IncrementalIndex buildOffheap(final NonBlockingPool<ByteBuffer> bufferPoo
Objects.requireNonNull(bufferPool, "bufferPool is null")
);
}
+
+ public IncrementalIndex buildOffheapOak()
+ {
+ if (maxRowCount <= 0) {
+ throw new IllegalArgumentException("Invalid max row count: " + maxRowCount);
+ }
+
+ return new OffheapOakIncrementalIndex(
+ Objects.requireNonNull(incrementalIndexSchema, "incrementalIndexSchema is null"),
+ deserializeComplexMetrics,
+ reportParseExceptions,
+ concurrentEventAdd,
+ maxRowCount
+ );
+ }
}
public boolean isRollup()
@@ -437,8 +444,6 @@ public boolean isRollup()
return rollup;
}
- public abstract FactsHolder getFacts();
-
public abstract boolean canAppendRow();
public abstract String getOutOfRowsReason();
@@ -450,35 +455,25 @@ public boolean isRollup()
boolean concurrentEventAdd
);
- // Note: This method needs to be thread safe.
- protected abstract Integer addToFacts(
- AggregatorFactory[] metrics,
- boolean deserializeComplexMetrics,
- boolean reportParseExceptions,
- InputRow row,
- AtomicInteger numEntries,
- TimeAndDims key,
- ThreadLocal<InputRow> rowContainer,
- Supplier<InputRow> rowSupplier,
- boolean skipMaxRowsInMemoryCheck
- ) throws IndexSizeExceededException;
-
public abstract int getLastRowIndex();
- protected abstract AggregatorType[] getAggsForRow(int rowOffset);
+ protected abstract AggregatorType[] getAggsForRow(TimeAndDims timeAndDims);
- protected abstract Object getAggVal(AggregatorType agg, int rowOffset, int aggPosition);
+ protected abstract Object getAggVal(AggregatorType agg, TimeAndDims timeAndDims, int aggPosition);
- protected abstract float getMetricFloatValue(int rowOffset, int aggOffset);
+ protected abstract float getMetricFloatValue(TimeAndDims timeAndDims, int aggOffset);
- protected abstract long getMetricLongValue(int rowOffset, int aggOffset);
+ protected abstract long getMetricLongValue(TimeAndDims timeAndDims, int aggOffset);
- protected abstract Object getMetricObjectValue(int rowOffset, int aggOffset);
+ protected abstract Object getMetricObjectValue(TimeAndDims timeAndDims, int aggOffset);
- protected abstract double getMetricDoubleValue(int rowOffset, int aggOffset);
+ protected abstract double getMetricDoubleValue(TimeAndDims timeAndDims, int aggOffset);
- protected abstract boolean isNull(int rowOffset, int aggOffset);
+ protected abstract boolean isNull(TimeAndDims timeAndDims, int aggOffset);
+ public abstract Iterable<TimeAndDims> timeRangeIterable(boolean descending, long timeStart, long timeEnd);
+
+ public abstract Iterable<TimeAndDims> keySet();
@Override
public void close()
@@ -519,23 +514,7 @@ public int add(InputRow row) throws IndexSizeExceededException
return add(row, false);
}
- public int add(InputRow row, boolean skipMaxRowsInMemoryCheck) throws IndexSizeExceededException
- {
- TimeAndDims key = toTimeAndDims(row);
- final int rv = addToFacts(
- metrics,
- deserializeComplexMetrics,
- reportParseExceptions,
- row,
- numEntries,
- key,
- in,
- rowSupplier,
- skipMaxRowsInMemoryCheck
- );
- updateMaxIngestedTime(row.getTimestamp());
- return rv;
- }
+ public abstract int add(InputRow row, boolean skipMaxRowsInMemoryCheck) throws IndexSizeExceededException;
@VisibleForTesting
TimeAndDims toTimeAndDims(InputRow row)
@@ -625,7 +604,7 @@ TimeAndDims toTimeAndDims(InputRow row)
return new TimeAndDims(Math.max(truncated, minTimestamp), dims, dimensionDescsList);
}
- private synchronized void updateMaxIngestedTime(DateTime eventTime)
+ protected synchronized void updateMaxIngestedTime(DateTime eventTime)
{
if (maxIngestedEventTime == null || maxIngestedEventTime.isBefore(eventTime)) {
maxIngestedEventTime = eventTime;
@@ -642,15 +621,9 @@ public int size()
return numEntries.get();
}
- private long getMinTimeMillis()
- {
- return getFacts().getMinTimeMillis();
- }
+ protected abstract long getMinTimeMillis();
- private long getMaxTimeMillis()
- {
- return getFacts().getMaxTimeMillis();
- }
+ protected abstract long getMaxTimeMillis();
public AggregatorType[] getAggs()
{
@@ -837,57 +810,7 @@ public Metadata getMetadata()
return iterableWithPostAggregations(null, false).iterator();
}
- public Iterable<Row> iterableWithPostAggregations(final List<PostAggregator> postAggs, final boolean descending)
- {
- return new Iterable<Row>()
- {
- @Override
- public Iterator<Row> iterator()
- {
- final List<DimensionDesc> dimensions = getDimensions();
-
- return Iterators.transform(
- getFacts().iterator(descending),
- timeAndDims -> {
- final int rowOffset = timeAndDims.getRowIndex();
-
- Object[] theDims = timeAndDims.getDims();
-
- Map<String, Object> theVals = Maps.newLinkedHashMap();
- for (int i = 0; i < theDims.length; ++i) {
- Object dim = theDims[i];
- DimensionDesc dimensionDesc = dimensions.get(i);
- if (dimensionDesc == null) {
- continue;
- }
- String dimensionName = dimensionDesc.getName();
- DimensionHandler handler = dimensionDesc.getHandler();
- if (dim == null || handler.getLengthOfEncodedKeyComponent(dim) == 0) {
- theVals.put(dimensionName, null);
- continue;
- }
- final DimensionIndexer indexer = dimensionDesc.getIndexer();
- Object rowVals = indexer.convertUnsortedEncodedKeyComponentToActualArrayOrList(dim, DimensionIndexer.LIST);
- theVals.put(dimensionName, rowVals);
- }
-
- AggregatorType[] aggs = getAggsForRow(rowOffset);
- for (int i = 0; i < aggs.length; ++i) {
- theVals.put(metrics[i].getName(), getAggVal(aggs[i], rowOffset, i));
- }
-
- if (postAggs != null) {
- for (PostAggregator postAgg : postAggs) {
- theVals.put(postAgg.getName(), postAgg.compute(theVals));
- }
- }
-
- return new MapBasedRow(timeAndDims.getTimestamp(), theVals);
- }
- );
- }
- };
- }
+ public abstract Iterable<Row> iterableWithPostAggregations(List<PostAggregator> postAggs, boolean descending);
public DateTime getMaxIngestedEventTime()
{
@@ -999,7 +922,7 @@ public ColumnCapabilitiesImpl getCapabilities()
* rowIndex is not checked in {@link #equals} and {@link #hashCode} on purpose. TimeAndDims acts as a Map key
* and "entry" object (rowIndex is the "value") at the same time. This is done to reduce object indirection and
* improve locality, and avoid boxing of rowIndex as Integer, when stored in JDK collection:
- * {@link RollupFactsHolder} needs concurrent collections, that are not present in fastutil.
+ * needs concurrent collections, that are not present in fastutil.
*/
private int rowIndex;
@@ -1040,7 +963,7 @@ public int getRowIndex()
return rowIndex;
}
- private void setRowIndex(int rowIndex)
+ public void setRowIndex(int rowIndex)
{
this.rowIndex = rowIndex;
}
@@ -1179,227 +1102,6 @@ private static boolean allNull(Object[] dims, int startPosition)
return true;
}
- interface FactsHolder
- {
- /**
- * @return the previous rowIndex associated with the specified key, or
- * {@code TimeAndDims#EMPTY_ROW_INDEX} if there was no mapping for the key.
- */
- int getPriorIndex(TimeAndDims key);
-
- long getMinTimeMillis();
-
- long getMaxTimeMillis();
-
- Iterator<TimeAndDims> iterator(boolean descending);
-
- Iterable<TimeAndDims> timeRangeIterable(boolean descending, long timeStart, long timeEnd);
-
- Iterable<TimeAndDims> keySet();
-
- /**
- * @return the previous rowIndex associated with the specified key, or
- * {@code TimeAndDims#EMPTY_ROW_INDEX} if there was no mapping for the key.
- */
- int putIfAbsent(TimeAndDims key, int rowIndex);
-
- void clear();
- }
-
- static class RollupFactsHolder implements FactsHolder
- {
- private final boolean sortFacts;
- // Can't use Set because we need to be able to get from collection
- private final ConcurrentMap<TimeAndDims, TimeAndDims> facts;
- private final List<DimensionDesc> dimensionDescsList;
-
- public RollupFactsHolder(boolean sortFacts, Comparator<TimeAndDims> timeAndDimsComparator, List<DimensionDesc> dimensionDescsList)
- {
- this.sortFacts = sortFacts;
- if (sortFacts) {
- this.facts = new ConcurrentSkipListMap<>(timeAndDimsComparator);
- } else {
- this.facts = new ConcurrentHashMap<>();
- }
- this.dimensionDescsList = dimensionDescsList;
- }
-
- @Override
- public int getPriorIndex(TimeAndDims key)
- {
- TimeAndDims timeAndDims = facts.get(key);
- return timeAndDims == null ? TimeAndDims.EMPTY_ROW_INDEX : timeAndDims.rowIndex;
- }
-
- @Override
- public long getMinTimeMillis()
- {
- if (sortFacts) {
- return ((ConcurrentNavigableMap<TimeAndDims, TimeAndDims>) facts).firstKey().getTimestamp();
- } else {
- throw new UnsupportedOperationException("can't get minTime from unsorted facts data.");
- }
- }
-
- @Override
- public long getMaxTimeMillis()
- {
- if (sortFacts) {
- return ((ConcurrentNavigableMap<TimeAndDims, TimeAndDims>) facts).lastKey().getTimestamp();
- } else {
- throw new UnsupportedOperationException("can't get maxTime from unsorted facts data.");
- }
- }
-
- @Override
- public Iterator<TimeAndDims> iterator(boolean descending)
- {
- if (descending && sortFacts) {
- return ((ConcurrentNavigableMap<TimeAndDims, TimeAndDims>) facts).descendingMap().keySet().iterator();
- }
- return keySet().iterator();
- }
-
- @Override
- public Iterable<TimeAndDims> timeRangeIterable(boolean descending, long timeStart, long timeEnd)
- {
- if (!sortFacts) {
- throw new UnsupportedOperationException("can't get timeRange from unsorted facts data.");
- }
- TimeAndDims start = new TimeAndDims(timeStart, new Object[]{}, dimensionDescsList);
- TimeAndDims end = new TimeAndDims(timeEnd, new Object[]{}, dimensionDescsList);
- ConcurrentNavigableMap<TimeAndDims, TimeAndDims> subMap =
- ((ConcurrentNavigableMap<TimeAndDims, TimeAndDims>) facts).subMap(start, end);
- final Map<TimeAndDims, TimeAndDims> rangeMap = descending ? subMap.descendingMap() : subMap;
- return rangeMap.keySet();
- }
-
- @Override
- public Iterable<TimeAndDims> keySet()
- {
- return facts.keySet();
- }
-
- @Override
- public int putIfAbsent(TimeAndDims key, int rowIndex)
- {
- // setRowIndex() must be called before facts.putIfAbsent() for visibility of rowIndex from concurrent readers.
- key.setRowIndex(rowIndex);
- TimeAndDims prev = facts.putIfAbsent(key, key);
- return prev == null ? TimeAndDims.EMPTY_ROW_INDEX : prev.rowIndex;
- }
-
- @Override
- public void clear()
- {
- facts.clear();
- }
- }
-
- static class PlainFactsHolder implements FactsHolder
- {
- private final boolean sortFacts;
- private final ConcurrentMap<Long, Deque<TimeAndDims>> facts;
-
- public PlainFactsHolder(boolean sortFacts)
- {
- this.sortFacts = sortFacts;
- if (sortFacts) {
- this.facts = new ConcurrentSkipListMap<>();
- } else {
- this.facts = new ConcurrentHashMap<>();
- }
- }
-
- @Override
- public int getPriorIndex(TimeAndDims key)
- {
- // always return EMPTY_ROW_INDEX to indicate that no prior key cause we always add new row
- return TimeAndDims.EMPTY_ROW_INDEX;
- }
-
- @Override
- public long getMinTimeMillis()
- {
- if (sortFacts) {
- return ((ConcurrentNavigableMap<Long, Deque<TimeAndDims>>) facts).firstKey();
- } else {
- throw new UnsupportedOperationException("can't get minTime from unsorted facts data.");
- }
- }
-
- @Override
- public long getMaxTimeMillis()
- {
- if (sortFacts) {
- return ((ConcurrentNavigableMap<Long, Deque<TimeAndDims>>) facts).lastKey();
- } else {
- throw new UnsupportedOperationException("can't get maxTime from unsorted facts data.");
- }
- }
-
- @Override
- public Iterator<TimeAndDims> iterator(boolean descending)
- {
- if (descending && sortFacts) {
- return concat(((ConcurrentNavigableMap<Long, Deque<TimeAndDims>>) facts)
- .descendingMap().values(), true).iterator();
- }
- return concat(facts.values(), false).iterator();
- }
-
- @Override
- public Iterable<TimeAndDims> timeRangeIterable(boolean descending, long timeStart, long timeEnd)
- {
- ConcurrentNavigableMap<Long, Deque<TimeAndDims>> subMap =
- ((ConcurrentNavigableMap<Long, Deque<TimeAndDims>>) facts).subMap(timeStart, timeEnd);
- final Map<Long, Deque<TimeAndDims>> rangeMap = descending ? subMap.descendingMap() : subMap;
- return concat(rangeMap.values(), descending);
- }
-
- private Iterable<TimeAndDims> concat(
- final Iterable<Deque<TimeAndDims>> iterable,
- final boolean descending
- )
- {
- return () -> Iterators.concat(
- Iterators.transform(
- iterable.iterator(),
- input -> descending ? input.descendingIterator() : input.iterator()
- )
- );
- }
-
- @Override
- public Iterable<TimeAndDims> keySet()
- {
- return concat(facts.values(), false);
- }
-
- @Override
- public int putIfAbsent(TimeAndDims key, int rowIndex)
- {
- Long time = key.getTimestamp();
- Deque<TimeAndDims> rows = facts.get(time);
- if (rows == null) {
- facts.putIfAbsent(time, new ConcurrentLinkedDeque<>());
- // in race condition, rows may be put by other thread, so always get latest status from facts
- rows = facts.get(time);
- }
- // setRowIndex() must be called before rows.add() for visibility of rowIndex from concurrent readers.
- key.setRowIndex(rowIndex);
- rows.add(key);
- // always return EMPTY_ROW_INDEX to indicate that we always add new row
- return TimeAndDims.EMPTY_ROW_INDEX;
- }
-
- @Override
- public void clear()
- {
- facts.clear();
- }
- }
-
private class LongMetricColumnSelector implements LongColumnSelector
{
private final TimeAndDimsHolder currEntry;
@@ -1415,7 +1117,7 @@ public LongMetricColumnSelector(TimeAndDimsHolder currEntry, int metricIndex)
public long getLong()
{
assert NullHandling.replaceWithDefault() || !isNull();
- return getMetricLongValue(currEntry.getValue(), metricIndex);
+ return getMetricLongValue(currEntry.get(), metricIndex);
}
@Override
@@ -1427,7 +1129,7 @@ public void inspectRuntimeShape(RuntimeShapeInspector inspector)
@Override
public boolean isNull()
{
- return IncrementalIndex.this.isNull(currEntry.getValue(), metricIndex);
+ return IncrementalIndex.this.isNull(currEntry.get(), metricIndex);
}
}
@@ -1452,7 +1154,7 @@ public ObjectMetricColumnSelector(
@Override
public Object getObject()
{
- return getMetricObjectValue(currEntry.getValue(), metricIndex);
+ return getMetricObjectValue(currEntry.get(), metricIndex);
}
@Override
@@ -1483,7 +1185,7 @@ public FloatMetricColumnSelector(TimeAndDimsHolder currEntry, int metricIndex)
public float getFloat()
{
assert NullHandling.replaceWithDefault() || !isNull();
- return getMetricFloatValue(currEntry.getValue(), metricIndex);
+ return getMetricFloatValue(currEntry.get(), metricIndex);
}
@Override
@@ -1495,7 +1197,7 @@ public void inspectRuntimeShape(RuntimeShapeInspector inspector)
@Override
public boolean isNull()
{
- return IncrementalIndex.this.isNull(currEntry.getValue(), metricIndex);
+ return IncrementalIndex.this.isNull(currEntry.get(), metricIndex);
}
}
@@ -1514,13 +1216,13 @@ public DoubleMetricColumnSelector(TimeAndDimsHolder currEntry, int metricIndex)
public double getDouble()
{
assert NullHandling.replaceWithDefault() || !isNull();
- return getMetricDoubleValue(currEntry.getValue(), metricIndex);
+ return getMetricDoubleValue(currEntry.get(), metricIndex);
}
@Override
public boolean isNull()
{
- return IncrementalIndex.this.isNull(currEntry.getValue(), metricIndex);
+ return IncrementalIndex.this.isNull(currEntry.get(), metricIndex);
}
@Override
diff --git a/processing/src/main/java/io/druid/segment/incremental/IncrementalIndexAdapter.java b/processing/src/main/java/io/druid/segment/incremental/IncrementalIndexAdapter.java
index 8eb148d6bb4..bea8d3f9827 100644
--- a/processing/src/main/java/io/druid/segment/incremental/IncrementalIndexAdapter.java
+++ b/processing/src/main/java/io/druid/segment/incremental/IncrementalIndexAdapter.java
@@ -91,7 +91,7 @@ public IncrementalIndexAdapter(
}
int rowNum = 0;
- for (IncrementalIndex.TimeAndDims timeAndDims : index.getFacts().keySet()) {
+ for (IncrementalIndex.TimeAndDims timeAndDims : index.keySet()) {
final Object[] dims = timeAndDims.getDims();
for (IncrementalIndex.DimensionDesc dimension : dimensions) {
@@ -174,7 +174,7 @@ public int getNumRows()
* iterator() call to ensure the counter starts at 0.
*/
return Iterators.transform(
- index.getFacts().keySet().iterator(),
+ index.keySet().iterator(),
new Function<IncrementalIndex.TimeAndDims, Rowboat>()
{
int count = 0;
@@ -200,7 +200,7 @@ public Rowboat apply(IncrementalIndex.TimeAndDims timeAndDims)
Object[] metrics = new Object[index.getMetricAggs().length];
for (int i = 0; i < metrics.length; i++) {
- metrics[i] = index.getMetricObjectValue(rowOffset, i);
+ metrics[i] = index.getMetricObjectValue(timeAndDims, i);
}
return new Rowboat(
diff --git a/processing/src/main/java/io/druid/segment/incremental/IncrementalIndexStorageAdapter.java b/processing/src/main/java/io/druid/segment/incremental/IncrementalIndexStorageAdapter.java
index 2f9b21ed486..d3ed7e3aec2 100644
--- a/processing/src/main/java/io/druid/segment/incremental/IncrementalIndexStorageAdapter.java
+++ b/processing/src/main/java/io/druid/segment/incremental/IncrementalIndexStorageAdapter.java
@@ -240,7 +240,7 @@ public Metadata getMetadata()
numAdvanced = -1;
maxRowIndex = index.getLastRowIndex();
final long timeStart = Math.max(interval.getStartMillis(), actualInterval.getStartMillis());
- cursorIterable = index.getFacts().timeRangeIterable(
+ cursorIterable = index.timeRangeIterable(
descending,
timeStart,
Math.min(actualInterval.getEndMillis(), gran.increment(interval.getStart()).getMillis())
diff --git a/processing/src/main/java/io/druid/segment/incremental/InternalDataIncrementalIndex.java b/processing/src/main/java/io/druid/segment/incremental/InternalDataIncrementalIndex.java
new file mode 100644
index 00000000000..4c59f5f8186
--- /dev/null
+++ b/processing/src/main/java/io/druid/segment/incremental/InternalDataIncrementalIndex.java
@@ -0,0 +1,36 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets 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 io.druid.segment.incremental;
+
+
+public abstract class InternalDataIncrementalIndex<AggregatorType> extends IncrementalIndex<AggregatorType>
+{
+
+ protected InternalDataIncrementalIndex(
+ IncrementalIndexSchema incrementalIndexSchema,
+ boolean deserializeComplexMetrics,
+ boolean reportParseExceptions,
+ boolean concurrentEventAdd
+ )
+ {
+ super(incrementalIndexSchema, deserializeComplexMetrics, reportParseExceptions, concurrentEventAdd);
+ }
+
+}
diff --git a/processing/src/main/java/io/druid/segment/incremental/OffheapIncrementalIndex.java b/processing/src/main/java/io/druid/segment/incremental/OffheapIncrementalIndex.java
index d377634e0bc..101e4e14e80 100644
--- a/processing/src/main/java/io/druid/segment/incremental/OffheapIncrementalIndex.java
+++ b/processing/src/main/java/io/druid/segment/incremental/OffheapIncrementalIndex.java
@@ -43,7 +43,7 @@
/**
*/
-public class OffheapIncrementalIndex extends IncrementalIndex<BufferAggregator>
+public class OffheapIncrementalIndex extends ExternalDataIncrementalIndex<BufferAggregator>
{
private static final Logger log = new Logger(OffheapIncrementalIndex.class);
@@ -95,7 +95,7 @@
}
@Override
- public FactsHolder getFacts()
+ protected FactsHolder getFacts()
{
return facts;
}
@@ -264,60 +264,66 @@ public String getOutOfRowsReason()
}
@Override
- protected BufferAggregator[] getAggsForRow(int rowOffset)
+ protected BufferAggregator[] getAggsForRow(TimeAndDims timeAndDims)
{
return getAggs();
}
@Override
- protected Object getAggVal(BufferAggregator agg, int rowOffset, int aggPosition)
+ protected Object getAggVal(BufferAggregator agg, TimeAndDims timeAndDims, int aggPosition)
{
- int[] indexAndOffset = indexAndOffsets.get(rowOffset);
+ int rowIndex = timeAndDims.getRowIndex();
+ int[] indexAndOffset = indexAndOffsets.get(rowIndex);
ByteBuffer bb = aggBuffers.get(indexAndOffset[0]).get();
return agg.get(bb, indexAndOffset[1] + aggOffsetInBuffer[aggPosition]);
}
@Override
- public float getMetricFloatValue(int rowOffset, int aggOffset)
+ public float getMetricFloatValue(TimeAndDims timeAndDims, int aggOffset)
{
BufferAggregator agg = getAggs()[aggOffset];
- int[] indexAndOffset = indexAndOffsets.get(rowOffset);
+ int rowIndex = timeAndDims.getRowIndex();
+ int[] indexAndOffset = indexAndOffsets.get(rowIndex);
ByteBuffer bb = aggBuffers.get(indexAndOffset[0]).get();
return agg.getFloat(bb, indexAndOffset[1] + aggOffsetInBuffer[aggOffset]);
}
@Override
- public long getMetricLongValue(int rowOffset, int aggOffset)
+ public long getMetricLongValue(TimeAndDims timeAndDims, int aggOffset)
{
BufferAggregator agg = getAggs()[aggOffset];
- int[] indexAndOffset = indexAndOffsets.get(rowOffset);
+ int rowIndex = timeAndDims.getRowIndex();
+ int[] indexAndOffset = indexAndOffsets.get(rowIndex);
ByteBuffer bb = aggBuffers.get(indexAndOffset[0]).get();
return agg.getLong(bb, indexAndOffset[1] + aggOffsetInBuffer[aggOffset]);
}
@Override
- public Object getMetricObjectValue(int rowOffset, int aggOffset)
+ public Object getMetricObjectValue(TimeAndDims timeAndDims, int aggOffset)
{
BufferAggregator agg = getAggs()[aggOffset];
- int[] indexAndOffset = indexAndOffsets.get(rowOffset);
+ int rowIndex = timeAndDims.getRowIndex();
+ int[] indexAndOffset = indexAndOffsets.get(rowIndex);
ByteBuffer bb = aggBuffers.get(indexAndOffset[0]).get();
return agg.get(bb, indexAndOffset[1] + aggOffsetInBuffer[aggOffset]);
}
@Override
- public double getMetricDoubleValue(int rowOffset, int aggOffset)
+ public double getMetricDoubleValue(TimeAndDims timeAndDims, int aggOffset)
{
BufferAggregator agg = getAggs()[aggOffset];
- int[] indexAndOffset = indexAndOffsets.get(rowOffset);
+ int rowIndex = timeAndDims.getRowIndex();
+ int[] indexAndOffset = indexAndOffsets.get(rowIndex);
ByteBuffer bb = aggBuffers.get(indexAndOffset[0]).get();
return agg.getDouble(bb, indexAndOffset[1] + aggOffsetInBuffer[aggOffset]);
}
@Override
- public boolean isNull(int rowOffset, int aggOffset)
+ public boolean isNull(TimeAndDims timeAndDims, int aggOffset)
{
BufferAggregator agg = getAggs()[aggOffset];
- int[] indexAndOffset = indexAndOffsets.get(rowOffset);
+ int rowIndex = timeAndDims.getRowIndex();
+ int[] indexAndOffset = indexAndOffsets.get(rowIndex);
ByteBuffer bb = aggBuffers.get(indexAndOffset[0]).get();
return agg.isNull(bb, indexAndOffset[1] + aggOffsetInBuffer[aggOffset]);
}
diff --git a/processing/src/main/java/io/druid/segment/incremental/OffheapOakComputeConsumer.java b/processing/src/main/java/io/druid/segment/incremental/OffheapOakComputeConsumer.java
new file mode 100644
index 00000000000..73c1464fc68
--- /dev/null
+++ b/processing/src/main/java/io/druid/segment/incremental/OffheapOakComputeConsumer.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets 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 io.druid.segment.incremental;
+
+import io.druid.data.input.InputRow;
+import io.druid.query.aggregation.AggregatorFactory;
+import io.druid.query.aggregation.BufferAggregator;
+import oak.WritableOakBuffer;
+
+import java.util.function.Consumer;
+
+/**
+ * For sending as input parameter to the Oak's putifAbsentComputeIfPresent method
+ */
+public class OffheapOakComputeConsumer implements Consumer<WritableOakBuffer>
+{
+ AggregatorFactory[] metrics;
+ boolean reportParseExceptions;
+ InputRow row;
+ ThreadLocal<InputRow> rowContainer;
+ int[] aggOffsetInBuffer;
+ BufferAggregator[] aggs;
+ boolean executed; // for figuring out whether a put or a compute was executed
+
+ public OffheapOakComputeConsumer(
+ AggregatorFactory[] metrics,
+ boolean reportParseExceptions,
+ InputRow row,
+ ThreadLocal<InputRow> rowContainer,
+ int[] aggOffsetInBuffer,
+ BufferAggregator[] aggs
+ )
+ {
+ this.metrics = metrics;
+ this.reportParseExceptions = reportParseExceptions;
+ this.row = row;
+ this.rowContainer = rowContainer;
+ this.aggOffsetInBuffer = aggOffsetInBuffer;
+ this.aggs = aggs;
+ this.executed = false;
+ }
+
+ @Override
+ public void accept(WritableOakBuffer writableOakBuffer)
+ {
+ OffheapOakIncrementalIndex.aggregate(metrics, reportParseExceptions, row, rowContainer,
+ writableOakBuffer.getByteBuffer(), aggOffsetInBuffer, aggs);
+ executed = true;
+
+ }
+
+ public boolean executed()
+ {
+ return executed;
+ }
+}
diff --git a/processing/src/main/java/io/druid/segment/incremental/OffheapOakCreateValueConsumer.java b/processing/src/main/java/io/druid/segment/incremental/OffheapOakCreateValueConsumer.java
new file mode 100644
index 00000000000..ac37344a394
--- /dev/null
+++ b/processing/src/main/java/io/druid/segment/incremental/OffheapOakCreateValueConsumer.java
@@ -0,0 +1,81 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets 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 io.druid.segment.incremental;
+
+import io.druid.data.input.InputRow;
+import io.druid.query.aggregation.AggregatorFactory;
+import io.druid.query.aggregation.BufferAggregator;
+import io.druid.segment.ColumnSelectorFactory;
+
+import java.util.Map;
+import java.util.function.Consumer;
+import java.nio.ByteBuffer;
+
+public class OffheapOakCreateValueConsumer implements Consumer<ByteBuffer>
+{
+
+ AggregatorFactory[] metrics;
+ boolean reportParseExceptions;
+ InputRow row;
+ ThreadLocal<InputRow> rowContainer;
+ BufferAggregator[] aggs;
+ Map<String, ColumnSelectorFactory> selectors;
+ int[] aggOffsetInBuffer;
+
+ public OffheapOakCreateValueConsumer(
+ AggregatorFactory[] metrics,
+ boolean reportParseExceptions,
+ InputRow row,
+ ThreadLocal<InputRow> rowContainer,
+ BufferAggregator[] aggs,
+ Map<String, ColumnSelectorFactory> selectors,
+ int[] aggOffsetInBuffer
+ )
+ {
+ this.metrics = metrics;
+ this.reportParseExceptions = reportParseExceptions;
+ this.row = row;
+ this.rowContainer = rowContainer;
+ this.aggs = aggs;
+ this.selectors = selectors;
+ this.aggOffsetInBuffer = aggOffsetInBuffer;
+ }
+
+ @Override
+ public void accept(ByteBuffer byteBuffer)
+ {
+ if (metrics.length > 0 && aggs[0] == null) {
+ // note: creation of Aggregators is done lazily when at least one row from input is available
+ // so that FilteredAggregators could be initialized correctly.
+ rowContainer.set(row);
+ for (int i = 0; i < metrics.length; i++) {
+ final AggregatorFactory agg = metrics[i];
+ aggs[i] = agg.factorizeBuffered(selectors.get(agg.getName()));
+ }
+ rowContainer.set(null);
+ }
+
+ for (int i = 0; i < metrics.length; i++) {
+ aggs[i].init(byteBuffer, aggOffsetInBuffer[i]);
+ }
+
+ OffheapOakIncrementalIndex.aggregate(metrics, reportParseExceptions, row, rowContainer, byteBuffer, aggOffsetInBuffer, aggs);
+ }
+}
diff --git a/processing/src/main/java/io/druid/segment/incremental/OffheapOakIncrementalIndex.java b/processing/src/main/java/io/druid/segment/incremental/OffheapOakIncrementalIndex.java
new file mode 100644
index 00000000000..30f85df090a
--- /dev/null
+++ b/processing/src/main/java/io/druid/segment/incremental/OffheapOakIncrementalIndex.java
@@ -0,0 +1,668 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets 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 io.druid.segment.incremental;
+
+import com.google.common.base.Supplier;
+import com.google.common.collect.Iterators;
+import com.google.common.collect.Maps;
+import com.google.common.primitives.Ints;
+import com.google.common.primitives.Longs;
+import io.druid.data.input.InputRow;
+import io.druid.data.input.MapBasedRow;
+import io.druid.data.input.Row;
+import io.druid.java.util.common.StringUtils;
+import io.druid.java.util.common.logger.Logger;
+import io.druid.java.util.common.parsers.ParseException;
+import io.druid.query.aggregation.AggregatorFactory;
+import io.druid.query.aggregation.BufferAggregator;
+import io.druid.query.aggregation.PostAggregator;
+import io.druid.segment.ColumnSelectorFactory;
+import io.druid.segment.DimensionHandler;
+import io.druid.segment.DimensionIndexer;
+import io.druid.segment.column.ColumnCapabilitiesImpl;
+import io.druid.segment.column.ValueType;
+
+import java.nio.ByteBuffer;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.List;
+import java.util.function.Function;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import oak.OakMapOffHeapImpl;
+import oak.WritableOakBufferImpl;
+import oak.OakMap;
+import oak.CloseableIterator;
+
+import javax.annotation.Nullable;
+
+
+/**
+ */
+public class OffheapOakIncrementalIndex extends
+ io.druid.segment.incremental.InternalDataIncrementalIndex<BufferAggregator>
+{
+
+ private static final Logger log = new Logger(OffheapOakIncrementalIndex.class);
+
+ // When serializing an object from TimeAndDims.dims, we use:
+ // 1. 4 bytes for representing its type (Double, Float, Long or String)
+ // 2. 8 bytes for saving its value or the array position and length (in the case of String)
+ static final Integer ALLOC_PER_DIM = 12;
+ static final Integer NO_DIM = -1;
+ static final Integer TIME_STAMP_INDEX = 0;
+ static final Integer DIMS_LENGTH_INDEX = TIME_STAMP_INDEX + Long.BYTES;
+ static final Integer DIMS_INDEX = DIMS_LENGTH_INDEX + Integer.BYTES;
+
+ OakMapOffHeapImpl oak;
+ private volatile Map<String, ColumnSelectorFactory> selectors;
+ private volatile int[] aggOffsetInBuffer;
+ private volatile int aggsTotalSize;
+ private final int maxRowCount;
+ private String outOfRowsReason = null;
+
+ OffheapOakIncrementalIndex(
+ io.druid.segment.incremental.IncrementalIndexSchema incrementalIndexSchema,
+ boolean deserializeComplexMetrics, boolean reportParseExceptions,
+ boolean concurrentEventAdd,
+ int maxRowCount
+ )
+ {
+ super(incrementalIndexSchema, deserializeComplexMetrics, reportParseExceptions,
+ concurrentEventAdd);
+ oak = new OakMapOffHeapImpl(new TimeAndDimsByteBuffersComp(dimensionDescsList), getMinTimeAndDimsByteBuffer());
+ this.maxRowCount = maxRowCount;
+ }
+
+ @Override
+ public Iterable<Row> iterableWithPostAggregations(List<PostAggregator> postAggs, boolean descending)
+ {
+ return new Iterable<Row>()
+ {
+ @Override
+ public Iterator<Row> iterator()
+ {
+ OakMap oakMap = descending ? oak.descendingMap() : oak;
+ Function<Map.Entry<ByteBuffer, ByteBuffer>, Row> transformer = new Transformer(postAggs);
+ return oakMap.entriesTransformIterator(transformer);
+ }
+ };
+ }
+
+ // for oak's transform iterator
+ private class Transformer implements Function<Map.Entry<ByteBuffer, ByteBuffer>, Row>
+ {
+ List<PostAggregator> postAggs;
+ final List<DimensionDesc> dimensions;
+
+ public Transformer(List<PostAggregator> postAggs)
+ {
+ this.postAggs = postAggs;
+ this.dimensions = getDimensions();
+ }
+
+ @Nullable
+ @Override
+ public Row apply(@Nullable Map.Entry<ByteBuffer, ByteBuffer> entry)
+ {
+ TimeAndDims key = timeAndDimsDeserialization(entry.getKey());
+ ByteBuffer value = entry.getValue();
+ Object[] dims = key.getDims();
+
+ Map<String, Object> theVals = Maps.newLinkedHashMap();
+ for (int i = 0; i < dims.length; ++i) {
+ Object dim = dims[i];
+ DimensionDesc dimensionDesc = dimensions.get(i);
+ if (dimensionDesc == null) {
+ continue;
+ }
+ String dimensionName = dimensionDesc.getName();
+ DimensionHandler handler = dimensionDesc.getHandler();
+ if (dim == null || handler.getLengthOfEncodedKeyComponent(dim) == 0) {
+ theVals.put(dimensionName, null);
+ continue;
+ }
+ final DimensionIndexer indexer = dimensionDesc.getIndexer();
+ Object rowVals = indexer.convertUnsortedEncodedKeyComponentToActualArrayOrList(dim, DimensionIndexer.LIST);
+ theVals.put(dimensionName, rowVals);
+ }
+
+ BufferAggregator[] aggs = getAggs();
+ for (int i = 0; i < aggs.length; ++i) {
+ theVals.put(metrics[i].getName(), aggs[i].get(value, aggOffsetInBuffer[i]));
+ }
+
+ if (postAggs != null) {
+ for (PostAggregator postAgg : postAggs) {
+ theVals.put(postAgg.getName(), postAgg.compute(theVals));
+ }
+ }
+
+ return new MapBasedRow(key.getTimestamp(), theVals);
+ }
+ };
+
+ @Override
+ protected long getMinTimeMillis()
+ {
+ ByteBuffer minKey = oak.getMinKey();
+ return getTimestamp(minKey);
+ }
+
+ @Override
+ protected long getMaxTimeMillis()
+ {
+ ByteBuffer maxKey = oak.getMaxKey();
+ return getTimestamp(maxKey);
+ }
+
+ @Override
+ public int getLastRowIndex()
+ {
+ return 0; // Oak doesn't use the row indexes
+ }
+
+ @Override
+ protected BufferAggregator[] getAggsForRow(TimeAndDims timeAndDims)
+ {
+ return getAggs();
+ }
+
+ @Override
+ protected Object getAggVal(BufferAggregator agg, TimeAndDims timeAndDims, int aggPosition)
+ {
+ ByteBuffer serializedKey = timeAndDimsSerialization(timeAndDims);
+ WritableOakBufferImpl oakValue = (WritableOakBufferImpl) oak.get(serializedKey);
+ ByteBuffer aggBuffer = oakValue.getByteBuffer();
+ return agg.get(aggBuffer, aggBuffer.position() + aggOffsetInBuffer[aggPosition]);
+ }
+
+ @Override
+ protected float getMetricFloatValue(TimeAndDims timeAndDims, int aggOffset)
+ {
+ BufferAggregator agg = getAggs()[aggOffset];
+ ByteBuffer serializedKey = timeAndDimsSerialization(timeAndDims);
+ WritableOakBufferImpl oakValue = (WritableOakBufferImpl) oak.get(serializedKey);
+ ByteBuffer aggBuffer = oakValue.getByteBuffer();
+ return agg.getFloat(aggBuffer, aggBuffer.position() + aggOffsetInBuffer[aggOffset]);
+ }
+
+ @Override
+ protected long getMetricLongValue(TimeAndDims timeAndDims, int aggOffset)
+ {
+ BufferAggregator agg = getAggs()[aggOffset];
+ ByteBuffer serializedKey = timeAndDimsSerialization(timeAndDims);
+ WritableOakBufferImpl oakValue = (WritableOakBufferImpl) oak.get(serializedKey);
+ ByteBuffer aggBuffer = oakValue.getByteBuffer();
+ return agg.getLong(aggBuffer, aggBuffer.position() + aggOffsetInBuffer[aggOffset]);
+ }
+
+ @Override
+ protected Object getMetricObjectValue(TimeAndDims timeAndDims, int aggOffset)
+ {
+ BufferAggregator agg = getAggs()[aggOffset];
+ ByteBuffer serializedKey = timeAndDimsSerialization(timeAndDims);
+ WritableOakBufferImpl oakValue = (WritableOakBufferImpl) oak.get(serializedKey);
+ ByteBuffer aggBuffer = oakValue.getByteBuffer();
+ return agg.get(aggBuffer, aggBuffer.position() + aggOffsetInBuffer[aggOffset]);
+ }
+
+ @Override
+ protected double getMetricDoubleValue(TimeAndDims timeAndDims, int aggOffset)
+ {
+ BufferAggregator agg = getAggs()[aggOffset];
+ ByteBuffer serializedKey = timeAndDimsSerialization(timeAndDims);
+ WritableOakBufferImpl oakValue = (WritableOakBufferImpl) oak.get(serializedKey);
+ ByteBuffer aggBuffer = oakValue.getByteBuffer();
+ return agg.getDouble(aggBuffer, aggBuffer.position() + aggOffsetInBuffer[aggOffset]);
+ }
+
+ @Override
+ protected boolean isNull(TimeAndDims timeAndDims, int aggOffset)
+ {
+ BufferAggregator agg = getAggs()[aggOffset];
+ ByteBuffer serializedKey = timeAndDimsSerialization(timeAndDims);
+ WritableOakBufferImpl oakValue = (WritableOakBufferImpl) oak.get(serializedKey);
+ ByteBuffer aggBuffer = oakValue.getByteBuffer();
+ return agg.isNull(aggBuffer, aggBuffer.position() + aggOffsetInBuffer[aggOffset]);
+ }
+
+ @Override
+ public Iterable<TimeAndDims> timeRangeIterable(
+ boolean descending, long timeStart, long timeEnd)
+ {
+ if (timeStart > timeEnd) {
+ return null;
+ }
+ ByteBuffer from = timeAndDimsSerialization(new TimeAndDims(timeStart, null, dimensionDescsList));
+ ByteBuffer to = timeAndDimsSerialization(new TimeAndDims(timeEnd + 1, null, dimensionDescsList));
+ OakMap subMap = oak.subMap(from, true, to, false);
+ if (descending == true) {
+ subMap = subMap.descendingMap();
+ }
+
+ CloseableIterator<ByteBuffer> keysIterator = subMap.keysIterator();
+
+ return new Iterable<TimeAndDims>() {
+ @Override
+ public Iterator<TimeAndDims> iterator()
+ {
+ return Iterators.transform(
+ keysIterator,
+ byteBuffer -> timeAndDimsDeserialization(byteBuffer)
+ );
+ }
+ };
+ }
+
+ @Override
+ public Iterable<IncrementalIndex.TimeAndDims> keySet()
+ {
+ CloseableIterator<ByteBuffer> keysIterator = oak.keysIterator();
+
+ return new Iterable<TimeAndDims>() {
+ @Override
+ public Iterator<TimeAndDims> iterator()
+ {
+ return Iterators.transform(
+ keysIterator,
+ byteBuffer -> timeAndDimsDeserialization(byteBuffer)
+ );
+ }
+ };
+ }
+
+ @Override
+ public boolean canAppendRow()
+ {
+ final boolean canAdd = size() < maxRowCount;
+ if (!canAdd) {
+ outOfRowsReason = StringUtils.format("Maximum number of rows [%d] reached", maxRowCount);
+ }
+ return canAdd;
+ }
+
+ @Override
+ public String getOutOfRowsReason()
+ {
+ return outOfRowsReason;
+ }
+
+ private Integer addToOak(
+ AggregatorFactory[] metrics,
+ boolean reportParseExceptions,
+ InputRow row,
+ AtomicInteger numEntries,
+ IncrementalIndex.TimeAndDims key,
+ ThreadLocal<InputRow> rowContainer,
+ boolean skipMaxRowsInMemoryCheck
+ ) throws IndexSizeExceededException
+ {
+ ByteBuffer serializedKey = timeAndDimsSerialization(key);
+ OffheapOakCreateValueConsumer valueCreator = new OffheapOakCreateValueConsumer(metrics, reportParseExceptions,
+ row, rowContainer, getAggs(), selectors, aggOffsetInBuffer);
+ OffheapOakComputeConsumer func = new OffheapOakComputeConsumer(metrics, reportParseExceptions, row, rowContainer,
+ aggOffsetInBuffer, getAggs());
+ if (numEntries.get() < maxRowCount || skipMaxRowsInMemoryCheck) {
+ oak.putIfAbsentComputeIfPresent(serializedKey, valueCreator, aggsTotalSize, func);
+ if (func.executed() == false) { // a put operation was executed
+ numEntries.incrementAndGet();
+ }
+ } else {
+ if (oak.computeIfPresent(serializedKey, func) == false) { // the key wasn't in oak
+ throw new IndexSizeExceededException("Maximum number of rows [%d] reached", maxRowCount);
+ }
+ }
+ return numEntries.get();
+ }
+
+ @Override
+ protected BufferAggregator[] initAggs(
+ AggregatorFactory[] metrics,
+ Supplier<InputRow> rowSupplier,
+ boolean deserializeComplexMetrics,
+ boolean concurrentEventAdd
+ )
+ {
+ selectors = Maps.newHashMap();
+ aggOffsetInBuffer = new int[metrics.length];
+
+ for (int i = 0; i < metrics.length; i++) {
+ AggregatorFactory agg = metrics[i];
+
+ ColumnSelectorFactory columnSelectorFactory = makeColumnSelectorFactory(
+ agg,
+ rowSupplier,
+ deserializeComplexMetrics
+ );
+
+ selectors.put(
+ agg.getName(),
+ new OnheapIncrementalIndex.ObjectCachingColumnSelectorFactory(columnSelectorFactory, concurrentEventAdd)
+ );
+
+ if (i == 0) {
+ aggOffsetInBuffer[i] = metrics[i].getMaxIntermediateSize();
+ } else {
+ aggOffsetInBuffer[i] = aggOffsetInBuffer[i - 1] + metrics[i - 1].getMaxIntermediateSize();
+ }
+ }
+ aggsTotalSize += aggOffsetInBuffer[metrics.length - 1] + metrics[metrics.length - 1].getMaxIntermediateSize();
+
+ return new BufferAggregator[metrics.length];
+ }
+
+ public static void aggregate(
+ AggregatorFactory[] metrics,
+ boolean reportParseExceptions,
+ InputRow row,
+ ThreadLocal<InputRow> rowContainer,
+ ByteBuffer aggBuffer,
+ int[] aggOffsetInBuffer,
+ BufferAggregator[] aggs
+ )
+ {
+ rowContainer.set(row);
+
+ for (int i = 0; i < metrics.length; i++) {
+ final BufferAggregator agg = aggs[i];
+
+ synchronized (agg) {
+ try {
+ agg.aggregate(aggBuffer, aggBuffer.position() + aggOffsetInBuffer[i]);
+ }
+ catch (ParseException e) {
+ // "aggregate" can throw ParseExceptions if a selector expects something but gets something else.
+ if (reportParseExceptions) {
+ throw new ParseException(e, "Encountered parse error for aggregator[%s]", metrics[i].getName());
+ } else {
+ log.debug(e, "Encountered parse error, skipping aggregator[%s].", metrics[i].getName());
+ }
+ }
+ }
+ }
+ rowContainer.set(null);
+ }
+
+ @Override
+ public int add(InputRow row, boolean skipMaxRowsInMemoryCheck) throws IndexSizeExceededException
+ {
+ TimeAndDims key = toTimeAndDims(row);
+ final int rv = addToOak(
+ metrics,
+ reportParseExceptions,
+ row,
+ numEntries,
+ key,
+ in,
+ skipMaxRowsInMemoryCheck
+ );
+ updateMaxIngestedTime(row.getTimestamp());
+ return rv;
+ }
+
+ ByteBuffer timeAndDimsSerialization(TimeAndDims timeAndDims)
+ {
+ int allocSize;
+ ByteBuffer buf;
+ Object[] dims = timeAndDims.getDims();
+
+ if (dims == null) {
+ allocSize = Long.BYTES + Integer.BYTES;
+ buf = ByteBuffer.allocate(allocSize);
+ buf.putLong(timeAndDims.getTimestamp());
+ buf.putInt(0);
+ buf.position(0);
+ return buf;
+ }
+
+ // When the dimensionDesc's capabilities are of type ValueType.STRING,
+ // the object in timeAndDims.dims is of type int[].
+ // In this case, we need to know the array size before allocating the ByteBuffer.
+ int sumOfArrayLengths = 0;
+ for (int i = 0; i < dims.length; i++) {
+ Object dim = dims[i];
+ DimensionDesc dimensionDesc = getDimensions().get(i);
+ if (dimensionDesc != null) {
+ }
+ if (dim == null) {
+ continue;
+ }
+ if (getDimValueType(i) == ValueType.STRING) {
+ sumOfArrayLengths += ((int[]) dim).length;
+ }
+ }
+
+ // The ByteBuffer will contain:
+ // 1. the timeStamp
+ // 2. dims.length
+ // 3. the serialization of each dim
+ // 4. the array (for dims with capabilities of a String ValueType)
+ allocSize = Long.BYTES + Integer.BYTES + ALLOC_PER_DIM * dims.length + Integer.BYTES * sumOfArrayLengths;
+ buf = ByteBuffer.allocate(allocSize);
+ buf.putLong(timeAndDims.getTimestamp());
+ buf.putInt(dims.length);
+ int currDimsIndex = DIMS_INDEX;
+ int currArrayIndex = DIMS_INDEX + ALLOC_PER_DIM * dims.length;
+ for (int dimIndex = 0; dimIndex < dims.length; dimIndex++) {
+ ValueType valueType = getDimValueType(dimIndex);
+ if (valueType == null || dims[dimIndex] == null) {
+ buf.putInt(currDimsIndex, NO_DIM);
+ currDimsIndex += ALLOC_PER_DIM;
+ continue;
+ }
+ switch (valueType) {
+ case LONG:
+ buf.putInt(currDimsIndex, valueType.ordinal());
+ currDimsIndex += Integer.BYTES;
+ buf.putLong(currDimsIndex, (Long) dims[dimIndex]);
+ currDimsIndex += Long.BYTES;
+ break;
+ case FLOAT:
+ buf.putInt(currDimsIndex, valueType.ordinal());
+ currDimsIndex += Integer.BYTES;
+ buf.putFloat(currDimsIndex, (Float) dims[dimIndex]);
+ currDimsIndex += Long.BYTES;
+ break;
+ case DOUBLE:
+ buf.putInt(currDimsIndex, valueType.ordinal());
+ currDimsIndex += Integer.BYTES;
+ buf.putDouble(currDimsIndex, (Double) dims[dimIndex]);
+ currDimsIndex += Long.BYTES;
+ break;
+ case STRING:
+ buf.putInt(currDimsIndex, valueType.ordinal()); // writing the value type
+ currDimsIndex += Integer.BYTES;
+ buf.putInt(currDimsIndex, currArrayIndex); // writing the array position
+ currDimsIndex += Integer.BYTES;
+ if (dims[dimIndex] == null) {
+ buf.putInt(currDimsIndex, 0);
+ currDimsIndex += Integer.BYTES;
+ break;
+ }
+ int[] array = (int[]) dims[dimIndex];
+ buf.putInt(currDimsIndex, array.length); // writing the array length
+ currDimsIndex += Integer.BYTES;
+ for (int j = 0; j < array.length; j++) {
+ buf.putInt(currArrayIndex, array[j]);
+ currArrayIndex += Integer.BYTES;
+ }
+ break;
+ default:
+ buf.putInt(currDimsIndex, NO_DIM);
+ currDimsIndex += ALLOC_PER_DIM;
+ }
+ }
+ buf.position(0);
+ return buf;
+ }
+
+ TimeAndDims timeAndDimsDeserialization(ByteBuffer buff)
+ {
+ long timeStamp = getTimestamp(buff);
+ int dimsLength = getDimsLength(buff);
+ Object[] dims = new Object[dimsLength];
+ for (int dimIndex = 0; dimIndex < dimsLength; dimIndex++) {
+ Object dim = getDimValue(buff, dimIndex);
+ dims[dimIndex] = dim;
+ }
+ return new TimeAndDims(timeStamp, dims, dimensionDescsList, TimeAndDims.EMPTY_ROW_INDEX);
+ }
+
+ private ValueType getDimValueType(int dimIndex)
+ {
+ DimensionDesc dimensionDesc = getDimensions().get(dimIndex);
+ if (dimensionDesc == null) {
+ return null;
+ }
+ ColumnCapabilitiesImpl capabilities = dimensionDesc.getCapabilities();
+ if (capabilities == null) {
+ return null;
+ }
+ return capabilities.getType();
+ }
+
+ static long getTimestamp(ByteBuffer buff)
+ {
+ return buff.getLong(buff.position() + TIME_STAMP_INDEX);
+ }
+
+ static int getDimsLength(ByteBuffer buff)
+ {
+ return buff.getInt(buff.position() + DIMS_LENGTH_INDEX);
+ }
+
+ static Object getDimValue(ByteBuffer buff, int dimIndex)
+ {
+ Object dimObject = null;
+ int dimsLength = getDimsLength(buff);
+ if (dimIndex >= dimsLength) {
+ return null;
+ }
+ int dimType = buff.getInt(getDimIndexInBuffer(buff, dimIndex));
+ if (dimType == NO_DIM) {
+ return null;
+ } else if (dimType == ValueType.DOUBLE.ordinal()) {
+ dimObject = buff.getDouble(getDimIndexInBuffer(buff, dimIndex) + Integer.BYTES);
+ } else if (dimType == ValueType.FLOAT.ordinal()) {
+ dimObject = buff.getFloat(getDimIndexInBuffer(buff, dimIndex) + Integer.BYTES);
+ } else if (dimType == ValueType.LONG.ordinal()) {
+ dimObject = buff.getLong(getDimIndexInBuffer(buff, dimIndex) + Integer.BYTES);
+ } else if (dimType == ValueType.STRING.ordinal()) {
+ int arrayIndex = buff.position() + buff.getInt(getDimIndexInBuffer(buff, dimIndex) + Integer.BYTES);
+ int arraySize = buff.getInt(getDimIndexInBuffer(buff, dimIndex) + 2 * Integer.BYTES);
+ int[] array = new int[arraySize];
+ for (int i = 0; i < arraySize; i++) {
+ array[i] = buff.getInt(arrayIndex);
+ arrayIndex += Integer.BYTES;
+ }
+ dimObject = array;
+ }
+
+ return dimObject;
+ }
+
+ static boolean checkDimsAllNull(ByteBuffer buff)
+ {
+ int dimsLength = getDimsLength(buff);
+ for (int index = 0; index < dimsLength; index++) {
+ if (buff.getInt(getDimIndexInBuffer(buff, index)) != NO_DIM) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ static int getDimIndexInBuffer(ByteBuffer buff, int dimIndex)
+ {
+ int dimsLength = getDimsLength(buff);
+ if (dimIndex >= dimsLength) {
+ return NO_DIM;
+ }
+ return buff.position() + DIMS_INDEX + dimIndex * ALLOC_PER_DIM;
+ }
+
+ private ByteBuffer getMinTimeAndDimsByteBuffer()
+ {
+ Object[] dims = new Object[dimensionDescsList.size()];
+ for (int i = 0; i < dims.length; i++) {
+ dims[i] = null;
+ }
+ TimeAndDims minTimeAndDims = new TimeAndDims(this.minTimestamp, dims, dimensionDescsList);
+ ByteBuffer minTimeAndDimsByteBuffer = timeAndDimsSerialization(minTimeAndDims);
+ return minTimeAndDimsByteBuffer;
+ }
+
+ public final Comparator<ByteBuffer> dimsByteBufferComparator()
+ {
+ return new TimeAndDimsByteBuffersComp(dimensionDescsList);
+ }
+
+ static final class TimeAndDimsByteBuffersComp implements Comparator<ByteBuffer>
+ {
+ private List<DimensionDesc> dimensionDescsList;
+
+ public TimeAndDimsByteBuffersComp(List<DimensionDesc> dimensionDescsList)
+ {
+ this.dimensionDescsList = dimensionDescsList;
+ }
+
+ @Override
+ public int compare(ByteBuffer lhs, ByteBuffer rhs)
+ {
+ int retVal = Longs.compare(getTimestamp(lhs), getTimestamp(rhs));
+ int numComparisons = Math.min(getDimsLength(lhs), getDimsLength(rhs));
+
+ int dimIndex = 0;
+ while (retVal == 0 && dimIndex < numComparisons) {
+ int lhsType = lhs.getInt(getDimIndexInBuffer(lhs, dimIndex));
+ int rhsType = rhs.getInt(getDimIndexInBuffer(rhs, dimIndex));
+
+ if (lhsType == NO_DIM) {
+ if (rhsType == NO_DIM) {
+ ++dimIndex;
+ continue;
+ }
+ return -1;
+ }
+
+ if (rhsType == NO_DIM) {
+ return 1;
+ }
+
+ final DimensionIndexer indexer = dimensionDescsList.get(dimIndex).getIndexer();
+ Object lhsObject = getDimValue(lhs, dimIndex);
+ Object rhsObject = getDimValue(rhs, dimIndex);
+ retVal = indexer.compareUnsortedEncodedKeyComponents(lhsObject, rhsObject);
+ ++dimIndex;
+ }
+
+ if (retVal == 0) {
+ int lengthDiff = Ints.compare(getDimsLength(lhs), getDimsLength(rhs));
+ if (lengthDiff == 0) {
+ return 0;
+ }
+ ByteBuffer largerDims = lengthDiff > 0 ? lhs : rhs;
+ return checkDimsAllNull(largerDims) ? 0 : lengthDiff;
+ }
+ return retVal;
+ }
+ }
+}
diff --git a/processing/src/main/java/io/druid/segment/incremental/OnheapIncrementalIndex.java b/processing/src/main/java/io/druid/segment/incremental/OnheapIncrementalIndex.java
index 1a377ea826b..47d3a3f02b1 100644
--- a/processing/src/main/java/io/druid/segment/incremental/OnheapIncrementalIndex.java
+++ b/processing/src/main/java/io/druid/segment/incremental/OnheapIncrementalIndex.java
@@ -44,7 +44,7 @@
/**
*/
-public class OnheapIncrementalIndex extends IncrementalIndex<Aggregator>
+public class OnheapIncrementalIndex extends ExternalDataIncrementalIndex<Aggregator>
{
private static final Logger log = new Logger(OnheapIncrementalIndex.class);
@@ -73,7 +73,7 @@
}
@Override
- public FactsHolder getFacts()
+ protected FactsHolder getFacts()
{
return facts;
}
@@ -251,45 +251,51 @@ public String getOutOfRowsReason()
}
@Override
- protected Aggregator[] getAggsForRow(int rowOffset)
+ protected Aggregator[] getAggsForRow(TimeAndDims timeAndDims)
{
- return concurrentGet(rowOffset);
+ int rowIndex = timeAndDims.getRowIndex();
+ return concurrentGet(rowIndex);
}
@Override
- protected Object getAggVal(Aggregator agg, int rowOffset, int aggPosition)
+ protected Object getAggVal(Aggregator agg, TimeAndDims timeAndDims, int aggPosition)
{
return agg.get();
}
@Override
- public float getMetricFloatValue(int rowOffset, int aggOffset)
+ public float getMetricFloatValue(TimeAndDims timeAndDims, int aggOffset)
{
- return concurrentGet(rowOffset)[aggOffset].getFloat();
+ int rowIndex = timeAndDims.getRowIndex();
+ return concurrentGet(rowIndex)[aggOffset].getFloat();
}
@Override
- public long getMetricLongValue(int rowOffset, int aggOffset)
+ public long getMetricLongValue(TimeAndDims timeAndDims, int aggOffset)
{
- return concurrentGet(rowOffset)[aggOffset].getLong();
+ int rowIndex = timeAndDims.getRowIndex();
+ return concurrentGet(rowIndex)[aggOffset].getLong();
}
@Override
- public Object getMetricObjectValue(int rowOffset, int aggOffset)
+ public Object getMetricObjectValue(TimeAndDims timeAndDims, int aggOffset)
{
- return concurrentGet(rowOffset)[aggOffset].get();
+ int rowIndex = timeAndDims.getRowIndex();
+ return concurrentGet(rowIndex)[aggOffset].get();
}
@Override
- protected double getMetricDoubleValue(int rowOffset, int aggOffset)
+ protected double getMetricDoubleValue(TimeAndDims timeAndDims, int aggOffset)
{
- return concurrentGet(rowOffset)[aggOffset].getDouble();
+ int rowIndex = timeAndDims.getRowIndex();
+ return concurrentGet(rowIndex)[aggOffset].getDouble();
}
@Override
- public boolean isNull(int rowOffset, int aggOffset)
+ public boolean isNull(TimeAndDims timeAndDims, int aggOffset)
{
- return concurrentGet(rowOffset)[aggOffset].isNull();
+ int rowIndex = timeAndDims.getRowIndex();
+ return concurrentGet(rowIndex)[aggOffset].isNull();
}
/**
diff --git a/processing/src/main/java/io/druid/segment/incremental/TimeAndDimsHolder.java b/processing/src/main/java/io/druid/segment/incremental/TimeAndDimsHolder.java
index 3e080e4a877..32b21c11821 100644
--- a/processing/src/main/java/io/druid/segment/incremental/TimeAndDimsHolder.java
+++ b/processing/src/main/java/io/druid/segment/incremental/TimeAndDimsHolder.java
@@ -36,13 +36,4 @@ public void set(IncrementalIndex.TimeAndDims currEntry)
{
this.currEntry = currEntry;
}
-
- /**
- * This method doesn't have well-defined semantics ("value" of what?), should be removed in favor of chaining
- * get().getRowIndex().
- */
- public int getValue()
- {
- return currEntry.getRowIndex();
- }
}
diff --git a/processing/src/test/java/io/druid/segment/incremental/OffheapOakIncrementalIndexTest.java b/processing/src/test/java/io/druid/segment/incremental/OffheapOakIncrementalIndexTest.java
new file mode 100644
index 00000000000..6c508b58431
--- /dev/null
+++ b/processing/src/test/java/io/druid/segment/incremental/OffheapOakIncrementalIndexTest.java
@@ -0,0 +1,323 @@
+/*
+ * Licensed to Metamarkets Group Inc. (Metamarkets) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Metamarkets 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 io.druid.segment.incremental;
+
+import com.google.common.collect.Maps;
+import io.druid.data.input.MapBasedInputRow;
+import io.druid.data.input.Row;
+import io.druid.data.input.impl.StringDimensionSchema;
+import io.druid.data.input.impl.DimensionSchema;
+import io.druid.data.input.impl.DimensionsSpec;
+import io.druid.java.util.common.logger.Logger;
+import io.druid.query.aggregation.AggregatorFactory;
+import io.druid.query.aggregation.FilteredAggregatorFactory;
+import io.druid.query.aggregation.CountAggregatorFactory;
+import io.druid.query.filter.SelectorDimFilter;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.ArrayList;
+import java.util.Map;
+
+import static io.druid.segment.incremental.IncrementalIndex.TimeAndDims;
+import java.util.function.Consumer;
+
+public class OffheapOakIncrementalIndexTest
+{
+
+ private static final Logger log = new Logger(OffheapOakIncrementalIndexTest.class);
+
+ @Test
+ public void testOffHeapOakIncrementalIndexBasics() throws Exception
+ {
+ OffheapOakIncrementalIndex index = getIndex();
+ MapBasedInputRow[] rows = new MapBasedInputRow[6];
+ long minTime = System.currentTimeMillis() - 1000 * rows.length;
+
+ // creating rows
+ rows[0] = toMapRow(minTime);
+ rows[1] = toMapRow(minTime + 1000, "StringA", "A");
+ rows[2] = toMapRow(minTime + 1000, "StringA", "B");
+ rows[3] = toMapRow(minTime + 4000, "StringA", "A", "StringB", "B", "StringC", "C", "StringD", "D");
+ rows[4] = toMapRow(minTime + 2000, "StringA", "A", "StringB", "B");
+ rows[5] = toMapRow(minTime + 3000, "StringD", "D");
+
+ for (int i = 0; i < rows.length; i++) {
+ for (int j = 0; j < 5; j++) {
+ index.add(rows[i]);
+ }
+ }
+
+ Assert.assertEquals(index.size(), rows.length);
+ Assert.assertEquals(index.getMinTimeMillis(), minTime);
+ Assert.assertEquals(index.getMaxTimeMillis(), minTime + 4000);
+ }
+
+ @Test
+ public void testOffHeapOakIncrementalIndexNoSchema() throws Exception
+ {
+ IncrementalIndex index = new OffheapOakIncrementalIndex.Builder()
+ .setSimpleTestingIndexSchema(new CountAggregatorFactory("cnt"))
+ .setMaxRowCount(1000)
+ .buildOffheapOak();
+
+ MapBasedInputRow[] rows = new MapBasedInputRow[6];
+ long minTime = System.currentTimeMillis() - 1000 * rows.length;
+
+ // creating rows
+ rows[0] = toMapRow(minTime);
+ rows[1] = toMapRow(minTime + 1000, "StringA", "A");
+ rows[2] = toMapRow(minTime + 1000, "StringA", "B");
+ rows[3] = toMapRow(minTime + 4000, "StringA", "A", "StringB", "B", "StringC", "C", "StringD", "D");
+ rows[4] = toMapRow(minTime + 2000, "StringA", "A", "StringB", "B");
+ rows[5] = toMapRow(minTime + 3000, "StringD", "D");
+
+ for (int i = 0; i < rows.length; i++) {
+ for (int j = 0; j < 5; j++) {
+ index.add(rows[i]);
+ }
+ }
+
+ Assert.assertEquals(index.size(), rows.length);
+ Assert.assertEquals(index.getMinTimeMillis(), minTime);
+ Assert.assertEquals(index.getMaxTimeMillis(), minTime + 4000);
+ Assert.assertEquals(index.getDimensionNames().size(), 4);
+ }
+
+ @Test
+ public void testOffHeapOakIncrementalIndexKeysIterator() throws Exception
+ {
+ OffheapOakIncrementalIndex index = getIndex();
+ MapBasedInputRow[] rows = new MapBasedInputRow[10];
+
+ long time = System.currentTimeMillis();
+
+ // creating rows
+ rows[0] = toMapRow(time - 5000);
+ rows[1] = toMapRow(time - 3000);
+ rows[2] = toMapRow(time - 4000, "StringA", "A");
+ rows[3] = toMapRow(time - 3000, "StringA", "ABC", "StringB", "B");
+ rows[4] = toMapRow(time - 3000, "StringD", "A", "StringC", "B");
+ rows[5] = toMapRow(time - 2000, "StringD", "D");
+ rows[6] = toMapRow(time - 1000, "StringA", "AA", "StringB", "BH", "StringC", "C", "StringD", "D");
+ rows[7] = toMapRow(time - 4000, "StringC", "C");
+ rows[8] = toMapRow(time - 1000, "StringA", "C", "StringB", "A", "StringC", "BGD", "StringD", "B");
+ rows[9] = toMapRow(time - 3000, "StringD", "A", "StringB", "B");
+
+ for (int j = 0; j < 5; j++) {
+ for (int i = 0; i < rows.length; i++) {
+ index.add(rows[i]);
+ }
+ }
+ Assert.assertEquals(index.size(), rows.length);
+
+ Iterable<TimeAndDims> keySet = index.keySet();
+ Consumer<TimeAndDims> keySetConsumer = new Consumer<TimeAndDims>() {
+
+ TimeAndDims prev = index.toTimeAndDims(toMapRow(time - 6000));
+
+ @Override
+ public void accept(TimeAndDims timeAndDims)
+ {
+ Assert.assertTrue(0 > index.dimsComparator().compare(prev, timeAndDims));
+ prev = timeAndDims;
+ }
+ };
+
+ keySet.forEach(keySetConsumer);
+ }
+
+ @Test
+ public void testOffHeapOakIncrementalIndexKeysTimeRangeIterable() throws Exception
+ {
+ OffheapOakIncrementalIndex index = getIndex();
+ MapBasedInputRow[] rows = new MapBasedInputRow[10];
+
+ long time = System.currentTimeMillis();
+
+ // creating rows
+ rows[0] = toMapRow(time - 5000);
+ rows[1] = toMapRow(time - 3000);
+ rows[2] = toMapRow(time - 4000, "StringA", "A");
+ rows[3] = toMapRow(time - 3000, "StringA", "ABC", "StringB", "B");
+ rows[4] = toMapRow(time - 3000, "StringD", "A", "StringC", "B");
+ rows[5] = toMapRow(time - 2000, "StringD", "D");
+ rows[6] = toMapRow(time - 1000, "StringA", "AA", "StringB", "BH", "StringC", "C", "StringD", "D");
+ rows[7] = toMapRow(time - 3000, "StringB", "ABC", "StringA", "B");
+ rows[8] = toMapRow(time - 1000, "StringA", "C", "StringB", "A", "StringC", "BGD", "StringD", "B");
+ rows[9] = toMapRow(time - 3000, "StringD", "A", "StringB", "B");
+
+ for (int j = 0; j < 5; j++) {
+ for (int i = 0; i < rows.length; i++) {
+ index.add(rows[i]);
+ }
+ }
+
+ Iterable<TimeAndDims> timeRangeIterable;
+ Consumer<TimeAndDims> timeRangeConsumer;
+
+ // An ascending iterator
+ timeRangeIterable = index.timeRangeIterable(false, time - 5000, time - 2000);
+ timeRangeConsumer = new Consumer<TimeAndDims>() {
+
+ TimeAndDims prev = index.toTimeAndDims(toMapRow(time - 6000));
+
+ @Override
+ public void accept(TimeAndDims timeAndDims)
+ {
+ Assert.assertTrue(0 > index.dimsComparator().compare(prev, timeAndDims));
+ Assert.assertTrue(time - 1999 > timeAndDims.getTimestamp());
+ Assert.assertTrue(time - 5001 < timeAndDims.getTimestamp());
+ prev = timeAndDims;
+ }
+ };
+
+ timeRangeIterable.forEach(timeRangeConsumer);
+
+ // A descending iterator
+ timeRangeIterable = index.timeRangeIterable(true, time - 5000, time - 2000);
+ timeRangeConsumer = new Consumer<TimeAndDims>() {
+
+ TimeAndDims prev = index.toTimeAndDims(toMapRow(time - 1000));
+
+ @Override
+ public void accept(TimeAndDims timeAndDims)
+ {
+ Assert.assertTrue(0 < index.dimsComparator().compare(prev, timeAndDims));
+ Assert.assertTrue(time - 1999 > timeAndDims.getTimestamp());
+ Assert.assertTrue(time - 5001 < timeAndDims.getTimestamp());
+ prev = timeAndDims;
+ }
+ };
+
+ timeRangeIterable.forEach(timeRangeConsumer);
+
+ }
+
+ @Test
+ public void testOffHeapOakIncrementalIndexAggs() throws Exception
+ {
+ OffheapOakIncrementalIndex index = getIndex();
+ MapBasedInputRow[] rows = new MapBasedInputRow[10];
+ int insertionTrials = 5;
+
+ long time = System.currentTimeMillis();
+
+ // creating rows
+ rows[0] = toMapRow(time - 5000);
+ rows[1] = toMapRow(time - 3000);
+ rows[2] = toMapRow(time - 4000, "StringA", "A");
+ rows[3] = toMapRow(time - 3000, "StringA", "ABC", "StringB", "B");
+ rows[4] = toMapRow(time - 3000, "StringD", "A", "StringC", "B");
+ rows[5] = toMapRow(time - 2000, "StringD", "D");
+ rows[6] = toMapRow(time - 1000, "StringA", "AA", "StringB", "BH", "StringC", "C", "StringD", "D");
+ rows[7] = toMapRow(time - 3000, "StringB", "ABC", "StringA", "B");
+ rows[8] = toMapRow(time - 1000, "StringA", "C", "StringB", "A", "StringC", "BGD", "StringD", "B");
+ rows[9] = toMapRow(time - 3000, "StringD", "A", "StringB", "B");
+
+ for (int j = 0; j < insertionTrials; j++) {
+ for (int i = 0; i < rows.length; i++) {
+ index.add(rows[i]);
+ }
+ }
+
+ Iterable<Row> iterable = index.iterableWithPostAggregations(null, false);
+ Consumer<Row> rowConsumer = new Consumer<Row>() {
+
+ @Override
+ public void accept(Row row)
+ {
+ // insertion trials counters
+ long count = Long.valueOf(row.getDimension("Count").get(0));
+ long countA = Long.valueOf(row.getDimension("CountStringA=A").get(0));
+ long countB = Long.valueOf(row.getDimension("CountStringB=B").get(0));
+ long countC = Long.valueOf(row.getDimension("CountStringC=C").get(0));
+ long countD = Long.valueOf(row.getDimension("CountStringD=D").get(0));
+
+ Assert.assertEquals(insertionTrials, count);
+ Assert.assertEquals(countA, row.getDimension("StringA").size() > 0 && row.getDimension("StringA").get(0) == "A" ? insertionTrials : 0);
+ Assert.assertEquals(countB, row.getDimension("StringB").size() > 0 && row.getDimension("StringB").get(0) == "B" ? insertionTrials : 0);
+ Assert.assertEquals(countC, row.getDimension("StringC").size() > 0 && row.getDimension("StringC").get(0) == "C" ? insertionTrials : 0);
+ Assert.assertEquals(countD, row.getDimension("StringD").size() > 0 && row.getDimension("StringD").get(0) == "D" ? insertionTrials : 0);
+ }
+ };
+
+ iterable.forEach(rowConsumer);
+ }
+
+ private OffheapOakIncrementalIndex getIndex()
+ {
+ DimensionsSpec dimensions = new DimensionsSpec(
+ Arrays.<DimensionSchema>asList(
+ new StringDimensionSchema("StringA"),
+ new StringDimensionSchema("StringB"),
+ new StringDimensionSchema("StringC"),
+ new StringDimensionSchema("StringD")
+ ), null, null
+ );
+
+ AggregatorFactory[] metrics = {
+ new CountAggregatorFactory("Count"),
+ new FilteredAggregatorFactory(
+ new CountAggregatorFactory("CountStringA=A"),
+ new SelectorDimFilter("StringA", "A", null)
+ ),
+ new FilteredAggregatorFactory(
+ new CountAggregatorFactory("CountStringB=B"),
+ new SelectorDimFilter("StringB", "B", null)
+ ),
+ new FilteredAggregatorFactory(
+ new CountAggregatorFactory("CountStringC=C"),
+ new SelectorDimFilter("StringC", "C", null)
+ ),
+ new FilteredAggregatorFactory(
+ new CountAggregatorFactory("CountStringD=D"),
+ new SelectorDimFilter("StringD", "D", null)
+ )
+ };
+
+ final IncrementalIndexSchema schema = new IncrementalIndexSchema.Builder()
+ .withDimensionsSpec(dimensions)
+ .withMetrics(metrics)
+ .build();
+
+ OffheapOakIncrementalIndex index = (OffheapOakIncrementalIndex) new IncrementalIndex.Builder()
+ .setIndexSchema(schema)
+ .setDeserializeComplexMetrics(false)
+ .setMaxRowCount(1000)
+ .buildOffheapOak();
+
+ return index;
+ }
+
+ private MapBasedInputRow toMapRow(long time, Object... dimAndVal)
+ {
+ Map<String, Object> data = Maps.newHashMap();
+ List<String> dims = new ArrayList<>();
+ for (int i = 0; i < dimAndVal.length; i += 2) {
+ data.put((String) dimAndVal[i], dimAndVal[i + 1]);
+ dims.add((String) dimAndVal[i]);
+ }
+
+ return new MapBasedInputRow(time, dims, data);
+ }
+}
diff --git a/processing/src/test/java/io/druid/segment/incremental/OnheapIncrementalIndexTest.java b/processing/src/test/java/io/druid/segment/incremental/OnheapIncrementalIndexTest.java
index 4021f1e3025..3274693507b 100644
--- a/processing/src/test/java/io/druid/segment/incremental/OnheapIncrementalIndexTest.java
+++ b/processing/src/test/java/io/druid/segment/incremental/OnheapIncrementalIndexTest.java
@@ -84,8 +84,10 @@ public void run()
public void run()
{
while (!Thread.interrupted()) {
- for (IncrementalIndex.TimeAndDims row : index.getFacts().keySet()) {
- if (index.getMetricLongValue(row.getRowIndex(), 0) != 1) {
+ Iterable<io.druid.segment.incremental.IncrementalIndex.TimeAndDims> iterKeySet =
+ index.keySet();
+ for (io.druid.segment.incremental.IncrementalIndex.TimeAndDims row : iterKeySet) {
+ if (index.getMetricLongValue(row, 0) != 1) {
checkFailedCount.addAndGet(1);
}
}
diff --git a/processing/src/test/java/io/druid/segment/incremental/TimeAndDimsCompTest.java b/processing/src/test/java/io/druid/segment/incremental/TimeAndDimsCompTest.java
index 8677dc18628..6bf1c39f4ab 100644
--- a/processing/src/test/java/io/druid/segment/incremental/TimeAndDimsCompTest.java
+++ b/processing/src/test/java/io/druid/segment/incremental/TimeAndDimsCompTest.java
@@ -26,6 +26,7 @@
import org.junit.Assert;
import org.junit.Test;
+import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Comparator;
import java.util.Map;
@@ -76,6 +77,85 @@ public void testBasic() throws IndexSizeExceededException
Assert.assertTrue(comparator.compare(td5, td4) > 0);
}
+ @Test
+ public void testTimeAndDimsSerialization() throws IndexSizeExceededException
+ {
+ IncrementalIndex index = new IncrementalIndex.Builder()
+ .setSimpleTestingIndexSchema(new CountAggregatorFactory("cnt"))
+ .setMaxRowCount(1000)
+ .buildOffheapOak();
+
+ OffheapOakIncrementalIndex oakIndex = (OffheapOakIncrementalIndex) index;
+
+ long time = System.currentTimeMillis();
+ TimeAndDims[] origTimeAndDims = new TimeAndDims[6];
+ ByteBuffer[] serializedTimeAndDims = new ByteBuffer[6];
+ TimeAndDims[] deserializedTimeAndDims = new TimeAndDims[6];
+
+ Comparator<TimeAndDims> comparator = index.dimsComparator();
+
+ origTimeAndDims[0] = index.toTimeAndDims(toMapRow(time, "billy", "A", "joe", "B"));
+ origTimeAndDims[1] = index.toTimeAndDims(toMapRow(time, "billy", "A", "joe", "A"));
+ origTimeAndDims[2] = index.toTimeAndDims(toMapRow(time, "billy", "A"));
+ origTimeAndDims[3] = index.toTimeAndDims(toMapRow(time + 1, "billy", "A", "joe", "B"));
+ origTimeAndDims[4] = index.toTimeAndDims(toMapRow(time + 1, "billy", "A", "joe", "A,B"));
+ origTimeAndDims[5] = index.toTimeAndDims(toMapRow(time + 1));
+
+ for (int i = 0; i < 6; i++) {
+ serializedTimeAndDims[i] = oakIndex.timeAndDimsSerialization(origTimeAndDims[i]);
+ deserializedTimeAndDims[i] = oakIndex.timeAndDimsDeserialization(serializedTimeAndDims[i]);
+ Assert.assertEquals(0, comparator.compare(origTimeAndDims[i], deserializedTimeAndDims[i]));
+ }
+ }
+
+ @Test
+ public void testTimeAndDimsByteBufferComparator() throws IndexSizeExceededException
+ {
+ IncrementalIndex index = new OffheapOakIncrementalIndex.Builder()
+ .setSimpleTestingIndexSchema(new CountAggregatorFactory("cnt"))
+ .setMaxRowCount(1000)
+ .buildOffheapOak();
+
+ OffheapOakIncrementalIndex oakIndex = (OffheapOakIncrementalIndex) index;
+
+ long time = System.currentTimeMillis();
+
+ TimeAndDims[] timeAndDimsArray = new TimeAndDims[6];
+ timeAndDimsArray[0] = index.toTimeAndDims(toMapRow(time, "billy", "A", "joe", "B"));
+ timeAndDimsArray[1] = index.toTimeAndDims(toMapRow(time, "billy", "A", "joe", "A"));
+ timeAndDimsArray[2] = index.toTimeAndDims(toMapRow(time, "billy", "A"));
+ timeAndDimsArray[3] = index.toTimeAndDims(toMapRow(time + 1, "billy", "A", "joe", "B"));
+ timeAndDimsArray[4] = index.toTimeAndDims(toMapRow(time + 1, "billy", "A", "joe", "B,A"));
+ timeAndDimsArray[5] = index.toTimeAndDims(toMapRow(time + 1));
+
+ ByteBuffer[] timeAndDimsByteBufferArray = new ByteBuffer[6];
+ for (int i = 0; i < 6; i++) {
+ timeAndDimsByteBufferArray[i] = oakIndex.timeAndDimsSerialization(timeAndDimsArray[i]);
+ }
+
+ Comparator<ByteBuffer> comparator = oakIndex.dimsByteBufferComparator();
+
+ Assert.assertEquals(0, comparator.compare(timeAndDimsByteBufferArray[0], timeAndDimsByteBufferArray[0]));
+ Assert.assertEquals(0, comparator.compare(timeAndDimsByteBufferArray[1], timeAndDimsByteBufferArray[1]));
+ Assert.assertEquals(0, comparator.compare(timeAndDimsByteBufferArray[2], timeAndDimsByteBufferArray[2]));
+
+ Assert.assertTrue(comparator.compare(timeAndDimsByteBufferArray[0], timeAndDimsByteBufferArray[1]) > 0);
+ Assert.assertTrue(comparator.compare(timeAndDimsByteBufferArray[1], timeAndDimsByteBufferArray[0]) < 0);
+ Assert.assertTrue(comparator.compare(timeAndDimsByteBufferArray[1], timeAndDimsByteBufferArray[2]) > 0);
+ Assert.assertTrue(comparator.compare(timeAndDimsByteBufferArray[2], timeAndDimsByteBufferArray[1]) < 0);
+ Assert.assertTrue(comparator.compare(timeAndDimsByteBufferArray[0], timeAndDimsByteBufferArray[2]) > 0);
+ Assert.assertTrue(comparator.compare(timeAndDimsByteBufferArray[2], timeAndDimsByteBufferArray[0]) < 0);
+
+ Assert.assertTrue(comparator.compare(timeAndDimsByteBufferArray[5], timeAndDimsByteBufferArray[0]) > 0);
+ Assert.assertTrue(comparator.compare(timeAndDimsByteBufferArray[5], timeAndDimsByteBufferArray[1]) > 0);
+ Assert.assertTrue(comparator.compare(timeAndDimsByteBufferArray[5], timeAndDimsByteBufferArray[2]) > 0);
+
+ Assert.assertTrue(comparator.compare(timeAndDimsByteBufferArray[3], timeAndDimsByteBufferArray[5]) > 0);
+ Assert.assertTrue(comparator.compare(timeAndDimsByteBufferArray[4], timeAndDimsByteBufferArray[5]) > 0);
+ Assert.assertTrue(comparator.compare(timeAndDimsByteBufferArray[3], timeAndDimsByteBufferArray[4]) < 0);
+ Assert.assertTrue(comparator.compare(timeAndDimsByteBufferArray[4], timeAndDimsByteBufferArray[3]) > 0);
+ }
+
private MapBasedInputRow toMapRow(long time, Object... dimAndVal)
{
Map<String, Object> data = Maps.newHashMap();
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org