Apache Druid

Hi, is there any reference for multi data center setup in druid? we use hdfs and i found these forum but can’t find where is yahoo druid video <https://www.druidforum.org/t/multi-datacenter-setup-druid/2338>

D you know where is yahoo druid cluster setup video?

I also found the broken link, so I can't help there. But <@U03149ALHN3> posted about such a configuration in <https://apachedruidworkspace.slack.com/archives/C0309C9L90D/p1683172472097029|this thread>.

broker-jvm.config

middleManager-jvm.config

router-jvm.config

historical-jvm.config

coordinator-overlord-jvm.config

When processes mysteriously die, the first suspect is memory issues. Thanks for the jvm settings, we would also need runtime settings to understand how direct memory has been allocated

You might be able to see this with
```dmesg -T | egrep -i 'killed process'```

middleManager-runtime.conf

coordinator-overlord-runtime.conf

historical-runtime.conf

broker-runtime.conf

router-runtime.conf

I have attached runtime settings for the reference. Can you provide the proper configurations for the services according to the server size?

Is this all running on one host? Just a quick bit of math shows too much memory has been allocated / not enough memory on the host
Broker process: 12GB heap + 9GB Direct memory (16 merge buffers + 1 processing buffer @ 0.5GB each)
Historical process: 8GB heap + 12.5GB Direct memory (8 merge buffers + 16 processing buffers @ 0.5GB each)
Coordinator process: 15GB heap
so we’re at 56GB allocated and we haven’t run a task yet.

Thank you sir, Yes all these services are running on one host (new-production server). The host has only 48gb RAM and zookeeper and kafka also running on this host. kafka has 7 ingestion for tables also.

The Druid image entrypoint is `druid.sh`  it uses the environment variable (from the environment file) `DRUID_SINGLE_NODE_CONF=micro-quickstart` which overwrites the paths to the configuration files as `/tmp/conf/druid/single-server/$DRUID_SINGLE_NODE_CONF/...` . This is probably why you are not seeing your changes.

It copies the configuration files from `$DRUID_CONFIG_COMMON` into the `/tmp/conf/druid/single-server/...` .I think that if you specify `DRUID_CONFIG_COMMON=/opt/druid/conf/druid/cluster/_common` in your environment file, it will take your common config.

<@U030K4UM3H7> hey, thanks for responding. I have question - `$DRUID_CONFIG_COMMON` should be declared inside the same environment file where the extension loadlist is present or a separate .env ??

177abf33b6b9e62259634b93576e8f2.png

Hi Kevin, who owns that directory that Druid is trying to create the FIFO in, and what are the permissions?

Also may I remark that it’s maybe not a good idea to run Druid as root

image.png

If you use SQL Based Ingestion, you can do something like:
```REPLACE INTO "inline_data" OVERWRITE ALL
WITH "ext" AS (SELECT *
FROM TABLE(
  EXTERN(
    '{"type":"inline","data":"{\"timestamp\":\"2023-01-01 00:00:00\", \"array_col\":[\"a\",\"b\",\"c\"], \"metric\":5},\n{\"timestamp\":\"2023-01-01 00:00:00\", \"array_col\":[\"b\",\"c\",\"d\"], \"metric\":50},\n{\"timestamp\":\"2023-01-01 00:00:00\", \"array_col\":[\"e\",\"f\",\"d\"], \"metric\":500}"}',
    '{"type":"json"}'
  )
) EXTEND ("timestamp" VARCHAR, "array_col" VARCHAR, "metric" BIGINT))
SELECT
  TIME_PARSE(TRIM("timestamp")) AS "__time",
  "array_col" as "array_col",
  SUM("metric") AS "sum_metric"
FROM "ext"
GROUP BY 1,2
PARTITIONED BY DAY```
You will need to set the query context parameter `groupByEnableMultiValueUnnesting:true`:

Actually we are using Batch ingestion with `index_parallel` setting. Is there any proper way to explode the arrays based on this ingestion method?

I don't think `index_parallel` has that behavior. If you are on 26.0.0, you could also consider loading it as an array column (only available though <https://druid.apache.org/docs/latest/ingestion/schema-design.html#type-aware-schema-discovery|schema auto discovery >for now) and use <https://druid.apache.org/docs/latest/querying/sql.html#unnest|UNNEST> to aggregate at query time.

Got it, thank you so much, I'll see what I can do :+1:

Going through the logs, its likely caused by this:

com.amazonaws.services.s3.internal.S3AbortableInputStream - Not all bytes were read from the S3ObjectInputStream, aborting HTTP connection. This is likely an error and may result in sub-optimal behavior. Request only the bytes you need via a ranged GET or drain the input stream after use.

This is an existing problem:  <https://github.com/apache/druid/issues/6905>

