Apache Druid

I guess something like a RANK function would solve it, but till we have it, I wonder if there is solution for this kind of problem in Druid

```select dim1,
  s,
m1,
  __time
from (
    SELECT dim1,
      __time,
      sum(m1) as s
    FROM datasource
    where __time &gt;= TIMESTAMP '2022-01-01 00:00:00' AND __time &lt; TIMESTAMP '2022-10-02 00:00:00'
    group by 1,
      2
    order by s desc
    limit 1
  )
UNION ALL

select dim1,
  s,
  __time,
m1
from (
    SELECT dim1,
      __time,
      sum(m1) as s
    FROM datasource
    where __time &gt;= TIMESTAMP '2022-01-02 00:00:00' AND __time &lt; TIMESTAMP '2022-10-03 00:00:00'
    group by 1,
      2
    order by s desc
    limit 1
  )```

^^ Union query that can achieve what I am looking for. It doesn't scale well though for 100s or 1000s of intervals unless there is a way to parallelize their execution.

There isn't a clean way to express it through SQL (because of the limitation of what the currently understood Druid SQL can express) but.... fun story! The EXACT use case that you describe is supported by the native topN query

<https://druid.apache.org/docs/latest/querying/topnquery.html>

Well, topn doesn’t work when I want to group by on multiple dimensions or does it? 

if you look at the example query in the docs (or any of the queries that SQL plans to you will always see `granularity = all` ) but if you set granularity to, say, "day" you will get a topN per timechunk (of day).

My original use case has a group by on multiple dimensions. I removed that detail in my example. 

I am not sure if it is feasible for you to issue native queries in your use case but FYI

i don’t think the native group by query pushes limit into the query granularity buckets

I guess if you wanted to group on multiple dimensions you could make a virtual column that concatenates them `d1###d2###f3`

back in the day before there was a groupBy query that is what the kids used to do

hmm, is it correct that group by limit doesn’t push the limit into the query granularity buckets?

aha! That’s a good point. I would need to then delimit the dimension values returned in the query results, I presume <@U030W50CNAC> ? 

like if query granularity is ‘all’ the current behavior is fine, which it basically always is set to if using sql

<@U030C8H59T8> - there is this config
```druid.query.groupBy.applyLimitPushDownToSegment	```
which probably does what you are looking for? Not sure, TBH.

```If Broker pushes limit down to queryable data server (historicals, peons) then limit results during segment scan. If typically there are a large number of segments taking part in a query on a data server, this setting may counterintuitively reduce performance if enabled.	```

we would at least need a mode to have that behavior if we ever want to replace the topn engine with specialized mode of the more general grouping engine

it just pushes the limit down to the segment scan, but doesn’t make it per query granularity

that topn with query granularity is pretty useful, am surprised there is no way to do with grouping engine with limits

hmm, the console for some reason doesn't like me specifying virtual columns.

console does not care about what you put IN your query

```{
  "queryType": "topN",
  "dataSource": "datasource",
  "dimension": "combined_column",
  "threshold": 5,
  "metric": "some_metric",
  "granularity": "HOUR",
  "virtualColumns": [
    {
      "type": "expression",
      "name": "combined_column",
      "expression": "concat(dim1,dim2)",
      "outputType": "STRING"
    }
  "aggregations": [
    {
      "type": "doubleSum",
      "name": "some_metric",
      "fieldName": "m1"
    }
  ],
  "intervals": [
    "2022-10-01T00:00:00.000/2022-10-02T00:00:00.000"
  ]
}```


&gt; console does not care about what you put IN your query
yeah, I am surprised too

yeah it won't let you submit if it can not parse what you wrote as hjson

Fun Friday Fact for everyone: did you know that if you are entering a native (json) query into the web console it actually parses it as <https://hjson.github.io/> and converts it to JSON for posting. So feel free to leave out `"` and `,` ! `]` are still required.

Thanks much, <@U030W50CNAC>! Virtual columns is a hidden gem in Druid, I believe. I haven't used it often, but I can see how powerful they are.

Will Druid SQL also use top n under the hood with the virtual column if I were to use concat?

yes, and you can always verify for yourself by writing a short query with concat like `SELECT a || b FROM some_table` and clicking `Explain`

like: `SELECT page || user FROM wikipedia`

Yeah, I know of the EXPLAIN feature. I recommend internally to a lot of our developers to validate their native queries.

I wanted to check if Druid will be using the top n query with virtual column if I formulate by query as below:
```select dim1 || dim2,
  sum(m1) s
from datasource
where __time &gt;= TIMESTAMP '2022-10-01 00:00:00.000'
  and __time &lt; TIMESTAMP '2022-10-02 00:00:00.000'
group by (dim1 || dim2)
order by s desc
limit 10```
and it did except that it set the granularity to "all" which as you explained above is a known behavior.

I wonder if there is a config to override the granularity setting for the sql query.

```{
	"queryType": "topN",
	"dataSource": {
		"type": "table",
		"name": "datasource"
	},
	"virtualColumns": [{
		"type": "expression",
		"name": "v0",
		"expression": "concat(\"dim1\",\"dim2\")",
		"outputType": "STRING"
	}],
	"dimension": {
		"type": "default",
		"dimension": "v0",
		"outputName": "d0",
		"outputType": "STRING"
	},
	"metric": {
		"type": "numeric",
		"metric": "a0"
	},
	"threshold": 10,
	"intervals": {
		"type": "intervals",
		"intervals": ["2022-10-01T00:00:00.000Z/2022-10-02T00:00:00.000Z"]
	},
	"filter": null,
	"granularity": {
		"type": "all".   --------&gt; need this to be HOUR. 
	},
	"aggregations": [{
		"type": "longSum",
		"name": "a0",
		"fieldName": "m1",
		"expression": null
	}],
	"postAggregations": [],
	"context": {
		"sqlOuterLimit": 100,
		"sqlQueryId": "6436af92-d524-4134-a5bd-127dc8e712e7"
	},
	"descending": false
}```


I think SQL will always set it to `all` I am not sure if there is an override

image.png

what about trailing semicolon vs no trailing semicolon?

if JSON and YAML ever produce a child, this is what it will look like :smile:

