Enable auto minor compaction

[yuanlihan]

Description

Note that @jihoonson introduced the minor compaction feature to Druid at version 0.16.x, which helps to effectively compact small segments. As we know, the cluster may suffer from the too many small segments issue if there are big pipelines that have a long trickle of late data.

But after upgrading to version 0.16.1, I found that the auto minor compaction was not implemented then. But still I rushed to the feature by making an adaptation based on the implementation of minor compaction and it works well till now. Thanks @jihoonson and all others involved with this greate feature.

Nowadays, we decide to upgrade to newer version, but find that the auto minor compaction is still not available. May I know if @jihoonson or anyone else WIP about this feature now? And if not, maybe I can have a try to raise a PR about the adaptation.

PS: there was a detail discuss #8489 between @jihoonson and @himanshug about CompactionTask and the auto compaction.

Motivation

• It will be more convenient to issue minor compaction tasks if enabling auto minor compaction.

[jihoonson]

Hi @yuanlihan, thank you for trying out minor compaction!

The thing is that you can enable auto minor compaction now by setting forceTimeChunkLock = false in the tuningConfig of coordinator compaction configuration. However, with auto compaction, you can use only either minor compaction or major compaction for now. To support both major and minor compactions, CompactSegments should be improved to be able to run both of them.

Also, please note that the segment lock which is used in minor compaction is still experimental and there are a couple of known issues such as #9571.

Probably I could work on fixing those bugs with segment lock and improving auto compaction to support both major and minor compactions at some point this year, but am unsure when it could be yet.

[yuanlihan]

Thanks @jihoonson for the explanation.

Now technically, the auto minor compaction is available by setting forceTimeChunkLock = false in the context property of both coordinator compaction configuration and the supervisor spec.(using version 0.17.x)
But it seems that currently it does not work perfectly in my case. The minor compaction tasks issued by coordinator always grab all segments in the given time interval, which seems not minor enough to fully leverage the power of minor compaction.

Here is a typical arrangement of segments within an hour in my case:

click to expand list

