ES|QL: Chunk time-series aggregation output by leontyevdv · Pull Request #151670 · elastic/elasticsearch

leontyevdv · 2026-06-18T16:53:53Z

The time-series aggregation operator previously emitted its entire intermediate output as a single page, so a data node could send one jumbo page to the coordinator and trip the circuit breaker while merging.

This PR enables chunking of the partial/intermediate output of the time-series aggregation.

TimeSeriesAggregationOperator now slices its partial/intermediate output into pages of about targetChunkRows rows, bounding the size of each page sent to the coordinator. A _tsid may straddle a page boundary; the coordinator re-merges groups by key, so no _tsid-aligned slicing is needed. Final-mode output is not chunked.

Chunking is controlled by a single dedicated, time-series-only knob, decoupled from the regular aggregation emit settings. Cluster setting esql.time_series.target_chunk_rows (dynamic, default 100,000), with a matching per-query pragma of the same name.

Closes #147286

elasticsearchmachine · 2026-06-18T16:54:18Z

Pinging @elastic/es-storage-engine (Team:StorageEngine)

elasticsearchmachine · 2026-06-18T16:54:19Z

Hi @leontyevdv, I've created a changelog YAML for you.

github-actions · 2026-06-18T16:55:13Z

🔍 Preview links for changed docs

⏳ Building and deploying preview... View progress

This comment will be updated with preview links when the build is complete.

github-actions · 2026-06-18T16:57:58Z

ℹ️ Important: Docs version tagging

👋 Thanks for updating the docs! Just a friendly reminder that our docs are now cumulative. This means all 9.x versions are documented on the same page and published off of the main branch, instead of creating separate pages for each minor version.

We use applies_to tags to mark version-specific features and changes.

Expand for a quick overview

When to use applies_to tags:

✅ At the page level to indicate which products/deployments the content applies to (mandatory)
✅ When features change state (e.g. preview, ga) in a specific version
✅ When availability differs across deployments and environments

What NOT to do:

❌ Don't remove or replace information that applies to an older version
❌ Don't add new information that applies to a specific version without an applies_to tag
❌ Don't forget that applies_to tags can be used at the page, section, and inline level

🤔 Need help?

Check out the cumulative docs guidelines
Reach out in the #docs Slack channel

elasticsearchmachine · 2026-06-22T13:57:39Z

Hi @leontyevdv, I've updated the changelog YAML for you.

dnhatn · 2026-06-23T04:39:16Z

Thanks Dima! I have some high-level feedback:

Periodic emit can break the rate aggregation. Can you back out the periodic emit changes from this PR? We can follow up on it later.
I think DimensionValues doesn't support chunking correctly. Could you please take a look and fix it?
TimeSeriesBlockHash.buildOrdinalKeys uses the full dictionary for every chunked page. I think we should rebuild the dictionary with only the terms referenced in each page?
The tsid-aligned slicing logic is nice, but adds complexity and the benefit is unknown. Since the default chunk size is large enough (100K) to compensate if a time-series is split, I think it's fine to go with unless benchmarks show otherwise.

leontyevdv · 2026-06-26T14:02:28Z

Thanks Nhat!

1. Periodic emit can break the rate aggregation. Can you back out the periodic emit changes from this PR? We can follow up on it later.

Done! So, I understood there are two different things - periodic emit and chunking. I disabled the periodic emit.

2. I think DimensionValues doesn't support chunking correctly. Could you please take a look and fix it?

We now materialize DimensionValues's values block once and copy each page out of it.

3. TimeSeriesBlockHash.buildOrdinalKeys uses the full dictionary for every chunked page. I think we should rebuild the dictionary with only the terms referenced in each page?

I added a per-page tsid dictionary there.

4. The tsid-aligned slicing logic is nice, but adds complexity and the benefit is unknown. Since the default chunk size is large enough (100K) to compensate if a time-series is split, I think it's fine to go with unless benchmarks show otherwise.

Removed!

dnhatn · 2026-06-29T19:30:36Z