<@U05H1DNHM7C>, looking at the linked GitHub issue you posted, it seems like the issue was fixed in 0.21 and/or subsequent releases. What version of Druid are you on?

Hi <@U030K4UM3H7> <@U031A9D19NU>, I face this problem quite often. Maybe because of large number of segments. Problem is, once a historical node is stuck at one segment,it cannot move on which leads to unavailability of Datasources. The only way out I know is restart of cluster. Pulling segments from deep storage takes 4-5 hours. So there is a significant downtime. IS there any way to avoid this ?

Hey <@U05H1DNHM7C>, taking a dive into the code. That error message occurs when writing to the local segment cache folder on the node and an IOException occurs:
```      if (!segmentInfoCacheFile.exists()) {
        try {
          jsonMapper.writeValue(segmentInfoCacheFile, segment);
        }
        catch (IOException e) {
          removeSegment(segment, callback, false);
          throw new SegmentLoadingException(
              e,
              "Failed to write to disk segment info cache file[%s]",
              segmentInfoCacheFile
          );
        }```
Just verifying the basics before digging elsewhere:
• is there space on the node's /data/druid/segment-cache/ mount?
• does it have adequate permissions for the user that druid processes run with?

How did you make the connection from that to the "S3AbortableInputStream - Not all bytes were read from the S3ObjectInputStream" problem. What log do you see these messages in?

error.log

HI <@U030K4UM3H7> Thanks for revert. I have attached a log about how i make a connection. The Failed to write disk segment occurs only after S3 ObjectInputStream error.
Is there space on disk? Yes.
Adequate permissions? I start druid from systemd service as root user. So yes it has permissions.

I see... could you verify that deep storage contains the file `bucket='quantseye-poc', path='druid/segments/TBT_cash/2023-06-26T00:00:00.000Z_2023-06-27T00:00:00.000Z/2023-07-01T16:35:11.280Z/53/index.zip'` and could you unzip it to see if it contains the files that the log reports as missing:
`00000.smoosh`
`factory.json`

yes <@U030K4UM3H7>, manually downloading from s3, I can see smoosh and factory json file.

I had one more observation yesterday.  when I was using i3en.xlarge instance the reproduction rate was 1/8 times. But yesterday I switched to using i3en.3xlarge, and reproduction rate was 100%. The historicals never came up.

I use following script as systemd service as root user to mount ephemeral storage.
#!/usr/bin/bash
mkfs.ext4 -F /dev/nvme1n1
mkdir -p /data/druid
mount /dev/nvme1n1 /data

I am stumped. Not sure what's going on here. In the 100% reproduction case, are you still seeing the same errors?

Hi <@U030K4UM3H7>, In the 100% reproduction case I did not come across "S3AbortableInputStream - Not all bytes were read" in the logs. I see only "Failed to write to disk segment info cache file" errors.

I'm still stumped... but studying the error and code a bit more, it seems like it is trying to write to a folder that does not exist. I'm wondering if this has anything to do with `druid.segmentCache.lazyLoadOnStart`... do you have this set to `true`?

Sounds a bit like apples and oranges.... so I'm not sure about efficiency but here are a few thoughts:
• SQL Input sources will only be parallelized if you specify multiple SQL statements, each will created a distinct input stream that are executed in parallel. There is networking involved in transmission of the data from the source DB to the Druid ingestion tasks.
• Local input sources are read directly from disk and having multiple files will help parallelism, but local implies that the data is accessible to all MM nodes in a local mount, such that no matter where the task runs, it has visibility into the files locally. So if this is shared storage, the network transfer will still be needed. If it is local storage you would need to copy the files to all MM nodes. S3 or some other distributed storage is probably a better solution for source files. 
• I'm not sure if there is a significant difference in the record parsing time between the two, I suspect it will depend a lot on the format of the local files.
• there's probably a whole bunch of other considerations that don't come to mind right now :wink: 

#1 - yes, total memory for the pod should be greater than heap+direct memory setting
#2 - have you tried <https://druid.apache.org/docs/latest/operations/request-logging.html#configure-request-logging|request logging>, <https://druid.apache.org/docs/latest/operations/metrics.html#general-health|health metrics> and <https://druid.apache.org/docs/latest/operations/metrics.html#query-metrics|query metrics> as a means of troubleshooting?

<https://github.com/apache/druid/pull/14359> - This patch introduced a new system monitor that should support arm instances. I have not tried it myself yet

20230719_082426_HDR~2.jpg

Is your Druid endpoint URL correct? This seems to be a 10.x network which would be in a private network; you cannot reach that from outside that nework

Endpoint is correct only..if send some limit of data it is copying

<@U030K4UM3H7> can you please let me know if you can able to help on above issue . Thank you