segmentId	size
ds_seg_id	129 KB
ds_seg_id_1	132 KB
ds_seg_id_2	192 KB
ds_seg_id_3	160 KB
ds_seg_id_4	136 KB
ds_seg_id_5	168 KB
ds_seg_id_6	144 KB
ds_seg_id_7	103 KB
ds_seg_id_8	188 KB
ds_seg_id_9	164 KB
ds_seg_id_10	103 KB
ds_seg_id_11	173 KB
ds_seg_id_12	190 KB
ds_seg_id_13	119 KB
ds_seg_id_14	164 KB
ds_seg_id_15	99 KB
ds_seg_id_16	188 KB
ds_seg_id_17	430 MB
ds_seg_id_18	437 MB
ds_seg_id_19	438 MB
ds_seg_id_20	445 MB
ds_seg_id_21	452 MB
ds_seg_id_22	453 MB
ds_seg_id_23	466 MB
ds_seg_id_24	474 MB
ds_seg_id_25	472 MB
ds_seg_id_26	472 MB
ds_seg_id_27	481 MB
ds_seg_id_28	481 MB
ds_seg_id_29	486 MB
ds_seg_id_30	486 MB
ds_seg_id_31	486 MB
ds_seg_id_32	485 MB
ds_seg_id_33	523 MB
ds_seg_id_34	522 MB
ds_seg_id_35	522 MB
ds_seg_id_36	523 MB
ds_seg_id_37	521 MB
ds_seg_id_38	516 MB
ds_seg_id_39	515 MB
ds_seg_id_40	508 MB
ds_seg_id_41	509 MB
ds_seg_id_42	508 MB
ds_seg_id_43	503 MB
ds_seg_id_44	502 MB
ds_seg_id_45	494 MB
ds_seg_id_46	494 MB
ds_seg_id_47	482 MB
ds_seg_id_48	481 MB
ds_seg_id_49	30 MB
ds_seg_id_50	23 MB
ds_seg_id_51	23 MB
ds_seg_id_52	16 MB
ds_seg_id_53	8 MB
ds_seg_id_54	19 MB
ds_seg_id_55	18 MB
ds_seg_id_56	18 MB
ds_seg_id_57	18 MB
ds_seg_id_58	18 MB
ds_seg_id_59	18 MB
ds_seg_id_60	165 KB
ds_seg_id_61	1 MB
ds_seg_id_62	798 KB
ds_seg_id_63	806 KB
ds_seg_id_64	17 MB
ds_seg_id_65	2 MB
ds_seg_id_66	2 MB
ds_seg_id_67	2 MB
ds_seg_id_68	2 MB
ds_seg_id_69	1 MB
ds_seg_id_70	17 MB
ds_seg_id_71	17 MB
ds_seg_id_72	17 MB
ds_seg_id_73	402 KB
ds_seg_id_74	17 MB
ds_seg_id_75	17 MB
ds_seg_id_76	17 MB
ds_seg_id_77	17 MB
ds_seg_id_78	17 MB
ds_seg_id_79	17 MB
ds_seg_id_80	3 MB
ds_seg_id_81	2 MB
ds_seg_id_82	2 MB
ds_seg_id_83	2 MB
ds_seg_id_84	2 MB
ds_seg_id_85	2 MB
ds_seg_id_86	1 MB
ds_seg_id_87	1 MB
ds_seg_id_88	1 MB
ds_seg_id_89	1 MB
ds_seg_id_90	1 MB
ds_seg_id_91	1 MB
ds_seg_id_92	623 KB
ds_seg_id_93	615 KB
ds_seg_id_94	621 KB
ds_seg_id_95	1 MB
ds_seg_id_96	1 MB
ds_seg_id_97	1 MB
ds_seg_id_98	1 MB
ds_seg_id_99	1 MB
ds_seg_id_100	262 KB
ds_seg_id_101	1 MB
ds_seg_id_102	1 MB
ds_seg_id_103	1 MB
ds_seg_id_104	1 MB
ds_seg_id_105	1 MB
ds_seg_id_106	1 MB
ds_seg_id_107	1 MB
ds_seg_id_108	1 MB
ds_seg_id_109	1 MB
ds_seg_id_110	1 MB
ds_seg_id_111	1 MB
ds_seg_id_112	291 KB
ds_seg_id_113	260 KB
ds_seg_id_114	268 KB
ds_seg_id_115	227 KB
ds_seg_id_116	200 KB
ds_seg_id_117	146 KB
ds_seg_id_118	117 KB
ds_seg_id_119	100 KB
ds_seg_id_120	87 KB
ds_seg_id_121	84 KB
ds_seg_id_122	223 KB
ds_seg_id_123	211 KB
ds_seg_id_124	212 KB
ds_seg_id_125	226 KB
ds_seg_id_126	235 KB
ds_seg_id_127	230 KB
ds_seg_id_128	238 KB
ds_seg_id_129	23 KB
ds_seg_id_130	23 KB
ds_seg_id_131	240 KB
ds_seg_id_132	20 KB
ds_seg_id_133	244 KB
ds_seg_id_134	22 KB
ds_seg_id_135	254 KB
ds_seg_id_136	20 KB
ds_seg_id_137	23 KB
ds_seg_id_138	237 KB
ds_seg_id_139	238 KB
ds_seg_id_140	240 KB
ds_seg_id_141	248 KB
ds_seg_id_142	242 KB
ds_seg_id_143	245 KB
ds_seg_id_144	60 KB
ds_seg_id_145	58 KB
ds_seg_id_146	59 KB
ds_seg_id_147	53 KB
ds_seg_id_148	47 KB
ds_seg_id_149	32 KB
ds_seg_id_150	26 KB
ds_seg_id_151	17 KB
ds_seg_id_152	109 KB
ds_seg_id_153	113 KB
ds_seg_id_154	110 KB
ds_seg_id_155	115 KB
ds_seg_id_156	121 KB
ds_seg_id_157	124 KB
ds_seg_id_158	122 KB
ds_seg_id_159	8 KB
ds_seg_id_160	6 KB
ds_seg_id_161	123 KB
ds_seg_id_162	8 KB
ds_seg_id_163	9 KB
ds_seg_id_164	123 KB
ds_seg_id_165	121 KB
ds_seg_id_166	9 KB
ds_seg_id_167	12 KB
ds_seg_id_168	12 KB
ds_seg_id_169	11 KB
ds_seg_id_170	124 KB
ds_seg_id_171	123 KB
ds_seg_id_172	120 KB
ds_seg_id_173	124 KB
ds_seg_id_174	121 KB
ds_seg_id_175	121 KB
ds_seg_id_176	33 KB
ds_seg_id_177	22 KB
ds_seg_id_178	24 KB
ds_seg_id_179	23 KB
ds_seg_id_180	17 KB
ds_seg_id_181	86 KB
ds_seg_id_182	95 KB
ds_seg_id_183	90 KB
ds_seg_id_184	89 KB
ds_seg_id_185	84 KB
ds_seg_id_186	88 KB
ds_seg_id_187	88 KB
ds_seg_id_188	90 KB
ds_seg_id_189	87 KB
ds_seg_id_190	84 KB
ds_seg_id_191	90 KB
ds_seg_id_192	87 KB
ds_seg_id_193	81 KB
ds_seg_id_194	85 KB
ds_seg_id_195	85 KB
ds_seg_id_196	80 KB
ds_seg_id_197	44 KB
ds_seg_id_198	46 KB
ds_seg_id_199	48 KB
ds_seg_id_200	45 KB
ds_seg_id_201	44 KB
ds_seg_id_202	42 KB
ds_seg_id_203	47 KB
ds_seg_id_204	46 KB
ds_seg_id_205	44 KB
ds_seg_id_206	45 KB
ds_seg_id_207	45 KB
ds_seg_id_208	44 KB
ds_seg_id_209	40 KB
ds_seg_id_210	43 KB
ds_seg_id_211	41 KB
ds_seg_id_212	39 KB
ds_seg_id_213	32 KB
ds_seg_id_214	31 KB
ds_seg_id_215	30 KB
ds_seg_id_216	31 KB
ds_seg_id_217	29 KB
ds_seg_id_218	30 KB
ds_seg_id_219	27 KB
ds_seg_id_220	8 KB
ds_seg_id_221	7 KB
ds_seg_id_222	28 KB
ds_seg_id_223	7 KB
ds_seg_id_224	6 KB
ds_seg_id_225	6 KB
ds_seg_id_226	26 KB
ds_seg_id_227	27 KB
ds_seg_id_228	7 KB
ds_seg_id_229	7 KB
ds_seg_id_230	27 KB
ds_seg_id_231	27 KB
ds_seg_id_232	25 KB
ds_seg_id_233	27 KB
ds_seg_id_234	27 KB
ds_seg_id_235	27 KB
ds_seg_id_236	7 KB
ds_seg_id_237	5 KB
ds_seg_id_238	27 KB
ds_seg_id_239	27 KB
ds_seg_id_240	20 KB
ds_seg_id_241	17 KB
ds_seg_id_242	24 KB
ds_seg_id_243	23 KB
ds_seg_id_244	27 KB
ds_seg_id_245	22 KB
ds_seg_id_246	17 KB
ds_seg_id_247	18 KB
ds_seg_id_248	7 KB
ds_seg_id_249	22 KB
ds_seg_id_250	19 KB
ds_seg_id_251	18 KB
ds_seg_id_252	26 KB
ds_seg_id_253	27 KB
ds_seg_id_254	24 KB
ds_seg_id_255	13 KB
ds_seg_id_256	16 KB
ds_seg_id_257	14 KB
ds_seg_id_258	13 KB
ds_seg_id_259	11 KB
ds_seg_id_260	14 KB
ds_seg_id_261	7 KB
ds_seg_id_262	7 KB
ds_seg_id_263	7 KB
ds_seg_id_264	8 KB
ds_seg_id_265	7 KB
ds_seg_id_266	8 KB
ds_seg_id_267	8 KB
ds_seg_id_268	8 KB
ds_seg_id_269	8 KB
ds_seg_id_270	15 KB
ds_seg_id_271	7 KB
ds_seg_id_272	15 KB
ds_seg_id_273	15 KB
ds_seg_id_274	19 KB
ds_seg_id_275	17 KB
ds_seg_id_276	17 KB
ds_seg_id_277	17 KB
ds_seg_id_278	18 KB
ds_seg_id_279	17 KB
ds_seg_id_280	17 KB
ds_seg_id_281	19 KB
ds_seg_id_282	19 KB
ds_seg_id_283	16 KB
ds_seg_id_284	19 KB
ds_seg_id_285	17 KB
ds_seg_id_286	18 KB
ds_seg_id_287	12 KB
ds_seg_id_288	11 KB
ds_seg_id_289	10 KB
ds_seg_id_290	11 KB
ds_seg_id_291	9 KB
ds_seg_id_292	10 KB
ds_seg_id_293	9 KB
ds_seg_id_294	8 KB
ds_seg_id_295	8 KB
ds_seg_id_296	8 KB
ds_seg_id_297	8 KB
ds_seg_id_298	8 KB
ds_seg_id_299	8 KB
ds_seg_id_300	8 KB
ds_seg_id_301	8 KB
ds_seg_id_302	8 KB
ds_seg_id_303	12 KB
ds_seg_id_304	14 KB
ds_seg_id_305	13 KB
ds_seg_id_306	12 KB
ds_seg_id_307	15 KB
ds_seg_id_308	13 KB
ds_seg_id_309	15 KB
ds_seg_id_310	16 KB
ds_seg_id_311	15 KB
ds_seg_id_312	13 KB
ds_seg_id_313	20 KB
ds_seg_id_314	13 KB
ds_seg_id_315	17 KB
ds_seg_id_316	17 KB
ds_seg_id_317	16 KB
ds_seg_id_318	19 KB
ds_seg_id_319	14 KB
ds_seg_id_320	14 KB
ds_seg_id_321	14 KB
ds_seg_id_322	13 KB
ds_seg_id_323	13 KB
ds_seg_id_324	13 KB
ds_seg_id_325	12 KB
ds_seg_id_326	10 KB
ds_seg_id_327	11 KB
ds_seg_id_328	11 KB
ds_seg_id_329	11 KB
ds_seg_id_330	11 KB
ds_seg_id_331	9 KB
ds_seg_id_332	11 KB
ds_seg_id_333	10 KB
ds_seg_id_334	7 KB
ds_seg_id_335	10 KB
ds_seg_id_336	7 KB
ds_seg_id_337	6 KB
ds_seg_id_338	12 KB
ds_seg_id_339	12 KB
ds_seg_id_340	10 KB
ds_seg_id_341	13 KB
ds_seg_id_342	12 KB
ds_seg_id_343	13 KB
ds_seg_id_344	13 KB
ds_seg_id_345	12 KB
ds_seg_id_346	13 KB
ds_seg_id_347	12 KB
ds_seg_id_348	11 KB
ds_seg_id_349	12 KB
ds_seg_id_350	11 KB
ds_seg_id_351	9 KB
ds_seg_id_352	9 KB
ds_seg_id_353	9 KB
ds_seg_id_354	8 KB
ds_seg_id_355	9 KB
ds_seg_id_356	8 KB
ds_seg_id_357	11 KB
ds_seg_id_358	13 KB
ds_seg_id_359	13 KB
ds_seg_id_360	12 KB
ds_seg_id_361	8 KB
ds_seg_id_362	7 KB
ds_seg_id_363	6 KB
ds_seg_id_364	8 KB
ds_seg_id_365	6 KB
ds_seg_id_366	7 KB
ds_seg_id_367	8 KB
ds_seg_id_368	10 KB
ds_seg_id_369	13 KB
ds_seg_id_370	11 KB

