Apache Druid

Hi Saurav, _`druid.processing.buffer.sizeBytes`_ is the size of the in-memory buffer Druid uses for storing intermediate query results (that it then performs operations on).
GroupBy queries with large intermediate result sets can require more than this buffer. We only support a size of up to 2GB for this buffer; the default size is 1 GB.
For larger buffers, you can spill these results out of memory and onto disk. That is what _`druid.query.groupBy.maxOnDiskStorage`_ does.
You could first try increasing `druid.processing.buffer.sizeBytes` to 2G in Broker runtime properties and check

Have you added CacheMonitor?
Please refer to <https://druid.apache.org/docs/latest/configuration/index.html#metrics-monitors> for more details

i have add this
druid.monitoring.monitors=["org.apache.druid.client.cache.CacheMonitor", "org.apache.druid.java.util.metrics.SysMonitor", "org.apache.druid.java.util.metrics.JvmMonitor", "org.apache.druid.java.util.metrics.JvmCpuMonitor",
 "org.apache.druid.java.util.metrics.CpuAcctDeltaMonitor", "org.apache.druid.java.util.metrics.JvmThreadsMonitor", "org.apache.druid.server.metrics.QueryCountStatsMonitor"]

Do you see any metrics from your Broker at all?

the "druid.monitoring.monitors" above i have add it to /svc/druid/apache-druid-0.23.0/conf/druid/cluster/_common/common.runtime.properties in master and query node
that is ok?

Yes, Peons run on middle manager node but on different ports.

That would be a question for the turnilo people

First, to clarify, Druid segments cannot be "updated" ... they are immutable once created and can only be overshadowed, replaced, or deleted.

Batch ingestion supports both append and overwrite modes.

But to my knowledge streaming ingestion only supports append mode due to it's architecture of involving multiple ingestion tasks operating in parallel, each creating their own separate segments within the same time chunk, and the ability to ingest "late arrival" data which belongs to older time chunks

... there is no notion in this architecture to facilitate some sort of group-level overwriting of existing data.

You can always setup a restatement batch job to “backfill” older data. We do that a lot.

<@U04C4593VDY> <@U03149ALHN3> great!! thanks for the advice and clarification

some detail here on restatement ("overwrite"): <https://druid.apache.org/docs/latest/data-management/update.html#overwrite>

Just use your favorite http client library. All you have to do is send this payload `{"query": "..."}`

is there a wrapper that can do the parsing for the result and/or query ?

not that i'm aware of for native queries. for SQL you could always use JDBC: <https://druid.apache.org/docs/latest/querying/sql-jdbc.html>

Thanks guys.
I was successfully managed to work with Scruid scala lib for Druid.

I think RDS/mysql/mariadb might be more common, not sure.  Scaling issues often have to do with metadata cleanup, you don't want any tables growing TOO large.  Enabling automatic metadata cleanup can help.

Aurora on AWS. The metadata db isn't really hit that hard.

I think both mysql and postgresql are pretty widely used. at Imply in AWS we're using the mysql extension &amp; a mix of proper mysql and aurora, both through RDS

Overwhelmed? how? Is there an error? Where do you see the bottleneck? How many segments are you dealing with ? min/avg/max size?
You can pre-load segments into a historical and it will announce what it has. But the coordinator is still calling the shots about which segments should be where, so it might be difficult to override that.
There are a few parameters that control the speed of balancing and replication.
*On Historicals:*
• `druid.segmentCache.numLoadingThreads` - controls the size of the thread pool used for downloading segments from Deep Storage. 
*On Coordinator:*
• `druid.coordinator.loadqueuepeon.type` = `http`  (this is the new default over zookeeper based curator method.)
*Coordinator dynamic config:*
• `maxSegmentsToMove` - defaults to 5, in larger clusters 100-200 is normal. When historicals are added, segments need to move around historicals, this controls the number of segments moved per coordination cycle (1 minute bey default).
• `replicationThrottleLimit` - defaults to 10, this controls the rate at which segments are replicated per cycle. Like above, 10 is quite small for larger clusters. 