Sorry for the delayed response.
There could be many tuning elements. Here are a few thoughts:
• Are these nodes MM only nodes or are they running MM+Historical. If MM only, it seems like you could go higher on worker capacity. If mixed, it depends on how you are balancing resources between the two.
• You can control how much memory each Peon is allocating and perhaps reduce it in `javaOptsArray` so that you can have more workers per MM server and still fit in memory. ( <https://druid.apache.org/docs/latest/configuration/index.html#middlemanager-configuration|druid.indexer.runner.javaOptsArray in Middle Manager config>) -XMS -XMX for Heap and -XX:MaxDirectMemorySize for direct memory allocation. Remember that max direct memory should be allocated based on the formula `druid.processing.buffer.sizeBytes * (druid.processing.numMergeBuffers + druid.processing.numThreads + 1)` so you may also want to adjust those values.
• Along with adjusting Peon configs, control the use of memory within each task by setting:
    ◦ maxRowsInMemory / maxBytesInMemory - when these limits are hit for the row buffers, it will spill to disk
    ◦ maxColumnsToMerge is also important if you have a lot of columns and a lot of intermediate persists. By default it merges all columns in direct memory buffers at the same time, by setting this value you can control the memory footprint of the merge operation.
Let us know how it goes.

Thanks <@U030K4UM3H7> for the inputs . These nodes are only MM nodes we runs historicals on separate EC2 machines . if we enable `druid.indexer.runner.javaOptsArray`  this in MM runtime properties we don't need jvm.config setting right ?

jvm.config for the Middle Manager is still needed <https://druid.apache.org/docs/latest/operations/basic-cluster-tuning.html#middlemanager-heap-sizing|but it doesn't use a lot of resources>. By default, if you do not specify `druid.indexer.runner.javaOptsArray` , Peons will be instantiated using the MM jvm settings which is almost always the wrong choice since Peons require much more resources.

ok .. we have very high heap in jvm.config so peons are also using that . i will play around with different -XMS and -XMX settings

<@U030K4UM3H7> wanted to provide an update . i started tested with the recommendations you provided .
1. had xms:128m and xmx 5g with direct memory as 10g as per the formula and worker capacity as 20 . tasks getting failed during merge phase with Direct out of memory . So increased it to 15g and  Ec2 utilization also increased and is around 80% for memory and 65% for CPU . we are planning to increase worker capacity to 25 and start some more testing . 
2. i have tried adding maxColumnsToMerge to 300 job took an hour more so removed this for now  


one more question worker capacity can go as high as no.of cores - 1 but on EC2 it is VCpu for example if we have 48 VCPU which is basically 24 core so can we go more than 23 worker capacity if we see host utilization is less

The formula is meant for VCPUs, not physical cores, so if you have enough memory you could run more workers. In theory up to 47. but 47*15GB per task = 705GB, so you might need to adjust settings.

<@U030K4UM3H7> one more question on this topic . can we pass this `javaOptsArray` as part of ingestion spec ? if we keep this parameter in the runtime properties all the tasks use same memory config but some tasks need very less memory so wanted to check if we can overwrite with by passing this in ingestion spec

That would be nice but I don't think that exists today. Would you like to <https://github.com/apache/druid/issues/new/choose|write an improvement request>?

If you are on k8s, I think some <https://druid.apache.org/docs/latest/development/extensions-contrib/k8s-jobs|semblance of that >can be achieved if you are using `kubernetes-overlord-extension` <https://druid.apache.org/docs/latest/development/extensions-contrib/k8s-jobs|extension> which run tasks as kubernetes jobs and provides elasticity for tasks, including custom pod templates which can vary by task type. Not the same as "by job" but closer :smile:

Screenshot 2023-07-20 at 1.25.12 pm.png

Screenshot 2023-07-20 at 1.24.36 pm.png

I think this might be because of the hourly Supervisor handover - all Kafka ingestion tasks for a given supervisor hand over and restart at the same time, which creates a load spike. This is addressed in <https://github.com/apache/druid/pull/14396>, which randomizes the task duration a bit so not all tasks hand over at the same time.

The pattern definitely looks like something that is happening at task handover. The kafka lag pattern is what I'd expect to see when there is not enough worker capacity to start new tasks while the old ones are spinning down and handing data off.

<https://druid.apache.org/docs/latest/development/extensions-core/kafka-supervisor-operations.html#capacity-planning> - Here are recommendations on what additional capacity is needed for smooth task handover.

Actually I’m not sure this is causing the issue in our circumstance. We have already set each of our kafka indexing tasks to a duration which was chosen to avoid them from starting at the same time, and none of them are a multiple of 1 hour (e.g. we use PT10M31S, PT11M, PT11M11S, PT11M33S, etc). We also have plenty of spare worker capacity - on the indexers that are having the most issues we have 768 total slots and 207 in-use.

Hmm, the other thing that runs hourly by default is coordinator metadata management (configuration: `druid.coordinator.period.metadataStoreManagementPeriod`), maybe this is what causes the spikes?

we’ve also tried changing that to 65 minutes in the past and we still noticed the hourly lag jumps

Is it possible that there is a cron job running that is dumping a lot of messages on to the stream every hour? Looking at the events processed metric from druid would help confirm this suspicion

Screenshot 2023-08-07 at 9.56.36 am.png

Screenshot 2023-08-07 at 9.52.47 am.png

i tend to see a dip in events processed on the hour and then a peak right after that as the indexers “catch up” from whatever causes the dip. All the events are from a kafka stream that has messages coming in pretty steadily so to me it seems like the peak is just a side-effect of the dip.

I found the issue related: <https://github.com/apache/druid/issues/11758>

Are your group by fields numeric? Can you try casting them to VARCHAR to see if the behavior changes?

0.17.1. Yes, if I cast to varchar, result is correct.

Is the system really busy? Could it be a network problem?
Your settings are in line with the recommendations.
Any real-time ingestion? threads on those? Are you <https://druid.apache.org/docs/latest/operations/metrics.html#jetty|collecting metrics>?

• System is not busy at all
• Not sure if this is a network problem, since it seemed to happen after I changed the numThread etc to a higher configuration since our queries were slow. i.e. not network related changes were made and it was working before
• There is no real-time ingestion. Only batch
• We not collecting jetty metrics. But am not sure what exactly to check here. Any suggestions? 

I am trying with reduced http.numThreads and numConnections.

Seem that above change did the trick. Not getting that error now.

but those numbers were not very high, I wonder if this is a ulimit problem. You might just need to adjust ulimit : <https://druid.apache.org/docs/latest/operations/basic-cluster-tuning.html#ulimit>

Thanks for sharing this. I will try this out!

Trying to understand... what caught my eye is that the timestamp for the row it cannot add is outside the segment's time interval. Can you share the ingestion spec or SQL you are using for this?

```{
  "type": "single_phase_sub_task",
  "id": "single_phase_sub_task_payment_action_v2_amiceajj_2023-07-20T14:58:23.250Z",
  "groupId": "index_parallel_payment_action_v2_confjfem_2023-07-20T13:05:52.868Z",
  "resource": {
    "availabilityGroup": "single_phase_sub_task_payment_action_v2_amiceajj_2023-07-20T14:58:23.250Z",
    "requiredCapacity": 1
  },
  "supervisorTaskId": "index_parallel_payment_action_v2_confjfem_2023-07-20T13:05:52.868Z",
  "subtaskSpecId": "index_parallel_payment_action_v2_confjfem_2023-07-20T13:05:52.868Z_32",
  "numAttempts": 2,
  "spec": {
    "dataSchema": {
      "dataSource": "payment_action_v2",
      "timestampSpec": {
        "column": "event_time",
        "format": "iso",
        "missingValue": null
      },
      "dimensionsSpec": {
        "dimensions": [

           ......... 
     
        ],
        "dimensionExclusions": [
          "__time",
          "event_time"
        ],
        "includeAllDimensions": false
      },
      "metricsSpec": [],
      "granularitySpec": {
        "type": "uniform",
        "segmentGranularity": "DAY",
        "queryGranularity": {
          "type": "none"
        },
        "rollup": false,
        "intervals": []
      },
      "transformSpec": {
        "filter": null,
        "transforms": []
      }
    },
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "s3",
        "objects": [
          {
            "bucket": "dw",
            "path": "backfill-payment/partkey_year=2023/partkey_month=07/partkey_day=19/part-00028-0337956a-5d56-456d-8131-6defebaccf99.c000.snappy.parquet"
          },
          {
            "bucket": "dw",
            "path": "backfill-payment/partkey_year=2023/partkey_month=07/partkey_day=20/part-00028-5dd4b8bc-ccf6-4faa-89e4-815bb37b358e.c000.snappy.parquet"
          }
        ]
      },
      "inputFormat": {
        "type": "parquet"
      },
      "appendToExisting": true,
      "dropExisting": false
    },
    "tuningConfig": {
      "type": "index_parallel",
      "maxRowsPerSegment": 5000000,
      "appendableIndexSpec": {
        "type": "onheap",
        "preserveExistingMetrics": false
      },
      "maxRowsInMemory": 1000000,
      "maxBytesInMemory": 0,
      "skipBytesInMemoryOverheadCheck": false,
      "maxTotalRows": null,
      "numShards": null,
      "splitHintSpec": null,
      "partitionsSpec": {
        "type": "dynamic",
        "maxRowsPerSegment": 5000000,
        "maxTotalRows": null
      },
      "indexSpec": {
        "bitmap": {
          "type": "roaring",
          "compressRunOnSerialization": true
        },
        "dimensionCompression": "lz4",
        "stringDictionaryEncoding": {
          "type": "utf8"
        },
        "metricCompression": "lz4",
        "longEncoding": "longs"
      },
      "indexSpecForIntermediatePersists": {
        "bitmap": {
          "type": "roaring",
          "compressRunOnSerialization": true
        },
        "dimensionCompression": "lz4",
        "stringDictionaryEncoding": {
          "type": "utf8"
        },
        "metricCompression": "lz4",
        "longEncoding": "longs"
      },
      "maxPendingPersists": 0,
      "forceGuaranteedRollup": false,
      "reportParseExceptions": false,
      "pushTimeout": 0,
      "segmentWriteOutMediumFactory": null,
      "maxNumConcurrentSubTasks": 5,
      "maxRetry": 3,
      "taskStatusCheckPeriodMs": 1000,
      "chatHandlerTimeout": "PT10S",
      "chatHandlerNumRetries": 5,
      "maxNumSegmentsToMerge": 100,
      "totalNumMergeTasks": 10,
      "logParseExceptions": false,
      "maxParseExceptions": 2147483647,
      "maxSavedParseExceptions": 0,
      "maxColumnsToMerge": -1,
      "awaitSegmentAvailabilityTimeoutMillis": 0,
      "maxAllowedLockCount": -1,
      "partitionDimensions": []
    }
  },
  "context": {
    "forceTimeChunkLock": true,
    "useLineageBasedSegmentAllocation": true
  },
  "dataSource": "payment_action_v2"
}```

Druid version?
I don't see anything odd in the spec.
Is there any real-time ingestion going on as well on this table or is this purely batch? I'm thinking maybe locking given that you are loading data  for today (07/20/2023)... just a thought.
I noticed you are using `dynamic` partitioning, not that it would solve this, but given that this is a batch process you might want to look into `range`<https://druid.apache.org/docs/latest/ingestion/native-batch.html#multi-dimension-range-partitioning| partitioning> (and there's a chance that it turns out to be a workaround, given the different data shuffling that it does).

disregard the range comment, I just realized you are appending data.

the version is 25.
I’m currently deleting and reloading the same datasource repeatedly, could that be causing the problem?

I always have deleted segments (not datasource entirely) using REST API (delete)

So, I figured out the data is still left in deep storage.

That shouldn't be a problem on its own. Segments are marked as unused and a kill job will get rid of the deep storage segments you've "deleted".
Given that you are deleting and then reloading... does append make sense? or replace?
Assuming you do want to append, let me see if I can reproduce this. Can we confirm the sequence:
• ingest parquet files using append to existing
• delete segments, but no kill executed
• ingest parquet files again using append
Is that correct?  did you see the error on the first ingestion?

But I checked my deep storage, the segment I deleted is left.
And at first, there was no error like that.

I just tried ingesting to new datasource name.
I will let you know the result.

I would also recommend you to try out sql based ingestion. We have made a lot of improvements in it so that most things run out of the box.

and yes, segments that are deleted are marked as unused... they are only cleaned up (actually deleted) when you run a kill task. <https://druid.apache.org/docs/latest/configuration/index.html#coordinator-operation|This can be configured to automatically run on a periodic basis in the coordinator> (look at the `druid.coordinator.kill...` properties.

I second Karan's recommendation. <https://druid.apache.org/docs/latest/multi-stage-query/index.html|Sql Based Ingestion docs>

Ok , I’ll see SQL-based ingestion.
and the result is success when I changed the name of datasource.

hmm, I don’t see `druid.indexer.task.inputSource.type` in the docs - where did you find the config? Afaik, once the s3 extension is loaded, you can submit native ingestion specs or DML-based ingestion with s3 as the input source right out of the box (no additional config needed)

thanks for reply <@U04CLDXP7HC>. I just found that option and tried.
after adding `druid-s3-extensions` I have restarted the druid service, still getting same error. I am trying native spec for loading s3 data.

Can you check the list of loaded extensions in console to see if s3 extension is loaded properly.

Hmmm ... I just tried that syntax on the wikipedia demo datasource and it worked for me:

```select * from "wikipedia_hash"
 where channel in ('#en.wikipedia','#sv.wikipedia')
  or channel in (select channel from wikipedia where flags = 'NB')```
and
```select * from "wikipedia_hash"
 where channel in ('#en.wikipedia','#sv.wikipedia')
  or channel in (select channel from wikipedia where flags = 'no_match')```
both seem to do what they are supposed to do.

My native query for the second statement is below.

You may want to try one or more of the following:
• compare your native query to below to see if it's any different
• Pull your OR conditions apart into separate queries, make sure each does what you expect it to do
• Load up the wikipedia datasource and try these queries out on them to see if you get the same result.
• Also check your useDefaultIfNull setting in case that is affecting behavior here ... mine is set to FALSE.
Thanks.  John
```{
  "queryType": "scan",
  "dataSource": {
    "type": "join",
    "left": {
      "type": "join",
      "left": {
        "type": "table",
        "name": "wikipedia_hash"
      },
      "right": {
        "type": "query",
        "query": {
          "queryType": "timeseries",
          "dataSource": {
            "type": "table",
            "name": "wikipedia"
          },
          "intervals": {
            "type": "intervals",
            "intervals": [
              "-146136543-09-08T08:23:32.096Z/146140482-04-24T15:36:27.903Z"
            ]
          },
          "filter": {
            "type": "selector",
            "dimension": "flags",
            "value": "no_match"
          },
          "granularity": {
            "type": "all"
          },
          "aggregations": [
            {
              "type": "count",
              "name": "a0"
            }
          ],
          "context": {
            "queryId": "dccb3b1b-6387-48ce-87c7-dae1833d2600",
            "sqlOuterLimit": 1001,
            "sqlQueryId": "dccb3b1b-6387-48ce-87c7-dae1833d2600",
            "useNativeQueryExplain": true
          }
        }
      },
      "rightPrefix": "j0.",
      "condition": "1",
      "joinType": "INNER"
    },
    "right": {
      "type": "query",
      "query": {
        "queryType": "groupBy",
        "dataSource": {
          "type": "table",
          "name": "wikipedia"
        },
        "intervals": {
          "type": "intervals",
          "intervals": [
            "-146136543-09-08T08:23:32.096Z/146140482-04-24T15:36:27.903Z"
          ]
        },
        "virtualColumns": [
          {
            "type": "expression",
            "name": "v0",
            "expression": "1",
            "outputType": "LONG"
          }
        ],
        "filter": {
          "type": "selector",
          "dimension": "flags",
          "value": "no_match"
        },
        "granularity": {
          "type": "all"
        },
        "dimensions": [
          {
            "type": "default",
            "dimension": "channel",
            "outputName": "d0",
            "outputType": "STRING"
          },
          {
            "type": "default",
            "dimension": "v0",
            "outputName": "d1",
            "outputType": "LONG"
          }
        ],
        "limitSpec": {
          "type": "NoopLimitSpec"
        },
        "context": {
          "queryId": "dccb3b1b-6387-48ce-87c7-dae1833d2600",
          "sqlOuterLimit": 1001,
          "sqlQueryId": "dccb3b1b-6387-48ce-87c7-dae1833d2600",
          "useNativeQueryExplain": true
        }
      }
    },
    "rightPrefix": "_j0.",
    "condition": "(\"channel\" == \"_j0.d0\")",
    "joinType": "LEFT"
  },
  "intervals": {
    "type": "intervals",
    "intervals": [
      "-146136543-09-08T08:23:32.096Z/146140482-04-24T15:36:27.903Z"
    ]
  },
  "resultFormat": "compactedList",
  "limit": 1001,
  "filter": {
    "type": "or",
    "fields": [
      {
        "type": "and",
        "fields": [
          {
            "type": "not",
            "field": {
              "type": "selector",
              "dimension": "j0.a0",
              "value": "0"
            }
          },
          {
            "type": "not",
            "field": {
              "type": "selector",
              "dimension": "_j0.d1",
              "value": null
            }
          },
          {
            "type": "not",
            "field": {
              "type": "selector",
              "dimension": "channel",
              "value": null
            }
          }
        ]
      },
      {
        "type": "in",
        "dimension": "channel",
        "values": [
          "#en.wikipedia",
          "#sv.wikipedia"
        ]
      }
    ]
  },
  "columns": [
    "__time",
    "added",
    "channel",
    "cityName",
    "comment",
    "commentLength",
    "countryIsoCode",
    "countryName",
    "deleted",
    "delta",
    "deltaBucket",
    "diffUrl",
    "flags",
    "isAnonymous",
    "isMinor",
    "isNew",
    "isRobot",
    "isUnpatrolled",
    "metroCode",
    "namespace",
    "page",
    "regionIsoCode",
    "regionName",
    "user"
  ],
  "legacy": false,
  "context": {
    "queryId": "dccb3b1b-6387-48ce-87c7-dae1833d2600",
    "sqlOuterLimit": 1001,
    "sqlQueryId": "dccb3b1b-6387-48ce-87c7-dae1833d2600",
    "useNativeQueryExplain": true
  },
  "granularity": {
    "type": "all"
  }
}```

Hi <@U04C4ADUGD9>,
I have pasted actual query below.
second part which is "manf_desc IN (
        SELECT manf_desc .... )" not giving any rows and first condition is giving rows but as a whole query if i run, not getting any result.

