Skip to content

Commit 77e6f2c

Browse files
[CI] Auto commit changes from spotless
1 parent 8fed649 commit 77e6f2c

File tree

6 files changed

+93
-83
lines changed

6 files changed

+93
-83
lines changed

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/ChangePointOperator.java

Lines changed: 18 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -38,7 +38,9 @@ public class ChangePointOperator implements Operator {
3838

3939
public static final int INPUT_VALUE_COUNT_LIMIT = 1000;
4040

41-
public record Factory(int metricChannel, List<Integer> partitionChannel, String sourceText, int sourceLine, int sourceColumn) implements OperatorFactory {
41+
public record Factory(int metricChannel, List<Integer> partitionChannel, String sourceText, int sourceLine, int sourceColumn)
42+
implements
43+
OperatorFactory {
4244
@Override
4345
public Operator get(DriverContext driverContext) {
4446
return new ChangePointOperator(driverContext, metricChannel, partitionChannel, sourceText, sourceLine, sourceColumn);
@@ -64,7 +66,14 @@ public String describe() {
6466

6567
// TODO: make org.elasticsearch.xpack.esql.core.tree.Source available here
6668
// (by modularizing esql-core) and use that instead of the individual fields.
67-
public ChangePointOperator(DriverContext driverContext, int metricChannel, List<Integer> partitionChannel, String sourceText, int sourceLine, int sourceColumn) {
69+
public ChangePointOperator(
70+
DriverContext driverContext,
71+
int metricChannel,
72+
List<Integer> partitionChannel,
73+
String sourceText,
74+
int sourceLine,
75+
int sourceColumn
76+
) {
6877
this.driverContext = driverContext;
6978
this.metricChannel = metricChannel;
7079
this.partitionChannel = partitionChannel;
@@ -128,7 +137,6 @@ private void createOutputPages() {
128137
}
129138
boolean tooManyValues = maxValuesCount > INPUT_VALUE_COUNT_LIMIT;
130139

131-
132140
List<MlAggsHelper.DoubleBucketValues> bucketValuesPerPartition = new ArrayList<>();
133141
boolean hasNulls = false;
134142
boolean hasMultivalued = false;
@@ -168,14 +176,14 @@ private void createOutputPages() {
168176
}
169177
}
170178
// Handle last partition separately
171-
// if (lastPartitionFieldValue != null) {
179+
// if (lastPartitionFieldValue != null) {
172180
MlAggsHelper.DoubleBucketValues bucketValues = new MlAggsHelper.DoubleBucketValues(
173181
null,
174182
values.stream().mapToDouble(Double::doubleValue).toArray(),
175183
bucketIndexes.stream().mapToInt(Integer::intValue).toArray()
176184
);
177185
bucketValuesPerPartition.add(bucketValues);
178-
// }
186+
// }
179187
}
180188

181189
List<ChangeType> changeTypes = new ArrayList<>();
@@ -237,7 +245,9 @@ private void insertChangePoints(Iterable<ChangeType> changeTypes) {
237245

238246
try {
239247
// TODO: How to handle case when there are no change points
240-
if (changeType != null && pageStartIndex <= changeType.changePoint() && changeType.changePoint() < pageStartIndex + inputPage.getPositionCount()) {
248+
if (changeType != null
249+
&& pageStartIndex <= changeType.changePoint()
250+
&& changeType.changePoint() < pageStartIndex + inputPage.getPositionCount()) {
241251
try (
242252
BytesRefBlock.Builder changeTypeBlockBuilder = blockFactory.newBytesRefBlockBuilder(inputPage.getPositionCount());
243253
DoubleBlock.Builder pvalueBlockBuilder = blockFactory.newDoubleBlockBuilder(inputPage.getPositionCount())
@@ -258,7 +268,7 @@ private void insertChangePoints(Iterable<ChangeType> changeTypes) {
258268
changeTypeBlock = blockFactory.newConstantNullBlock(inputPage.getPositionCount());
259269
changePvalueBlock = blockFactory.newConstantNullBlock(inputPage.getPositionCount());
260270
}
261-
outputPage = inputPage.appendBlocks(new Block[]{changeTypeBlock, changePvalueBlock});
271+
outputPage = inputPage.appendBlocks(new Block[] { changeTypeBlock, changePvalueBlock });
262272
if (pageStartIndex + inputPage.getPositionCount() > INPUT_VALUE_COUNT_LIMIT) {
263273
outputPage = outputPage.subPage(0, INPUT_VALUE_COUNT_LIMIT - pageStartIndex);
264274
}
@@ -310,7 +320,7 @@ private static String describe(int metricChannel, List<Integer> partitionChannel
310320
+ metricChannel
311321
+ ", partitionChannels="
312322
+ partitionChannel.stream().map(c -> c.toString()).collect(Collectors.joining(",", "[", "]"))
313-
+ "]";
323+
+ "]";
314324
}
315325

316326
private Warnings warnings(boolean onlyWarnings) {

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/topn/TopNOperator.java

Lines changed: 68 additions & 66 deletions
Original file line numberDiff line numberDiff line change
@@ -453,78 +453,78 @@ private Iterator<Page> toPages() {
453453
// TODO: optimize case where all the queues are empty
454454
try {
455455
for (var entry : inputQueues.entrySet()) {
456-
Queue inputQueue = entry.getValue();
456+
Queue inputQueue = entry.getValue();
457457

458-
list = new ArrayList<>(inputQueue.size());
459-
builders = null;
460-
while (inputQueue.size() > 0) {
461-
list.add(inputQueue.pop());
462-
}
463-
Collections.reverse(list);
464-
465-
int p = 0;
466-
int size = 0;
467-
for (int i = 0; i < list.size(); i++) {
468-
if (builders == null) {
469-
size = Math.min(maxPageSize, list.size() - i);
470-
builders = new ResultBuilder[elementTypes.size()];
471-
for (int b = 0; b < builders.length; b++) {
472-
builders[b] = ResultBuilder.resultBuilderFor(
473-
blockFactory,
474-
elementTypes.get(b),
475-
encoders.get(b).toUnsortable(),
476-
channelInKey(sortOrders, b),
477-
size
478-
);
479-
}
480-
p = 0;
458+
list = new ArrayList<>(inputQueue.size());
459+
builders = null;
460+
while (inputQueue.size() > 0) {
461+
list.add(inputQueue.pop());
481462
}
463+
Collections.reverse(list);
464+
465+
int p = 0;
466+
int size = 0;
467+
for (int i = 0; i < list.size(); i++) {
468+
if (builders == null) {
469+
size = Math.min(maxPageSize, list.size() - i);
470+
builders = new ResultBuilder[elementTypes.size()];
471+
for (int b = 0; b < builders.length; b++) {
472+
builders[b] = ResultBuilder.resultBuilderFor(
473+
blockFactory,
474+
elementTypes.get(b),
475+
encoders.get(b).toUnsortable(),
476+
channelInKey(sortOrders, b),
477+
size
478+
);
479+
}
480+
p = 0;
481+
}
482482

483-
Row row = list.get(i);
484-
BytesRef keys = row.keys.bytesRefView();
485-
for (SortOrder so : sortOrders) {
486-
if (keys.bytes[keys.offset] == so.nul()) {
483+
Row row = list.get(i);
484+
BytesRef keys = row.keys.bytesRefView();
485+
for (SortOrder so : sortOrders) {
486+
if (keys.bytes[keys.offset] == so.nul()) {
487+
keys.offset++;
488+
keys.length--;
489+
continue;
490+
}
487491
keys.offset++;
488492
keys.length--;
489-
continue;
493+
builders[so.channel].decodeKey(keys);
494+
}
495+
if (keys.length != 0) {
496+
throw new IllegalArgumentException("didn't read all keys");
490497
}
491-
keys.offset++;
492-
keys.length--;
493-
builders[so.channel].decodeKey(keys);
494-
}
495-
if (keys.length != 0) {
496-
throw new IllegalArgumentException("didn't read all keys");
497-
}
498-
499-
BytesRef values = row.values.bytesRefView();
500-
for (ResultBuilder builder : builders) {
501-
builder.decodeValue(values);
502-
}
503-
if (values.length != 0) {
504-
throw new IllegalArgumentException("didn't read all values");
505-
}
506498

507-
list.set(i, null);
508-
row.close();
499+
BytesRef values = row.values.bytesRefView();
500+
for (ResultBuilder builder : builders) {
501+
builder.decodeValue(values);
502+
}
503+
if (values.length != 0) {
504+
throw new IllegalArgumentException("didn't read all values");
505+
}
509506

510-
p++;
511-
if (p == size) {
512-
Block[] blocks = new Block[builders.length];
513-
try {
514-
for (int b = 0; b < blocks.length; b++) {
515-
blocks[b] = builders[b].build();
516-
}
517-
} finally {
518-
if (blocks[blocks.length - 1] == null) {
519-
Releasables.closeExpectNoException(blocks);
507+
list.set(i, null);
508+
row.close();
509+
510+
p++;
511+
if (p == size) {
512+
Block[] blocks = new Block[builders.length];
513+
try {
514+
for (int b = 0; b < blocks.length; b++) {
515+
blocks[b] = builders[b].build();
516+
}
517+
} finally {
518+
if (blocks[blocks.length - 1] == null) {
519+
Releasables.closeExpectNoException(blocks);
520+
}
520521
}
522+
result.add(new Page(blocks));
523+
Releasables.closeExpectNoException(builders);
524+
builders = null;
521525
}
522-
result.add(new Page(blocks));
523-
Releasables.closeExpectNoException(builders);
524-
builders = null;
525526
}
526-
}
527-
assert builders == null;
527+
assert builders == null;
528528
}
529529
success = true;
530530
return result.iterator();
@@ -578,9 +578,8 @@ public void close() {
578578
Releasables.closeExpectNoException(spare, Releasables.wrap(releasables));
579579
}
580580

581-
private static long SHALLOW_SIZE = RamUsageEstimator.shallowSizeOfInstance(TopNOperator.class)
582-
+ RamUsageEstimator.shallowSizeOfInstance(List.class) * 4
583-
+ RamUsageEstimator.shallowSizeOfInstance(Map.class);
581+
private static long SHALLOW_SIZE = RamUsageEstimator.shallowSizeOfInstance(TopNOperator.class) + RamUsageEstimator
582+
.shallowSizeOfInstance(List.class) * 4 + RamUsageEstimator.shallowSizeOfInstance(Map.class);
584583

585584
@Override
586585
public long ramBytesUsed() {
@@ -595,7 +594,8 @@ public long ramBytesUsed() {
595594
size += partitions.size() * Partition.SHALLOW_SIZE;
596595
size += RamUsageEstimator.alignObjectSize(arrHeader + ref * sortOrders.size());
597596
size += sortOrders.size() * SortOrder.SHALLOW_SIZE;
598-
long ramBytesUsedSum = inputQueues.entrySet().stream()
597+
long ramBytesUsedSum = inputQueues.entrySet()
598+
.stream()
599599
.mapToLong(e -> e.getKey().getBytes().length + e.getValue().ramBytesUsed())
600600
.sum();
601601
size += ramBytesUsedSum;
@@ -612,7 +612,9 @@ public Status status() {
612612
public String toString() {
613613
int queueSizeSum = inputQueues.values().stream().mapToInt(Queue::size).sum();
614614
return "TopNOperator[count="
615-
+ queueSizeSum + "/" + topCount
615+
+ queueSizeSum
616+
+ "/"
617+
+ topCount
616618
+ ", elementTypes="
617619
+ elementTypes
618620
+ ", encoders="

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/ChangePointByOperatorTests.java

Lines changed: 7 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@
1919

2020
import java.util.ArrayList;
2121
import java.util.List;
22-
import java.util.Optional;
2322

2423
import static org.hamcrest.Matchers.equalTo;
2524
import static org.hamcrest.Matchers.hasSize;
@@ -74,17 +73,19 @@ protected void assertSimpleOutput(List<Page> input, List<Page> output) {
7473
}
7574
assertThat(
7675
actualChangePoints,
77-
equalTo(List.of(
78-
Tuple.tuple(rowCount / 6, "step_change"),
79-
Tuple.tuple(rowCount / 2, "step_change"),
80-
Tuple.tuple(8 * rowCount / 9, "spike"))
76+
equalTo(
77+
List.of(
78+
Tuple.tuple(rowCount / 6, "step_change"),
79+
Tuple.tuple(rowCount / 2, "step_change"),
80+
Tuple.tuple(8 * rowCount / 9, "spike")
81+
)
8182
)
8283
);
8384
}
8485

8586
@Override
8687
protected Operator.OperatorFactory simple(SimpleOptions options) {
87-
return new ChangePointOperator.Factory(0, List.of(1), null,0, 0);
88+
return new ChangePointOperator.Factory(0, List.of(1), null, 0, 0);
8889
}
8990

9091
@Override

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/ChangePointOperatorTests.java

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@
1919

2020
import java.util.ArrayList;
2121
import java.util.List;
22-
import java.util.Optional;
2322

2423
import static org.hamcrest.Matchers.equalTo;
2524
import static org.hamcrest.Matchers.hasSize;

x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/topn/TopNOperatorTests.java

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -52,7 +52,6 @@
5252
import java.util.LinkedHashSet;
5353
import java.util.List;
5454
import java.util.Map;
55-
import java.util.Optional;
5655
import java.util.Set;
5756
import java.util.function.Function;
5857
import java.util.stream.Collectors;

x-pack/plugin/esql/compute/test/src/main/java/org/elasticsearch/compute/test/AbstractBlockSourceOperator.java

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,6 @@
1010
import org.elasticsearch.compute.data.BlockFactory;
1111
import org.elasticsearch.compute.data.Page;
1212
import org.elasticsearch.compute.operator.SourceOperator;
13-
import org.elasticsearch.test.ESTestCase;
1413

1514
/**
1615
* An abstract source operator. Implementations of this operator produce pages with a random

0 commit comments

Comments
 (0)