It would be better to only do minor compaction for the first M small segments and the tail N small segments.
And in my opinion, the base implementation of minor compaction is able to compact a subset of adjacent small segments. So would it better to have a fine grained policy to choose segments for minor compaction task? eg: possibly by adding an implementation of CompactionSegmentSearchPolicy for minor compaction tasks.

Also found that minor compaction tasks will always fail if the partitionId is not consecutive.

WARN [TaskQueue-Manager] org.apache.druid.indexing.overlord.TaskQueue - Exception thrown during isReady for task: compact_ds_name_pfjceoge_2020-04-20T03:25:14.299Z
org.apache.druid.java.util.common.ISE: Can't compact segments of non-consecutive rootPartition range

[jihoonson]

@yuanlihan your assessment is very correct! Yes, the coordinator should be able to run both minor compaction and major compactions; minor compaction for recent data and major compaction for old data. As you mentioned, minor compaction should be able to run on a subset of segments in a time chunk instead of grabbing all of them.

It would be better to only do minor compaction for the first M small segments and the tail N small segments.

This sounds nice, but I'm not sure how we can do it. Auto compaction algorithm used to use segment size as a trigger for compaction, but this caused a bunch of bugs since the segment size after compaction can be still small based on your configuration such as maxRowsPerSegment. Also parallel task will create at least one small segment in most cases since the last task will be likely assigned small number of segments. As a result, we changed the algorithm to be stateful in #8573. Do you have a good idea?

