Part-3: Working E2E Quickstart for Time Series Engine #14048

ankitsultana · 2024-09-20T16:53:28Z

Last PR to get us to a working Quickstart state. The PRs after this will be much smaller. Explaining the changes at a high-level below:

Adds a Prometheus Like Pinot Broker HTTP API

Adds two new APIs to the broker: 1 for range queries and 1 for instant-vector queries. The API is Prometheus Compatible, except that the top-level path is prefixed with timeseries/api.

Changes to Receive a Time Series Query Request in Broker and Send to Server

After the broker receives the query, we use the new TimeSeriesRequestHandler, which uses the pinot-timeseries-planner module's TimeSeriesQueryEnvironment to plan the query. The plan is finally dispatched via the QueryDispatcher. The implementation mimics the MSE very closely, but of course the exact details are quite different.

Code Duplication: I had to create the dispatch related classes again: AsyndTimeSeriesDispatchClient and the like; ideally we should just use their MSE equivalents but that will require us to consolidate more stuff and I plan to take it up as part of phase-2.

Executing Received Plan in Server

Once the plan is received in the server, we:

Deserialize the plan and compile the plan-tree to an operator-tree
Run the operator tree in QueryRunner's ExecutorService. The same executor service is used by OpChainSchedulerService as well.

Converting ScanFilterAndProject to TimeSeriesPhysicalTableScan

This is done as part of the plan compilation in the server, because we need to generate the LeafTimeSeriesOperator, which is a pinot-query-runtime construct, and ScanFilterAndProject only has a dependency to pinot-spi.

Dummy M3QL Implementation

For now I have added a dummy M3QL implementation so we can play with the Quickstart.

Instructions for Starting Quickstart

You can start the "TIME_SERIES" quick start, wait a minute or so for some data to be populated, and then run:

➜  ~ cat script.py
import requests
import time
import urllib


with open('query.txt', 'r') as of:
    query = of.read()
    request = {
            'language': 'm3ql',
            'start': int(time.time()),
            'end': int(time.time()) + 3600,
            'query': query
    }
    query_params = urllib.urlencode(request)
    resp = requests.get('http://localhost:8000/timeseries/api/v1/query_range?' + query_params)
    print(resp.text)

➜  ~ cat query.txt
fetch{table="meetupRsvp_REALTIME",filter="",ts_column="__metadata$recordTimestamp",ts_unit="MILLISECONDS",value="1"}
  | max{group_city}
  | transformNull{0}
  | keepLastValue{}

codecov-commenter · 2024-09-20T17:41:42Z

Codecov Report

Attention: Patch coverage is 2.19124% with 491 lines in your changes missing coverage. Please review.

Project coverage is 64.80%. Comparing base (59551e4) to head (8b15367).
Report is 1101 commits behind head on master.

Files with missing lines	Patch %	Lines
...pinot/tsdb/planner/TimeSeriesQueryEnvironment.java	0.00%	62 Missing ⚠️
...roker/requesthandler/TimeSeriesRequestHandler.java	0.00%	52 Missing ⚠️
...common/response/PinotBrokerTimeSeriesResponse.java	0.00%	51 Missing ⚠️
.../pinot/query/service/dispatch/QueryDispatcher.java	3.92%	49 Missing ⚠️
...va/org/apache/pinot/query/runtime/QueryRunner.java	0.00%	43 Missing and 1 partial ⚠️
...time/timeseries/PhysicalTimeSeriesPlanVisitor.java	0.00%	40 Missing ⚠️
.../pinot/tsdb/planner/physical/TableScanVisitor.java	0.00%	36 Missing ⚠️
...ache/pinot/common/utils/HumanReadableDuration.java	0.00%	32 Missing ⚠️
...pinot/broker/api/resources/PinotClientRequest.java	0.00%	19 Missing ⚠️
...ery/runtime/timeseries/LeafTimeSeriesOperator.java	0.00%	14 Missing ⚠️
... and 14 more

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #14048      +/-   ##
============================================
+ Coverage     61.75%   64.80%   +3.05%     
- Complexity      207     1534    +1327     
============================================
  Files          2436     2579     +143     
  Lines        133233   141262    +8029     
  Branches      20636    21640    +1004     
============================================
+ Hits          82274    91542    +9268     
+ Misses        44911    42975    -1936     
- Partials       6048     6745     +697

Flag	Coverage Δ
custom-integration1	`100.00% <ø> (+99.99%)`	⬆️
integration	`100.00% <ø> (+99.99%)`	⬆️
integration1	`100.00% <ø> (+99.99%)`	⬆️
integration2	`0.00% <ø> (ø)`
java-11	`64.76% <2.19%> (+3.05%)`	⬆️
java-21	`64.69% <2.19%> (+3.06%)`	⬆️
skip-bytebuffers-false	`64.79% <2.19%> (+3.04%)`	⬆️
skip-bytebuffers-true	`64.65% <2.19%> (+36.93%)`	⬆️
temurin	`64.80% <2.19%> (+3.05%)`	⬆️
unittests	`64.79% <2.19%> (+3.05%)`	⬆️
unittests1	`56.28% <0.72%> (+9.39%)`	⬆️
unittests2	`34.90% <1.99%> (+7.17%)`	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