-                timestamps.appendLong(p, finalHash.getKey2(p));
+                final int groupId = selected.getInt(p);
+                final long globalOrd = finalHash.getKey1(groupId);
+                if (localOrd < 0 || globalOrd != prevGlobalOrd) {


Here we rely on the ordering of TSIDs; however, this ordering is not always guaranteed. Can you keep a map of ordinals instead of solely relying on the previous ordinal?

Done! Added a test too: testTimeSeriesBlockHashChunkedPageDeduplicatesUnsortedTsids

dnhatn · 2026-06-29T19:31:29Z

        final Block[] blocks;
        if (OrdinalBytesRefBlock.isDense(positionCount, tsidHash.size())) {
-            blocks = buildOrdinalKeys(positionCount);
+            blocks = buildOrdinalKeys(selected);


Can we have a non-chunking version that reuses the full dictionary as we do today?

Yes, I returned the existing (in main) logic for cases when there is only one page (non-chunked). Thanks for pointing out on that!

dnhatn · 2026-06-29T19:34:04Z

@@ -241,26 +243,29 @@ public GroupingAggregatorFunction.PreparedForEvaluation prepareEvaluateIntermedi

    private void evaluate(Block[] blocks, int offset, IntVector selectedInPage) {
        int positionCount = selectedInPage.getPositionCount();


I think we lose the optimization for non-chunked cases. How about something like this:

@Override public GroupingAggregatorFunction.PreparedForEvaluation prepareEvaluateIntermediate( IntVector selected, GroupingAggregatorEvaluationContext ctx ) { values = builder.build(); return (blocks, offset, selectedInPage) -> { if (selected == selectedInPage && values.getPositionCount() == selected.getPositionCount()) { values.incRef(); blocks[offset] = values; return; } final int positionCount = selectedInPage.getPositionCount(); final BytesRef scratch = new BytesRef(); try (var outputBuilder = driverContext.blockFactory().newBytesRefBlockBuilder(positionCount)) { for (int p = 0; p < positionCount; p++) { int groupId = selectedInPage.getInt(p); if (groupId < values.getPositionCount()) { outputBuilder.copyFrom(values, groupId, scratch); } else { outputBuilder.appendNull(); } } blocks[offset] = outputBuilder.build(); } }; } @Override public GroupingAggregatorFunction.PreparedForEvaluation prepareEvaluateFinal( IntVector selected, GroupingAggregatorEvaluationContext ctx ) { return prepareEvaluateIntermediate(selected, ctx); }

Thanks! I use your method almost as is but instead of selected == selectedInPage I added an explicit check that the groups are not reordered or dropped by TimeSeriesAggregationOperator.selectedForValuesAggregator when window is not an exact multiple of the bucket. See this method. Does that make sense?

dnhatn · 2026-06-29T19:35:36Z

-            final boolean outputFinal = aggregatorMode.isOutputPartial() == false;
+            final boolean outputPartial = aggregatorMode.isOutputPartial();
+            final boolean outputFinal = outputPartial == false;
+            final int effectiveTargetChunkRows = outputPartial ? targetChunkRows : Integer.MAX_VALUE;


I think we can enable chunking in both partial and final, but let’s enable it for partial first as this PR does.

dnhatn

Thanks for all iterations, Dima!

dnhatn · 2026-06-30T23:13:48Z

+     * each page sent to the coordinator. Specific to the time-series operator and independent of the regular
+     * aggregation emit settings.
+     */
+    public static final Setting<Integer> TIME_SERIES_TARGET_CHUNK_SIZE = Setting.intSetting(


Can we name this setting with ROWS instead of SIZE? Right now we use "row" and "size" interchangeably, but at some point we'll need "size" to refer to actual byte size.

dnhatn · 2026-06-30T23:19:06Z

+                final int groupId = selected.getInt(p);
+                final int globalOrd = (int) finalHash.getKey1(groupId);
+                final int localOrd;
+                if (globalToLocalOrd.containsKey(globalOrd)) {


Nit: Maybe use indexOf / indexGet / indexInsert for a single lookup/insert instead.

Yes, it's globalToLocalOrd.indexExists(slot). Fixed this and the insert part. Thanks!

leontyevdv · 2026-07-02T13:13:04Z

Here are benchmarking results.

scenario: v1.0.2-hostmetrics_8clientsort-270m

avg_avgot_memory_by_host_5m

Q_MEM='TS metrics-hostmetricsreceiver.otel-default | WHERE @timestamp >= "2025-01-01T00:30:00Z" AND @timestamp <= "2025-01-01T04:30:00Z" | STATS AVG(AVG_OVER_TIME(`metrics.system.memory.utilization`)) BY host.name, TBUCKET(5m) | LIMIT 10000'

avg_rate_cpu_by_host_5m

Q_RATE='TS metrics-hostmetricsreceiver.otel-default | WHERE @timestamp >= "2025-01-01T00:30:00Z" AND @timestamp <= "2025-01-01T04:30:00Z" | STATS AVG(RATE(`metrics.system.cpu.time`)) BY host.name, TBUCKET(5m) | LIMIT 10000'

sum_rate_sys_cpu_time_large_clause_5m

Q_WIDE='TS metrics-hostmetricsreceiver.otel-default | STATS SUM(RATE(system.cpu.time)) BY host.name, host.ip, host.mac, host.arch, attributes.cpu, attributes.state, host.cpu.cache.l2.size, host.cpu.model.id, host.cpu.stepping, host.cpu.vendor.id, TBUCKET(5 minutes)'

Q_MEM — `avg_avgot_memory_by_host_5m` (G = 28,800)

Config	median (ms)	min	max	exch_pages	Chunking (exp_pages)
main · no-pragma	77	60	367	12	off (main)
branch · no-pragma	44	40	51	12	off
branch · chunk=2147483647	42	40	47	12	not-chunked (1)
branch · chunk=100000	41	39	50	12	not-chunked (1)
branch · chunk=50000	39	38	45	12	not-chunked (1)
branch · chunk=10000	41	38	45	12	CHUNKED (3)
branch · chunk=1000	41	38	58	37	CHUNKED (29)

Q_RATE — `avg_rate_cpu_by_host_5m` (G = 192,000)

Config	median (ms)	min	max	exch_pages	Chunking (exp_pages)
main · no-pragma	291	251	387	12	off (main)
branch · no-pragma	240	234	276	12	off
branch · chunk=2147483647	239	233	265	12	not-chunked (1)
branch · chunk=100000	235	227	266	12	CHUNKED (2)
branch · chunk=50000	241	229	254	12	CHUNKED (4)
branch · chunk=10000	238	231	261	25	CHUNKED (20)
branch · chunk=1000	242	232	271	199	CHUNKED (192)

Q_WIDE — `sum_rate_sys_cpu_time_large_clause_5m` (G = 192,000)

Config	median (ms)	min	max	exch_pages	Chunking (exp_pages)
main · no-pragma	499	484	518	12	off (main)
branch · no-pragma	477	465	505	12	off
branch · chunk=2147483647	476	466	501	12	not-chunked (1)
branch · chunk=100000	473	462	507	12	CHUNKED (2)
branch · chunk=50000	477	463	507	12	CHUNKED (4)
branch · chunk=10000	473	465	506	24	CHUNKED (20)
branch · chunk=1000	479	468	510	197	CHUNKED (192)

Summary — main vs branch (chunking off) and chunking overhead

Query	main median	branch off (chunk=MAX)	branch default (100000)	branch heavy (1000)	median spread across branch sweep
Q_MEM	77	42	41	41	39–44 (~5 ms)
Q_RATE	291	239	235	242	235–242 (~7 ms)
Q_WIDE	499	476	473	479	473–479 (~6 ms)

The time-series aggregation operator previously emitted its entire intermediate output as a single page, so a data node could send one jumbo page to the coordinator and trip the circuit breaker while merging. Enable chunking of the partial/intermediate output: - TimeSeriesAggregationOperator emits partial results periodically and slices each emission into _tsid-aligned pages: it fills up to targetChunkRows, then cuts at the next _tsid change so a _tsid that starts in a page is sent whole. maxChunkRows (2x the target) is a hard ceiling that only splits a single oversized _tsid. Final mode is unchanged. - Add a nextPageSliceEnd hook on HashAggregationOperator so the slicing policy can be specialized by the time-series subclass. - Fix TimeSeriesBlockHash#getKeys to project only the selected group ids instead of always returning all keys, which the slicer needs. Decouple the time-series knobs from the regular aggregation settings: add esql.time_series.partial_agg_emit_keys_threshold, esql.time_series.partial_agg_emit_unique_threshold and esql.time_series.target_chunk_size cluster settings, read only by the time-series factory (with matching query pragmas). Defaults of 100,000 / 0.1 / 100,000 leave regular STATS aggregations untouched.

Chunk the time-series aggregation's intermediate output so data nodes send bounded pages to the coordinator instead of one jumbo page, while keeping the final (coordinator) path unchanged. Drop the periodic-emit and tsid-alignment and reduce the knobs to a single chunk size.

Address review feedback

leontyevdv requested a review from dnhatn June 18, 2026 16:53

leontyevdv self-assigned this Jun 18, 2026

leontyevdv added >feature Team:StorageEngine :StorageEngine/ES|QL Timeseries / metrics / logsdb capabilities in ES|QL v9.5.0 labels Jun 18, 2026

leontyevdv changed the title ~~Esql/chunk time series aggregation output~~ ES|QL: Chunk time-series aggregation output Jun 18, 2026

leontyevdv requested review from a team as code owners June 19, 2026 08:48

leontyevdv closed this Jun 19, 2026

leontyevdv reopened this Jun 19, 2026

leontyevdv force-pushed the esql/chunk-time-series-aggregation-output branch from 9d92715 to 6954a81 Compare June 19, 2026 09:42

leontyevdv removed request for a team June 19, 2026 09:42

leontyevdv force-pushed the esql/chunk-time-series-aggregation-output branch from 6954a81 to 498a649 Compare June 22, 2026 11:42

leontyevdv force-pushed the esql/chunk-time-series-aggregation-output branch 2 times, most recently from 94607b1 to 4244259 Compare June 26, 2026 11:05

sidosera self-requested a review June 29, 2026 19:08

dnhatn reviewed Jun 29, 2026

View reviewed changes

leontyevdv force-pushed the esql/chunk-time-series-aggregation-output branch 2 times, most recently from 72f2104 to d87704f Compare June 30, 2026 12:35

leontyevdv requested a review from dnhatn June 30, 2026 15:06

dnhatn approved these changes Jun 30, 2026

View reviewed changes

leontyevdv force-pushed the esql/chunk-time-series-aggregation-output branch 2 times, most recently from 833e961 to 259c4d4 Compare July 1, 2026 13:56

leontyevdv added 8 commits July 2, 2026 16:10

Update docs/changelog/151670.yaml

9594abf

Update docs/changelog/151670.yaml

2d59354

ESQL: Refine time-series partial output chunking

e183da1

Address review feedback

Fix querypragmas

ae7963a

Address Nhat's feedback

455c319

Fix compilation errors after rebasing

592dbe6

leontyevdv force-pushed the esql/chunk-time-series-aggregation-output branch from 259c4d4 to 592dbe6 Compare July 2, 2026 14:10

leontyevdv merged commit d394fb6 into elastic:main Jul 3, 2026
42 checks passed

		@@ -241,26 +243,29 @@ public GroupingAggregatorFunction.PreparedForEvaluation prepareEvaluateIntermedi

		private void evaluate(Block[] blocks, int offset, IntVector selectedInPage) {
		int positionCount = selectedInPage.getPositionCount();

Conversation

leontyevdv commented Jun 18, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

elasticsearchmachine commented Jun 18, 2026

Uh oh!

elasticsearchmachine commented Jun 18, 2026

Uh oh!

github-actions Bot commented Jun 18, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

🔍 Preview links for changed docs

Uh oh!

github-actions Bot commented Jun 18, 2026

ℹ️ Important: Docs version tagging

When to use applies_to tags:

What NOT to do:

🤔 Need help?

Uh oh!

elasticsearchmachine commented Jun 22, 2026

Uh oh!

dnhatn commented Jun 23, 2026

Uh oh!

leontyevdv commented Jun 26, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

leontyevdv Jun 30, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

leontyevdv Jun 30, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dnhatn left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

leontyevdv commented Jul 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Q_MEM — avg_avgot_memory_by_host_5m (G = 28,800)

Q_RATE — avg_rate_cpu_by_host_5m (G = 192,000)

Q_WIDE — sum_rate_sys_cpu_time_large_clause_5m (G = 192,000)

Summary — main vs branch (chunking off) and chunking overhead

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

leontyevdv commented Jun 18, 2026 •

edited

Loading

github-actions Bot commented Jun 18, 2026 •

edited

Loading

leontyevdv Jun 30, 2026 •

edited

Loading

leontyevdv Jun 30, 2026 •

edited

Loading

leontyevdv commented Jul 2, 2026 •

edited

Loading

Q_MEM — `avg_avgot_memory_by_host_5m` (G = 28,800)

Q_RATE — `avg_rate_cpu_by_host_5m` (G = 192,000)

Q_WIDE — `sum_rate_sys_cpu_time_large_clause_5m` (G = 192,000)