SELECT distinct fe.manf_desc AS manf_desc,sales_year,sales_month,sales_quarter
FROM
denorm_fact_extrapolated_unspsc fe
WHERE
(LOWER(manf_desc) = LOWER('3M')
or
manf_desc IN (
        SELECT manf_desc as top5manf_desc FROM denorm_fact_extrapolated_unspsc fe where __time BETWEEN
        TIME_PARSE ('2019-01-31T00:00:00.000Z') AND TIME_PARSE (LOOKUP('max_end_date_monthly_user', 'max_min_dates'))
        AND LOWER(fe.manf_desc) != LOWER('3M') and fe.taxonomy_id in (
                                                                    SELECT unspsc_code FROM denorm_manf_entitlements
                                                                    where email_address= '<mailto:dev3@dresources-hidam1.com|dev3@dresources-hidam1.com>')
        AND fe.facility_type IN('HOSPITAL/HEALTH SYSTEM')
        AND fe.sku_manf IN('001505 - 3M','05113199524 - 3M')
        and fe.unspsc_3 IN ('Bandages and dressings and related products',
        'Medical tissue closure and related products') and fe.unspsc_4 IN ('Adhesive dry bandages or dressings','Alginate dressings',
        'Medical adhesive strip or topical glue for skin closure') group by manf_desc order by SUM(fe.dist_total_revenue) desc limit 5
        )
      )

