This message was deleted.

A shard spec of -inf to +inf is equivalent to no secondary partitioning at all. This probably means that given your time partitioning, there aren't enough rows to split each time chunk into multiple segment files. Perhaps you would benefit from coarser segment granularity. What are you PARTITIONED BY ?

Hi Sergio, The table is partitioned by day. The record count on weekdays is ~1.2MM++. On weekends it falls to around 300K. Overall the table has around 200MM++ records YTD. Thanks, AR.

Try month, is you are consistently filtering on PART_COL, it should benefit significantly on the query side. The time filter will select the appropriate month(s) segments and the PART_COL should prune them further. The segments will be bigger but the data is always time sorted within a segment, so filtering on time should not suffer. Overall you should end up reading less segments and that should help with performance and concurrency.

Hi Sergio, We load the data on a daily basis. Making the granularity monthly would complicate ingestion as we will have to merge with existing data for the month (MSQ doc states that REPLACE has to be used for segment pruning to work, also reindexing for any date will require merging). As we approach the latter part of the month, this will cause ingestion to slow down. Am I correct? There is only one segment per time chunk in the partitioned table and queries on this are running slower than on the unpartitioned table which has several segments per chunk (it is not compacted). So it looks like parallel reads of several segment files is faster than reading a larger single segment file. Looks to me like specifying a smaller value of "rowsPerSegment" in the MSQ query context would yield more files which can then be processed in parallel. Is this understanding correct? Regarding metrics, it is enabled on the brokers but we had to disable it on the historicals as it was clogging up the log file. We will have to develop some custom filter for it before enabling it again. Thanks, AR

Hi AR, How many processing threads on the historicals do you have? If you only have enough data for one segment per month, then yes reducing maxRows to force more segments to be created but still retaining range partitioning theoretically should do what you want it to do ... Let's say you reduced maxRows so that you end up with 30 segments per month. This is the same number of segments you would get per month if you used DAY segment granularity ... except in in your case these segments would be range partitioned, so any given query filter on a range key would look up only one (or a few) segment(s) (vs in the DAY granularity model you would still open up all 30 segments). Now if parallelism is what is going to give you performance, then possible DAY granularity would still be an advantage ... you would be opening up all 30 segments, but each segment would have only a small number of records satisfying your filter condition, so you might get more parallel performance this way. Performance and segment size varies widely by actual use case, so I suggest trying out both methods to see which one works better for you (and please report back so we can hear the results). Thanks. John

Like John says, there is a balance between parallelism for a single query where more segments can make it faster if you have enough processing threads to take advantage of the parallelism. Opening each segment has some overhead. As concurrency grows, it is also beneficial to limit the number of segments even if processing each one individually is slower. The reason for this is that it will use less threads to process the query and therefore have more threads available to process other concurrent queries.

Hi John, Sergio, We tested this last week and below are the results. Env: We used our UAT env for testing 5 historical nodes (4 - 16CPU/128GB & 1 - 8CPU/128GB) Middlemanagers are also running on the 4 hosts with 16CPU (8 slots per host). Few realtime ingestion jobs running along with frequent daily batch ingestion plus any other testing that is ongoing There are several tables on the cluster. Overall load is lower than PROD - especially query loads are much lower. We created 3 tables. TAB_UNPART -> Original table created using the "index-parallel" JSON spec. Has segment sizes varying from 50-800k rows (numbered partitions). Contains daily data for 8 months (202301 - 202308). Granularity (seg & query) is daily. TAB_PART -> Partitioned table created using the MSQ engine with default config (range partitioned on PART_COL). Has single segment for all time chunks (segment size 50K-1.4MM rows). Contains daily data for 8 months (202301 - 202308). Granularity (seg & query) is daily. TAB_PART_W_SEG -> Partitioned table created using the MSQ engine with specified segment rowsize config (range partitioned on PART_COL). We recreated this table with different segment rowsize - 500K, 300, 200K & 100K. Contains daily data for 2 months (202307 - 202308). Granularity (seg & query) is daily. We tested from the console using a "count(distinct HIGH_CARDINALITY_COL)" query for time interval ranging from 1 day to 2 months. Approximate count distinct and cache was disabled. Queries on TAB_PART_W_SEG were slightly faster or similar to TAB_PART. In all the tests, queries on TAB_UNPART were faster than the queries on the other 2 tables - sometimes taking half the time than the other 2. In the segments of TAB_PART_W_SEG, I see something like below in the shard spec: {"type":"range", "dimensions":["PART_COL"], "start": ["PART_COL_VAL", "0000000000121303"], "end": .... } The number after the partition col value seems to be some kind of a row counter for that partition. Is this right? The PARTITION_COL is low cardinality (6 or 7 unique values). There is skew in the data. Most of the rows falls within 2 buckets. We tested on different values of the PARTITION_COL and the results were similar. Is something wrong with our test? Why are queries on the random partitioned table running much faster than the neatly partitioned ones? We have several other datasets where we have decided to go with sub-partitioned tables rather than create separate tables for each subset of data to avoid table explosion. But if sub-partitioning isn't going to deliver a similar query performance then this is something we will have to relook at. Hoping someone can offer some advice on this. Ideally, we don't need range partitioning as we will always query for a specific value of PART_COL. But it seems like there is no way to specify the same in the MSQ engine config. Thanks, AR.

