Skip to content

Commit 6e46b81

Browse files
nik9000nik9000
committed
ESQL: Split large pages on load sometimes (#131053)
This adds support for splitting `Page`s of large values when loading from single segment, non-descending hits. This is hottest code path as it's how we load data for aggregation. So! We had to make very very very sure this doesn't slow down the fast path of loading doc values. Caveat - this only defends against loading large values via the row-by-row load mechanism that we use for stored fields and _source. That covers the most common kinds of large values - mostly `text` and geo fields. If we need to split further on docs values, we'll have to invent something for them specifically. For now, just row-by-row. This works by flipping the order in which we load row-by-row and column-at-a-time values. Previously we loaded all column-at-a-time values first because that was simpler. Then we loaded all of the row-by-row values. Now we save the column-at-a-time values and instead load row-by-row until the `Page`'s estimated size is larger than a "jumbo" size which defaults to a megabyte. Once we load enough rows that we estimate the page is "jumbo", we then stop loading rows. The Page will look like this: ``` | txt1 | int | txt2 | long | double | |------|-----|------|------|--------| | XXXX | | XXXX | | | | XXXX | | XXXX | | | | XXXX | | XXXX | | | | XXXX | | XXXX | | | | XXXX | | XXXX | | | | XXXX | | XXXX | | | <-- after loading this row | | | | | | we crossed to "jumbo" size | | | | | | | | | | | | | | | | | | <-- these rows are entirely empty | | | | | | | | | | | | ``` Then we chop the page to the last row: ``` | txt1 | int | txt2 | long | double | |------|-----|------|------|--------| | XXXX | | XXXX | | | | XXXX | | XXXX | | | | XXXX | | XXXX | | | | XXXX | | XXXX | | | | XXXX | | XXXX | | | | XXXX | | XXXX | | | ``` Then fill in the column-at-a-time columns: ``` | txt1 | int | txt2 | long | double | |------|-----|------|------|--------| | XXXX | 1 | XXXX | 11 | 1.0 | | XXXX | 2 | XXXX | 22 | -2.0 | | XXXX | 3 | XXXX | 33 | 1e9 | | XXXX | 4 | XXXX | 44 | 913 | | XXXX | 5 | XXXX | 55 | 0.1234 | | XXXX | 6 | XXXX | 66 | 3.1415 | ``` And then we return *that* `Page`. On the next `Driver` iteration we start from where we left off.
1 parent caeae5c commit 6e46b81

File tree

50 files changed

+766
-235
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

50 files changed

+766
-235
lines changed

‎benchmarks/README.md‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -152,11 +152,10 @@ exit
152152
Grab the async profiler from https://github.com/jvm-profiling-tools/async-profiler
153153
and run `prof async` like so:
154154
```
155-
gradlew -p benchmarks/ run --args 'LongKeyedBucketOrdsBenchmark.multiBucket -prof "async:libPath=/home/nik9000/Downloads/async-profiler-3.0-29ee888-linux-x64/lib/libasyncProfiler.so;dir=/tmp/prof;output=flamegraph"'
155+
gradlew -p benchmarks/ run --args 'LongKeyedBucketOrdsBenchmark.multiBucket -prof "async:libPath=/home/nik9000/Downloads/async-profiler-4.0-linux-x64/lib/libasyncProfiler.so;dir=/tmp/prof;output=flamegraph"'
156156
```
157157

158-
Note: As of January 2025 the latest release of async profiler doesn't work
159-
with our JDK but the nightly is fine.
158+
Note: As of July 2025 the 4.0 release of the async profiler works well.
160159

161160
If you are on Mac, this'll warn you that you downloaded the shared library from
162161
the internet. You'll need to go to settings and allow it to run.

‎benchmarks/src/main/java/org/elasticsearch/benchmark/_nightly/esql/ValuesSourceReaderBenchmark.java‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,8 +24,10 @@
2424
import org.apache.lucene.util.BytesRef;
2525
import org.apache.lucene.util.NumericUtils;
2626
import org.elasticsearch.common.breaker.NoopCircuitBreaker;
27+
import org.elasticsearch.common.logging.LogConfigurator;
2728
import org.elasticsearch.common.lucene.Lucene;
2829
import org.elasticsearch.common.settings.Settings;
30+
import org.elasticsearch.common.unit.ByteSizeValue;
2931
import org.elasticsearch.common.util.BigArrays;
3032
import org.elasticsearch.compute.data.BlockFactory;
3133
import org.elasticsearch.compute.data.BytesRefBlock;
@@ -85,6 +87,10 @@
8587
@State(Scope.Thread)
8688
@Fork(1)
8789
public class ValuesSourceReaderBenchmark {
90+
static {
91+
LogConfigurator.configureESLogging();
92+
}
93+
8894
private static final String[] SUPPORTED_LAYOUTS = new String[] { "in_order", "shuffled", "shuffled_singles" };
8995
private static final String[] SUPPORTED_NAMES = new String[] {
9096
"long",
@@ -344,6 +350,7 @@ public FieldNamesFieldMapper.FieldNamesFieldType fieldNames() {
344350
public void benchmark() {
345351
ValuesSourceReaderOperator op = new ValuesSourceReaderOperator(
346352
blockFactory,
353+
ByteSizeValue.ofMb(1).getBytes(),
347354
fields(name),
348355
List.of(new ValuesSourceReaderOperator.ShardContext(reader, () -> {
349356
throw new UnsupportedOperationException("can't load _source here");

‎docs/changelog/131053.yaml‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
pr: 131053
2+
summary: Split large pages on load sometimes
3+
area: ES|QL
4+
type: bug
5+
issues: []

‎server/src/main/java/org/elasticsearch/index/mapper/AbstractShapeGeometryFieldMapper.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -98,11 +98,11 @@ protected void writeExtent(BlockLoader.IntBuilder builder, Extent extent) {
9898
public BlockLoader.AllReader reader(LeafReaderContext context) throws IOException {
9999
return new BlockLoader.AllReader() {
100100
@Override
101-
public BlockLoader.Block read(BlockLoader.BlockFactory factory, BlockLoader.Docs docs) throws IOException {
101+
public BlockLoader.Block read(BlockLoader.BlockFactory factory, BlockLoader.Docs docs, int offset) throws IOException {
102102
var binaryDocValues = context.reader().getBinaryDocValues(fieldName);
103103
var reader = new GeometryDocValueReader();
104-
try (var builder = factory.ints(docs.count())) {
105-
for (int i = 0; i < docs.count(); i++) {
104+
try (var builder = factory.ints(docs.count() - offset)) {
105+
for (int i = offset; i < docs.count(); i++) {
106106
read(binaryDocValues, docs.get(i), reader, builder);
107107
}
108108
return builder.build();

‎server/src/main/java/org/elasticsearch/index/mapper/BlockDocValuesReader.java‎

Lines changed: 43 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -124,10 +124,10 @@ private static class SingletonLongs extends BlockDocValuesReader {
124124
}
125125

126126
@Override
127-
public BlockLoader.Block read(BlockFactory factory, Docs docs) throws IOException {
128-
try (BlockLoader.LongBuilder builder = factory.longsFromDocValues(docs.count())) {
127+
public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset) throws IOException {
128+
try (BlockLoader.LongBuilder builder = factory.longsFromDocValues(docs.count() - offset)) {
129129
int lastDoc = -1;
130-
for (int i = 0; i < docs.count(); i++) {
130+
for (int i = offset; i < docs.count(); i++) {
131131
int doc = docs.get(i);
132132
if (doc < lastDoc) {
133133
throw new IllegalStateException("docs within same block must be in order");
@@ -173,9 +173,9 @@ private static class Longs extends BlockDocValuesReader {
173173
}
174174

175175
@Override
176-
public BlockLoader.Block read(BlockFactory factory, Docs docs) throws IOException {
177-
try (BlockLoader.LongBuilder builder = factory.longsFromDocValues(docs.count())) {
178-
for (int i = 0; i < docs.count(); i++) {
176+
public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset) throws IOException {
177+
try (BlockLoader.LongBuilder builder = factory.longsFromDocValues(docs.count() - offset)) {
178+
for (int i = offset; i < docs.count(); i++) {
179179
int doc = docs.get(i);
180180
if (doc < this.docID) {
181181
throw new IllegalStateException("docs within same block must be in order");
@@ -259,10 +259,10 @@ private static class SingletonInts extends BlockDocValuesReader {
259259
}
260260

261261
@Override
262-
public BlockLoader.Block read(BlockFactory factory, Docs docs) throws IOException {
263-
try (BlockLoader.IntBuilder builder = factory.intsFromDocValues(docs.count())) {
262+
public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset) throws IOException {
263+
try (BlockLoader.IntBuilder builder = factory.intsFromDocValues(docs.count() - offset)) {
264264
int lastDoc = -1;
265-
for (int i = 0; i < docs.count(); i++) {
265+
for (int i = offset; i < docs.count(); i++) {
266266
int doc = docs.get(i);
267267
if (doc < lastDoc) {
268268
throw new IllegalStateException("docs within same block must be in order");
@@ -308,9 +308,9 @@ private static class Ints extends BlockDocValuesReader {
308308
}
309309

310310
@Override
311-
public BlockLoader.Block read(BlockFactory factory, Docs docs) throws IOException {
312-
try (BlockLoader.IntBuilder builder = factory.intsFromDocValues(docs.count())) {
313-
for (int i = 0; i < docs.count(); i++) {
311+
public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset) throws IOException {
312+
try (BlockLoader.IntBuilder builder = factory.intsFromDocValues(docs.count() - offset)) {
313+
for (int i = offset; i < docs.count(); i++) {
314314
int doc = docs.get(i);
315315
if (doc < this.docID) {
316316
throw new IllegalStateException("docs within same block must be in order");
@@ -408,10 +408,10 @@ private static class SingletonDoubles extends BlockDocValuesReader {
408408
}
409409

410410
@Override
411-
public BlockLoader.Block read(BlockFactory factory, Docs docs) throws IOException {
412-
try (BlockLoader.DoubleBuilder builder = factory.doublesFromDocValues(docs.count())) {
411+
public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset) throws IOException {
412+
try (BlockLoader.DoubleBuilder builder = factory.doublesFromDocValues(docs.count() - offset)) {
413413
int lastDoc = -1;
414-
for (int i = 0; i < docs.count(); i++) {
414+
for (int i = offset; i < docs.count(); i++) {
415415
int doc = docs.get(i);
416416
if (doc < lastDoc) {
417417
throw new IllegalStateException("docs within same block must be in order");
@@ -461,9 +461,9 @@ private static class Doubles extends BlockDocValuesReader {
461461
}
462462

463463
@Override
464-
public BlockLoader.Block read(BlockFactory factory, Docs docs) throws IOException {
465-
try (BlockLoader.DoubleBuilder builder = factory.doublesFromDocValues(docs.count())) {
466-
for (int i = 0; i < docs.count(); i++) {
464+
public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset) throws IOException {
465+
try (BlockLoader.DoubleBuilder builder = factory.doublesFromDocValues(docs.count() - offset)) {
466+
for (int i = offset; i < docs.count(); i++) {
467467
int doc = docs.get(i);
468468
if (doc < this.docID) {
469469
throw new IllegalStateException("docs within same block must be in order");
@@ -544,10 +544,10 @@ private static class DenseVectorValuesBlockReader extends BlockDocValuesReader {
544544
}
545545

546546
@Override
547-
public BlockLoader.Block read(BlockFactory factory, Docs docs) throws IOException {
547+
public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset) throws IOException {
548548
// Doubles from doc values ensures that the values are in order
549-
try (BlockLoader.FloatBuilder builder = factory.denseVectors(docs.count(), dimensions)) {
550-
for (int i = 0; i < docs.count(); i++) {
549+
try (BlockLoader.FloatBuilder builder = factory.denseVectors(docs.count() - offset, dimensions)) {
550+
for (int i = offset; i < docs.count(); i++) {
551551
int doc = docs.get(i);
552552
if (doc < iterator.docID()) {
553553
throw new IllegalStateException("docs within same block must be in order");
@@ -645,19 +645,19 @@ private BlockLoader.Block readSingleDoc(BlockFactory factory, int docId) throws
645645
if (ordinals.advanceExact(docId)) {
646646
BytesRef v = ordinals.lookupOrd(ordinals.ordValue());
647647
// the returned BytesRef can be reused
648-
return factory.constantBytes(BytesRef.deepCopyOf(v));
648+
return factory.constantBytes(BytesRef.deepCopyOf(v), 1);
649649
} else {
650-
return factory.constantNulls();
650+
return factory.constantNulls(1);
651651
}
652652
}
653653

654654
@Override
655-
public BlockLoader.Block read(BlockFactory factory, Docs docs) throws IOException {
656-
if (docs.count() == 1) {
657-
return readSingleDoc(factory, docs.get(0));
655+
public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset) throws IOException {
656+
if (docs.count() - offset == 1) {
657+
return readSingleDoc(factory, docs.get(offset));
658658
}
659-
try (BlockLoader.SingletonOrdinalsBuilder builder = factory.singletonOrdinalsBuilder(ordinals, docs.count())) {
660-
for (int i = 0; i < docs.count(); i++) {
659+
try (var builder = factory.singletonOrdinalsBuilder(ordinals, docs.count() - offset)) {
660+
for (int i = offset; i < docs.count(); i++) {
661661
int doc = docs.get(i);
662662
if (doc < ordinals.docID()) {
663663
throw new IllegalStateException("docs within same block must be in order");
@@ -700,9 +700,9 @@ private static class Ordinals extends BlockDocValuesReader {
700700
}
701701

702702
@Override
703-
public BlockLoader.Block read(BlockFactory factory, Docs docs) throws IOException {
704-
try (BytesRefBuilder builder = factory.bytesRefsFromDocValues(docs.count())) {
705-
for (int i = 0; i < docs.count(); i++) {
703+
public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset) throws IOException {
704+
try (BytesRefBuilder builder = factory.bytesRefsFromDocValues(docs.count() - offset)) {
705+
for (int i = offset; i < docs.count(); i++) {
706706
int doc = docs.get(i);
707707
if (doc < ordinals.docID()) {
708708
throw new IllegalStateException("docs within same block must be in order");
@@ -780,9 +780,9 @@ private static class BytesRefsFromBinary extends BlockDocValuesReader {
780780
}
781781

782782
@Override
783-
public BlockLoader.Block read(BlockFactory factory, Docs docs) throws IOException {
784-
try (BlockLoader.BytesRefBuilder builder = factory.bytesRefs(docs.count())) {
785-
for (int i = 0; i < docs.count(); i++) {
783+
public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset) throws IOException {
784+
try (BlockLoader.BytesRefBuilder builder = factory.bytesRefs(docs.count() - offset)) {
785+
for (int i = offset; i < docs.count(); i++) {
786786
int doc = docs.get(i);
787787
if (doc < docID) {
788788
throw new IllegalStateException("docs within same block must be in order");
@@ -879,9 +879,9 @@ private static class DenseVectorFromBinary extends BlockDocValuesReader {
879879
}
880880

881881
@Override
882-
public BlockLoader.Block read(BlockFactory factory, Docs docs) throws IOException {
883-
try (BlockLoader.FloatBuilder builder = factory.denseVectors(docs.count(), dimensions)) {
884-
for (int i = 0; i < docs.count(); i++) {
882+
public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset) throws IOException {
883+
try (BlockLoader.FloatBuilder builder = factory.denseVectors(docs.count() - offset, dimensions)) {
884+
for (int i = offset; i < docs.count(); i++) {
885885
int doc = docs.get(i);
886886
if (doc < docID) {
887887
throw new IllegalStateException("docs within same block must be in order");
@@ -963,10 +963,10 @@ private static class SingletonBooleans extends BlockDocValuesReader {
963963
}
964964

965965
@Override
966-
public BlockLoader.Block read(BlockFactory factory, Docs docs) throws IOException {
967-
try (BlockLoader.BooleanBuilder builder = factory.booleansFromDocValues(docs.count())) {
966+
public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset) throws IOException {
967+
try (BlockLoader.BooleanBuilder builder = factory.booleansFromDocValues(docs.count() - offset)) {
968968
int lastDoc = -1;
969-
for (int i = 0; i < docs.count(); i++) {
969+
for (int i = offset; i < docs.count(); i++) {
970970
int doc = docs.get(i);
971971
if (doc < lastDoc) {
972972
throw new IllegalStateException("docs within same block must be in order");
@@ -1012,9 +1012,9 @@ private static class Booleans extends BlockDocValuesReader {
10121012
}
10131013

10141014
@Override
1015-
public BlockLoader.Block read(BlockFactory factory, Docs docs) throws IOException {
1016-
try (BlockLoader.BooleanBuilder builder = factory.booleansFromDocValues(docs.count())) {
1017-
for (int i = 0; i < docs.count(); i++) {
1015+
public BlockLoader.Block read(BlockFactory factory, Docs docs, int offset) throws IOException {
1016+
try (BlockLoader.BooleanBuilder builder = factory.booleansFromDocValues(docs.count() - offset)) {
1017+
for (int i = offset; i < docs.count(); i++) {
10181018
int doc = docs.get(i);
10191019
if (doc < this.docID) {
10201020
throw new IllegalStateException("docs within same block must be in order");

‎server/src/main/java/org/elasticsearch/index/mapper/BlockLoader.java‎

Lines changed: 9 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,7 @@ interface ColumnAtATimeReader extends Reader {
4343
/**
4444
* Reads the values of all documents in {@code docs}.
4545
*/
46-
BlockLoader.Block read(BlockFactory factory, Docs docs) throws IOException;
46+
BlockLoader.Block read(BlockFactory factory, Docs docs, int offset) throws IOException;
4747
}
4848

4949
interface RowStrideReader extends Reader {
@@ -149,8 +149,8 @@ public String toString() {
149149
*/
150150
class ConstantNullsReader implements AllReader {
151151
@Override
152-
public Block read(BlockFactory factory, Docs docs) throws IOException {
153-
return factory.constantNulls();
152+
public Block read(BlockFactory factory, Docs docs, int offset) throws IOException {
153+
return factory.constantNulls(docs.count() - offset);
154154
}
155155

156156
@Override
@@ -183,8 +183,8 @@ public Builder builder(BlockFactory factory, int expectedCount) {
183183
public ColumnAtATimeReader columnAtATimeReader(LeafReaderContext context) {
184184
return new ColumnAtATimeReader() {
185185
@Override
186-
public Block read(BlockFactory factory, Docs docs) {
187-
return factory.constantBytes(value);
186+
public Block read(BlockFactory factory, Docs docs, int offset) {
187+
return factory.constantBytes(value, docs.count() - offset);
188188
}
189189

190190
@Override
@@ -261,8 +261,8 @@ public ColumnAtATimeReader columnAtATimeReader(LeafReaderContext context) throws
261261
}
262262
return new ColumnAtATimeReader() {
263263
@Override
264-
public Block read(BlockFactory factory, Docs docs) throws IOException {
265-
return reader.read(factory, docs);
264+
public Block read(BlockFactory factory, Docs docs, int offset) throws IOException {
265+
return reader.read(factory, docs, offset);
266266
}
267267

268268
@Override
@@ -408,13 +408,13 @@ interface BlockFactory {
408408
/**
409409
* Build a block that contains only {@code null}.
410410
*/
411-
Block constantNulls();
411+
Block constantNulls(int count);
412412

413413
/**
414414
* Build a block that contains {@code value} repeated
415415
* {@code size} times.
416416
*/
417-
Block constantBytes(BytesRef value);
417+
Block constantBytes(BytesRef value, int count);
418418

419419
/**
420420
* Build a reader for reading keyword ordinals.

‎server/src/main/java/org/elasticsearch/index/mapper/BooleanScriptBlockDocValuesReader.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -49,10 +49,10 @@ public int docId() {
4949
}
5050

5151
@Override
52-
public BlockLoader.Block read(BlockLoader.BlockFactory factory, BlockLoader.Docs docs) throws IOException {
52+
public BlockLoader.Block read(BlockLoader.BlockFactory factory, BlockLoader.Docs docs, int offset) throws IOException {
5353
// Note that we don't emit falses before trues so we conform to the doc values contract and can use booleansFromDocValues
54-
try (BlockLoader.BooleanBuilder builder = factory.booleans(docs.count())) {
55-
for (int i = 0; i < docs.count(); i++) {
54+
try (BlockLoader.BooleanBuilder builder = factory.booleans(docs.count() - offset)) {
55+
for (int i = offset; i < docs.count(); i++) {
5656
read(docs.get(i), builder);
5757
}
5858
return builder.build();

‎server/src/main/java/org/elasticsearch/index/mapper/DateScriptBlockDocValuesReader.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -49,10 +49,10 @@ public int docId() {
4949
}
5050

5151
@Override
52-
public BlockLoader.Block read(BlockLoader.BlockFactory factory, BlockLoader.Docs docs) throws IOException {
52+
public BlockLoader.Block read(BlockLoader.BlockFactory factory, BlockLoader.Docs docs, int offset) throws IOException {
5353
// Note that we don't sort the values sort, so we can't use factory.longsFromDocValues
54-
try (BlockLoader.LongBuilder builder = factory.longs(docs.count())) {
55-
for (int i = 0; i < docs.count(); i++) {
54+
try (BlockLoader.LongBuilder builder = factory.longs(docs.count() - offset)) {
55+
for (int i = offset; i < docs.count(); i++) {
5656
read(docs.get(i), builder);
5757
}
5858
return builder.build();

‎server/src/main/java/org/elasticsearch/index/mapper/DoubleScriptBlockDocValuesReader.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -49,10 +49,10 @@ public int docId() {
4949
}
5050

5151
@Override
52-
public BlockLoader.Block read(BlockLoader.BlockFactory factory, BlockLoader.Docs docs) throws IOException {
52+
public BlockLoader.Block read(BlockLoader.BlockFactory factory, BlockLoader.Docs docs, int offset) throws IOException {
5353
// Note that we don't sort the values sort, so we can't use factory.doublesFromDocValues
54-
try (BlockLoader.DoubleBuilder builder = factory.doubles(docs.count())) {
55-
for (int i = 0; i < docs.count(); i++) {
54+
try (BlockLoader.DoubleBuilder builder = factory.doubles(docs.count() - offset)) {
55+
for (int i = offset; i < docs.count(); i++) {
5656
read(docs.get(i), builder);
5757
}
5858
return builder.build();

0 commit comments

Comments
 (0)