Also found that minor compaction tasks will always fail if the partitionId is not consecutive.

WARN [TaskQueue-Manager] org.apache.druid.indexing.overlord.TaskQueue - Exception thrown during isReady for task: compact_ds_name_pfjceoge_2020-04-20T03:25:14.299Z
org.apache.druid.java.util.common.ISE: Can't compact segments of non-consecutive rootPartition range

Oh yeah, this is a known issue. Just opened #9768.

[yuanlihan]

Hi, @jihoonson

I like the stateful algorithm for auto compaction.

This sounds nice, but I'm not sure how we can do it. Auto compaction algorithm used to use segment size as a trigger for compaction, but this caused a bunch of bugs since the segment size after compaction can be still small based on your configuration such as maxRowsPerSegment. Also parallel task will create at least one small segment in most cases since the last task will be likely assigned small number of segments. As a result, we changed the algorithm to be stateful in #8573. Do you have a good idea?

I would like to introduce two extra properties to DataSourceCompactionConfig:

Property	Description	Default
skipSegmentWithSizeBytesGreaterThan	filter out segments with size greater than this value. Specify this property if you want to use minor compaction.	null
minNumSegmentsToCompact	Minimum number of segments to trigger compaction. Suggest set this value to >=2 if you want to use minor compaction.	1

