Skip to content
Merged
Show file tree
Hide file tree
Changes from 10 commits
Commits
Show all changes
20 commits
Select commit Hold shift + click to select a range
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Loading
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Original file line number Diff line number Diff line change
Expand Up @@ -185,6 +185,7 @@ static TransportVersion def(int id) {
public static final TransportVersion ESQL_THREAD_NAME_IN_DRIVER_PROFILE = def(9_027_0_00);
public static final TransportVersion INFERENCE_CONTEXT = def(9_028_0_00);
public static final TransportVersion ML_INFERENCE_DEEPSEEK = def(9_029_00_0);
public static final TransportVersion ESQL_AGGREGATE_METRIC_DOUBLE_LITERAL = def(9_030_0_00);

/*
* STOP! READ THIS FIRST! No, really,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -557,7 +557,6 @@ public static boolean isRepresentable(DataType t) {
&& t != SOURCE
&& t != HALF_FLOAT
&& t != PARTIAL_AGG
&& t != AGGREGATE_METRIC_DOUBLE
&& t.isCounter() == false;
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -7,9 +7,17 @@

package org.elasticsearch.compute.data;

import org.elasticsearch.TransportVersion;
import org.elasticsearch.TransportVersions;
import org.elasticsearch.common.io.stream.GenericNamedWriteable;
import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.core.Releasables;
import org.elasticsearch.index.mapper.BlockLoader;

import java.io.IOException;

public class AggregateMetricDoubleBlockBuilder extends AbstractBlockBuilder implements BlockLoader.AggregateMetricDoubleBuilder {

private DoubleBlockBuilder minBuilder;
Expand Down Expand Up @@ -161,11 +169,107 @@ public String getLabel() {
}
}

public record AggregateMetricDoubleLiteral(Double min, Double max, Double sum, Integer count) {
public record AggregateMetricDoubleLiteral(Double min, Double max, Double sum, Integer count) implements GenericNamedWriteable {
public AggregateMetricDoubleLiteral {
min = min.isNaN() ? null : min;
max = max.isNaN() ? null : max;
sum = sum.isNaN() ? null : sum;
}

public static final NamedWriteableRegistry.Entry ENTRY = new NamedWriteableRegistry.Entry(
GenericNamedWriteable.class,
"AggregateMetricDoubleLiteral",
AggregateMetricDoubleLiteral::new
);

@Override
public String getWriteableName() {
return "AggregateMetricDoubleLiteral";
}

public AggregateMetricDoubleLiteral(StreamInput input) throws IOException {
this(input.readOptionalDouble(), input.readOptionalDouble(), input.readOptionalDouble(), input.readOptionalInt());
}

@Override
public void writeTo(StreamOutput out) throws IOException {
out.writeOptionalDouble(min);
out.writeOptionalDouble(max);
out.writeOptionalDouble(sum);
out.writeOptionalInt(count);
}

@Override
public TransportVersion getMinimalSupportedVersion() {
return TransportVersions.ESQL_AGGREGATE_METRIC_DOUBLE_LITERAL;
}

}

public static class AggregateMetricDoubleVectorBuilder extends AbstractBlockBuilder {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we not extend from BlockBuilder but just a Releasable here. Also should we move this to ToAggregateMetricDouble class instead?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah I think keeping it in ToAggregateMetricDouble makes sense, I moved it there, but that class is getting quite heavy, I wonder if the factories should be moved into a separate file?

private final DoubleBlockBuilder valuesBuilder;

public AggregateMetricDoubleVectorBuilder(int estimatedSize, BlockFactory blockFactory) {
super(blockFactory);
valuesBuilder = new DoubleBlockBuilder(estimatedSize, blockFactory);
}

@Override
protected int valuesLength() {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Are these methods not used?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

They weren't necessary at this stage (and they're still not implemented on the original AggregateMetricDoubleBlockBuilder) and I had wanted to get eyes on this approach so I left them like that; I actually have a subtask in the meta-issue for figuring out what to do with these methods

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since I changed it to no longer extend AbstractBlockBuilder these methods don't need to be implemented for this anymore (they're still unimplemented for AggregateMetricDoubleBlockBuilder though)

throw new UnsupportedOperationException("Not available on aggregate_metric_double");
}

@Override
protected void growValuesArray(int newSize) {
throw new UnsupportedOperationException("Not available on aggregate_metric_double");
}

@Override
protected int elementSize() {
throw new UnsupportedOperationException("Not available on aggregate_metric_double");
}

@Override
public Block.Builder copyFrom(Block block, int beginInclusive, int endExclusive) {
// TODO
return null;
}

@Override
public Block.Builder mvOrdering(Block.MvOrdering mvOrdering) {
// TODO
return null;
}

public void appendValue(double value) {
valuesBuilder.appendDouble(value);
}

@Override
public Block build() {
Block[] blocks = new Block[4];
Block block = null;
IntBlock countVector = null;
boolean success = false;
try {
finish();
block = valuesBuilder.build();
countVector = blockFactory.newConstantIntBlockWith(1, block.getPositionCount());
blocks[Metric.MIN.getIndex()] = block;
blocks[Metric.MAX.getIndex()] = block;
block.incRef();
blocks[Metric.SUM.getIndex()] = block;
block.incRef();
blocks[Metric.COUNT.getIndex()] = countVector;
CompositeBlock compositeBlock = new CompositeBlock(blocks);
success = true;
return compositeBlock;
} finally {
if (success == false) {
Releasables.closeExpectNoException(blocks);
}
}
}

}
}
Original file line number Diff line number Diff line change
Expand Up @@ -436,6 +436,10 @@ public AggregateMetricDoubleBlockBuilder newAggregateMetricDoubleBlockBuilder(in
return new AggregateMetricDoubleBlockBuilder(estimatedSize, this);
}

public AggregateMetricDoubleBlockBuilder.AggregateMetricDoubleVectorBuilder newAggregateMetricDoubleVectorBuilder(int estimatedSize) {
return new AggregateMetricDoubleBlockBuilder.AggregateMetricDoubleVectorBuilder(estimatedSize, this);
}

public final Block newConstantAggregateMetricDoubleBlock(
AggregateMetricDoubleBlockBuilder.AggregateMetricDoubleLiteral value,
int positions
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -285,7 +285,19 @@ private static Object valueAtOffset(Block block, int offset) {
DocVector v = ((DocBlock) block).asVector();
yield new Doc(v.shards().getInt(offset), v.segments().getInt(offset), v.docs().getInt(offset));
}
case COMPOSITE -> throw new IllegalArgumentException("can't read values from composite blocks");
case COMPOSITE -> {
CompositeBlock compositeBlock = (CompositeBlock) block;
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can in a follow up the usage of CompositeBlock for aggregate metric double field be replaced with AggregateMetricDoubleBlockBuilder?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm not sure I understand? There's no block creation here, just fetching the value from the block that already exists, so no building is involved

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sorry, this is comment doesn't make sense here. I think I was more wondering about whether in AggregateMetricDoubleBlockBuilder we should use build a specialized block instead of building CompositeBlock. ComposteBlock is very generic and using this here now I think is ok, I'm just wondering whether in the future this makes sense? (when there are more usages of CompositeBlock)

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We could if we wanted, but I'm not sure it's all that important. The entire compute engine is pretty ok with runtime types and treating blocks in specific ways based on the input. We don't have a special type for utf-8 text - just bytes - so I'm not sure we'd need a special type for these blocks.

If we want it we can have it for sure, but I don't think it's all that important.

var minBlock = (DoubleBlock) compositeBlock.getBlock(AggregateMetricDoubleBlockBuilder.Metric.MIN.getIndex());
var maxBlock = (DoubleBlock) compositeBlock.getBlock(AggregateMetricDoubleBlockBuilder.Metric.MAX.getIndex());
var sumBlock = (DoubleBlock) compositeBlock.getBlock(AggregateMetricDoubleBlockBuilder.Metric.SUM.getIndex());
var countBlock = (IntBlock) compositeBlock.getBlock(AggregateMetricDoubleBlockBuilder.Metric.COUNT.getIndex());
yield new AggregateMetricDoubleLiteral(
minBlock.getDouble(offset),
maxBlock.getDouble(offset),
sumBlock.getDouble(offset),
countBlock.getInt(offset)
);
}
case UNKNOWN -> throw new IllegalArgumentException("can't read values from [" + block + "]");
};
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import org.elasticsearch.common.regex.Regex;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.util.BigArrays;
import org.elasticsearch.compute.data.AggregateMetricDoubleBlockBuilder;
import org.elasticsearch.compute.data.BlockFactory;
import org.elasticsearch.compute.data.BlockUtils;
import org.elasticsearch.compute.data.BytesRefBlock;
Expand Down Expand Up @@ -788,6 +789,12 @@ public static Literal randomLiteral(DataType type) {
case CARTESIAN_POINT -> CARTESIAN.asWkb(ShapeTestUtils.randomPoint());
case GEO_SHAPE -> GEO.asWkb(GeometryTestUtils.randomGeometry(randomBoolean()));
case CARTESIAN_SHAPE -> CARTESIAN.asWkb(ShapeTestUtils.randomGeometry(randomBoolean()));
case AGGREGATE_METRIC_DOUBLE -> new AggregateMetricDoubleBlockBuilder.AggregateMetricDoubleLiteral(
randomDouble(),
randomDouble(),
randomDouble(),
randomInt()
);
case NULL -> null;
case SOURCE -> {
try {
Expand All @@ -798,8 +805,9 @@ public static Literal randomLiteral(DataType type) {
throw new UncheckedIOException(e);
}
}
case UNSUPPORTED, OBJECT, DOC_DATA_TYPE, TSID_DATA_TYPE, PARTIAL_AGG, AGGREGATE_METRIC_DOUBLE ->
throw new IllegalArgumentException("can't make random values for [" + type.typeName() + "]");
case UNSUPPORTED, OBJECT, DOC_DATA_TYPE, TSID_DATA_TYPE, PARTIAL_AGG -> throw new IllegalArgumentException(
"can't make random values for [" + type.typeName() + "]"
);
}, type);
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -880,7 +880,12 @@ public enum Cap {
/**
* Do {@code TO_LOWER} and {@code TO_UPPER} process all field values?
*/
TO_LOWER_MV;
TO_LOWER_MV,

/**
* Support for to_aggregate_metric_double function
*/
AGGREGATE_METRIC_DOUBLE_CONVERT_TO(AGGREGATE_METRIC_DOUBLE_FEATURE_FLAG);

private final boolean enabled;

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
import org.elasticsearch.xpack.esql.expression.function.fulltext.FullTextWritables;
import org.elasticsearch.xpack.esql.expression.function.scalar.ScalarFunctionWritables;
import org.elasticsearch.xpack.esql.expression.function.scalar.convert.FromBase64;
import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToAggregateMetricDouble;
import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToBase64;
import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToBoolean;
import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToCartesianPoint;
Expand Down Expand Up @@ -180,6 +181,7 @@ public static List<NamedWriteableRegistry.Entry> unaryScalars() {
entries.add(StY.ENTRY);
entries.add(Tan.ENTRY);
entries.add(Tanh.ENTRY);
entries.add(ToAggregateMetricDouble.ENTRY);
entries.add(ToBase64.ENTRY);
entries.add(ToBoolean.ENTRY);
entries.add(ToCartesianPoint.ENTRY);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@
import org.elasticsearch.xpack.esql.expression.function.scalar.conditional.Greatest;
import org.elasticsearch.xpack.esql.expression.function.scalar.conditional.Least;
import org.elasticsearch.xpack.esql.expression.function.scalar.convert.FromBase64;
import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToAggregateMetricDouble;
import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToBase64;
import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToBoolean;
import org.elasticsearch.xpack.esql.expression.function.scalar.convert.ToCartesianPoint;
Expand Down Expand Up @@ -376,6 +377,7 @@ private static FunctionDefinition[][] functions() {
// conversion functions
new FunctionDefinition[] {
def(FromBase64.class, FromBase64::new, "from_base64"),
def(ToAggregateMetricDouble.class, ToAggregateMetricDouble::new, "to_aggregate_metric_double", "to_aggregatemetricdouble"),
def(ToBase64.class, ToBase64::new, "to_base64"),
def(ToBoolean.class, ToBoolean::new, "to_boolean", "to_bool"),
def(ToCartesianPoint.class, ToCartesianPoint::new, "to_cartesianpoint"),
Expand Down
Loading