raghavyadav01 · 2024-09-23T17:58:40Z

pinot-broker/src/main/java/org/apache/pinot/broker/api/resources/PinotClientRequest.java

+        asyncResponse.resume(response);
+      }
+    } catch (Exception e) {
+      LOGGER.error("Caught exception while processing POST request", e);


Where do we do translation for http errors codes like Invalid param like Validation error in query planner?

In the execution part we catch exception, and create a "PinotBrokerTimeSeriesResponse" with the error and errorType set. Right now we don't have good error categories but it's a good point, we should converge on a standard. Added an item to the tracker #13957

raghavyadav01 · 2024-09-23T18:02:53Z

pinot-broker/src/main/java/org/apache/pinot/broker/api/resources/PinotClientRequest.java

+  @Path("timeseries/api/v1/query_range")
+  @ApiOperation(value = "Prometheus Compatible API for Pinot's Time Series Engine")
+  @ManualAuthorization
+  public void processTimeSeriesQueryEngine(@Suspended AsyncResponse asyncResponse,


I noticed these API's are not part of Swagger endpoint. Do we need to make additional changes to include in swagger console in pinot?

Are you checking the pinot-controller swagger? I can take it as a follow-up.. I almost never use the pinot-broker swagger.

raghavyadav01 · 2024-09-23T18:11:12Z

pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/TimeSeriesRequestHandler.java

+    if (StringUtils.isNotBlank(timeoutStr)) {
+      timeout = HumanReadableDuration.from(timeoutStr);
+    }
+    // TODO: Pass full raw query param string to the request


This is handled now correct?

Yeah good point. Will raise a PR shortly to remove this and make some other minor improvements.

raghavyadav01 · 2024-09-23T18:12:34Z

pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/TimeSeriesRequestHandler.java

+    }
+    try {
+      return Long.parseLong(step);
+    } catch (NumberFormatException ignored) {


Why are we ignoring the exception? Will we revert to default in case of invalid step time?

Oh because the duration passed by the client could be something like step=10s, step=10. In the former case it's very easy to know what the expectation. In the latter case, we right now assume that the default unit the user is targeting is seconds.

raghavyadav01 · 2024-09-23T18:19:26Z

pinot-common/src/main/java/org/apache/pinot/common/response/PinotBrokerTimeSeriesResponse.java

+      return _result;
+    }
+
+    public static Data newMatrix(List<Value> result) {


We will need to add other types "vector", "scalar" , "string" as well for prometheus.

raghavyadav01 · 2024-09-23T18:36:43Z

pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/QueryRunner.java

  }

+  /**
+   * Receives a serialized plan sent by the broker, and runs it to completion, blocking the thread until the execution


Is there a timeout on the request in this thread or can it hang for ever?

raghavyadav01 · 2024-09-23T18:40:25Z

...time/src/main/java/org/apache/pinot/query/runtime/timeseries/TimeSeriesExecutionContext.java

+
+public class TimeSeriesExecutionContext {
+  private final String _language;
+  private final TimeBuckets _initialTimeBuckets;


Why initial can these TimeBuckets change when execution is happening ?

Yup. The time buckets generated by the broker are used by the operators starting at the leaf stage.

From there on, the operators themselves have control over the time buckets and they can also change the granularity of the time-buckets on the fly. e.g. M3 has the "summarize 1h sum" function which allows the same.

raghavyadav01 · 2024-09-23T18:45:37Z

pinot-query-runtime/src/main/java/org/apache/pinot/query/service/dispatch/QueryDispatcher.java

+      TimeSeriesQueryServerInstance queryServerInstance) {
+    String hostname = queryServerInstance.getHostname();
+    int port = queryServerInstance.getQueryServicePort();
+    String key = String.format("%s_%d", hostname, port);


Is this key generation similar to sql query dispatcher? I am wondering if hostname changes for pod specially in kubernetes cluster what would be the side effect?

This is the same as the Multistage Engine so all related constraints will apply. In case of a hostname change, that will involve a node restart which will anyways lead to a query failure. Right now our Multistage queries aren't able to still run if any of the involved server dies or gets restarted midway.

raghavyadav01 · 2024-09-23T18:51:32Z

...meseries-planner/src/main/java/org/apache/pinot/tsdb/planner/TimeSeriesQueryEnvironment.java

+        "Expected exactly one table name in the logical plan, got: %s",
+        tableNames);
+    String tableName = tableNames.iterator().next();
+    // Step-2: Compute routing table assuming all segments are selected. This is to perform the check to reject tables


To avoid this we will need to make sure in write path all segments for a table go to single server, till we implement multi server in phase2. Can we capture the limitations of phase1 in task list?

raghavyadav01 · 2024-09-23T18:58:54Z

...ies/pinot-timeseries-spi/src/main/java/org/apache/pinot/tsdb/spi/series/TimeSeriesBlock.java

  private final Map<Long, List<TimeSeries>> _seriesMap;

-  public TimeSeriesBlock(TimeBuckets timeBuckets, Map<Long, List<TimeSeries>> seriesMap) {
+  public TimeSeriesBlock(@Nullable TimeBuckets timeBuckets, Map<Long, List<TimeSeries>> seriesMap) {


Why timebucket is nullable? What will happen if TimeBucket is null? Is this for Instant query use case?

Yeah this is for that, and also for the case when we have to perform partial aggregates. in that case, we might not be able to bucket the values at a time granularity under the combine operator, and may have to instead return time values as a Long[] instead of TimeBuckets in the TimeSeries.

raghavyadav01 · 2024-09-23T19:12:14Z

test https://github.com/apache/pinot/actions/runs/10973640885/job/30479986608?pr=14048 is failing though it does not look to be related.

ankitsultana · 2024-09-24T04:56:10Z

Thanks folks for the quick review. Will be raising smaller PRs going forward for incremental improvements and bug fixes

gortiz · 2024-09-24T10:17:19Z

...timeseries-m3ql/src/main/java/org/apache/pinot/tsdb/m3ql/operator/KeepLastValueOperator.java

+  @Override
+  public TimeSeriesBlock getNextBlock() {
+    TimeSeriesBlock seriesBlock = _childOperators.get(0).nextBlock();
+    seriesBlock.getSeriesMap().values().parallelStream().forEach(unionOfSeries -> {


Are we sure we want to use parallel streams here? What is the advantage? Assuming we have a high enough QPS I can only see queries competing to each other to use these limited threads.

This entire package will be overwritten in the next few days, and you are right we shouldn't use it.

I have to go through a small review process internally before I can share the actual M3 Plugin implementation which doesn't have any of these hacks.

dang-stripe · 2026-01-07T00:12:45Z

pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/BaseBrokerStarter.java

      // multi-stage request handler uses both Netty and GRPC ports.
      // worker requires both the "Netty port" for protocol transport; and "GRPC port" for mailbox transport.
      // TODO: decouple protocol and engine selection.
+      queryDispatcher = createQueryDispatcher(_brokerConf);


hey @ankitsultana i noticed this creates a new query dispatcher even though the MultistageBrokerRequestHandler creates it's own: https://github.com/apache/pinot/blob/master/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java#L174 was there a reason for that?

Yeah the reason was that QueryDispatcher has an in-memory state so we wanted them to be decoupled.

Jackie just merged a PR that creates a separate class for the Time Series dispatcher: #17474

ankitsultana added 2 commits September 20, 2024 22:14

Part-4: Working E2E Quickstart for Time Series Engine

7d8a533

pinot-assembly change

2392a5c

ankitsultana added 2 commits September 20, 2024 23:12

fix pom/assembly

8cb8228

working quickstart

d7d3ccc

ankitsultana changed the title ~~[WIP] Part-4: Working E2E Quickstart for Time Series Engine~~ Part-4: Working E2E Quickstart for Time Series Engine Sep 21, 2024

ankitsultana added 2 commits September 21, 2024 15:14

fix pom + minor

4764b7f

fix pom

8b15367

raghavyadav01 reviewed Sep 23, 2024

View reviewed changes

raghavyadav01 approved these changes Sep 23, 2024

View reviewed changes

Jackie-Jiang approved these changes Sep 23, 2024

View reviewed changes

ankitsultana changed the title ~~Part-4: Working E2E Quickstart for Time Series Engine~~ Part-3: Working E2E Quickstart for Time Series Engine Sep 24, 2024

ankitsultana added the timeseries-engine Tracking tag for generic time-series engine work label Sep 24, 2024

ankitsultana merged commit c395d09 into apache:master Sep 24, 2024

gortiz reviewed Sep 24, 2024

View reviewed changes

dang-stripe reviewed Jan 7, 2026

View reviewed changes

dang-stripe mentioned this pull request Jan 7, 2026

[multistage] Reset GRPC connection backoff when server is re-enabled #17466

Open

Part-3: Working E2E Quickstart for Time Series Engine #14048

Part-3: Working E2E Quickstart for Time Series Engine #14048

Uh oh!

Conversation

ankitsultana commented Sep 20, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Adds a Prometheus Like Pinot Broker HTTP API

Changes to Receive a Time Series Query Request in Broker and Send to Server

Executing Received Plan in Server

Converting ScanFilterAndProject to TimeSeriesPhysicalTableScan

Dummy M3QL Implementation

Instructions for Starting Quickstart

Uh oh!

codecov-commenter commented Sep 20, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

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!

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!

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!

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!

raghavyadav01 commented Sep 23, 2024

Uh oh!

ankitsultana commented Sep 24, 2024

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!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants

ankitsultana commented Sep 20, 2024 •

edited

Loading

codecov-commenter commented Sep 20, 2024 •

edited

Loading