I think you might find what you need in <https://druid.apache.org/docs/24.0.0/operations/metrics.html|Metrics>. Druid can emit metrics about many of its internal processes, including query metrics (which includes `query/bytes` which I think is what you need). Imply's Clarity provides all metrics out of the box and provides a UI to navigate them. You can alternatively use one of the emitter <https://druid.apache.org/docs/24.0.0/development/extensions.html#community-extensions|extensions that the community has contributed  >(graphite-emitter, statsd-emitter, kafka-emitter, ambari-metrics-emitter, druid-influxdb-emitter, prometheus-emitter).

Druid Summit is coming to Tel Aviv! Join us on 27 October to hear leaders from ironSource, ZipRecruiter, Singular, AWS and more as they share real-world use cases of #ApacheDruid in production. <https://druidsummit.org/events/druid-summit-Tel%20Aviv-2022?utm_source=linkedin&amp;utm_medium=organic-social&amp;utm_campaign=druid-summit|Register> today!

cc <@U030W50CNAC> <@U030K4UM3H7> <@U03149ALHN3>

Btw, I am submitting a query from Looker via Looker Druid connection.

The "query bytes" metrics report the response size sent over the HTTP API

`query/bytes` isn't set in request logs for the native query, if the original query came from SQL, because the native query doesn't actually go over the wire via HTTP

There is `sqlQuery/bytes` for the SQL query, if you use the HTTP API. It isn't set for JDBC, though, for the same reason: with JDBC the SQL results do not go out directly over a single HTTP call

I'm not sure if this is documented, but if not it would be a good addition!

Hi <@U0344FW86DD>

Below is the request log for sqlQuery from Looker. `sqlQuery/bytes` also outputs as -1.
```2022-10-11T18:40:03.162Z                        {"sqlQuery/time":747,"sqlQuery/bytes":-1,"success":true,"context":{"sqlQueryId":"cdbb8016-0c07-43ee-a53d-b6b3ed3c8038","nativeQueryIds":"[8195ce07-bce5-465b-9079-d9e1a2bb8c5d]"},"identity":"allowAll"}        {"query":"-- Looker Query Context '{\"user_id\":3117,\"history_id\":59894910,\"instance_slug\":\"0045ad671f9e3537c627a1e669873855\"}' \n-- Looker limited the size of this query to avoid running out of memory.\n\nSELECT * FROM (\nSELECT * FROM druid.experiment_druid LIMIT 10\n) AS force_limit LIMIT 5000","context":{"sqlQueryId":"cdbb8016-0c07-43ee-a53d-b6b3ed3c8038","nativeQueryIds":"[8195ce07-bce5-465b-9079-d9e1a2bb8c5d]"}}```
All in all for my use case, there are users who create dashboards on looker and those dashboards query druid to get the data. I am looking to get `query/bytes` for all those dashboard queries. If for some reason, a user via dashboard tries to query druid to get huge data we want to notify that user.
Given this, I would need `query/bytes` or `sqlQuery/bytes`  for all those queries sent to druid via Looker triggered by user. How would I get this info, in this case?

<@U0344FW86DD> It would be great if you can provide input for me on this. Trying to figure out for a while now.

<@U03RH9440HY> I think what Gian is saying is that JDBC connections do not produce that metric.

<@U030K4UM3H7> So you are saying Looker dashboard queries to Druid are using JDBC connection and hence we do not see the `query/bytes` or `sqlQuery/bytes` . Is this correct?

Also can you suggest what is better way to solve my above use case?

That’s what I’m understanding from Gian’s post. I wonder if you can see those metrics in the  metrics emitter for Historicals and MMs that you could aggregate. I’ll have to try that out to find out. 

<@U0344FW86DD>

&gt; `query/bytes` isn't set in request logs for the native query,
In this doc - <https://druid.apache.org/docs/latest/querying/querying.html>, it mentions `Druid's native query language is JSON over HTTP,` . So isn't this a HTTP request and should be able to get `query/bytes`
?

