[9.3] Partition time-series source (#140475)

docs/changelog/140475.yaml

@@ -0,0 +1,5 @@
+pr: 140475
+summary: Partition time-series source
+area: ES|QL
+type: enhancement
+issues: []

x-pack/plugin/esql/qa/server/multi-clusters/src/javaRestTest/java/org/elasticsearch/xpack/esql/ccq/MultiClusterTimeSeriesIT.java

@@ -52,7 +52,7 @@
 @ThreadLeakFilters(filters = TestClustersThreadFilter.class)
 public class MultiClusterTimeSeriesIT extends ESRestTestCase {
 
-    static final List<String> REQUIRED_CAPABILITIES = List.of("ts_command_v0");
+    static final List<String> REQUIRED_CAPABILITIES = List.of("ts_command_v0", "rate_fix_resets_multiple_segments");
 
     static ElasticsearchCluster remoteCluster = Clusters.remoteCluster();
     static ElasticsearchCluster localCluster = Clusters.localCluster(remoteCluster);

x-pack/plugin/esql/qa/testFixtures/src/main/java/org/elasticsearch/xpack/esql/EsqlTestUtils.java

@@ -588,6 +588,7 @@ public static LogicalOptimizerContext unboundLogicalOptimizerContext() {
     public static final PlannerSettings TEST_PLANNER_SETTINGS = new PlannerSettings(
         DataPartitioning.AUTO,
         ByteSizeValue.ofMb(1),
+        ByteSizeValue.ofMb(2),
         10_000,
         ByteSizeValue.ofMb(1)
     );

x-pack/plugin/esql/qa/testFixtures/src/main/resources/k8s-timeseries.csv-spec

@@ -24,6 +24,7 @@ max_bytes:long | cluster: keyword
 maxRate
 required_capability: ts_command_v0
 required_capability: double_quotes_source_enclosing
+required_capability: rate_fix_resets_multiple_segments
 TS k8s | STATS max=max(rate(network.total_bytes_in)) | EVAL max=ROUND(max, 6) | KEEP max;
 
 max: double
@@ -33,6 +34,7 @@ max: double
 maxRatePerMinute
 required_capability: ts_command_v0
 required_capability: double_quotes_source_enclosing
+required_capability: rate_fix_resets_multiple_segments
 TS k8s | STATS max=max(60 * rate(network.total_bytes_in)) | EVAL max=ROUND(max, 6) | KEEP max;
 
 max: double
@@ -41,6 +43,7 @@ max: double
 
 maxCost
 required_capability: ts_command_v0
+required_capability: rate_fix_resets_multiple_segments
 TS k8s | STATS max_cost=max(rate(network.total_cost)) | EVAL max_cost=ROUND(max_cost, 6) | KEEP max_cost;
 
 max_cost: double
@@ -49,6 +52,7 @@ max_cost: double
 
 maxRateAndBytes
 required_capability: ts_command_v0
+required_capability: rate_fix_resets_multiple_segments
 TS k8s | STATS max_rate=max(60 * rate(network.total_bytes_in)), max_bytes=max(network.bytes_in) | EVAL max_rate=ROUND(max_rate, 6) | KEEP max_rate, max_bytes;
 
 max_rate: double | max_bytes: long
@@ -57,6 +61,7 @@ max_rate: double | max_bytes: long
 
 maxRateAndBytesExplicit
 required_capability: ts_command_v0
+required_capability: rate_fix_resets_multiple_segments
 TS k8s | STATS max_rate=max(60 * rate(network.total_bytes_in)), max_bytes=max(last_over_time(network.bytes_in)) | EVAL max_rate=ROUND(max_rate, 6) | KEEP max_rate, max_bytes;
 
 max_rate: double | max_bytes: long
@@ -65,6 +70,7 @@ max_rate: double | max_bytes: long
 
 maxRateAndMarkupBytes
 required_capability: ts_command_v0
+required_capability: rate_fix_resets_multiple_segments
 TS k8s | STATS max_rate=max(rate(network.total_bytes_in)), max_markup=max(network.bytes_in * 1.05) | EVAL max_rate=ROUND(max_rate, 6) | KEEP max_rate, max_markup;
 
 max_rate: double | max_markup: double
@@ -73,6 +79,7 @@ max_rate: double | max_markup: double
 
 maxRateAndMarkupBytesExplicit
 required_capability: ts_command_v0
+required_capability: rate_fix_resets_multiple_segments
 TS k8s | STATS max_rate=max(rate(network.total_bytes_in)), max_bytes_in = max(last_over_time(network.bytes_in) * 1.05) | EVAL max_rate=ROUND(max_rate, 6) | KEEP max_rate, max_bytes_in;
 
 max_rate: double | max_bytes_in: double
@@ -81,6 +88,7 @@ max_rate: double | max_bytes_in: double
 
 maxRateAndLastBytesIn
 required_capability: ts_command_v0
+required_capability: rate_fix_resets_multiple_segments
 TS k8s | STATS max_rate=max(rate(network.total_bytes_in)), max_bytes_in = max(last_over_time(network.bytes_in * 1.05)) | EVAL max_rate=ROUND(max_rate, 6) | KEEP max_rate, max_bytes_in;
 
 max_rate: double | max_bytes_in: double
@@ -89,6 +97,7 @@ max_rate: double | max_bytes_in: double
 
 maxRateAndBytesAndCost
 required_capability: ts_command_v0
+required_capability: rate_fix_resets_multiple_segments
 TS k8s | STATS max_rate=max(rate(network.total_bytes_in)), max_bytes=max(max_over_time(network.bytes_in)), max_cost=max(rate(network.total_cost)) | EVAL max_rate=ROUND(max_rate, 6), max_cost=ROUND(max_cost, 6) | KEEP max_rate, max_bytes, max_cost;
 
 max_rate:double | max_bytes:long | max_cost:double
@@ -97,6 +106,7 @@ max_rate:double | max_bytes:long | max_cost:double
 
 sumRate
 required_capability: ts_command_v0
+required_capability: rate_fix_resets_multiple_segments
 TS k8s | STATS bytes=sum(rate(network.total_bytes_in)), sum_cost=sum(rate(network.total_cost)) BY cluster | EVAL bytes=ROUND(bytes, 6), sum_cost=ROUND(sum_cost, 6) | KEEP bytes, sum_cost, cluster | SORT cluster;
 
 bytes: double | sum_cost: double | cluster: keyword
@@ -108,6 +118,7 @@ bytes: double | sum_cost: double | cluster: keyword
 oneRateWithBucket
 required_capability: ts_command_v0
 required_capability: rate_with_interpolation_v2
+required_capability: rate_fix_resets_multiple_segments
 
 // tag::rate[]
 TS k8s
@@ -125,8 +136,8 @@ max_rate: double | time_bucket:date
 
 oneRateWithTBucket
 required_capability: ts_command_v0
-required_capability: tbucket
 required_capability: rate_with_interpolation_v2
+required_capability: rate_fix_resets_multiple_segments
 
 TS k8s
 | STATS max_rate=max(rate(network.total_bytes_in)) BY time_bucket = tbucket(5minute)
@@ -141,6 +152,7 @@ max_rate: double | time_bucket:date
 oneRateWithPromql
 required_capability: promql_pre_tech_preview_v7
 required_capability: rate_with_interpolation_v2
+required_capability: rate_fix_resets_multiple_segments
 
 PROMQL index=k8s step=5m max(rate(network.total_bytes_in[5m]))
 | SORT step DESC | LIMIT 2;
@@ -153,7 +165,8 @@ max(rate(network.total_bytes_in[5m])):double | step:datetime
 
 oneRateWithSingleTBucket
 required_capability: ts_command_v0
-required_capability: tbucket
+required_capability: rate_fix_resets_multiple_segments
+
 TS k8s
 | STATS max_rate=max(rate(network.total_bytes_in)) BY time_bucket = tbucket(1h)
 | EVAL max_rate=ROUND(max_rate, 6) | KEEP max_rate, time_bucket;
@@ -164,6 +177,7 @@ max_rate: double | time_bucket:date
 
 oneRateWithSingleStepPromql
 required_capability: promql_pre_tech_preview_v7
+required_capability: rate_fix_resets_multiple_segments
 PROMQL index=k8s step=1h max(rate(network.total_bytes_in[1h]));
 
 max(rate(network.total_bytes_in[1h])):double | step:datetime
@@ -174,6 +188,7 @@ max(rate(network.total_bytes_in[1h])):double | step:datetime
 twoRatesWithBucket
 required_capability: ts_command_v0
 required_capability: rate_with_interpolation_v2
+required_capability: rate_fix_resets_multiple_segments
 
 TS k8s | STATS max_rate=max(rate(network.total_bytes_in)), sum_rate=sum(rate(network.total_bytes_in)) BY time_bucket = bucket(@timestamp,5minute) | EVAL max_rate=ROUND(max_rate, 6), sum_rate=ROUND(sum_rate, 6) | KEEP max_rate, sum_rate, time_bucket | SORT time_bucket DESC | LIMIT 3;
 
@@ -185,8 +200,9 @@ max_rate:double | sum_rate:double | time_bucket:datetime
 
 twoRatesWithTBucket
 required_capability: ts_command_v0
-required_capability: tbucket
 required_capability: rate_with_interpolation_v2
+required_capability: rate_fix_resets_multiple_segments
+
 TS k8s | STATS max_rate=max(rate(network.total_bytes_in)), sum_rate=sum(rate(network.total_bytes_in)) BY time_bucket = tbucket(5minute) | EVAL max_rate=ROUND(max_rate, 6), sum_rate=ROUND(sum_rate, 6) | KEEP max_rate, sum_rate, time_bucket | SORT time_bucket DESC | LIMIT 3;
 
 max_rate:double | sum_rate:double | time_bucket:datetime    
@@ -198,6 +214,8 @@ max_rate:double | sum_rate:double | time_bucket:datetime
 oneRateWithBucketAndCluster
 required_capability: ts_command_v0
 required_capability: rate_with_interpolation_v2
+required_capability: rate_fix_resets_multiple_segments
+
 TS k8s | STATS max_rate=max(rate(network.total_bytes_in)) BY time_bucket = bucket(@timestamp,5minute), cluster | EVAL max_rate=ROUND(max_rate, 6) | KEEP max_rate, time_bucket, cluster | SORT time_bucket DESC, cluster | LIMIT 6;
 
 max_rate:double | time_bucket:datetime     | cluster:keyword
@@ -212,6 +230,7 @@ max_rate:double | time_bucket:datetime     | cluster:keyword
 BytesAndCostByBucketAndCluster
 required_capability: ts_command_v0
 required_capability: rate_with_interpolation_v2
+required_capability: rate_fix_resets_multiple_segments
 
 TS k8s | STATS max_rate=max(rate(network.total_bytes_in)), max_cost=max(max_over_time(network.cost)) BY time_bucket = bucket(@timestamp,5minute), cluster | EVAL max_rate=ROUND(max_rate, 6) | KEEP max_rate, max_cost, time_bucket, cluster | SORT time_bucket DESC, cluster | LIMIT 6;
 
@@ -228,6 +247,7 @@ max_rate:double | max_cost:double | time_bucket:date         | cluster: keyword
 oneRateWithBucketAndClusterThenFilter
 required_capability: ts_command_v0
 required_capability: rate_with_interpolation_v2
+required_capability: rate_fix_resets_multiple_segments
 
 TS k8s | WHERE cluster=="prod" | STATS max_rate=max(rate(network.total_bytes_in)) BY time_bucket = bucket(@timestamp,5minute), cluster | EVAL max_rate=ROUND(max_rate, 6) | KEEP max_rate, time_bucket, cluster | SORT time_bucket DESC | LIMIT 3;
 
@@ -240,6 +260,8 @@ max_rate:double | time_bucket:datetime     | cluster:keyword
 
 maxRateWithInlineFilter
 required_capability: ts_command_v0
+required_capability: rate_fix_resets_multiple_segments
+
 TS k8s | STATS max_rate = max(rate(network.total_bytes_in)) WHERE cluster=="prod" BY cluster | EVAL max_rate=ROUND(max_rate, 6) | KEEP max_rate, cluster | SORT cluster;
 
 max_rate:double | cluster:keyword
@@ -250,6 +272,8 @@ null            | staging
 
 maxRateWithPreFilter
 required_capability: ts_command_v0
+required_capability: rate_fix_resets_multiple_segments
+
 TS k8s | WHERE cluster=="prod" | STATS max_rate = max(rate(network.total_bytes_in)) BY cluster | EVAL max_rate=ROUND(max_rate, 6) | KEEP max_rate, cluster | SORT cluster;
 
 max_rate:double | cluster:keyword
@@ -259,6 +283,7 @@ max_rate:double | cluster:keyword
 notEnoughSamples
 required_capability: ts_command_v0
 required_capability: rate_with_interpolation
+required_capability: rate_fix_resets_multiple_segments
 
 TS k8s | WHERE @timestamp <= "2024-05-10T00:06:14.000Z" | STATS max_rate=max(rate(network.total_bytes_in)) BY pod, time_bucket = bucket(@timestamp,1minute) | EVAL max_rate=ROUND(max_rate, 6) | KEEP max_rate, pod, time_bucket | SORT pod, time_bucket DESC | LIMIT 10;
 
@@ -516,6 +541,7 @@ distincts:long | distincts_imprecise:long | cluster:keyword | time_bucket:dateti
 two_rates
 required_capability: ts_command_v0
 required_capability: rate_with_interpolation
+required_capability: rate_fix_resets_multiple_segments
 
 TS k8s | STATS cost_per_mb=max(rate(network.total_bytes_in) / 1024 * 1024 * rate(network.total_cost)) BY cluster, time_bucket = bucket(@timestamp,5minute) | EVAL cost_per_mb=ROUND(cost_per_mb, 6) | KEEP cost_per_mb, cluster, time_bucket | SORT cost_per_mb DESC, cluster, time_bucket DESC | LIMIT 5;
 
@@ -716,6 +742,7 @@ sd:double | var:double | sd_squared:double | cluster:keyword | time_bucket:datet
 
 Max of Rate with Bucket
 required_capability: ts_command_v0
+required_capability: rate_fix_resets_multiple_segments
 
 TS k8s
 | STATS maxRate = max(rate(network.total_cost))  BY tbucket = bucket(@timestamp, 1hour) 
@@ -727,6 +754,7 @@ maxRate:double | tbucket:datetime
 
 Max of Rate with Bucket, rename _tsid
 required_capability: ts_command_v0
+required_capability: rate_fix_resets_multiple_segments
 
 TS k8s METADATA _tsid
 | RENAME _tsid AS newTsid
@@ -741,6 +769,7 @@ maxRate:double | tbucket:datetime
 
 Max of Rate with Bucket, drop _tsid
 required_capability: ts_command_v0
+required_capability: rate_fix_resets_multiple_segments
 
 TS k8s METADATA _tsid
 | DROP _tsid
@@ -943,6 +972,7 @@ max:double | inc:double | time_bucket:date_nanos   | cluster:keyword
 dateNanosOneRateIncreaseWithBucketAndClusterThenFilter
 required_capability: ts_command_v0
 required_capability: ts_rate_datenanos_2
+required_capability: rate_fix_resets_multiple_segments
 
 TS datenanos-k8s
 | WHERE cluster=="prod"
@@ -991,6 +1021,7 @@ required_capability: ts_command_v0
 required_capability: ts_linreg_derivative
 required_capability: ts_deriv_datenanos
 required_capability: rate_with_interpolation_v2
+required_capability: rate_fix_resets_multiple_segments
 
 TS k8s
 | STATS max_deriv = max(deriv(to_long(network.total_bytes_in))), max_rate = max(rate(network.total_bytes_in)) BY time_bucket = bucket(@timestamp,5minute), cluster

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java

@@ -1837,6 +1837,11 @@ public enum Cap {
          */
         CONDITIONAL_BLOCK_LOADER_FOR_TEXT_FIELDS,
 
+        /**
+         * Fixes reset calculation in rates where partitioning data into multiple slices can lead to incorrect results.
+         */
+        RATE_FIX_RESETS_MULTIPLE_SEGMENTS,
+
         // Last capability should still have a comma for fewer merge conflicts when adding new ones :)
         // This comment prevents the semicolon from being on the previous capability when Spotless formats the file.
         ;