Hi <@U04C4ADUGD9>,
Query is failing to give result when using group by clause in the subquery.

Query is failing in the line "....*group by manf_desc...*"

Ok I can reproduce it in wikipedia ...

```select * from "wikipedia_hash"
 where channel in ('#en.wikipedia','#sv.wikipedia')
  or channel in (select channel from wikipedia where flags = 'no_match' group by channel)```
and
```select * from "wikipedia_hash"
 where channel in ('#en.wikipedia','#sv.wikipedia')
  or channel in (select distinct channel from wikipedia where flags = 'no_match')```
return no records ...

... whereas
```select * from "wikipedia_hash"
 where channel in ('#en.wikipedia','#sv.wikipedia')
  or channel in (select channel from wikipedia where flags = 'no_match')```
returns records ... the only difference being the group by or distinct in the subquery.

In general I'm not a fan of putting join conditions on one side of the OR (an IN clause is essentially a join) so you could also play it safer and outer join to your subquery and then check for matches in the OR, e.g.:

```select w1.* from "wikipedia_hash" w1
  left join (select channel from wikipedia where flags = 'no_match' group by channel) w2 on w1.channel = w2.channel
 where w1.channel in ('#en.wikipedia','#sv.wikipedia')
    or w1.channel = w2.channel```
would that work for your query?