thanks for the reply. we ended up updating the parameters you listed above from other tips we found, wiped the historicals and let the coordinator do all the work. after about 15 hours we had all segments online and much less problems.

one last question, do replicas improve query performance when people are querying the same interval ?

Yes. All replicas of a segment are fair game for the brokers to use, so when concurrency on the same data grows, having more historicals able to answer queries on that data helps a lot.

I don't think so ... MSQ SQL ingestion is currently only implemented for batch ingestion.

no it is not, Kafka doesn’t speak SQL either; what’s your use case?

Not a use case really, was just hoping to create a supervisor using MSQ

write a small kafka producer script to poll the db and inject rows , i guess. That’s what we do 

Screenshot 2023-04-30 at 11.40.09 AM.png

could you share a screenshot of the `Services` tab?

Screenshot 2023-05-01 at 11.18.48 AM.png

hi <@U030W50CNAC> please find it below
its showing historical as 100% but My PVC have enough storage

your historicals are full - that is the issue

I don't know what PVC is in this context (I only know the polymer) but do you maybe need to adjust the `druid.server.maxSize` ?

see: <https://druid.apache.org/docs/latest/operations/basic-cluster-tuning.html>

`*historical*:`
  `## If false, historical will not be installed`
  `##`
 `*enabled*: true`
 `*name*: historical`
 `*replicaCount*: 2`
 `*port*: 8083`
 `*serviceType*: ClusterIP`

  `# The Historical will use any available free system memory (i.e., memory not used by the Historical JVM and heap/direct memory buffers or other processes on the system) for memory-mapping of segments on disk. For better query performance, you will want to ensure a good (free system memory / total size of all druid.segmentCache.locations) ratio so that a greater proportion of segments can be kept in memory.`
 `*config*:`
  `*DRUID_XMX*: 2048m # Heap: (0.5GiB * number of CPU cores) + (2 * total size of lookup maps) + druid.cache.sizeInBytes`
  `*DRUID_XMS*: 2048m`
  `*DRUID_MAXDIRECTMEMORYSIZE*: 2000m # (druid.processing.numThreads + druid.processing.numMergeBuffers + 1) * druid.processing.buffer.sizeBytes)`
  `*druid_processing_buffer_sizeBytes*: '100000000'`
  `*druid_processing_numMergeBuffers*: 2 # a 1:4 ratio of merge buffers to processing threads is a reasonable choice for general use.`
  `*druid_processing_numThreads*: 7`


`This is what my hostorical conf is`

Do i  need to add any confirmations in it please suggest

Hi <@U04VDGHUKTQ> although you may have enough disk space in your PVC it is only finding 1 GB storage. That is why your historicals are running out of space. You need to configure `druid.server.maxSize` and `druid.segmentCache.locations` appropriately to fix this issue. You can find more information about these configurations here: <https://druid.apache.org/docs/latest/configuration/index.html#historical-general-configuration>

It is not clear to me that you can configure those settings in k8s. There should be a portion of the file for historicals:
```runtime.properties: |
  druid.processing.numMergeBuffers=2
  druid.segmentCache.locations=[{\"path\":\"var/druid/data/segments\",\"maxSize\":"${HISTORICAL_VOLUME_SIZE}"}]```

Where I'm using an environment variable but you could just hard code it "500Gi" for example.

Hi <@U04P4A4J3JP> you need to change the maxSize. What is it currently?

I don't set that setting. It is inferred from my setting for segmentCache.locations (it totals them).

Hmm in that case you probably need to check the templates and find out what is defining `${HISTORICAL_VOLUME_SIZE}`

As I said, that is an environment variable I set. You can just not use that and set your size into the config if you want.

I see. In that case why is Historical size showing up as 1 GB?

There is a good chance that this setting isn't getting recognised

No idea. I assume it is writing to the root disk space that k8s makes available to your pod. You definitely do not want to use that. It has to point to the location you are mounting a volume.

