Apache Druid

You could try using a time function.
<https://druid.apache.org/docs/latest/misc/math-expr.html#time-functions>

getting this err when used in transformspec
{
  "ingestionState": "BUILD_SEGMENTS",
  "unparseableEvents": {},
  "rowStats": {
    "buildSegments": {
      "processed": 0,
      "processedWithError": 0,
      "thrownAway": 0,
      "unparseable": 0
    }
  },
  "errorMsg": "org.apache.druid.java.util.common.ISE: Could not transform value for kafka.timestamp reason: Function[timestamp] first argument should be a STRING but got LONG instead\n\tat org.apache.druid.segment.transform.ExpressionTransform$ExpressionRowFunction.eval(ExpressionTransform.java:117)\n\tat org.apache.druid.segment.transform.Transformer$TransformedInputRow.getRaw(Transformer.java:218)\n\tat org.apache.druid.segment.incremental.IncrementalIndex.toIncrementalIndexRow(IncrementalIndex.java:581)\n\tat org.apache.druid.segment.incremental.IncrementalIndex.add(IncrementalIndex.java:516)\n\tat org.apache.druid.segment.realtime.plumber.Sink.add(Sink.java:184)\n\tat org.apache.druid.segment.realtime.appenderator.StreamAppenderator.add(StreamAppenderator.java:276)\n\tat org.apache.druid.segment.realtime.appenderator.BaseAppenderatorDriver.append(BaseAppenderatorDriver.java:411)\n\tat org.apache.druid.segment.realtime.appenderator.StreamAppenderatorDriver.add(StreamAppenderatorDriver.java:189)\n\tat org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskRunner.runInternal(SeekableStreamIndexTaskRunner.java:654)\n\tat org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskRunner.run(SeekableStreamIndexTaskRunner.java:266)\n\tat org.apache.druid.indexing.seekablestream.SeekableStreamIndexTask.run(SeekableStreamIndexTask.java:151)\n\tat org.apache.druid.indexing.overlord.SingleTaskBackgroundRunner$SingleTaskBackgroundRunnerCallable.call(SingleTaskBackgroundRunner.java:477)\n\tat org.apache.druid.indexing.overlord.SingleTaskBackgroundRunner$SingleTaskBackgroundRunnerCallable.call(SingleTaskBackgroundRunner.java:449)\n\tat java.util.concurrent.FutureTask.run(FutureTask.java:266)\n\tat java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)\n\tat java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)\n\tat java.lang.Thread.run(Thread.java:750)\n\tSuppressed: java.lang.RuntimeException: java.lang.IllegalArgumentException: fromIndex(0) &gt; toIndex(-1)\n\t\tat org.apache.druid.segment.realtime.appenderator.StreamAppenderatorDriver.persist(StreamAppenderatorDriver.java:241)\n\t\tat org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskRunner.runInternal(SeekableStreamIndexTaskRunner.java:772)\n\t\t... 8 more\n\tCaused by: java.lang.IllegalArgumentException: fromIndex(0) &gt; toIndex(-1)\n\t\tat java.util.ArrayList.subListRangeCheck(ArrayList.java:1016)\n\t\tat java.util.ArrayList.subList(ArrayList.java:1006)\n\t\tat org.apache.druid.segment.realtime.appenderator.StreamAppenderator.persistAll(StreamAppenderator.java:589)\n\t\tat org.apache.druid.segment.realtime.appenderator.StreamAppenderatorDriver.persist(StreamAppenderatorDriver.java:233)\n\t\t... 9 more\nCaused by: org.apache.druid.math.expr.ExpressionValidationException: Function[timestamp] first argument should be a STRING but got LONG instead\n\tat org.apache.druid.math.expr.NamedFunction.validationFailed(NamedFunction.java:46)\n\tat org.apache.druid.math.expr.Function$TimestampFromEpochFunc.apply(Function.java:2829)\n\tat org.apache.druid.math.expr.FunctionExpr.eval(FunctionalExpr.java:183)\n\tat org.apache.druid.segment.transform.ExpressionTransform$ExpressionRowFunction.eval(ExpressionTransform.java:113)\n\t... 16 more\n",
  "segmentAvailabilityConfirmed": false,
  "segmentAvailabilityWaitTimeMs": 0
}

"transformSpec": {
        "filter": null,
        "transforms": [
          {
            "type": "expression",
            "name": "kafka.timestamp",
            "expression": "timestamp(\"kafka.timestamp\")"
          }
        ]
      }
    },

From the log it says it is long, but expecting string , not sure how it is changed to LONG

I think it already comes out as long from Kafka?

You should be able to define it as a dimension of type `long` without using a transform

it is coming as string, is there any possibility to do it instead of adding in dimension.
since we are not aware of all dimensions so we kept our ingestion spec like this
      "dimensionsSpec": {
        "dimensions": [],
        "dimensionExclusions": [
          "__time",
          "ts"
        ],
        "includeAllDimensions": true
      }
if we define only  kafka.timestamp in dimension other dimensions are not seen in the query