I think we would need more details to figure this out. The number of segments needed in each case will be a factor, individual segment scan time too. Broker processing time could also be significant given the high cardinality. Other workload on the data servers at the time can cause variability in your results if the data server processing threads are busy and some of your segment scans need to wait. Do you have metrics turned on? If you use sqlQueryId identify your queries with you can examine processing in more detail, some metrics that are useful: On Historicals • query/segment/time: min, avg and max of this in each case will tell help you compare efficiency of the segment processing in each case. if max is much higher than avg, it indicates segment imbalance. Where the biggest segment may be limiting the performance. • query/wait/time: this records how much time a segment scan operation waited before being scanned. Ideally zero, but likely not zero, high values indicate resource exhaustion. Look for differences across historicals, I see that one of your historical nodes is less powerful. It might be that given the distribution of segments to historicals, the ones that just happen to fall on the small historical make the query slower. On Broker: • query/node/time - how much time the request took to respond from each historical back to broker. Is there a slower historical? Is it because of the segment size imbalance of each query? • query/node/ttfb - time to first byte - also by historical, how long the processing took without considering the time it takes to stream all the results back. Some historicals may return more data than others, so this measures actual processing time and not data transmission time. • query/segments/count - total number of segments needed for the query • query/time - overall wall clock time taken to process the query, if you subtract the max(query/node/time) from this you'll get the amount of time spent on the broker to finalize the query. This could be where the bottleneck lies, if there is high cardinality in the distinct count column, it may be that one partitioning scheme produces many more results than another from the historicals, leaving more work to the broker. For example, if you range partition on the distinct count column, this would make each individual segment scan return less values to the broker. One more thought: If the distinct count is the target, why not use an approximation? Seems like a good case for it. Is the exact distinct count relevant? Given the low cardinality of the partitioning column and its skew, I think hash partitioning is creating very unbalanced segments, but maybe less segments to read. The metrics for the query on each partitioning strategy should provide more insight. I gravitate toward less segments to process for each query because it keeps the processing thread pool usage lower which helps significantly when concurrency grows. Using approximations also keeps the broker processing significantly lower since the rows returned by the historicals will be aggregated with a theta sketch instead of the individual values.

If your goal is to product the fastest query for select count(distinct HIGH_CARD_COL), then I would think to try setting segmentGranularity to month, and range partition on HIGH_CARD_COL only. This will shard the dictionary for the HIGH_CARD_COL across many segments for the month, eliminating a ton of dictionary duplication across segments ... and the distinct counts can be computed locally in "perfect rollup" manner. The distinct lists will still have to be sent up to the Broker for final merge, but this might involved less processing if there is no overlap between lists. If you try this I am very curious to hear the results. Thanks. John

Hi John, Sergio, Metrics are enabled on the broker but not on the historical. I will try to run the tests again and extract the metrics. Slack access is blocked at my workplace hence the delay in my responses. Thanks, AR

Hi AR, do you have access to the segment scan metrics? Avg and Max scan times and # segments scanned?

Hi John, Are these the same things that Sergio also mentioned - query/segment/time & query/wait/time ? I am working on enabling the historical metrics and extracting these. Will share them as soon as i get them. Thanks, AR.