I am fairly certain I have seen `OPTIMIZE TABLE` have some large benefits even with the cluster running. I don’t think you need to do a dump/restore…

Under what conditions are you seeing this? High Concurrency?
Or is this from a single connection that is running multiple statements?

Yes, high concurrency as part of load testing. Should be from different connections

Could this issue be related to <https://druid.apache.org/docs/latest/operations/basic-cluster-tuning.html#ulimit> too?

I took a quick dive in the code. This happens in the JDBC createStatement or createPrepatedStatement. Is the target application closing its statements?

ulimit can be an issue, but I'm not sure it is related.

Target application is using connection pooling and uses a With statement. So ideally the connection should be released back to the pool. I am a bit lost on this one, as you mentioned there is something wrong between the interaction between client and druid. Even after I stopped the locust performance test, I could see the same error until a few minutes. I believe during this the connections drained out and hence works after a period. If I kill the client pod the issue goes away immediately but repeats after 1-2 mins into the performance test.

By the way, when I increased this - druid.sql.avatica.maxStatementsPerConnection=150 .. I started getting the error at 150. "Maximum number of simultaneous open statements per Avatica client connection". I dont see many queries with bad queryTime in metric, which implies queries are executing fast enough. Hence still not sure why it would reach 150 open statements per connection in a connection pool tried with 10 - 30 connections.

