Skip to content

Commit

Permalink
Star Tree Implementation [OnHeap] (opensearch-project#14512)
Browse files Browse the repository at this point in the history
---------
Signed-off-by: Sarthak Aggarwal <sarthagg@amazon.com>
  • Loading branch information
sarthakaggarwal97 authored and wangdongyu.danny committed Aug 22, 2024
1 parent 96e98e0 commit 6914036
Show file tree
Hide file tree
Showing 32 changed files with 3,586 additions and 37 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
import org.apache.lucene.index.MergeState;
import org.apache.lucene.index.SegmentWriteState;
import org.opensearch.common.annotation.ExperimentalApi;
import org.opensearch.index.compositeindex.datacube.startree.builder.StarTreesBuilder;
import org.opensearch.index.mapper.CompositeMappedFieldType;
import org.opensearch.index.mapper.MapperService;
import org.opensearch.index.mapper.StarTreeMapper;
Expand Down Expand Up @@ -98,7 +99,9 @@ private void createCompositeIndicesIfPossible(DocValuesProducer valuesProducer,
if (compositeFieldSet.isEmpty()) {
for (CompositeMappedFieldType mappedType : compositeMappedFieldTypes) {
if (mappedType instanceof StarTreeMapper.StarTreeFieldType) {
// TODO : Call StarTree builder
try (StarTreesBuilder starTreesBuilder = new StarTreesBuilder(fieldProducerMap, state, mapperService)) {
starTreesBuilder.build();
}
}
}
}
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,34 @@
/*
* SPDX-License-Identifier: Apache-2.0
*
* The OpenSearch Contributors require contributions made to
* this file be licensed under the Apache-2.0 license or a
* compatible open source license.
*/

package org.opensearch.index.compositeindex.datacube.startree;

import org.opensearch.common.annotation.ExperimentalApi;

import java.util.Arrays;

/**
* Star tree document
*
* @opensearch.experimental
*/
@ExperimentalApi
public class StarTreeDocument {
public final Long[] dimensions;
public final Object[] metrics;

public StarTreeDocument(Long[] dimensions, Object[] metrics) {
this.dimensions = dimensions;
this.metrics = metrics;
}

@Override
public String toString() {
return Arrays.toString(dimensions) + " | " + Arrays.toString(metrics);
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,66 @@
/*
* SPDX-License-Identifier: Apache-2.0
*
* The OpenSearch Contributors require contributions made to
* this file be licensed under the Apache-2.0 license or a
* compatible open source license.
*/
package org.opensearch.index.compositeindex.datacube.startree.aggregators;

import org.opensearch.index.compositeindex.datacube.MetricStat;
import org.opensearch.index.compositeindex.datacube.startree.aggregators.numerictype.StarTreeNumericType;

/**
* Count value aggregator for star tree
*
* @opensearch.experimental
*/
public class CountValueAggregator implements ValueAggregator<Long> {
public static final StarTreeNumericType VALUE_AGGREGATOR_TYPE = StarTreeNumericType.LONG;
public static final long DEFAULT_INITIAL_VALUE = 1L;

@Override
public MetricStat getAggregationType() {
return MetricStat.COUNT;
}

@Override
public StarTreeNumericType getAggregatedValueType() {
return VALUE_AGGREGATOR_TYPE;
}

@Override
public Long getInitialAggregatedValueForSegmentDocValue(Long segmentDocValue, StarTreeNumericType starTreeNumericType) {
return DEFAULT_INITIAL_VALUE;
}

@Override
public Long mergeAggregatedValueAndSegmentValue(Long value, Long segmentDocValue, StarTreeNumericType starTreeNumericType) {
return value + 1;
}

@Override
public Long mergeAggregatedValues(Long value, Long aggregatedValue) {
return value + aggregatedValue;
}

@Override
public Long getInitialAggregatedValue(Long value) {
return value;
}

@Override
public int getMaxAggregatedValueByteSize() {
return Long.BYTES;
}

@Override
public Long toLongValue(Long value) {
return value;
}

@Override
public Long toStarTreeNumericTypeValue(Long value, StarTreeNumericType type) {
return value;
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,130 @@
/*
* SPDX-License-Identifier: Apache-2.0
*
* The OpenSearch Contributors require contributions made to
* this file be licensed under the Apache-2.0 license or a
* compatible open source license.
*/
package org.opensearch.index.compositeindex.datacube.startree.aggregators;

import org.opensearch.index.compositeindex.datacube.MetricStat;
import org.opensearch.index.compositeindex.datacube.startree.aggregators.numerictype.StarTreeNumericType;
import org.opensearch.index.compositeindex.datacube.startree.utils.SequentialDocValuesIterator;
import org.opensearch.index.fielddata.IndexNumericFieldData;

import java.util.Comparator;
import java.util.Objects;

/**
* Builds aggregation function and doc values field pair to support various aggregations
*
* @opensearch.experimental
*/
public class MetricAggregatorInfo implements Comparable<MetricAggregatorInfo> {

public static final String DELIMITER = "_";
private final String metric;
private final String starFieldName;
private final MetricStat metricStat;
private final String field;
private final ValueAggregator valueAggregators;
private final StarTreeNumericType starTreeNumericType;
private final SequentialDocValuesIterator metricStatReader;

/**
* Constructor for MetricAggregatorInfo
*/
public MetricAggregatorInfo(
MetricStat metricStat,
String field,
String starFieldName,
IndexNumericFieldData.NumericType numericType,
SequentialDocValuesIterator metricStatReader
) {
this.metricStat = metricStat;
this.valueAggregators = ValueAggregatorFactory.getValueAggregator(metricStat);
this.starTreeNumericType = StarTreeNumericType.fromNumericType(numericType);
this.metricStatReader = metricStatReader;
this.field = field;
this.starFieldName = starFieldName;
this.metric = toFieldName();
}

/**
* @return metric type
*/
public MetricStat getMetricStat() {
return metricStat;
}

/**
* @return field Name
*/
public String getField() {
return field;
}

/**
* @return the metric stat name
*/
public String getMetric() {
return metric;
}

/**
* @return aggregator for the field value
*/
public ValueAggregator getValueAggregators() {
return valueAggregators;
}

/**
* @return star tree aggregated value type
*/
public StarTreeNumericType getAggregatedValueType() {
return starTreeNumericType;
}

/**
* @return metric value reader iterator
*/
public SequentialDocValuesIterator getMetricStatReader() {
return metricStatReader;
}

/**
* @return field name with metric type and field
*/
public String toFieldName() {
return starFieldName + DELIMITER + field + DELIMITER + metricStat.getTypeName();
}

@Override
public int hashCode() {
return Objects.hashCode(toFieldName());
}

@Override
public boolean equals(Object obj) {
if (this == obj) {
return true;
}
if (obj instanceof MetricAggregatorInfo) {
MetricAggregatorInfo anotherPair = (MetricAggregatorInfo) obj;
return metricStat == anotherPair.metricStat && field.equals(anotherPair.field);
}
return false;
}

@Override
public String toString() {
return toFieldName();
}

@Override
public int compareTo(MetricAggregatorInfo other) {
return Comparator.comparing((MetricAggregatorInfo o) -> o.field)
.thenComparing((MetricAggregatorInfo o) -> o.metricStat)
.compare(this, other);
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,97 @@
/*
* SPDX-License-Identifier: Apache-2.0
*
* The OpenSearch Contributors require contributions made to
* this file be licensed under the Apache-2.0 license or a
* compatible open source license.
*/
package org.opensearch.index.compositeindex.datacube.startree.aggregators;

import org.apache.lucene.util.NumericUtils;
import org.opensearch.index.compositeindex.datacube.MetricStat;
import org.opensearch.index.compositeindex.datacube.startree.aggregators.numerictype.StarTreeNumericType;
import org.opensearch.search.aggregations.metrics.CompensatedSum;

/**
* Sum value aggregator for star tree
*
* @opensearch.experimental
*/
public class SumValueAggregator implements ValueAggregator<Double> {

public static final StarTreeNumericType VALUE_AGGREGATOR_TYPE = StarTreeNumericType.DOUBLE;
private double sum = 0;
private double compensation = 0;
private CompensatedSum kahanSummation = new CompensatedSum(0, 0);

@Override
public MetricStat getAggregationType() {
return MetricStat.SUM;
}

@Override
public StarTreeNumericType getAggregatedValueType() {
return VALUE_AGGREGATOR_TYPE;
}

@Override
public Double getInitialAggregatedValueForSegmentDocValue(Long segmentDocValue, StarTreeNumericType starTreeNumericType) {
kahanSummation.reset(0, 0);
kahanSummation.add(starTreeNumericType.getDoubleValue(segmentDocValue));
compensation = kahanSummation.delta();
sum = kahanSummation.value();
return kahanSummation.value();
}

@Override
public Double mergeAggregatedValueAndSegmentValue(Double value, Long segmentDocValue, StarTreeNumericType starTreeNumericType) {
assert kahanSummation.value() == value;
kahanSummation.reset(sum, compensation);
kahanSummation.add(starTreeNumericType.getDoubleValue(segmentDocValue));
compensation = kahanSummation.delta();
sum = kahanSummation.value();
return kahanSummation.value();
}

@Override
public Double mergeAggregatedValues(Double value, Double aggregatedValue) {
assert kahanSummation.value() == aggregatedValue;
kahanSummation.reset(sum, compensation);
kahanSummation.add(value);
compensation = kahanSummation.delta();
sum = kahanSummation.value();
return kahanSummation.value();
}

@Override
public Double getInitialAggregatedValue(Double value) {
kahanSummation.reset(0, 0);
kahanSummation.add(value);
compensation = kahanSummation.delta();
sum = kahanSummation.value();
return kahanSummation.value();
}

@Override
public int getMaxAggregatedValueByteSize() {
return Double.BYTES;
}

@Override
public Long toLongValue(Double value) {
try {
return NumericUtils.doubleToSortableLong(value);
} catch (Exception e) {
throw new IllegalStateException("Cannot convert " + value + " to sortable long", e);
}
}

@Override
public Double toStarTreeNumericTypeValue(Long value, StarTreeNumericType type) {
try {
return type.getDoubleValue(value);
} catch (Exception e) {
throw new IllegalStateException("Cannot convert " + value + " to sortable aggregation type", e);
}
}
}
Loading

0 comments on commit 6914036

Please sign in to comment.