Sorry <@U04P4A4J3JP> I guess there is a misunderstanding. Maybe you are trying to answer the original question that <@U04VDGHUKTQ> asked and I misunderstood the answer thinking that both of you work in the same team

Sorry I thought you were the original question asker.

Hopefully this will be good enough for <@U04VDGHUKTQ>

I personally set an env variable in my ci pipeline that I can pull the historical segment size from so that I can change it easily without changing the config, so that I can use different values between staging and prod environments without changing the config file.

Makes sense. I used to run a Druid Kubernetes cluster on spot instances. It was fun.

That's why this issue is so familiar. It always used to happen if `maxSize`  wasn't defined properly

I think I remember seeing an issue where they started defaulting it because people ran into that issue a lot.

Or having different values between the two causing issues.

Yea I used these almost 2 years ago. The official helm charts were deprecated around that time. After Druid 0.20. So, makes sense that it may have changed

As you have 24 segments that suggest, you might have segment granularity as Hour, if you compact it using Day granularity, then you should see 1 segment only

```{
  "type": "compact",
  "dataSource": "test",
  "ioConfig": {
    "type": "compact",
    "inputSpec": {
      "type": "interval",
      "interval": "Provide your interval"
    }
  },
  "granularitySpec": {
    "segmentGranularity": "DAY",
  }
}```

If you already have DAY granularity in the ingestion then check the segment name you should be able to see on what timechunk/interval segments are created and accordingly you can change in the compaction spec  `segmentGranularity`  to Day/week/month/quarter/year.

okay great thanks,
can we do the compaction configuration using the UI  instead of spec.?

Auto compaction, you can do from console but you will not be able to change all parameters, ssystem will do the best compaction available, If you want to  manually change something like here you want 24 segments to be 1 then you should do manual compaction through SPEC.

In the spec you provided, what is the interval.?

i mean what does it signify which interval I need to put in that.?

The interval will tell the compaction task to compact _all segments_ within the interval.

Complete details on compaction with example is given in below link : <https://druid.apache.org/docs/latest/data-management/compaction.html>

this spec should be submitted in the compaction section in data source for that table right.?

You can submit it as a normal task from the ingestion section, the datasource is their in SPEC.

Hi, i used compaction and it was succesfull but my no of rows and size of the new table is significantly reduced, is there a data loss.?

i can see the latest data available but is there any loss of data in between.?

okay,thats a relief, thanks, but what about the reduced size and no of rows, my no of rows  reduced to almost half of that before compaction

That is expected if you have rollup enabled or query granularity changed.

Below links has the details on the roll up and query granularity, check both ingestion and compaction spec. or you can query to the table now and see if there is any data loss.
<https://druid.apache.org/docs/latest/data-management/compaction.html#setting-up-manual-compaction>
<https://druid.apache.org/docs/latest/data-management/compaction.html>

Snippet from attached link:
```By default, compaction does not modify the underlying data of the segments. However, there are cases when you may want to modify data during compaction to improve query performance:

If, after ingestion, you realize that data for the time interval is sparse, you can use compaction to increase the segment granularity.
If you don't need fine-grained granularity for older data, you can use compaction to change older segments to a coarser query granularity. For example, from minute to hour or hour to day. This reduces the storage space required for older data.
You can change the dimension order to improve sorting and reduce segment size.
You can remove unused columns in compaction or implement an aggregation metric for older data.
You can change segment rollup from dynamic partitioning with best-effort rollup to hash or range partitioning with perfect rollup. For more information on rollup, see perfect vs best-effort rollup.```

```

  "type": "compact",
  "dataSource": "test",
  "ioConfig": {
    "type": "compact",
    "inputSpec": {
      "type": "interval",
      "interval": "Provide your interval"
    }
  },
  "granularitySpec": {
    "segmentGranularity": "DAY",
  }
}```
i used this spec which you gave and changed the segment granularity to month cause it is set to month in my original table
the task duration for my original table is 1H