<https://druid.apache.org/docs/latest/querying/sql-jdbc.html#connection-stickiness> &lt;- If we have not configured stickiness. Could this issue happen since the state is not maintained?

That's definitely something you need to use JDBC correctly. I'm not sure that would cause this though. In principal, what is happening is that a single connection is creating many statements but not closing them. So I think the problem is on the client side.

I just increased druid.sql.avatica.maxStatementsPerConnection=10000 and this error is gone. Seems like in one connection itself a lot of statements are pushed and closing connection at client side doesnt matter. This is a weird behavior for avatica JDBC based connections.

Did you set up stickiness? It just occurred to me that if the `close statement` goes to another broker, then it will likely just error out, and the statement it was meant to close stays open on a different broker. Do you have many brokers? Are you going through the router?

Yes, we are are going through router hence sticky load balancers are not required as per documentation. I will performance test more soon. As of now, I think raising maxStatementsPerConnection worked and does not seem to have a bad side-effect.

Hi team,

Clarification about a config on historical

```druid.coordinator.loadqueuepeon.curator.numCallbackThreads

Number of threads for executing callback actions associated with loading or dropping of segments. One might want to increase this number when noticing clusters are lagging behind w.r.t. balancing segments across historical nodes.```
*Any idea how do i measure that cluster is lagging behind w.r.t balancing segments ?  and what is the meaning of lagging behind ?* 