ah i see, this is the _schemaless_ dimensions setup (<https://druid.apache.org/docs/latest/ingestion/ingestion-spec.html#inclusions-and-exclusions>)

in this case everything is loaded as a string. it's a limitation with the schemaless feature today

we are working on type detection so they will be loaded as the proper type, <@U03NP2W7VF1> or <@U030C8H59T8> should know more

til then you may also experiment with the <https://druid.apache.org/docs/latest/querying/nested-columns.html> feature — not sure if it will satisfy your needs, but it also provides a way of loading data without knowing all the columns in advance, so it's worth a look

and it is possible to combine this with explicitly-specified dimensions, so you can get the correct type for your kafka timestamp

Can we do it using transformSpec, so it will be simple?
A final option that we see is using a query, just if data is already converted to the required format we don't need conversions in query :)

we saw there is a parameter includeAllDimensions (<https://druid.apache.org/docs/24.0.0/ingestion/ingestion-spec.html#dimensionsspec>) which says "You can set `includeAllDimensions` it to true to ingest both explicit dimensions in the `dimensions` field and other dimensions that the ingestion task discovers from input data", tried setting this parameter but ended up only with kafka.timestamp column and other columns are missing.

by enabling this are we going to have schemaless dimensions plus "kafka.timestamp" with required type ?
tried the same and we see no changes, can u please the spec and let me know if it is correct?

Hi <@U0344FW86DD> can you please check and help with this? are we missing something ?

Hrm...  From reading the code, that includeAllDimensions *should* work.  When you say you are only getting kafka.timestamp and nothing else, is that from the sampling in the wizard in the console or are you actually running the ingestion and not getting the columns?

I ask because we sometimes see bugs where the console shows one thing due to something weird the Sampler is doing while the ingestion jobs themselves do the right thing.  I'm curious if this is going to be one of those cases.

this our sample data: {"ts":"2021-11-29T17:45:00+08:00","mt":"pwr","p1":"BLR","p2":"767","p3":"767","p4":"171","p5":"244","p6":"08","p7":"001","p8":"001","p9":"test","p12345":930,"p233434":900,"duration":15}

when we ingest this data with the above spec in the SQL console we see only "kafka.timestamp"(so basically it is showing only 'kafka.timestamp' and 'ts' and other columns are not shown in the SQL query tab)

Oh, do you have the supervisor/task still running?

That should cause it to end the task and that will cause a hand-off of the segment.  I suspect that will fix the query console

Also, make sure that you are refreshing your browser.  The schema in the browser is grabbed once and not updated.

we paused(suspended) the supervisor and tried refreshing browser, but still, we see the same.

pause.PNG

sql_1.PNG

Can you go to the segments view, click on the magnifying glass to the right of the line for that one segment and copy&amp;paste or screenshot the Payload section of that?

That is the one... and hrm.  That is weird.  I'm not sure exactly what is going on, it's definitely not what I expect.  If it's possible, you can use the nested column to do auto-discovered columns, which will also identify types as well

can u plz share some links or examples for this nested column?

<https://druid.apache.org/docs/latest/querying/nested-columns.html>

Hi Naresh, one way to do this is to pre-process the data. In other words, send the data from Kafka to a real time stream processor, convert the string to long, and deliver to Druid.

Unused segments can be automatically cleaned up from deep storage. You can checkout <https://druid.apache.org/docs/latest/operations/clean-metadata-store.html#segment-records-and-segments-in-deep-storage-kill-task> to configure the automation

When a druid process announces itself, it will add the following labels to a pod:
```druidDiscoveryAnnouncement-&lt;node-role&gt;
druidDiscoveryAnnouncement-id-hash
druidDiscoveryAnnouncement-cluster-identifier```
Similarly, when it unannounces itself, it will remove the above labels. On the discovery side, we initialize a K8s watcher which looks for pods with all the above labels.

The problem lies with the fact that we can have peons running in a single pod with middle managers. When a peon dies, it will unannounce itself and that removes the common labels (`druidDiscoveryAnnouncement-id-hash`, `druidDiscoveryAnnouncement-cluster-identifier`) which were also being used by the middle manager process to announce itself. This leads to the K8s watcher filtering out the middle manager process running on the same pod since the required label does not exist anymore. So the middle manager ends up zombie'ing out because the overlord doesnt see it anymore.

The simplest solution would be to update the labels to add the node-role to the key:
```druidDiscoveryAnnouncement-&lt;node-role&gt;
druidDiscoveryAnnouncement-id-hash-&lt;node-role&gt;
druidDiscoveryAnnouncement-cluster-identifier-&lt;node-role&gt;```
so that there are no common labels being used by any druid process and that keeps announces and unannounces clean.

Unfortunately, this will break backwards compatibility for existing deployments using this extension. This is because the K8s watcher will start looking for nodes with the `druidDiscoveryAnnouncement-cluster-identifier-&lt;node-role&gt;`  label and already running nodes will not follow that convention.

Implementing a backwards compatible solution is possible but will be quite messy because I think we will have to initialize 2 K8s watchers (<https://github.com/apache/druid/blob/master/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProvider.java#L219-L251|code for reference>) because afaik K8s labels dont really support an OR on keys. So im open to any other ideas folks might have about how we can fix this. Also happy to contribute a fix for this if we can come up with a good solution.

throwing this out there and may be completely off base.  Could this be adapted to indexer process which is experimental leaving the existing labels alone?  Then there is an option to switch to non zookeeper indexers leaving backwards compatability?  Not sure but was just thinking about this and it was interesting

For our use case we want to stick to using middle managers and peons. If I am understanding your suggestion correctly, using indexers wouldnt really solve the problem right? It would just not run into this edge case.

is it jdbc or http query? ive seen some evidence that heavy jdbc workloads can seem to clog the broker parallel merge pool, i suspect there might be some not great behavior, especially when some jdbc clients are not fetching all of their results or fetching them very slowly (since jdbc shifts the query completing onto the client, and on the broker the query is pulling stuff out of a blocking queue of the parallel merge pool)

in the flame graphs i saw the merge pool itself appeared to be blocked trying to put results into these blocking queues, indicating that it was waiting for downstream stuff to finish consuming the sequence and close everything out

it could be something simple like waiting for a merge buffer: this time isn't counted in `query/cpu/time` since we're just waiting

if that's in, increasing the # of merge buffers would be useful

you can use some CDC tool to import into kafka, then ingest in 'real-time'

there's also some option to import via jdbc_connector, but there's always some issues doing that way

Druid is opinionated when it comes to ingestion.
Maybe one day, when multi-stage-query is matured enough, it can ingest directly using SQL.

<https://druid.apache.org/docs/latest/ingestion/native-batch-input-sources.html#sql-input-source> - This might work too

Thanks for your responds.
My point is why we have to push data to csv or tsv first then indexing them?
<@U031A9D19NU> it is great to have this new solution now. But, this tool still will store data into disk then it will read data from local disk on druid to segment them. The following phrase from druid documentation:
&gt; During indexing, each sub-task would execute one of the SQL queries and the results are stored locally on disk. The sub-tasks then proceed to read the data from these local input files and generate segments.

all batch ingestion do generate intermediate data though some thinking will be needed to write a list of queries that lets you parallelize the execution.

I believe this is where it happens: <https://github.com/apache/druid/blob/8296123d895db7d06bc4517db5e767afb7862b83/core/src/main/java/org/apache/druid/java/util/emitter/core/HttpPostEmitter.java#L662|https://github.com/apache/druid/blob/8296123d895db7d06bc4517db5e767afb7862b83/core[…]va/org/apache/druid/java/util/emitter/core/HttpPostEmitter.java>
it does retry up to MAX_SEND_RETRIES (3) times

it will retry forever, until either:

• druid server is shut down
• or we run out of buffer space for metrics; this is some % of the heap

in either of these cases we will discard metrics that haven't yet been emitted successfully

for #2, yes, we'll retry if we get `SC_INTERNAL_SERVER_ERROR`

Are you also capturing metrics emitted by your broker? Does the cpu or memory usage seem to be high?

i think exit code 3 can happen when the jvm exits due to OutOfMemoryError

You can also run `kubectl logs ... -p` to get the logs from the previous execution of the the pod to see if there are any error messages associated with the exit.

This is why we deploy all of our daemons as StatefulSet.
And thus, you can simply do `kubectl logs --previous` to see what bombed the daemon.

<@U0318CX1NH4> <@U0344FW86DD>, indeed the memory usage is almost full for the container.

And looking at the logs from the previous run, indeed it is out of memory.

```kai.sun@kaisun-ltmtukn script % kubectl -n xxxx logs druid-monc-ra-common-lab1-brokers-0 -c druid-monc-ra-common-lab1-brokers --previous 
2022-11-01T03:36:49+0000 startup service broker
Setting druid.host=10.64.213.220 in /tmp/conf/druid/cluster/query/broker/runtime.properties
Terminating due to java.lang.OutOfMemoryError: Java heap space```

i think on some platforms `javax.net.ssl` may be backed by openssl. it would depend on the jvm you use

i don't think this is common though. i heard of it happening on some older android distros

i assume you are not running on android :slightly_smiling_face:

hahaha that would be a very special android

are you using `druid.dynamic.config.provider`?

i think it's from this block

```    Object dynamicConfigProviderJson = consumerProperties.get(KafkaSupervisorIOConfig.DRUID_DYNAMIC_CONFIG_PROVIDER_KEY);
    if (dynamicConfigProviderJson != null) {
      DynamicConfigProvider dynamicConfigProvider = configMapper.convertValue(dynamicConfigProviderJson, DynamicConfigProvider.class);
      Map&lt;String, String&gt; dynamicConfig = dynamicConfigProvider.getConfig();
      for (Map.Entry&lt;String, String&gt; e : dynamicConfig.entrySet()) {
        properties.setProperty(e.getKey(), e.getValue());
      }
    }```

so, there's some null key or value in the dynamic config provided by `druid.dynamic.config.provider`

i see there's no nice error message for that case; would be good to add one!

ah, this is good enough hint for me. Thank you!

Ah this is due to the escaped quotes. After I unescaped it I got a different error.

Which is likely due to mismatched username and password.

I forgot to add that on our dev environment, the solution is running on a *single node*.
This is why, we would like to use local storage in this case.

the deep storage needs to be shared between all servers, so if it is `local` then it does need to be on a shared filesystem

i am not sure if the operator does this or not; my guess is it probably doesn't, since it's designed to work well on multi-server environments too

i think you can fix your issue by either:

• modifying your setup to point the `local` deep storage at a shared filesystem
• or, using a remote deep storage like `hdfs`, `s3`, or google, azure equivalents

Thank you for your response.

Since we are *not* on a multi-server environment, I thought it would work just by using the pvc storage of the Druid Historical pod.

What do you mean by using a shared filesystem ? Is it a same PVC used by all Druid pods ?

i'm not super familiar with the k8s concepts, so i am not sure exactly how a shared PVC works exactly.

if it's a single filesystem that is mounted across all pods, that would be OK

Thanks, we'll try to figure out how to make it works

Flame graphs might shed some light:

<https://support.imply.io/hc/en-us/articles/360033747953-Profiling-Druid-queries-using-flame-graphs>

Hi Shubham! I locally tried a small experiment and was able to connect `trino-402` installation with my local druid setup.
I followed the steps mentioned in <https://trino.io/docs/current/connector/druid.html>
My druid setup was the latest apache master. Which druid version are you using?

Hi Rohan I am on the latest master branch. Are you using it via bearer auth ?  Can you share you are setting the config.

I was using un-secured setup for the connector. The properties were :
```connector.name=druid
connection-url=jdbc:avatica:remote:url=<http://localhost:8888/druid/v2/sql/avatica/>```
`8888` was my local router
are you using some authentication setup as well?

Yes I am using the basic authentication in druid but wasn't able to connect with that.

I think the doc is also not updated that well. Are you seeing the catalog name as druiddb as mentioned in the doc ?
<https://trino.io/docs/current/connector/druid.html>

yeah, I think that's a mistake in the docs. the catalog name should be `druid` in the docs

also I can try to check the basic auth setup as well

I tried basic authentication and was successfully able to connect. For druid side auth, I followed <https://druid.apache.org/docs/latest/development/extensions-core/druid-basic-security.html>
and on the trino side added the `connection-user` and `connection-password` properties in the catalog properties

Were you able to query the tables also ? For some reason I am always getting org.apache.http.client.ClientProtocolException.

Yes I was able to query the tables too. is there more info in the exception you see? maybe in the trino log files? Also, are you able to query druid independently?

Screenshot 2022-11-05 at 8.23.55 AM.png

Hi Rohan as below you can see, what i am facing is that a simple query is running in my local for more than 30 minutes, seems like this is stuck and no logs in the trino container. I am running this in a dockerized container.

For delta catalog I am able to query. This is happening only for druid catalog.

Screenshot 2022-11-05 at 8.54.02 AM.png

Also in the console all i see is WAITING_FOR_RESOURCES while executing any simple query.

Thanks for the update! Is your druid cluster also in the same container as the trino deployment? Also, are you able to query druid directly ie without trino?

The druid is hosted via imply for us. So we are able to query druid. I am running trino in my local docker container.

I tried connecting with an imply managed hybrid cluster as well and was able to do it. The steps I followed were :
1. Download the `Root certificate` from Imply's API page and create a java truststore from it 
2. Add the truststore in the jdbc url using the following format : `Imply-Private-JDBC-connection-string;truststore=&lt;path-to-truststore-file&gt;;truststore_password=&lt;truststore-password&gt;` (ref : <https://calcite.apache.org/avatica/docs/client_reference.html> - druid uses calcite avatica for JDBC driver). Also added user/password from the Imply API page in the trino catalog properties
3. Connect to the VPN where the druid cluster is before running trino queries
Were you also following a similar way to connect to druid? A couple of things to check outside druid could be :
1. your docker container does allow outside connection
2. your trino container has both coordinator and worker configured in it. a lot of times `waiting-for-resources` in trino could also mean no workers available
I hope that helps!

hi :slightly_smiling_face: is there a way to see why as part of an indexing job things are "thrownAway"?

i.e. we are seeing in the report:
```{
  "ingestionState": "COMPLETED",
  "unparseableEvents": {},
  "rowStats": {
    "determinePartitions": {
      "processed": 0,
      "processedWithError": 0,
      "thrownAway": 0,
      "unparseable": 0
    },
    "buildSegments": {
      "processed": 2097447,
      "processedWithError": 0,
      "thrownAway": 553,
      "unparseable": 0
    }
  },
  "errorMsg": null,
  "segmentAvailabilityConfirmed": false
}```

you can set logParseExceptions to true in your spec. Any parse errors will show up in the log

and we are checking that :slightly_smiling_face:

I am pretty sure that events outside the interval is the only reason you would have data thrown away - so that is the reason

From my understanding the requests should be timed out at 5 min mark. Internally, the processing classes keep a track of the timeout and in case it is exceeded cancel the execution and throw the desired error
Are you facing any troubles while using the query timeout or seeing any conflicting behaviour to above?

It’s not an exact science, but I was comparing the query timeout clocked by Superset and charts on Datadog.
After the 5 minutes mark, it looks like there are still high volume activity on Historical side.

Does the volume reduce after a while, i.e. while the load is high at 5 min mark, does the load subside after say 2-3 mins?

it didn’t subside after 2-3 minutes. More like 15-20 minutes

Which version of Druid are you on? And are there any specific type of queries that you are running which are showing this behavior? It might be a bug similar to <https://github.com/apache/druid/pull/12271>.

I haven’t delve deep into what type of query yet

Ohh okay. Are there any dubious errors or warnings in the logs which might point to the queries not cancelling?

So this is interesting. We just had a major outage due to a number of bad “LIKE” queries.

The most interesting relevant part of this is that all of Historical threads were stuck spinning forever performing groupBy.

After 2 hours the cluster didn’t recover and the same threads that were stuck still stuck.

We had to perform complete Historical restart to recover from this issue.

I will paste relevant jstack when I am in front of my computer later.

Starting to think that extremely long G1 GC pause is the culprit.
(Which will happen if the user runs crazy big `LIKE` queries)

Jetty metrics are documented here - <https://druid.apache.org/docs/24.0.0/operations/metrics.html#jetty>

set the channel description: Troubleshooting and problem-solving - Please check the druid docs - <https://druid.apache.org/docs/latest/design/> first.

Couple of things.
1. We generally try to mask IP_addresses while sharing stuff.
2. How is the coordinator health. Is it gcing a lot ?
3. `druid.manager.lookups.hostTimeout` increasing this might help. 

interesting. <@U03149ALHN3> do you know if there was a major change preceding it? e.g. maybe an upgrade

Task logs didn’t say anything because it bombed earlier.

As it turned out, the Kafka I am connecting to enforces authz via consumer group.
After adding `group.id` in `consumerProperties`, it worked.

Nice. I am curious about how did it work. did you just set same group id (kafka-supervisor-xxx) in consumer properties?

yes, I just set the same `group.id` as what the Kafka team gave me.

I suppose that they also had to change some ACLs on their end.

was it a fresh cluster installation or an upgrade?

I see similar issue  "errorMsg": "org.apache.kafka.common.errors.GroupAuthorizationException: Not authorized to access group: kafka-su...

how to fix this.  Tried setting group.id in ingestion spec

how we fixed it is setting the `group.id` in the supervisor spec.

try testing on default Kafka CLI first and see.

at the risk of sounding silly, it improves performance in the situations where your performance is limited by lack of replication :slightly_smiling_face:

in particular, if all your Historical process CPUs are at 100% during a query, then increasing replication won't do anything useful

but if they are less than that, it may help, if it means you get to use more Historicals

classic case where replication is helpful for query perf is where you have a small number of hotspot segments (certain segments that get more queries than others)

like, if you have 5 hotspot segments, with the default replicas of 2, they will replicate to at most 10 servers.

in this scenario, let's say you have 50 servers. then, increasing replicas can improve perf b/c those hotspot segments can go out to more servers

essentially -- APPROX_COUNT_DISTINCT_DS_THETA (and its cousin DS_HLL) is an aggregation function, whereas THETA_SKETCH_ESTIMATE is a regular function

in the example you gave, i think you want an aggregation, so APPROX_COUNT_DISTINCT_DS_THETA is the valid choice

THETA_SKETCH_ESTIMATE is meant for use after doing intersections and stuff

for example: <https://druid.apache.org/docs/latest/tutorials/tutorial-sketches-theta.html#set-operations>

```SELECT THETA_SKETCH_ESTIMATE(
         THETA_SKETCH_INTERSECT(
           DS_THETA(theta_uid) FILTER(WHERE "show" = 'Bridgerton' AND "episode" = 'S1E1'),
           DS_THETA(theta_uid) FILTER(WHERE "show" = 'Bridgerton' AND "episode" = 'S1E2')
         )
       ) AS users
FROM ts_tutorial```

As per my understanding, the information should have been deleted from the INFORMATION_SCHEMA.TABLES by themselves. Are there any segments for the deleted datasource that are still present in the cluster? And if it’s not a hassle, can you try restarting the cluster to check if the INFORMATION_SCHEMA gets refreshed?

do you see any segments for this datasource in `sys.segments`? generally datasources disappear from `INFORMATION_SCHEMA.TABLES` once the last available segment goes away

generally these operations are limited by metadata store performance: scaling up your metadata store generally helps

like, using a server type with more CPUs/memory

we're looking into improvements in this area that will limit the load on metadata store as the number of allocations increases. hope to have something to ship here in the next release or two!

btw, i'm wondering, why not use hour segment granularity? i've seen that as the most common segment granularity for streaming data

<@U03HF9GLL1X>, thanks for reporting the issue. We have seen the issue with the allocation of pending segments on some other clusters as well. We are actively working to resolve this issue, hopefully in the upcoming Druid release.

&gt; overlord logs are continuously allocating pending segments
I don't think a 15min granularity would help with the segment allocation slowness. For your cluster, I agree with Gian, you should try going with an HOUR granularity.

You also mention that with a granularity higher than 15 min, you end up with too many segments per time chunk (after compaction). You could try increasing the number of rows per segment specified in the auto-compaction config, and use `range` partitioning on 3 to 4 dimensions for better data distribution.

Could you please also elaborate on how the "coordinator slows down"? If it's segment loading and datasource availability that takes time, you can try increasing the load queue size through the coordinator dynamic config.

Hope this helps :slightly_smiling_face:

&gt; generally these operations are limited by metadata store performance: scaling up your metadata store generally helps
We're running the bitnami postgresql-ha deployment for metadata. It's a cluster of 3 (1 master and 2 replicas where read requests are load balanced and write requests are forced to the master). CPU and memory does not seem to be close to exhausted according to the metrics. And it's backed by some decent disks and the latency doesn't seem to be an issue either.

What are others using for metadata? Does mysql offer better performance?

&gt; btw, i'm wondering, why not use hour segment granularity? i've seen that as the most common segment granularity for streaming data
We found that when using hour segment granularity, we ran into issue with the coordinator cycles taking too long due to the number of segments in the timechunk (post compaction we have upwards of 3000 segments at an hourly granularity). I think we're hitting this <https://github.com/apache/druid/issues/11700>

&gt; Could you please also elaborate on how the "coordinator slows down"? If it's segment loading and datasource availability that takes time, you can try increasing the load queue size through the coordinator dynamic config.
The coordinator cycle takes a while due to the VersionedIntervalTimeline issue above (or was on 0.23 when we decided to move to 15 minute granularity). The cycles took so long that eventually it stopped handing off segments during ingest with the default completion timeout

After compaction, what is the typical size of each of your segments in terms of rows and bytes?

Reducing the granularity from hour to 15 min will only increase the number of total segments. You can also look at the `coordinator/time` metric to identify the problematic coordinator duty.

The timeline build happens on a separate thread, so it shouldn't affect the coordinator cycles, except maybe through the `MarkAsUnusedOvershadowedSegments` duty, we can be sure once we look at the above metric. If it does turn out to be the `MarkAsUnused` duty that is taking most of the time, you can temporarily disable it by setting `millisToWaitBeforeDeleting` in the coordinator dynamic config to a very high value. (a fix for this duty has already been merged <https://github.com/apache/druid/pull/13287> and will be released in Druid 25).

&gt; After compaction, what is the typical size of each of your segments in terms of rows and bytes?
Around 5 million rows with each row around 200 bytes so each segment is ~1GB in size

&gt; The timeline build happens on a separate thread, so it shouldn't affect the coordinator cycles, except maybe through the `MarkAsUnusedOvershadowedSegments` duty, we can be sure once we look at the above metric. If it does turn out to be the `MarkAsUnused` duty that is taking most of the time, you can temporarily disable it by setting `millisToWaitBeforeDeleting` in the coordinator dynamic config to a very high value. (a fix for this duty has already been merged <https://github.com/apache/druid/pull/13287> and will be released in Druid 25).
This duty is indeed the one that takes the most time in the coordinator duty cycle. Looking forward to Druid 25 as that will fix one of the biggest issues we've seen with druid.

Does that patch change the acceptable number of segments per timechunk (2000 as per <https://github.com/apache/druid/issues/11700#issuecomment-918374527>)?

The patch significantly reduces the number of segments that the duty has to check for marking as unused. So the runtime of the duty also goes down.

&gt; Around 5 million rows with each row around 200 bytes so each segment is ~1GB in size
&gt; 
So you are getting around 1TB data per hour?

Since your segments already seem big enough, you cannot increase the segment size further to reduce the total number of segments. Instead, try enabling roll-up if you are not already using it.

&gt; So you are getting around 1TB data per hour?
&gt; 
&gt; Since your segments already seem big enough, you cannot increase the segment size further to reduce the total number of segments. Instead, try enabling roll-up if you are not already using it.
Yeah, around 1TB / hour at the moment but that will increase when we can stabilise things at this speed.

Roll-up is on the horizon but in this early stage of deployment is not yet in place and won't be for a while I don't think

<@U03HF9GLL1X> - you may want to look at your pending segments table and see if it needs to be cleaned up. 

It’s configured to auto cleanup but at a 15 minute segment granularity will grow to be a little under a million records in a day or so. 

I think the auto cleanup looks at the time at which the earliest task was submitted for a datasource. If you happen to have any zombie tasks or kill tasks for a datasource that are blocked, auto cleanup may not be cleaning up. Alternatively, you can increase the frequency of cleanup.

You may also want to look at creating indices in your metadata store to improve query perf.

an FYI about allocating segments— there's another patch from <@U0318CX1NH4> that helps here: <https://github.com/apache/druid/pull/13369>

we've been running it on a cluster with nearly 1000 tasks for the last couple of days, &amp; 98%ile action runtimes are down from 3 minutes to 13 seconds. Maximum (over this time period) is 15s; we had seen over 5 minutes prior to rolling out this patch.

he's really on fire recently with the scalability patches! :astronaut: :rocket:

there's been three others related to kafka supervisor performance that went up recently:

• <https://github.com/apache/druid/pull/13334> :arrow_left:  more aggressive deduping of work in the main loop
• <https://github.com/apache/druid/pull/13328> :arrow_left:  fewer metadata store calls per run in the main loop
• <https://github.com/apache/druid/pull/13354> :arrow_left:  async comms from supervisor -&gt; task; contact all tasks at once and don't block worker threads

we're targeting 25.0 for all of these and for the Coordinator patch he mentioned a couple weeks ago (<https://github.com/apache/druid/pull/13287>)

These all look very promising. When is 25.0 due for release?

The release is scheduled for the mid of December. The release branch has already been cut, backports are in progress. We should have a release candidate by next week.

After some measures in our production dataset, we saw the groupBy query  result is sent back is approximatively 8-9s

where the topN (without currency conversion) takes less than 2s.

So any hints about the possibility to use lookups with topN queries would be helpful :slightly_smiling_face:

Well, it is possible.

We tried using a virtual column and use lookup expression.
```{
      "type": "expression",
      "name": "convertedSaleValue",
      "outputType": "DOUBLE",
      "expression": "localCurrencySalesValue * lookup(CONCAT(currency_short_desc, '_USD'),'currencies_rate_fromto')"
    },```
The compute is done, but the result are not so accurate.

We have a huge factor difference between topN &amp; groupBy.
Does anyone know when is this virtual column created  ? (before/after/during) aggregation ?

It seems like the currency rate used is not the proper one (since for 1 client, there could be multiple currencies used)

TopN will use an approximate algorithm where each historical will limit its results to `max(1000, threshold)` top results prior to merging it in the broker to calculate the TopN globally. This means that the result is an approximation. GroupBy will do the full calculation.
Another thought on your use case, if the lookup is used to query events that span multiple days, won't the exchange rates all be different by day? Perhaps this would require upstream data enhancement to add the USD conversion with the specific date based on the time of the transaction?

Well, I thought about that but it didn't explain that big differences. 
Actually, that was due to my mistake in the calculation :sweat_smile: 
Now we have something similar (at least something that could be explained by approximation). 

For the other point, no, our lookup is refreshed once a year, it does not reflect public rates. 
But we thought about adding most common currencies in the upstream data, which would really simplify queries. 
We could even reuse group by (not nested one) to get accurate results if needed with less impacts on query time.

out of curiosity, do you have the full query you ran? wondering if we can suggest some other restructuring after seeing all the logic

Yes, sure.

Here is the initial request :
```{
  "queryType": "groupBy",
  "dataSource": {
    "type": "query",
    "query": {
      "queryType": "groupBy",
      "dataSource": {
        "type": "table",
        "name": "sale"
      },
      "intervals": {
        "type": "intervals",
        "intervals": [
          "2022-01-01T00:00:00.000Z/2022-10-10T23:59:59.999Z"
        ]
      },
      "virtualColumns": [
        {
          "type": "expression",
          "name": "ts",
          "outputType": "LONG",
          "expression": "timestamp(transaction_time)"
        }
      ],
      "filter": {
        "type": "and",
        "fields": [
          {
            "type": "selector",
            "dimension": "brand_code",
            "value": "53",
            "extractionFn": null
          },
          {
            "type": "not",
            "field": {
              "type": "selector",
              "dimension": "operation_flag",
              "value": "D",
              "extractionFn": null
            }
          },
          {
            "type": "not",
            "field": {
              "type": "selector",
              "dimension": "operation_flag",
              "value": "true",
              "extractionFn": null
            }
          },
          {
            "type": "not",
            "field": {
              "type": "selector",
              "dimension": "unknown_flg",
              "value": "Y",
              "extractionFn": null
            }
          },
          {
            "type": "not",
            "field": {
              "type": "selector",
              "dimension": "transaction_time",
              "value": null,
              "extractionFn": null
            }
          }
        ]
      },
      "granularity": {
        "type": "all"
      },
      "dimensions": [
        {
          "type": "default",
          "dimension": "client_id",
          "outputName": "clientId",
          "outputType": "STRING"
        },
        {
          "type": "extraction",
          "dimension": "currency_short_desc",
          "outputName": "currency_rate",
          "outputType": "DOUBLE",
          "extractionFn": {
            "type": "cascade",
            "extractionFns": [
              {
                "type": "stringFormat",
                "format": "%s_USD"
              },
              {
                "type": "registeredLookup",
                "lookup": "currencies_rate_fromto",
                "retainMissingValue": true
              }
            ]
          }
        }
      ],
      "aggregations": [
        {
          "type": "doubleSum",
          "name": "localCurrencySalesValue",
          "fieldName": "amt_discounted",
          "expression": null
        },
        {
          "type": "longSum",
          "name": "quantity",
          "fieldName": "qty",
          "expression": null
        },
        {
          "type": "longMax",
          "name": "lastPurchaseDate",
          "fieldName": "ts",
          "expression": null
        },
        {
          "type": "stringLast",
          "name": "firstName",
          "fieldName": "client_name_first_local",
          "expression": null
        },
        {
          "type": "stringLast",
          "name": "lastName",
          "fieldName": "client_name_last_local",
          "expression": null
        }
      ],
      "postAggregations": [
        {
          "type": "arithmetic",
          "name": "convertedSalesValue",
          "fn": "*",
          "fields": [
            {
              "type": "fieldAccess",
              "name": "localCurrencySalesValue",
              "fieldName": "localCurrencySalesValue"
            },
            {
              "type": "fieldAccess",
              "name": "currency_rate",
              "fieldName": "currency_rate"
            }
          ]
        }
      ]
    }
  },
  "intervals": {
    "type": "intervals",
    "intervals": [
      "2022-01-01T00:00:00.000Z/2022-10-10T23:59:59.999Z"
    ]
  },
  "virtualColumns": [],
  "filter": null,
  "granularity": {
    "type": "all"
  },
  "dimensions": [
    {
      "type": "default",
      "dimension": "clientId",
      "outputName": "clientId",
      "outputType": "STRING"
    }
  ],
  "aggregations": [
    {
      "type": "doubleSum",
      "name": "salesValue",
      "fieldName": "convertedSalesValue",
      "expression": null
    },
    {
      "type": "longSum",
      "name": "quantity",
      "fieldName": "quantity",
      "expression": null
    },
    {
      "type": "longMax",
      "name": "lastPurchaseDate",
      "fieldName": "lastPurchaseDate",
      "expression": null
    },
    {
      "type": "stringLast",
      "name": "firstName",
      "fieldName": "firstName",
      "expression": null
    },
    {
      "type": "stringLast",
      "name": "lastName",
      "fieldName": "lastName",
      "expression": null
    }
  ],
  "having": null,
  "limitSpec": {
    "type": "default",
    "limit": 10,
    "columns": [
      {
        "dimension": "salesValue",
        "direction": "descending",
        "dimensionOrder": "numeric"
      }
    ]
  },
  "context": {
    "groupByStrategy": "v2"
  },
  "descending": false
}```
and here is the topN equivalent one :
```{
  "queryType": "topN",
  "metric": "salesValue",
  "threshold": 10,
  "dataSource": {
    "type": "table",
    "name": "sale"
  },
  "intervals": {
    "type": "intervals",
    "intervals": [
      "2022-01-01T00:00:00.000Z/2022-10-10T23:59:59.999Z"
    ]
  },
  "virtualColumns": [
    {
      "type": "expression",
      "name": "convertedSaleValue",
      "outputType": "DOUBLE",
      "expression": "amt_discounted * lookup(CONCAT(currency_short_desc, '_USD'),'currencies_rate_fromto')"
    },
    {
      "type": "expression",
      "name": "ts",
      "outputType": "LONG",
      "expression": "timestamp(transaction_time)"
    }
  ],
  "filter": {
    "type": "and",
    "fields": [
      {
        "type": "selector",
        "dimension": "brand_code",
        "value": "53",
        "extractionFn": null
      },
      {
        "type": "not",
        "field": {
          "type": "selector",
          "dimension": "operation_flag",
          "value": "D",
          "extractionFn": null
        }
      },
      {
        "type": "not",
        "field": {
          "type": "selector",
          "dimension": "operation_flag",
          "value": "true",
          "extractionFn": null
        }
      },
      {
        "type": "not",
        "field": {
          "type": "selector",
          "dimension": "unknown_flg",
          "value": "Y",
          "extractionFn": null
        }
      },
      {
        "type": "not",
        "field": {
          "type": "selector",
          "dimension": "transaction_time",
          "value": null,
          "extractionFn": null
        }
      }
    ]
  },
  "granularity": {
    "type": "all"
  },
  "dimension": "client_id",
  "aggregations": [
    {
      "type": "stringLast",
      "name": "firstName",
      "fieldName": "client_name_first_local",
      "expression": null
    },
    {
      "type": "stringLast",
      "name": "lastName",
      "fieldName": "client_name_last_local",
      "expression": null
    },
    {
      "type": "doubleSum",
      "name": "salesValue",
      "fieldName": "convertedSaleValue",
      "expression": null
    },
    {
      "type": "longSum",
      "name": "quantity",
      "fieldName": "qty",
      "expression": null
    },
    {
      "type": "longMax",
      "name": "lastPurchaseDate",
      "fieldName": "ts",
      "expression": null
    }
  ]
}```
Don't hesitate if you need some other inputs.

oh man, the topN version is likely going to be super slow since `lookup` is a string function, and it's being fed into arithmetic. that means we need to do a string parse on every row!

That's what I thought until I run it against our prod dataset. 
Group by : ~10s
TopN : ~1s 
:man-shrugging:

It may be "druid slow" but still better for our users :smile:

applying `timestamp(transaction_time)` on each row is also going to be pretty rough

possibly because you have a lot of `client_id`?

Yes
Don't have the exact in head, but yes

that's generally when topN has an advantage, if there are a lot of distinct values of the dimension (`client_id`) in this case

out of curiosity -- can you try this one? I think it's equivalent to your topN version, but replaces the `lookup` with a join on a subquery, which might be more efficient since it avoids doing string-to-number conversion for each row. It does add a subquery, though, so I'm not 100% sure which way it'll go efficiency-wise :slightly_smiling_face:

the `TIMESTAMP_TO_MILLIS(TIME_PARSE(transaction_time))` is the equivalent of the expr `timestamp(transaction_time)` you have. this is probably taking a bunch of time too, so it'd be better to store that time as a number instead of a string. i.e. do the transformation on ingest instead of during query

~well, I fixed some errors thrown back in the druid query ui, but now I get a 503 error~ :confused:
Damn, my cluster just went down, I forgot about the time :sweat_smile:

My QA cluster is destroyed during nights.
So I run the query in production, annnnd

```Error: Unknown exception
Cannot build plan for query: WITH currencies AS ( SELECT REPLACE(k, '_USD', '') AS currency, CAST(v AS DOUBLE) AS rate FROM lookup.currencies_rate_fromto WHERE k LIKE '%_USD' ) SELECT client_id AS clientId, SUM(amt_discounted * currencies.rate) AS convertedSalesValue, SUM(qty) AS quantity, MAX(TIMESTAMP_TO_MILLIS(TIME_PARSE(transaction_time))) AS lastPurchaseDate, LATEST(client_name_first_local, 1024) AS firstName, LATEST(client_name_last_local, 1024) AS lastName FROM sale LEFT JOIN currencies ON sale.currency_short_desc = currencies.currency WHERE __time &gt;= TIMESTAMP '2022-01-01 00:00:00' AND __time &lt;= TIMESTAMP '2022-10-10 23:59:59' AND brand_code = '53' AND operation_flag NOT IN ('D', 'true') AND unknown_flg &lt;&gt; 'Y' AND transaction_time IS NOT NULL GROUP BY client_id ORDER BY SUM(convertedSalesValue) DESC LIMIT 10
org.apache.druid.java.util.common.ISE```
I guessed your function "LATEST_VALUE" was "LATEST" and just fixed some minor typos.
I'll go in logs to see if I get more insights

```org.apache.calcite.plan.RelOptPlanner$CannotPlanException: There are not enough rules to produce a node with desired properties: convention=DRUID, sort=[6 DESC].
Missing conversion is LogicalSort[convention: NONE -&gt; DRUID]
There is 1 empty subset: rel#9046781:Subset#9.DRUID.[6 DESC], the relevant part of the original plan is as follows
9046779:LogicalSort(sort0=[$6], dir0=[DESC], fetch=[10])
  9046777:LogicalAggregate(subset=[rel#9046778:Subset#8.NONE.[]], group=[{0}], convertedSalesValue=[SUM($1)], quantity=[SUM($2)], lastPurchaseDate=[MAX($3)], firstName=[LATEST($4, $5)], lastName=[LATEST($6, $5)], agg#5=[SUM($7)])
    9046775:LogicalProject(subset=[rel#9046776:Subset#7.NONE.[]], clientId=[$3], $f1=[*($1, $12)], qty=[$8], $f3=[TIMESTAMP_TO_MILLIS(TIME_PARSE($9))], client_name_first_local=[$4], $f5=[1024], client_name_last_local=[$5], $f7=[SUM(*($1, $12))])
      9046773:LogicalFilter(subset=[rel#9046774:Subset#6.NONE.[]], condition=[AND(&gt;=($0, 2022-01-01 00:00:00), &lt;=($0, 2022-10-10 23:59:59), =($2, '53'), &lt;&gt;($7, 'D'), &lt;&gt;($7, 'true'), &lt;&gt;($10, 'Y'), IS NOT NULL($9))])
        9046771:LogicalJoin(subset=[rel#9046772:Subset#5.NONE.[]], condition=[=($6, $11)], joinType=[left])
          9046764:LogicalProject(subset=[rel#9046765:Subset#1.NONE.[]], __time=[$0], amt_discounted=[$4], brand_code=[$11], client_id=[$19], client_name_first_local=[$20], client_name_last_local=[$21], currency_short_desc=[$42], operation_flag=[$69], qty=[$76], transaction_time=[$124], unknown_flg=[$129])
            9046710:LogicalTableScan(subset=[rel#9046763:Subset#0.NONE.[]], table=[[druid, enriched_sale]])
          9046769:LogicalProject(subset=[rel#9046770:Subset#4.NONE.[]], currency=[REPLACE($0, '_USD', '')], rate=[CAST($1):DOUBLE])
            9046767:LogicalFilter(subset=[rel#9046768:Subset#3.NONE.[]], condition=[LIKE($0, '%_USD')])
              9046711:LogicalTableScan(subset=[rel#9046766:Subset#2.NONE.[]], table=[[lookup, currencies_rate_fromto]])```
btw, I never noticed that kind of logs... is it related to sql queries only ? (because we only run native json queries here)

oops, i meant `ANY_VALUE` not `LATEST_VALUE`

as to the error, it means the SQL planner can't figure out how to make a native query -- sometimes due to some unsupported feature. wonder what the reason is…

I'm in 0.22.1, maybe that is the reason ?

```SELECT
    REPLACE(k, '_USD', '') AS currency,
    CAST(v AS DOUBLE) AS rate
  FROM lookup.currencies_rate_fromto
  WHERE k LIKE '%_USD'```

try replacing `ORDER BY SUM(convertedSalesValue) DESC` with `ORDER BY SUM(amt_discounted * currencies.rate) DESC`

from some executions, it's faster than groupBy, but slower than topN

can't win everytime ^^
Anyway, thanks for the time.
I'll also take a look tomorrow at this transaction_time and its type

it supports `AggregateFn(...) filter (where Y)` eg:
 `count(1) filter ( where httpRspCode not like '5%') as successRate`

Then do we support the first, aka
```SELECT 
  COUNT(CASE WHEN httpRspCode not like '5%' THEN 1 ELSE NULL END) 
From table```

I think yes, but the null may throw out all the results, usually I do sum(case when X then 1 else 0 end) when doing these

Right. The intentions is to throw out when WHEN condition is not met. Same effects as `sum(case when X then 1 else 0 end)`

Do they have same performance or any other performance difference?

```COUNT(CASE WHEN mm not like '5%' THEN 1 ELSE NULL END)  as a,
sum(CASE WHEN mm not like '5%' THEN 1 ELSE 0 END)  as b,
COUNT(1)  as c,
sum(CASE WHEN mm not like '5%' THEN 1 ELSE 1 END)  as d```


&gt;   Do they have same performance or any other performance difference?
As far as I'm aware, it should be very little difference, most of the work will depends on your entire query condition and the how many segments you can exclude using the __time column
the `col like 'Y%'` is very fast because is done on the bitmap index just once for each string, then a bitmap index will be used to match the rows , maybe count(1) with filter can be a little faster than using a case-when

image.png

both A and B column generates the same execution plan, so there's no difference

```count(1) filter (where mm not like '5%') as X```
same plan as well, calcite is very smart

it uses Java regular expression syntax, since its calling `Pattern.compile` internally and then using a `Matcher` to find the stuff. <https://docs.oracle.com/javase/8/docs/api/java/util/regex/Pattern.html>

its also available in SQL, <https://druid.apache.org/docs/latest/querying/sql-scalar.html#string-functions>

Thanks <@U030C8H59T8> for the info. Then can I suppose the above query is supported, right?

might be some minor syntax differences, but it should be possible