First filter out big segments from candidates(SegmentsToCompact) then find a subset of consecutive segments and then create a minor compaction task with segment type inputSpec. In the meanwhile, it relieves the impact of non-consecutive segments.
Since the type of partitionSpec of minor compaction task should be dynamic only(is it?), we can adjust maxRowsPerSegment if segments created by minor compaction tasks is still too small.

[jihoonson]

@yuanlihan thanks for the suggestion. I agree that the auto compaction should be able to skip compaction for segments which are already in a good size. Maybe my previous concern is not a problem anymore if we check both the segment size and whether the segment is already compacted. For example we can skip compaction even for small segments if they are created by compaction.

However, the first property skipSegmentWithSizeBytesGreaterThan seems possible to introduce a couple of issues. For example, minor compaction couldn't compact segments if their size shows a pattern of (small segment, large segment, small segment, large segment, ...) when large segments are greater than skipSegmentWithSizeBytesGreaterThan. Another issue is that the compaction can be used for splitting big segments as well as merging small segments. This might not be practically an issue since minor compaction would be used mostly with streaming ingestion which usually creates small segments, but seems nice if we can still support this case. Maybe we can add targetRowsPerSegment, so that auto compaction can skip if a segment has a similar number of rows to it.

For the second property, is there a use case where you don't want to compact segments using minor compaction? Or can we always compact if there are 2 or more segments?

Since the type of partitionSpec of minor compaction task should be dynamic only(is it?), we can adjust maxRowsPerSegment if segments created by minor compaction tasks is still too small.

This is true for now, but I think it should support all partitionsSpec types. Hash and range partitioning will be a primary use case when they are supported. But there are also needs for supporting maxRowsPerSegment for hash and range partitioning to handle data skew.

[yuanlihan]

@jihoonson

However, the first property skipSegmentWithSizeBytesGreaterThan seems possible to introduce a couple of issues. For example, minor compaction couldn't compact segments if their size shows a pattern of (small segment, large segment, small segment, large segment, ...) when large segments are greater than skipSegmentWithSizeBytesGreaterThan. Another issue is that the compaction can be used for splitting big segments as well as merging small segments.

In these cases, I would like to increase skipSegmentWithSizeBytesGreaterThan a bit, perhaps 1.2 * skipSegmentWithSizeBytesGreaterThan ≈ inputSegmentSizeBytes, and then maybe we can filter the big segment or the pair of segments(a big and a small) for further minor compaction with the specific partitionSpec which may be a spec for splitting. But still not sure if it is feasible😂.

Maybe we can add targetRowsPerSegment, so that auto compaction can skip if a segment has a similar number of rows to it.

Personally, I prefer specifying by size in bytes per segment because it may be more obvious for users.

For the second property, is there a use case where you don't want to compact segments using minor compaction? Or can we always compact if there are 2 or more segments?

No special use case here. In my opinion, it's acceptable to omit the occasional small segments mixed in segments with regular size, especially when using minor compaction. This property may make it more flexible for users to configure. And I agree that we should consider the use case of splitting a big segment which may caused by data skew.

[github-actions]

This issue has been marked as stale due to 280 days of inactivity.
It will be closed in 4 weeks if no further activity occurs. If this issue is still
relevant, please simply write any comment. Even if closed, you can still revive the
issue at any time or discuss it on the [email protected] list.
Thank you for your contributions.

[github-actions]

This issue has been closed due to lack of activity. If you think that
is incorrect, or the issue requires additional review, you can revive the issue at
any time.