I've noticed same issue recently with coordinator metrics. But didn't look at it closely yet - initially I thought that this is coordinator specific, since we have few other issues with coordinator metric emissions (eg. frequent `NullPointerException`s). Do you have any other examples?

Any of the query metrics from the historicals (query/time or query/segment/time) when enabling at least the 'id' dimension

<@U032CA4JJV6>: I am using Range Partitioning with targetRowsPerSegment set to *1000000*, still I can see one partition of rows *7856095*
Can you explain the reason for it?
```{
            "type": "index_parallel",
            "partitionsSpec": {
                "type": "range",
                "partitionDimensions": ["category", "gender","brand_name","seller_id"],
                "targetRowsPerSegment": 1000000
            },
            "maxNumConcurrentSubTasks": 5,
            "forceGuaranteedRollup" : true
        }```

Also, range partitioning in native batch can be susceptible to skew. Is there a combination of ["category", "gender","brand_name","seller_id"] that has many more rows than others?
<https://druid.apache.org/docs/latest/multi-stage-query/index.html|SQL Based Ingestion> (REPLACE without GROUP BY) is enhanced to address this by adding an artificial/virtual hash column to the partitioning strategy, such that skewed values are further partitioned and more balanced segments are created.

You need to use timestamps, for example:
```TIME_FLOOR(CURRENT_TIMESTAMP - INTERVAL '12' MONTH, 'P1D')```
which results in the start of the day that is 12 months ago, if you want it to be the start of the month you can change `P1D` to `P1M`

and I just looked at your data and realized your _time column is not populated, so you are looking for a condition using both columns `yea_r_v` and `month_v` which sounds more complicated... why not load the data using the _time column set to the beginning  of the month i.e. `year__v-month_v-01 00:00:00` and then you can use the condition
```__time &gt; TIME_FLOOR(CURRENT_TIMESTAMP - INTERVAL '12' MONTH, 'P1D')```

Sure; that sounds good idea. let me check the data.

if you open the "Developer console" in your browser (Chrome), you can see the whole request that gets this error in response. there will be a query in that response.

yeh when it tries to run select * from datasource it gets that error, and also that datasource is not available for query, and information_schema.COLUMNS dose not have that entry.
but those are present in metadata RDS table.

this means that while segments for that datasource were created at some point, they are not loaded on any historical.

A datasource is queryable in SQL only when its loaded on at least one historical

data is available on historical as it matches the total data size from service page and datasource page

Thanks <@U031A9D19NU> restarting the coordinator and overlord did the trick.

This is an example of what I am looking for: <https://trino.io/docs/current/sql/explain-analyze.html>

Yes, you'll need to<https://druid.apache.org/docs/latest/configuration/index.html#enabling-metrics| enable metrics> to see <https://druid.apache.org/docs/latest/operations/metrics.html#query-metrics|query metrics>.

Although an EXPLAIN ANALYZE would be really cool. 

&gt; Yes, you’ll need to<https://druid.apache.org/docs/latest/configuration/index.html#enabling-metrics| enable metrics> to see <https://druid.apache.org/docs/latest/operations/metrics.html#query-metrics|query metrics>.
Metrics are not very helpful for debugging specific queries.

But better than nothing for sure.
Thanks sergio.

The metrics are perfect for debugging specific queries if you enable the right dimensions within the metrics. query/time and query/segment/time (once you enable the id dimension) carry the query ID as a label you can trace through your system to find how much time it's spent and where.

The hard part is consuming those sorts of metrics at scale. In my opinion those dimensions are not enabled by default as the cardinality of the metrics at scale overwhelms the consuming system (prometheus in our case).

what version are you on? are these properties set on broker?

<@U031A9D19NU> 26.0.0
yes the properties are set

Now I am getting -  java.nio.file.NoSuchFileException
I am using the absolute path `/home/ubuntu/_druid/local/v2/apache-druid-26.0.0/truststore.jks`

I passed the properties via environment file

```# Setup client side TLS
druid.client.https.protocol=TLSv1.2
druid.client.https.trustStoreType=jks
druid.client.https.trustStorePath=/path/to/file/truststore.jks # replace with correct trustStore file
druid.client.https.trustStorePassword=udjc@1  # replace with your own password```