Oops, sorry, yes, those are the same as what Sergio was referring to .. I missed your last comment on not having set that up yet. Those are the key metrics to help understand if your segments are sized and configured well for the queries you are running.

Hi John, Sergio, I enabled the metrics on the historical. It looks like both the metrics are logged for every segment that is read. For 1 month of data, that is a lot of segments. Do we need every value from all historicals for analysis? Regards, AR

We usually look at avg and 98%ile for one minute intervals, so yes you can do aggregates but try to keep the intervals as short as possible. FYI good avg scan times are in the. 10-20ms and under range, 98% under 100-200ms.

Most of the data is at daily intervals. So the smallest interval would be 1 day. I will try to pull the data for 1 day and extract the stats. Regards, AR.

For the metrics? Metrics should be emitted in a very granular timeframe ...

Hi John, I did not get it. The "query/segment/time" & "query/wait/time" metrics are emitted only when a query is executed, correct? I will execute the query for an interval of 1 day and extract the avg, min, max & 98% from each historical for the query. I will try to repeat the test for longer query intervals (1 week, 1 month). Will this help with the analysis? Or is something additional needed? Regards, AR.

Oh, I see ... you are only executing one query at a time. Yes that is fine then, with the Historical stats emitting you can gather info on how many segments were scanned, and get the min, max, avg, and hopefully 98%ile scan times. Also if you have streaming ingestion running, then please separate out the metrics by "Node Type" (Peon = real-time, vs Historical)

Hi John, Sergio, Below are the metrics for 1 month interval query. I did not include the metrics for 1 day or 1 week queries as the difference in query time b/w the tables is not high. Looks to me like we are being limited by disk I/O. Or is it insufficient memory allocated to the historicals? If I didn't mention it earlier, the cluster is running on VMs and the local disk used for the segment cache on the historical nodes is not a SSD. Historicals have 128 GB of RAM. Middle Managers are also running on these hosts (except the 8 CPU host). Historical process heap: 8GB Historical process DM: 34GB Historical process NumThreads/MergeBuffers: 15 On the historical nodes: Free memory: 50 - 60 GB Buffered/Cache: 20 - 32 GB Swap size: 15 GB on all 16 CPU hosts, 2 GB on the one 8 CPU host. Utilization is 0. Another thing I noticed is that the brokers prefer the same set of historicals. They don't change their preference even when the target table is different. Why is this so? Please let me know if anything else is obvious from the metrics. Thanks, AR.

Theoretically ... If the data is grouped better and you are focusing on a particular slice of data, it is possible that the data you are querying on is concentrated into a few segments, which skews the query performance of the Historicals. Since in most casess the Broker's query response time is dependent on when the last Historical sends its data up, it will have to wait for the slowest Historical. In a Dynamically partitioned set of segments the data you are looking for is likely scattered more evenly across the segments, so the historicals will return a more uniform, average response time, which might be significantly better than the "slowest" Historical in a skewed scenario. Does this seem to explain what you are experiencing?

Hi John, The query used for testing covers the full data in the interval specified in the query. The query is a count distinct grouping on the partition column. SELECT __time, <PART_COL>, COUNT(DISTINCT <HIGH_CARD_COL>) FROM <TAB> WHERE TIME_IN_INTERVAL(__time, '2023-07-01/2023-08-01') GROUP BY 1, 2 Basically this is running slow if the segments are well partitioned (range or hash) by the <PART_COL>. I have also tried the query by adding a filter on the <PART_COL> and that is also slower than the dynamically partitioned table. This is what initiated our testing in the first place. So I don't think the skew in the data is reason for the query slowness. Thanks, AR.

Sergio may have more knowledge of the behavior here. My theory was that the PART_COL with high cardality underneath it is all concentrated in one segment, so that segment takes a lot longer to process. One way you should be able to check this is to compare the Avg and 98% Historical scan times ... for the range/hash partitioned you should see 98% be much higher than Avg, and for Dynamic you should see the two closer together.

Hi John, I don't have the AVG times handy (only took the min, max & 98%). 😞 But what does it mean in terms of query performance? Should we not partition the tables? Instead, should we go with separate tables for each partition and keep them dynamically partitioned? Thanks, AR