in this spec where is the rollup setting.?

In this spec it is not present, it would be a separate parameter.

Was your initial ingestion does have roll up, You can check the task logs to see more information on what happened

when you run the task beside to the task there will be option to see the logs, let me provide a screen shot

Hi, i have been using the spec format that you gave to apply compaction.

But when i make some changes in the spec and submit a new task for compaction it shows "fully compacted" and does not run compaction

If it shows fully compacted that means the compaction has run, I have seen in some situation the first ever compaction does not show iin the task log, if you want you can run again and see the task log. Just test with any small datasource.

compaction has run, but it is not making the no of segments i wish to make, i need a segment size of around 400mb I am getting around 1 -2 mb

can you suggest the changes or addition in the config that you gave so that I will have a segment size of around 400-500 mb after compaction and i want to apply this on tables which have already ran compaction

This totally depends on your data, you are using segmentGranularity as day in compaction, and if you have less data for each day, then compacting with day granularity will not help, may be try with month granularity. This totally depends on data and there isn’t only one setting, we need to understand the parameters that is present for the compaction and accordingly configure manual compaction.

I had set it to month for tables with less data

can we not set it based on size of data.?

I think during ingestion you can set based on size but if you have less data per day and you are using day segment granularity then one segment will get created for that day, if you have more data only size will come into picture.

Is this what you are looking for: <https://druid.apache.org/docs/latest/querying/lookups.html#update-lookup>
&amp;
<https://druid.apache.org/docs/latest/querying/lookups.html#api-for-lookup-status>

I could not find a way to manually trigger the refresh of the lookup. But you can control the poll period of the <https://druid.apache.org/docs/latest/development/extensions-core/lookups-cached-global.html#jdbc-lookup|JDBC lookup>. By default it will load only once, but by setting the `pollPeriod` it will reload periodically. If you also specify the `tsColumn` is can do incremental updates from the source by only retrieving lookup changes since the last poll.

Hi <@U030K4UM3H7>
In general yes.  At the end I’ve used sequence of delete-create to replace content of static map lookups for regression tests.
Thank you

you should first narrow it down to how much time, an individual sub-task is taking. if that's high and volume processed is low, then I will suggest optimizing that task runtime

you can look at the sub tasks for your main task by filtering on the task group id

I have only one task per compaction interval

actually trying to edit the value for `maxNumConcurrentSubTasks` causes the task to fail without any meaningful error in the logs

if you're having issues with classic native batch ingest or compaction, I'd suggest upgrading and trying out the SQL-based approach. It has been totally rewritten to be more reliable

a relevant example: <https://druid.apache.org/docs/latest/multi-stage-query/examples.html#replace-for-replacing-a-specific-time-segment>

this example is basically equivalent to a compaction: it does a `REPLACE` from a datasource into itself for a specific time range

<@U0344FW86DD> Thank you
but upgrade is not an option right now

<https://superset.apache.org/docs/databases/druid/>

image.png

i was trying to create a user, but that didnt work

Are you familiar  with superset ? <@U04EWB06GR4>

I recently connected Superset to a Druid cluster - when you say “didn’t work” what kind of issues were you seeing?

<@U054JARM13R> do you authenticated enabled on druid side for querying ? if you have basic enabled then you can enter username and pass. if you do not have any auth enabled then you can use
```druid://&lt;Host&gt;:&lt;Port-default-9088&gt;/druid/v2/sql```

created the users, now getting this error which is needs to install druid drivers to superset,