Looker uses SQL to query Druid (<https://help.looker.com/hc/en-us/articles/4420185407123-Apache-Druid> - Here it enables to enable SQL when using with Looker). Given this, are you also saying that we do not get `query/bytes` when using Druid SQL as well from Looker?

Also, can you also help clarify this where you mention `sqlQuery/bytes` is also not populated because Looker dashboard queries to Druid are based of JDBC connection?  Please note that we are using this - <https://help.looker.com/hc/en-us/articles/4420185407123-Apache-Druid> for establishing a connection to Druid from Looker.

&gt; The "query bytes" metrics report the response size sent over the HTTP API
Can you give an example, where I can issue SQL Query over HTTP API and have `query/bytes`  or `sqlQuery/bytes` populated?

Also, wanted to know whether the below is possible?

&gt; I would need `query/bytes` or `sqlQuery/bytes`  for all those queries (triggered via Looker dashboards) sent to druid via Looker triggered by user.

From the <https://help.looker.com/hc/en-us/articles/4420185407123-Apache-Druid> posted, there are configurations for Avatica JDBC parameters suggested. I suspect that Looker might be querying Druid using JDBC.
If there is a way to access the Druid cluster via HTTP, can you please check if running a simple query using `/druid/v2/`  or ``/druid/v2/sql`  emits this metric? It might help us understand if JDBC is the problem or if there is a bug or configuration error.

<@U031AD802MN> Tried submitting the query to druid via HTTP request

```curl -XPOST -H'Content-Type: application/json' <http://ROUTER:8082/druid/v2/sql> -d @query.json```
Here i was able to see `query/bytes` is still `-1` but `sqlQuery/bytes` is populated now.

```2022-10-12T17:23:31.591Z		{"query/time":10,"query/bytes":-1,"success":true,"identity":"allowAll"}


2022-10-12T17:23:31.582Z	10.220.14.235		{"sqlQuery/time":19,"sqlQuery/bytes":1226,"success":true,"}```

<@U031AD802MN> <@U0344FW86DD> Is there a way to get the bytes scanned metric for druid?

For example, when a Query is submitted to Druid (like a Group By clause) then the druid might scan multiple bytes on druid datasources (lets say 100 GB) before returning the actual output (which might just be 1 GB). So I am assuming `query/bytes` (or) `sqlQuery/bytes` would capture it as 1 GB, but I am actually looking for `bytes scanned` metric which will the druid process has scanned `100 GB` of data.

<@U031AD802MN> <@U0344FW86DD> Also trying to understand how does Imply charge its customers for such scenarios where for example a user query will scan the entire `100 GB` bytes data in druid datasources (which means high cost on druid to process the query) but the actual query result can only be `10 MB` .

How would you attribute the costs in such cases?

&gt; For example, when a Query is submitted to Druid (like a Group By clause) then the druid might scan multiple bytes on druid datasources (lets say 100 GB) before returning the actual output (which might just be 1 GB). So I am assuming `query/bytes` (or) `sqlQuery/bytes` would capture it as 1 GB, but I am actually looking for `bytes scanned` metric which will the druid process has scanned `100 GB` of data.
Regarding this, I see there are a few metrics that might be of relevance:
1. On Broker, `query/node/time`  and `query/node/bytes`  seems the most useful, however there would be a group by on the broker level as well, so this might represent less data then needed. 
2. On Historicals and Realtime servers, `query/time`  which can tell the time required to complete the query on respective nodes. 
3. On Broker, `query/segments/count` , multiplied with the average query size, however there is a room for variation here.
I didn’t find a clear cut metric which solves the query that you were looking for. May I know the use case that you are looking for. There is some information here <https://druid.apache.org/docs/latest/querying/multitenancy.html#supporting-high-query-concurrency>, if that’s the aim.

If you think a metric like `query/bytes`  would be useful for Historical/Realtime servers, can you please create a GH issue with the relevant information and the usecase. It would reach a wider audience, and suggestions if I have missed out on any.

&gt; 1. On Broker, `query/node/time`  and `query/node/bytes`  seems the most useful, however there would be a group by on the broker level as well, so this might represent less data then needed.
`query/node/bytes`  represents `Number of bytes returned from querying individual historical/realtime processes`. And `query/node/time` represents `Milliseconds taken to query individual historical/realtime processes.`

So `query/node/bytes` only provides the bytes returned for the query but not the amount of bytes processed for the query. I am actually looking for the amount of bytes processed internally for the query rather than bytes in result query.

&gt; May I know the use case that you are looking for.
We are trying to figure out a cost attribution model for Druid. For us, Druid cluster is deployed on AWS and we know the overall monthly costs for the Druid cluster.

Data will be ingested to Druid datasources from S3. And users build Looker dashboards which internally query Druid.

So we thought `bytes scanned` (or) `bytes processed` is a good metric for the queries submitted to Druid. The reason we have selected this is because a complex SQL query might scan/process a lot of bytes internally `which will attribute to processing costs` for Druid.

User1 -&gt; Druid Query (Query1) -&gt; Get Byte scanned from Druid
User 1 -&gt; Druid Query (Query2) -&gt; Get Byte scanned from Druid
User 2 -&gt; Druid Query (Query3) -&gt; Get Byte scanned from Druid

This way we have all the bytes scanned for all queries for all different users and we can assign a cost to each of these users out of the total cost.

`Note:` Please note that `bytes scanned` (or) `bytes processed` is different from `bytes returned` for query response.

Also can you provide your input on how Imply does cost attribution for Druid? What kind of metrics do you use for cost attribution?

Using bytes scanned is actually an interesting idea.

For us, we do cost attribution by mapping datasources to internal teams as owners.

Then I broke down the datasource by data size %, query volume %, and top slowest query volume %.

<@U03149ALHN3> Just wondering how data size %, query volume %, and top slowest query volume % can attribute to cost. Is there a rough equation or scenario you can explain how it can attribute to cost?

Basically, I am looking to see if it can done  on our side.

Since we are on AWS we know roughly how much our monthly spend is, right?

And since we are an internal cloud team, we eat the cost first and then charge back our internal customers based on the percentage breakdown.

It’s not an exact math, but everyone involved is happy enough 

&gt; Since we are on AWS we know roughly how much our monthly spend is, right?
&gt; 
&gt; And since we are an internal cloud team, we eat the cost first and then charge back our internal customers based on the percentage breakdown.
This is almost exactly the same scenario we have as well.

For cost attribution, perhaps doing it by CPU time is best?

I suggest it because generally, CPU is the most precious resource in a Druid cluster

Good way to do this would be to set up Druid metrics emitting, and then use the `query/cpu/time` metric. You can break it down by datasource, or by user (if you are using Druid-based authentication), and in various other ways

I think it's better than bytes, because sometimes a very small result set takes a lot of CPU to compute, and sometimes a big one is easy to compute

Thanks <@U0344FW86DD> for your input. Yes, we have decided to use that metric `query/cpu/time` from the <https://druid.apache.org/docs/24.0.0/operations/metrics.html#historical|historical metrics> for our cost attribution. So a couple of questions

1. I don't see this metric being displayed when I fire a query.
```2022-10-14T22:23:03.624Z	10.220.14.235	{"queryType":"scan","dataSource":{"type":"table","name":"experiment_druid"},"intervals":{"type":"segments","segments":[{"itvl":"2020-11-01T00:00:00.000Z/2020-11-02T00:00:00.000Z","ver":"2022-07-19T08:38:54.286Z","part":0}]},"virtualColumns":[],"resultFormat":"compactedList","batchSize":20480,"limit":10,"order":"none","filter":null,"columns":["__time","client","create_date","end_date","id","identifier"],"legacy":false,"context":{"defaultTimeout":300000,"finalize":false,"maxQueuedBytes":50000000,"maxScatterGatherBytes":9223372036854775807,"queryFailTime":1665786483616,"queryId":"30033a71-6eac-4f7f-8cd4-b4983fd7d8d9","scanOutermost":false,"sqlQueryId":"418f1863-48c6-4557-8fef-30cb3320bdff","timeout":300000},"descending":false,"granularity":{"type":"all"}}	{"query/time":93,"query/bytes":2506,"success":true,"identity":"allowAll"}```
I have enabled RequestLogging as below on the data server node (`common.runtime.properties` firle)
```druid.request.logging.type=file
druid.request.logging.dir=/opt/druid/var/request
druid.request.logging.durationToRetain=P1D```
Is there anything i need to enable to output this metric in the logs?

2. Is there a possibility that a single query request might be sent to multiple historical nodes depending on the query and each historical node give its own `query/cpu/time` ? In that case, should we aggregate the `query/cpu/time`  from all historical nodes to get the CPU time for the entire query?

It's in the metrics, not the request logs

And yes, you should aggregate across all servers

Documentation on metrics is here btw: <https://druid.apache.org/docs/latest/operations/metrics.html>

Typically they are emitted to a metric collection system using one of the core emitters <https://druid.apache.org/docs/latest/configuration/index.html#metrics-emitters> or the kafka emitter <https://druid.apache.org/docs/latest/development/extensions-contrib/kafka-emitter.html>

at Imply we use Druid itself: the metrics are sent to Kafka and then Druid reads from Kafka, and then we can query the metrics using Druid itself

(We use a separate cluster from the production cluster for this)

Here's an example of what it looks like for one of our clusters once aggregated. Datasource names hidden, but each line is a separate datasource

Hi Gian. Thanks for the detailed explanation.

As an experiment, I want to first try by logging data to a local file on the node. Given this, on the `data server` node in the `common.runtime.properties` file, I have added the following

```druid.emitter=logging
druid.emitter.logging.logLevel=debug```
and specified the `log4j2.xml` file

```&lt;?xml version="1.0" encoding="UTF-8"?&gt;
&lt;Configuration status="warn" name="Imply" packages=""&gt;
    &lt;Appenders&gt;
        &lt;RollingFile name="RollingFile" fileName="var/sv/${sys:service}-service.log"
                     filePattern="var/sv/${sys:service}-service-%d{yyyy-MM-dd}.%i.log.gz"&gt;
            &lt;PatternLayout&gt;
                &lt;Pattern&gt;%d{ISO8601} %p [%t] %c - %m%n&lt;/Pattern&gt;
            &lt;/PatternLayout&gt;
            &lt;Policies&gt;
                &lt;TimeBasedTriggeringPolicy interval="1" modulate="true"/&gt;
                &lt;SizeBasedTriggeringPolicy size="100 MB"/&gt;
            &lt;/Policies&gt;
            &lt;DefaultRolloverStrategy max="10"/&gt;
        &lt;/RollingFile&gt;
    &lt;/Appenders&gt;
    &lt;Loggers&gt;
        &lt;Root level="info"&gt;
            &lt;AppenderRef ref="RollingFile"/&gt;
        &lt;/Root&gt;
    &lt;/Loggers&gt;
&lt;/Configuration&gt;```
I made the settings and restarted the `historical` process on the `data server` nodes. However, when I fire the query now I don't see any metrics being output to `var/sv/historical-service.log` file now. Is there anything I am missing now?

The loglevel you set is the one that the message go out at

So "info" is better if your overall log level is info. Otherwise you won't see them

You’ll need to add monitors to the services you want to produce metrics from:
<https://druid.apache.org/docs/24.0.0/configuration/index.html#enabling-metrics|https://druid.apache.org/docs/24.0.0/configuration/index.html#enabling-metrics>

So in this case, I would need to add `org.apache.druid.server.metrics.HistoricalMetricsMonitor` ? For example,

```druid.monitoring.monitors=["org.apache.druid.server.metrics.HistoricalMetricsMonitor"]
druid.emitter=logging
druid.emitter.logging.logLevel=info```
Please correct me if I am wrong.

There is no monitor needed for `query/cpu/time`. It's always emitted regardless of monitors

Thanks <@U0344FW86DD> for all the information you provided.

Given the above info, I am trying the following use case where I am trying to *map the user submitted query to total CPU time taken to complete the query*. Can you please confirm if my understanding below is correct?

1. From request logs, Map Query to NativeQueryIds.
2. For each NativeQueryId, compute the cumulative sum of `query/cpu/time` across all the nodes (nodes include historical and brokers).
3. Total CPU time for query = [Total CPU Time for NativeQueryId1 + Total CPU Time for NativeQueryId2 + Total CPU Time for NativeQueryId3 .......]
For example see below,

1. I get the info about `query -&gt; nativeQueryIds`
```{"feed":"druidRequestLogFeed","timestamp":"2022-10-18T18:43:42.321Z","service":"druid_dev/broker","remoteAddr":"","sql":"-- Looker Query Context '{\"user_id\":3222,\"history_id\":60223411,\"instance_slug\":\"0045ad671f9e3537c737a1e669873855\"}' \n-- Looker limited the size of this query to avoid running out of memory.\n\nSELECT * FROM (\n&lt;&lt;&lt; SAMPLE QUERY &gt;&gt;&gt;\n) AS force_limit LIMIT 5000","queryStats":{"sqlQuery/time":284,"sqlQuery/bytes":-1,"success":true,"context":{"sqlQueryId":"7db189df-019c-4f5d-96ae-b05b1e0af2b9","nativeQueryIds":"[eda63995-8b3d-4dd5-8e2e-acb8aed40089, 9fade51c-b844-463b-a811-61d1e329ea2c]"},"identity":"allowAll"},"query":null,"host":"druid-dev-query-server:8082"}```
2. Get the `query/cpu/time` metric from all historical and broker service nodes for `nativeQueryId`  -&gt; `eda63995-8b3d-4dd5-8e2e-acb8aed40089`. CPU time taken for `nativeQueryId`  -&gt; `eda63995-8b3d-4dd5-8e2e-acb8aed40089` is the sum of all `query/cpu/time`  values from all nodes for this nativeQueryId.

```{"feed":"metrics","timestamp":"2022-10-18T18:43:42.479Z","service":"druid_dev/historical","host":"druid-dev-data-server3:8083","version":"0.19.0","metric":"query/cpu/time","value":49668,"dataSource":"dev_table","duration":"PT86400S","hasFilters":"false","id":"eda63995-8b3d-4dd5-8e2e-acb8aed40089","interval":["2022-09-20T00:00:00.000Z/2022-09-21T00:00:00.000Z"],"type":"scan"}```
```{"feed":"metrics","timestamp":"2022-10-18T18:43:42.544Z","service":"druid_dev/historical","host":"druid-dev-data-server2:8083","version":"0.19.0","metric":"query/cpu/time","value":100262,"dataSource":"dev_table","duration":"PT172800S","hasFilters":"false","id":"eda63995-8b3d-4dd5-8e2e-acb8aed40089","interval":["2022-06-25T00:00:00.000Z/2022-06-26T00:00:00.000Z","2022-09-10T00:00:00.000Z/2022-09-11T00:00:00.000Z"],"type":"scan"}```
```{"feed":"metrics","timestamp":"2022-10-18T18:43:42.597Z","service":"druid_dev/historical","host":"druid-dev-data-server1:8083","version":"0.19.0","metric":"query/cpu/time","value":151986,"dataSource":"dev_table","duration":"PT259200S","hasFilters":"false","id":"eda63995-8b3d-4dd5-8e2e-acb8aed40089","interval":["2022-05-31T00:00:00.000Z/2022-06-02T00:00:00.000Z","2022-09-25T00:00:00.000Z/2022-09-26T00:00:00.000Z"],"type":"scan"}```
```{"feed":"metrics","timestamp":"2022-10-18T18:43:42.605Z","service":"druid_dev/broker","host":"druid-dev-query-server:8082","version":"0.18.1","metric":"query/cpu/time","value":12562,"dataSource":"dev_table","duration":"PT9223372036854775.807S","hasFilters":"false","id":"eda63995-8b3d-4dd5-8e2e-acb8aed40089","interval":["-146136543-09-08T08:23:32.096Z/146140482-04-24T15:36:27.903Z"],"type":"scan"}```
3. Similarly get the `query/cpu/time` metric from all historical and broker service nodes for another `nativeQueryId`  -&gt; `9fade51c-b844-463b-a811-61d1e329ea2c`. CPU time taken for `nativeQueryId`  -&gt; `9fade51c-b844-463b-a811-61d1e329ea2c` is the sum of all `query/cpu/time`  values from all nodes for this nativeQueryId.

So total CPU taken for query is `(2) + (3)`.

Is this understanding correct?

cc <@U030K4UM3H7>

For completeness, If your queries also hit real-time segments (from streaming ingestion) you will also want to gather metrics from Indexers/MM tasks. 

&gt; also hit real-time segments (from streaming ingestion)
Looking at the architecture diagram here - <https://druid.apache.org/docs/latest/design/architecture.html>. I have changed the `common.runtime.properties` on Query server nodes and Data server nodes to emit metrics/request logging. Where does real-time segments come into picture here?

&gt; you will also want to gather metrics from Indexers/MM tasks.
Does Indexer/MM also emit these metrics? I don't see anything from here - <https://druid.apache.org/docs/latest/operations/metrics.html>.

I see this section <https://druid.apache.org/docs/24.0.0/operations/metrics.html#real-time> in the docs, but it does not contain `query/cpu/time`.

Looks like <@U0344FW86DD> has a PR - <https://github.com/apache/druid/pull/13229> which adds to `query/cpu/time` in real-time metrics. I think its just missing from doc.

right! So it should be emitted. I think if you are using Middle Managers, it will be the Peons that emit them. Not sure how that plays with the logging emitter as that would go to the individual task logs.

So just to confirm again *`overall CPU total time taken for the query`* = Cumulative for `query/cpu/time` of `nativeQueryId1` on all `historicals, brockers and real-time process` + Cumulative for `query/cpu/time` of `nativeQueryId2` on all `historicals, brockers and real-time process` + Cumulative for `query/cpu/time` of `nativeQueryId3` on all `historicals, brockers and real-time process` + ..... so on.....

that's accurate but I think you're making it more complicated than it needs to be. All of the native query metrics are also tagged with sqlQueryId, so you can use that directly

also, if you wanted to find CPU use for all queries done by a specific user, that's tagged as `identity` in the metrics. no need to deal with individual query IDs

Hi <@U0344FW86DD>.

&gt; All of the native query metrics are also tagged with sqlQueryId,
This is not happening.

If you see my example above

```{"feed":"metrics","timestamp":"2022-10-18T18:43:42.479Z","service":"druid_dev/historical","host":"druid-dev-data-server3:8083","version":"0.19.0","metric":"query/cpu/time","value":49668,"dataSource":"dev_table","duration":"PT86400S","hasFilters":"false","id":"eda63995-8b3d-4dd5-8e2e-acb8aed40089","interval":["2022-09-20T00:00:00.000Z/2022-09-21T00:00:00.000Z"],"type":"scan"}```
This metric emitted by historical for `query/cpu/time` doesn't include `sqlQueryId` . It only includes `id` which maps to `nativeQueryId` .

See below `nativeQueryId` in request log for the query.
```{"feed":"druidRequestLogFeed","timestamp":"2022-10-18T18:43:42.321Z","service":"druid_dev/broker","remoteAddr":"","sql":"-- Looker Query Context '{\"user_id\":3222,\"history_id\":60223411,\"instance_slug\":\"0045ad671f9e3537c737a1e669873855\"}' \n-- Looker limited the size of this query to avoid running out of memory.\n\nSELECT * FROM (\n&lt;&lt;&lt; SAMPLE QUERY &gt;&gt;&gt;\n) AS force_limit LIMIT 5000","queryStats":{"sqlQuery/time":284,"sqlQuery/bytes":-1,"success":true,"context":{"sqlQueryId":"7db189df-019c-4f5d-96ae-b05b1e0af2b9","nativeQueryIds":"[eda63995-8b3d-4dd5-8e2e-acb8aed40089, 9fade51c-b844-463b-a811-61d1e329ea2c]"},"identity":"allowAll"},"query":null,"host":"druid-dev-query-server:8082"}```


oh, sorry, i was mistaken. we have that customization added to our imply version of druid, but i don't think it's there in apache druid

if you wanted something similar, &amp; if you build your own version, you can add `setDimension("sqlQueryId", query.getSqlQueryId());` to `sqlQueryId` of `DefaultQueryMetrics`

we could also consider adding this to the standard apache version; i'm not sure if there is a reason why it's not there. perhaps the idea was to save space in the metrics

<@U0344FW86DD> It would have been really great if this would have been part of Apache druid. Is there any specific reason why it was originally not part of apache druid

there's a bunch of metrics that are off by default in the standard apache build; i think the idea was to save space since some people have limited metric storage available

Oh got it. But as far for now with current Apache druid version , this hold true right - <https://apachedruidworkspace.slack.com/archives/C0303FDCZEZ/p1666142039444669?thread_ts=1665514948.145529&amp;cid=C0303FDCZEZ> ?

we could certainly re-open the discussion

yes, in the standard apache build, you would need to map the sql query to native query IDs manually

i feel this is a pain though and it would be easier to build your own version with the patch i mentioned

or, reopen the discussion about including it by default

but, it's up to you what you think is easiest for you

&gt; i feel this is a pain though and it would e easier to build your own version with the patch i mentioned
The problem is that we have to maintain and build a custom apache druid version every time for every new release. it would have been more simple if its directly part of the community release.

&gt; reopen the discussion about including it by default
I will try this one.

But thanks <@U0344FW86DD> and <@U030K4UM3H7> for all the discussion on the thread. Both of your input was very helpful to me.

Hi <@U0344FW86DD> <@U030K4UM3H7>, I have deployed my changes in our Prod env. based on this model explained here- <https://apachedruidworkspace.slack.com/archives/C0303FDCZEZ/p1666144576378839?thread_ts=1665514948.145529&amp;cid=C0303FDCZEZ> .

However I see 2 issues:

1. Issue with `Event too large to emit` error. Because of this, there is a possibility I might miss some important metrics.
2. Also i see some `query/cpu/time` metric which doesn't have a corresponding Request Log metric which contains the sqlQueryId -&gt; nativeQueryId[] mapping

for example, there is no sqlQueryid associated with the nativeQueryId `dfe408fc-790a-4eae-b2b6-39503fdd8501` as mentioned below. I am not sure if I lost the metric because of 1st issue.

```[INFO ] 2022-11-11 08:29:45.134 [qtp1490509465-17] HttpServerMain - Query CPU metric is: {"feed":"metrics","timestamp":"2022-11-11T08:28:34.844Z","service":"druid-blue/broker","host":"<http://druid-prod-blue-query-server-1025.bdp.roku.com:8082|druid-prod-blue-query-server-1025.bdp.roku.com:8082>","version":"0.18.1","metric":"query/cpu/time","value":54659,"dataSource":"hello","duration":"PT6998400S","hasFilters":"false","id":"dfe408fc-790a-4eae-b2b6-39503fdd8501","interval":["2022-02-11T00:00:00.000Z/2022-04-07T22:00:00.000Z","2022-04-07T23:00:00.000Z/2022-05-03T01:00:00.000Z"],"type":"segmentMetadata"}```
Ca you please provide your inputs on these?

&gt; Issue with `Event too large to emit` error.
More info about logs can be found here on this thread - <https://apachedruidworkspace.slack.com/archives/C0309C9L90D/p1668156960221749?thread_ts=1668122316.719779&amp;cid=C0309C9L90D>

Also, Can data ingestion into druid cause any `query/cpu/time` metric to be fired as part of its ingestion process? Can that be one of the reason of point 2 above?

<@U030K4UM3H7> do you have any thoughts on Question 2 above?

tagging <@U0344FW86DD> to see if he can provide some input on this.

<@U03RH9440HY> the query  `"type":"segmentMetadata"` is are native query, it does not have a direct corresponding SQL version. So I think these are expected to only have nativeQueryId's.

Could there be Native groupBy queries submitted on your cluster? or is it all SQL?

No., its only SQL queries submitted via Looker to Druid

Seeing this for some `groupby` , `topN` queries as well. there is no request log which corresponds to the sqlquery id associated with it.

For example, nativeQueryId - `5be37dc3-8028-4803-8a35-27d6b0c4b995`  is amounting to ``10127715324`  micro seconds of CPU time which is not attributed to any user fired query.
This is making it difficult to attribute the CPU time to any user fired query.

Please note that in our case, user fires query to druid via Looker only.

&gt; 2. Also i see some query/cpu/time metric which doesn't have a corresponding Request Log metric which contains the sqlQueryId -&gt; nativeQueryId[] mapping
for (2), the query type is `segmentMetadata`. this is possibly one of the internally-generated `segmentMetadata` queries that druid issues to itself

it is not associated with a SQL query log since it was made directly as a native query

&gt; 1. Issue with Event too large to emit error. Because of this, there is a possibility I might miss some important metrics.
for (1), i'm surprised you see so many metrics bigger than the 1 MB limit. that's a pretty big metric! any idea why they're so large? i think we have two paths here:

• if they are large for some weird reason, we can fix that reason and make them fit under the limit
• if they are large for a good reason, we can make the limit configurable and you can raise it in your environment

Thanks Gian for your input.

&gt; for (2), the query type is `segmentMetadata`. this is possibly one of the internally-generated `segmentMetadata` queries that druid issues to itself (edited)
I am also seeing this for some `groupby` , `topN` queries which doesn't have a corresponding sqlQuery. Not sure if this is expected behavior?

&gt; for (1), i'm surprised you see so many metrics bigger than the 1 MB limit. that's a pretty big metric! any idea why they're so large? i think we have two paths here:
We have a pretty large cluster. i am not sure if its because of that. I will check it out again.

do you know whether there's ever been any discussion around it, whether it's been considered to be added?

for my use-case I can unfortunately only consume with fan-out

Could you please share how you plan to ingest data? There are a few tuning configurations which may be able to help.

the fan-out requirement comes from another team providing the stream, for all tenants. is the implication that certain tuning configs could negate the detrimental effect of consuming without fan-out? let's just say I'm looking to consume 500 mbps of data over 3000 shards as a rough example, and I was originally hoping to do so with the core kinesis extension

You could try to increase `recordsPerFetch`  while also increasing `fetchDelayMillis` .
This may require you to increase properties such as `recordBuffer*`  while also increasing the task memory

Given the number of shards, you could also try using a relatively higher task count which may require even less tuning.

thanks! it's good to know tuning can reduce the load per shard (if I understand right). I guess though, no matter how much I tune, there will be a load &gt; 0 and thus a reduction of capacity available for other stream tenants, so it seems somewhat unlikely that that alone would be able to solve my problem. it may well turn out to be a hard requirement on fan-out

No, a single supervisor can only read from a single topic, and you can have just one supervisor for a datasource.
A (not-so-great) workaround could be to have one datasource per topic and then periodically append this data into one common datasource. For the append job, you might find the new multi-stage-query engine very helpful.
<https://druid.apache.org/docs/24.0.0/multi-stage-query/concepts.html#load-data-with-insert>

One more workaround could be to do a UNION query over the multiple datasources you have per kafka topic / cluster

thanks！because our druid version is under 24.0.0，so we plan to use “union all” query to satisfy this case.

I think this would be a good feature ask.
I have raised a github issue : <https://github.com/apache/druid/issues/13222>
Support for that helps with priority.

&gt; How can I write a custom expression
Are you looking for examples?

You can't add custom syntax to the expression grammar. You can add custom functions though. So you can add something like custom `SUM` but not custom `+`

examples - <https://github.com/apache/druid/blob/master/extensions-core/druid-bloom-filter/src/main/java/org/apache/druid/query/expressions/BloomFilterExpressions.java|https://github.com/apache/druid/blob/master/extensions-core/druid-bloom-filter/src[…]/org/apache/druid/query/expressions/BloomFilterExpressions.java>

It's meant to help you migrate from a local setup (Derby) to a clustered setup (metadata store)

If you're migrating from mysql, you can use `mysqldump`

but we need to modify payload loadspec to match with s3 . is there any tool that can do it or we need to write our own script to modify ?

<@U0344FW86DD> i copied one hour data to s3 and inserted metadata into mysql . after that i am able to see the segments from router ui but when i query the table for that time interval i am not getting results . what could be the issue ?

in the Datasources tab do you see the datasource as available or not? if it has segments, but is not available, it may be some issue with loading them. perhaps the Historicals cannot fetch the segmentt files rom S3? checking their logs may give you some clues

<@U0344FW86DD> i see segments are available and i checked on historicals i see the smoosh files

hmm. odd. in that case you should be able to query the table just fine.

if you do `select count(*) from the_table` does that show a count?

yes it is showing but we have some segments which are directly loaded on to s3 instead of moving from hdfs to s3 .

bit confused about what is wrong? are you still seeing a problem with not getting results?

i see issue for the hour where i copied data from hdfs to s3 and inserted rows in to metadata db . i changed only loadspec part . do i have to change anything else ?

nah, changing the loadspec is enough. however you do need to make sure you got it right: if  not then the historicals will fail to download the files from s3

i was suggesting checking for those specific files on the historicals, or for error logs about not being able to pull segments

No, `transformSpec` takes a single row as input and outputs a single value. It cannot be used to explode rows.

On a related note, if you have nested data, you need not explode your rows and can access the nested fields easily. This would typically address many such use cases.
Say, if had something like `year = 2022` , `counts={"jan": 1, "feb": 1, "oct": 1}` .
<https://druid.apache.org/docs/latest/querying/nested-columns.html>

If you don't have realtime requirements, you should consider batch. In particular, you could try the new multi-stage-query engine:
<https://druid.apache.org/docs/24.0.0/multi-stage-query/concepts.html#multi-stage-query-task-engine>

The advantage of batch over real-time is that you can organize the data in most read-optimized way from the get-go.

the realtime ingestion pipelines ingest using dynamic partitioning which is great for faster write. There is no shuffling involved in creating the segments. Later on the compaction kicks in, which is a native batch job that shuffles and arranges the data in a way that you most prefer.

If you are using batch ingestion the first time itself, you can shuffle the data and write it using range or hash partitioning. Such segments would also not need to be compacted.

The batch will certainly be more resource efficient than real-time ingestion if you use dynamic partitioning in batch. If you are using range or hash partitioning in batch, then your batch jobs are doing some extra work and could potentially take more resources.

batch ingestion has some optimizations built-in to take less resources since batch ingestion doesn't need to serve queries.

Though, you should definitely avoid dynamic partitioning as it is not recommended for good query performance.

Thank you both for the detailed answers.  This is helpful and answers my question.

Hi Everyone, not sure if this is the right forum but Druid’s been winning some awards lately. Just found out this morning Druid is part of InfoWorld’s 2022 The best open source software of 2022: <https://www.infoworld.com/article/3676829/the-best-open-source-software-of-2022.html#slide24>

Also, Druid was also nominated in the 2022 Datanami Readers Choice Awards. We’re up against Pinot and Clickhouse (and several others) in the Best Data and AI Product or Technology: Analytics Database category. Polls close on Thursday at 11:59 PT: <https://www.datanami.com/2022-datanami-readers-choice-awards-polls-are-open/>

I think there is some discrepancy in the documentation for the config (<https://druid.apache.org/docs/latest/configuration/index.html>). The default value is set as 100TB in the latest version to avoid running into data limit problems - since the limit must be greater than the time chunk being compacted.
So, I think you can try with the current default value or not set it if you're on the latest version.
Meanwhile, the documentation could be clearer.

yes, unless you enable `forceLimitPushDown`

check here for some information: <https://druid.apache.org/docs/latest/data-management/update.html|https://druid.apache.org/docs/latest/data-management/update.html>

Oct 2022 CMotM.jpg

Thank you very much for the recognition! Thanks a bunch to the support from the community!

The query you've written seems fine to me! Coalesce should work as you expect it, but there are some catches around OUTER/RIGHT join results (ref : <https://druid.apache.org/docs/latest/querying/datasource.html#future-work-for-joins>).

This might be a reasonable starting point:

<https://druid.apache.org/docs/24.0.0/operations/security-overview.html>

This is a bit old and generic but I came across:

<https://www.druidforum.org/t/datasketches-tuple-sketch-module/5367>

i'm also not super familiar with tuple sketches :slightly_smiling_face:

i gather from the docs that a tuple sketch is like a theta sketch where, instead of just being able to do counts, we can also sample some vectors (arrays of doubles) associated with the original items, and compute means/variance of the columns of those vectors

so, you could use this with `metricColumns = 1` (just the `value`). they would be 1-vectors

if you want the sum, mean, and variance of your `value` then it seems like a good solution

(sum, I suppose, would be the _estimate_ times the _mean_)

although i have to admit i am not very familiar and could have this incorrect. the <https://datasketches.apache.org/|datasketches folks> have their own mailing list and slack if you'd like to chat with them directly

thanks <@U0344FW86DD> and <@U030AQ8AR8E> I’ll try to play with it and see if it’s working and gives good approximation
(yes, sum probably will be expressed and estimate * mean)

<@U047P93LEUT> did you manage with the tuple sketches? i'm facing the same now i'm trying to get a variance approximation and finding hard to find any examples of using tuple sketches in druid, do you have any example or can share info?

Hi <@U04DRDGD8UF> we left this direction and implemented some deduplication logic before Druid, so at the end we don’t count distincts with druid

indeed I had hard time to figure out how it should be used

Hi everyone, just a quick note: Datanami extended their 2022 Readers’ Choice Awards voting until tomorrow, Oct 28th. Druid is nominated for Best Data and AI Product or Technology: Analytics Database (which happens to be the first category on the ballot). If you’d like to cast your vote, here’s the link: <https://www.datanami.com/2022-datanami-readers-choice-awards-polls-are-open/>

Also, quick question: would y’all prefer a separate channel here for this type of news?

IMG_7133.jpg

Had a great time at Druid Summit - Tel Aviv today. Thank you everyone that came!

The ability to ingest sketches directly does exist in the tuple sketches module as well - I think it is probably not documented as well as the theta sketch ones.
For trying that, you should mention the `metricColumns` as empty in the aggregator definition (ref : <https://druid.apache.org/docs/latest/development/extensions-core/datasketches-tuple.html>)

Thank you for your help Rohan! I’ll try that.

With the latest version of Druid (24.0), you can utilize the new MSQ engine to run the queries like above. Please refer to the documentation here: <https://druid.apache.org/docs/latest/multi-stage-query/index.html>.

A note here: every INSERT query will generate at minimum 1 segment (if there is data to insert). So you should definitely not use it to ingest 1 or a handful of rows at a time. It would lead to poor performance.

there is an openid connect extension: <https://druid.apache.org/docs/latest/development/extensions-core/druid-pac4j.html>

not the same thing, exactly, although it does use oauth. It may be interesting to you.

I wonder if the message gap is actually computed at each emission period, or if it's only computed every several minutes, once some "batch" of messages from within a task is completed

and then, why always 30s and 90s? the numbers are just too clean

I tried digging into the Druid code, and the message gap seems to be emitted from somewhere in the `SeekableStreamIndexTaskRunner`, but I haven't been able to fully understand what the code does, it's quite complex

Hi <@U03MTCKS1JM>, I took a quick dive in the code. As far as I understand it...
`ingest/events/messageGap` is retrieved <https://github.com/apache/druid/blob/9e5f0109fd00ffbed2ee80ebb0dfae253660ed5f/indexing-service/src/main/java/org/apache/druid/indexing/common/stats/TaskRealtimeMetricsMonitor.java#L145|here>.  Which in turn depends on <https://github.com/apache/druid/blob/9e5f0109fd00ffbed2ee80ebb0dfae253660ed5f/server/src/main/java/org/apache/druid/segment/realtime/FireDepartmentMetrics.java#L267|this>.  This is calculating the  "last completion timestamp" - "max message timestamp". "Last completion timestamp" is set to current timestamp <https://github.com/apache/druid/blob/09991dcf17fb9984b65f291f0db47e38058253e4/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java#L725|when all messages an input stream are fully caught up to>. "Max message timestamp" is the max of all timestamps mapped to the __time column so far in the ingestion. So the gap is relative to the timestamps in the data. Assuming you are all caught up, are the timestamps in the data potentially a bit behind? any upstream processing that would put them at a 30 or 90 seconds behind when published? in other words, what is the __time column mapped to and how does that behave upstream?

it's computed once per minute as the delta between the max seen event time and the current timestamp

it's odd for it to have the behavior you describe… i wonder if something is rounding it off somewhere between emission and visualization?

Hi and thanks for the input!

To give a bit more context: the event time we use is the collection time as determined by an upstream processing task. However, the latency added by that task is quite small, usually less than 5 seconds, so it does not fully account for the reported message gap.

For debugging purposes, I disabled rollup (set query granularity to none), and then ran a bunch of queries to see exactly just how far behind the ingestion is. It turns out that, quite consistently, data is ingested with only around 5 seconds of latency:

Data is generally produced in near-real time and in order, so for us, as long as we see messages with a certain timestamp, we can reasonably assume that processing and ingestion are up to date and everything has been processed up to that point

But I do think I'm getting quite close to figuring this out. On a dev cluster with only 2 middle managers and very little data, it looks like the load on the middle managers alternates. Always, the middle manager that is currently ingesting data reports a correct message gap of around 3 seconds. At the same time, the middle manager that is "idle" briefly reports a larger gap, closer to 1 minute. I guess that this is simply because it actually finished ingesting data at some point and then it has nothing to do until the next emission period, which is why the reported gap is large

So I think that it's simply related to the 1 minute emission period and how work is distributed among the middle managers. Previously, I would only look at the maximum reported messageGap. I probably need to find a better aggregated value

Sorry if this is not making a lot of sense :smile:

Hello! There's any way to get most recent value given a group-by, without scanning all the segments?
So far I'm thinking in implementing a 'retry when missing' and run queries multiple times, with larger window each retry until I get all the values I expect

i'm wondering what you mean by "most recent value given a group-by"? do you have a SQL query in mind?

initially,  something like this:

```select  dim1, SUBSTR(replace(dim2, ' ', ''), 13, 2) as dim2_group_by, LATEST(dim2, 20) as last_value, 
max(__time) as last_seen

from stream
where 
__time &gt; '2022-09-31T15:35:45.732Z' -- since X days behind, the further I go, the slower the responses
and dim1 in (10 and 20 ids)
group by 1,2```
I can alter the stream to add a new dim with the dim2 already calculated with the dim1, so I only need the last value given a dim1 (ID)
if I run this query with 1 day, it's really fast (200ms max), but the further I go back, the slower it gets, sometimes it returns in 300ms, other times it goes to 12 seconds
I think it depends on the caching, as I have values coming late all the time, the cache gets invalidated fast

if there's a backward scanning in native query mode, I can run it for each dim1, instead of running a group-by query

I guess, until I alter the pipeline to keep this result in memory somewhere, the best approach is: run with  ~ 2hours (hopefully only hits MM), exclude dim1 where results were found, submit the query again with 24 hours, then with 7 days, and keep until all results are found (or when X &gt; retention)

I think this is too application specific for Druid to effectively cache.

We had similar idea to cache rolling windowed result if the group-by has __time component.
(But we abandoned that thought since there are a lot of other more important priorities).