Each partitioning scheme has its own strengths ... I see clusters with lots of datasources left with Dynamic partitioning ... it is much faster to compact, in some cases you don't have to compact at all ... and dynamic also tends to provide more even distribution of work for some queries. Likewise range partitioning has its strengths, but it is not "universally better" than dynamic ... it can help a lot in specific situations to speed up queries and/or reduce segment sizes. If you stick with dynamic, then one thing you can also look at is how you are creating segments up front ... if you can create the right number and size of segments initially for your queries to perform well, then you may not have to run compaction at all.

Hi John, Range partitioning creates larger segments per time chunk (just one in most cases) and reading through this large segment in a single thread seems to be slower. It looks like either hash or dynamic partitioning with target segment rows of ~300K seems to be a better option. In dynamic partitioning, will the data in the segments still be sorted by the <PART_COL> if its is the first column in the data? Does this hold true for segments created using the MSQ query as well (data sorted by order of columns in the SELECT statement)? Thanks, AR.

In dynamic partitioning I do not know that the rows are explicitly sorted on any column other than __time ... so if your queryGranularity is high and is heavily truncating the __time values, then the rows may all be unsorted on any of the remaining columns. I don't know if they are clustered though through any form of default hashing that might be applied. (@Sergio Ferragut?) Regardless of the partitioning strategy though, the dictionaries for all of the string dimension columns will always be sorted for quick binary lookup.

My understanding is that all rows within a segment are sorted by __time and by all the other dimensions in the order that the dimensions are specified.

Thanks Sergio, John. Looks like we will have to play around with the segment sizes here since there isn't any clear direction we can take.

So, with Range partitioning you are getting one bad range? One that is bigger than the rest? I think that is a bug that has been addressed. I’ll try to get back to this and get you a link on that PR.

Hi Sergio, There is only one partition with range partitioning (using the default segment rows) but the queries are much slower than dynamic or hash partitioned tables. Summary below: Dynamic partitioned table: 2-6 segment files per time chunk in random sizes. Range partitioned table (default segment rows): 1 segment file per time chunk Hash partitioned table (300k segment rows): 2-4 segment files per time chunk in varied sizes but better than dynamic partitioned table. Query performance (Time taken): Dynamic partitioned < Hash partitioned < Range partitioned I have shared the metrics from broker and historical nodes previously. Thanks, AR.

Sorry, trying to follow too many threads. I'm assuming you are filtering on the partitioning column and that parallelism is helping the dynamic case. Given your summary, perhaps one more test would make sense. Range partitioned with smaller rows per segment using the average of # of rows you are seeing in the dynamic case for the maxRowsPerSegment setting. My thought is that this will result in the same level of parallelism that the dynamic segments is providing but also provide better pruning and individual segment scan times that are similar to the dynamic case. I believe this would still help as concurrency grows because less segments will need to be scanned for each query.

Hi Sergio, I was thinking on the same lines. Range partitioning with ~300k segment row size. This should increase the parallelism while also providing some segment pruning where applicable. I will create a table with this config and revert with the metrics for comparison. Really appreciate your & John's help on this. Thanks, AR.

I am a bit surprised that Hashed gives better performance than Dynamic for the same number of segments. But again this is highly dependent on the parallelism (i.e. number of processing threads you have on the Historicals) and the time intervals being queried on. If you are using streaming ingestion, one additional thing you could focus on is creating the right number and size of segments up front, by varying the number of ingestion tasks, taskDuration and maxRowsPerSegment, to try to generate a segment structure that doesn't require compaction. I've done that before successfully, but only for specific use cases that were friendly to that configuration.

Hi John, Hashed does not give a better performance than dynamic. It matches or comes close to it. It is way better than range partitioned. Thanks, AR.

Since you mentioned that the column has low cardinality and most of the data is concentrated in a couple of entries itself, perhaps the underlying issue is something else 🤔 . What are the number of segments generated with the different partitioning schemes?

Hi Laksh, Dynamic partitioned table - 5 segments per time chunk in most cases. Could be because 5 subtasks were used during ingestion. Range partitioned table - Depends on rows per segment. With default value (3MM) only one segment file per time chunk. With 300K rows per segment, varies b/w 1 - 6 segments per time chunk. Hash partitioned table - Varies b/w 1 - 4 segments per time chunk. 3 in most cases. Regards, AR.