ERROR: (builtins.NoneType) None
(Background on this error at: <https://sqlalche.me/e/14/dbapi>)

superset.exceptions.SupersetErrorsException: [SupersetError(message='(builtins.NoneType) None\n(Background on this error at: <https://sqlalche.me/e/14/dbapi)>', error_type=&lt;SupersetErrorType.GENERIC_DB_ENGINE_ERROR: 'GENERIC_DB_ENGINE_ERROR'&gt;, level=&lt;ErrorLevel.ERROR: 'error'&gt;, extra={'engine_name': 'Apache Druid', 'issue_codes': [{'code': 1002, 'message': 'Issue 1002 - The database returned an unexpected error.'}]})]

Tried below and its working: `"batchSize":30480`

``` {
   "queryType": "scan",
   "dataSource": "wikipedia",
   "resultFormat": "list",
   "columns":[],
   "intervals": [
     "2013-01-01/2023-01-02"
   ],
   "batchSize":30480,
   "limit":3
 }```

this is a native request, but I don't know how to do that in SQL request

I don't know of a way to do this, would be a useful feature request (like a new SQL context parameter)

yes, could be nice as new SQL context parameter. I'll take some time to create a ticket on that point in the github, if it does not already exists.

created the ticket:
<https://github.com/apache/druid/issues/14215>

The simplest way to achieve this is likely using compaction jobs with a changing `queryGranularity` <https://druid.apache.org/docs/latest/data-management/compaction.html#query-granularity-handling>

you can also use SQL-based approach: <https://druid.apache.org/docs/latest/multi-stage-query/examples.html#replace-for-reindexing-an-existing-datasource-into-itself>

The PR for it is <https://github.com/apache/druid/pull/13144|here>. It explains that the supervisor in an idle state will stop creating new ingestion tasks. So I would expect those remaining tasks to finish their regular lifetime. The supervisor will still check for lag on the stream every so often (30seconds by default) and go back to a running state when lag&gt;0, back to creating tasks.

I see. Also, looks like the inflight tasks remain active until `taskDuration` , which makes sense.

16 may work but we do not explicitly test this setup

But not sure when last time it was updated.. so is 17 still experimental?

ah yeah although it's working pretty well as of druid 25. we're planning on marking it fully supported soon with no further code changes

nice!… IO assume then 0.26 will be fully supporting it? (whenever 26 comes out)

I'm not 100% sure if that made it for 26, I would think either 26 or 27

that being said, even 25 support has been working well, we have many folks using it in production

thats good news.. any eta on when 0.26 is coming out? I am about to setup a new cluster and wanted to go with the latest greatest

:slightly_smiling_face: it should be pretty soon

I believe it supports zstd compression as mentioned here <https://druid.apache.org/docs/latest/ingestion/native-batch.html#supported-compression-formats>. Not sure if it is supported in 0.19 though.

How do we set the property to accept zstd compressed parquet data in druid server? What is the property and where do we set it?

I see it in the supported list for 0.20.0: <https://druid.apache.org/docs/0.20.0/ingestion/native-batch.html#compression-formats-supported>

I don't see a property where you specify the incoming compression

IIRC for Parquet decompression is automatic as part of the reader

I have some follow-up questions:
1. What property is needed to be set on druid ingestion spec to ingest zstd compressed parquet data on druid version 0.19?
2. Do we need to install any packages on druid server for druid to parse the zstd compressed parquet data?

unfortunately I don't know the answers since I haven't use parquet with zstd before :sweat_smile:

Also, Druid version 0.19 is really old :slightly_smiling_face:

One path here could be to add a zstd Parquet file to <https://github.com/apache/druid/blob/master/extensions-core/parquet-extensions/example/compat>

And add a test case in `CompatParquetReaderTest`

This'll validate it works in the latest version, and in case there is some reason it doesn't, it would provide a test case to help figure out a fix

No problem. Maybe someone else from the community might know the answer. Yeah we are moving to 0.25 in a new cluster, but for now we are maintaining 0.19.

if the above way you mentioned can be done, that would be nice. Do we need to open a pull request for the same?

If you do; it'd be greatly appreciated and would help ensure it always works going forward!

If you do, post the PR link here and I'll have a look

ok, I mostly look at the infra side, I am not that comfortable with it. Let me check with other team member if he knows how to do it.

<@U03RH9440HY> will you be able to try it out?

For this data source, are the segments available already? If no segments are available, then you will get this error

Segments are available, but there also when we check the data, it gives query returned no data.

can you post screenshots of different tabs (datasources, query) in console that also shows the error you are seeing

`0.0% available`  - none of them are available

do you see `historicals` in `Services` tab?

you should also check that the historicals are healthy

I can see these things in services tab, can you guide me how to see if they are correct and in healthy state.

there is no historical running. the logs of historical will be in the same directory where druid is installed. I will suggest going through those logs

you mean to check the log file, if there is any issue, right?

How are you starting Druid? The issue is, like <@U031A9D19NU> said, is that your historical is not running

if you followed all the steps, it should be running already. it will be better to check the logs

I can see single warning in the log file as  Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
2023-05-03T00:00:03,654 WARN [main] org.apache.hadoop.util.NativeCodeLoader - Unable to load native-hadoop library for your platform... using builtin-java classes where applicable

do you see this line
``` Successfully started lifecycle [module]```
in the historical logs?

you can do a few more things. you can check if the process is running or not through `ps -eaf | grep historical` the command. Most likely, the process just died.

don't think it is since process id is constantly changing. you can check historical.stdout.log in the same log directory

its getting restarted continuously. it's weird that you don't see an error in the logs

what I need to do about it now, can you please help me solve the issue

As I checked the service, for me it shows running. also, I checked the grep historical once again with team and found out that task id is different for 2 different threads, sharing the highlighted snap below, the one with supervisor is actual service running for historical in druid, the other one is thread which got created during execution of logs probably.

Hi everyone,
1. Does anyone know how to use druid-extension functions in the druid console while writing sql druid queries ? I have created a custom function but unable to use them
2. When I get a doubleSketch value from druid in terms of String as druid query Response. How can I convert that String to a doubleSketch object in JAVA ?

I think <https://druid.apache.org/docs/latest/operations/request-logging.html#configure-request-logging|Request Logging> and <https://druid.apache.org/docs/latest/operations/request-logging.html#configure-metrics-emission|metrics> should be able to track<https://druid.apache.org/docs/latest/operations/metrics.html#query-metrics| each query in the broker and historical> by their query ID.

Thanks <@U030K4UM3H7>. While the document suggests broker tiering have a dependency on historical tiering, I don’t see there is a linkage between broker tiers and historical tiers. Is the tierToBrokerMap used to link between broker tiers and historical tiers?

I think that there are two levels of setup for this (three if you include the query context). The <https://druid.apache.org/docs/latest/design/router.html#router-strategies|router and how it decides to select a broker> (`druid.router.strategies`) for a given request and <https://druid.apache.org/docs/latest/operations/mixed-workloads.html#service-tiering|the broker/historical config which decides which set of historicals> to use to resolve a query. The query context also plays a role if you want specific queries to default to running on a particular tier (by specifying the `brokerService` ) or by giving the query a `priority` when using <https://druid.apache.org/docs/latest/design/router.html#priority|priority based routing strategy> where the brokers have a priority designation. Broker priority is based on `druid.router.tierToBrokerMap`  where highest priority is broker tier that is listed first. I haven't tried this myself, but that is how I understand this. Let us know how it goes.

I think I missed your specific question. For each broker set the `druid.broker.select.tier` to define strategy when selecting historical tier to use:
•  `highestPriority` would select the historical tier with the highest priority value (as set with `druid.server.priority` in each historical runtime.properties),
• `lowestPriority` selects historical tier with lowest priority 
• `custom` allows you to define historical tier selection with specific priority designations.

not yet. There is a PR here - <https://github.com/apache/druid/pull/9116> - if you are interested in contributing the feature, the PR is a good place to start.

got it, thanks for your response <@U031A9D19NU>!

I am not aware of any. I know someone was looking to connect springboot orm with druid - <https://www.linen.dev/s/apache-druid/t/10141550/hey-folks-anyone-working-on-the-druid-jdbc-connector-for-jav#38a658f0-f98d-4ca3-9aeb-fd8792c3eca1> - I had some generic advice that you may find useful.

It is probably much easier and quicker if you just connect to Druid using requests and mapped the results to objects yourself.

Yeah, but we have existing system developed using Django ORM storing data in SQL tables. I was trying to evaluate how easily can we migrate few tables to Druid and not modifying majority part of the code which uses the ORM..

as soon as it's ready :slightly_smiling_face:

<@U030C8H59T8> is release managering for it, I think there's still a few bug fixes we're looking to get in. for example <https://github.com/apache/druid/pull/14139>

full list at <https://github.com/apache/druid/milestone/50>

Hi <@U047P93LEUT>, please have a look at Kafka based lookups: <https://druid.apache.org/docs/latest/development/extensions-core/kafka-extraction-namespace.html>. With a Kafka lookup, you would always get the latest dimension value for your key.

I want something a bit different: I want Druid to send me updates to some topic

i’m aggregating data by dimensions in druid, so each time there is update in some dimension instance i would like get notification of that

oh wait, so you want to get data _out of_ druid?

Druid queries are normally pull-only. (At Imply we have alerts in out Pivot tool that send push notifications when something changes in Druid, though.)

ok, thanks. I think it might be nice feature :slightly_smiling_face:

Not sure why the historicals are failing. Looking at their logs should shed some light. The speed with which historicals will load segments depends on a few factors. Here's a <https://apachedruidworkspace.slack.com/archives/C0309C9L90D/p1682974111833419|recent thread on this>. Also, it makes sense to use a stateful set for the historicals such that their local storage of cached segments files is persistent across restarts.

Hello Sergio,
Thanks for the reply.
But I am unable to open the link you shared.

Do the logs indicate the historicals are loading from s3 on each start?

I ask that because 6 hours for that much data with even just one historical seems like a long time if the data already exists in a k8s volume on disk after a prior restart.

6 hours does seem like a super-excessive amount of time to load data from an already-existing PV

I wonder what it's logging during this time?

You probably see a bunch of lines like `Loading segment cache file [%d/%d]`

Normally a historical would have up to tens of thousands of those (possibly low-hundreds-of-thousands for very dense historicals; rarely more)

while this is happening do you see I/O maxing out on your PVs?

Hello <@U0344FW86DD>,
As you mentioned above we can see the Loading segment cache file [%d/%d] and Loading segment for data source in the historical logs during this time.
At some point historical pods restart and it will start loading the segment cache files from the beginning.
This cycle will repeat and will never ends.
Finally we ended up in clearing all data sources and started loading them from Kafka again.

MicrosoftTeams-image (1).png

MicrosoftTeams-image.png

Hello <@U04813M7M27>,
We are using the Azure storage account as the deep storage.
We didn't get anything from the logs to confirm it is loading the segments from the deep storage on each start.
But we can see two series of logs in the historical pods. The first is 'Loading segment Cache file' and the second is loading segment

Ignore literally everything I just said. It just looks like you have a ton of segments, so it takes a long time.

A few of thoughts:
• it would be good to understand why the historicals are failing. Is this kubernetes? Is kubernetes killing the pods because they are using too many resources? If so changing the resources available to each historical pod should help, or changing their config to use less resources.
• <https://druid.apache.org/docs/latest/operations/segment-optimization.html|Perhaps you've already done this>, but if not segment size optimization is always good to pay attention to.
• Historicals should not need to reload all of their segments after a quick restart. If you are using kubernetes, deploy them as a stateful set. Relevant info toward the <https://druid.apache.org/docs/latest/operations/segment-optimization.html|last paragraph in this section of the docs>. Essentially the historical pod that restarts looks at its local segment cache to announce the segments it already has. If you use statefulset for historicals, their local storage should still have the segments they need after the restart.


I don't think it is loading from s3, based on the screenshot. But it is loading about five segments per second. That is 4.5 hours to pull its segments from its cache. My best guess is whatever volumes you have mounted are very, very slow.

not exactly why would you want to do that.
Most storage classes support read write once on cloud.
For historicals, you should use statefulsets. Brokers are stateless you wont need pvc.

Hi Adheip,
Thank you for your response.
So you mean we don't need to use PVCs to historicals?
We have one data set with 1.5TB of data. We are using an azure storage account as deep storage.
So historical will try to load this data in the segment cache.
We are using persistent volume as the segment cache.
Is this the correct way of doing it?
Could you please advise?

`For historicals, you should use statefulsets.` , IMHO for brokers you dont need pvc.
The way you have described is correct.

Okay, so if there are two replicas for historical can we point both to the same PVC?

If we use two tiers say hot and default, do we need two PVC for these?

For tiering you would have 2 separate statefulsets, each statefulset can be scaled to N replicas.
So for hot 1 sts and default 1 sts,
You dont need to get confused on pvc's. `when you use statefulsets in k8s, for each replica you will have a pvc created.` 
Just clearly identify the abstraction  ie druid as an application and k8s as a platform. For druid its a historical tier, for k8s its a sts.

Oh got it. Thank you so much for making it clear.
We are facing issues with the data load in historical.
Each time the historical get restarted, it will start to load the data from deep storage.
It is taking more than 6 hours to load around 2TB data.
After the data load is completed Historicals will again restart and this is repeated.
Could you please help me understand the reason and solution for this?
I have put a separate post with more details on this issue. <https://apachedruidworkspace.slack.com/archives/C0303FDCZEZ/p1683213538747139>

I am definitely not an expert in this, but I looked at the code a bit and it seems to me that this is not related to the number of parameters. I think it may be a bug related to the function definition here where the parameter expects a literal value: <https://github.com/apache/druid/blob/fb38085ddbb439974325b230739e129ebcbf9ba1/extensions-core/datasketches/src/main/java/org/apache/druid/query/aggregation/datasketches/quantiles/sql/DoublesSketchApproxQuantileSqlAggregator.java#LL231C88-L231C88|https://github.com/apache/druid/blob/fb38085ddbb439974325b230739e129ebcbf9ba1/exte[…]hes/quantiles/sql/DoublesSketchApproxQuantileSqlAggregator.java>
but like I said, I am not an expert, I may be misreading how calcite interprets `OperandTypes.LITERAL`.

In <https://calcite.apache.org/javadocAggregate/org/apache/calcite/sql/type/OperandTypes.html> it describes the validation as:
```Operand type-checking strategy type must be a non-NULL literal.```
but I don't know if that would exclude use of parameters.

Does it work if you use a literal value in the SQL template ?

Nope, even Literal value in template did not work.

I noticed that when I try the following directly via query UI of druid
`SELECT APPROX_QUANTILE_DS( "duration" , CAST(0.99 as DOUBLE)) as p99 ...`
I get
```Error: Plan validation failed

org.apache.calcite.runtime.CalciteContextException: From line 13, column 8 to line 13, column 61: Cannot apply 'APPROX_QUANTILE_DS' to arguments of type 'APPROX_QUANTILE_DS(&lt;BIGINT&gt;, &lt;DOUBLE&gt;)'. Supported form(s): 'APPROX_QUANTILE_DS(column, probability)' 'APPROX_QUANTILE_DS(column, probability, k)'

org.apache.calcite.tools.ValidationException```
Looks like somewhere there is an implicit type conversion and it does not like double value

Interesting... what happens if you don't cast it, just `0.99`?

Then it works, `SELECT APPROX_QUANTILE_DS( "duration", 0.99)` is valid.

So, it essentially doesn't like expressions for the parameter... I think this is a bug. I will investigate further and create a github issue to track this.

PR for this: <https://github.com/apache/druid/pull/14249>

The issue is that this parameter must be a literal. We eliminate CASTs around literals, but CAST elimination happens after operator validation. So, at the point the operator args are checked, the CAST is still there and validation fails. The PR adjusts validation to accept literals wrapped in CAST (since they'll be eliminated later anyway)