You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "cecemei (via GitHub)" <gi...@apache.org> on 2026/01/30 08:44:47 UTC
[PR] wip (druid)
cecemei opened a new pull request, #18968:
URL: https://github.com/apache/druid/pull/18968
<!-- Thanks for trying to help us make Apache Druid be the best it can be! Please fill out as much of the following information as is possible (where relevant, and remove it when irrelevant) to help make the intention and scope of this PR clear in order to ease review. -->
<!-- Please read the doc for contribution (https://github.com/apache/druid/blob/master/CONTRIBUTING.md) before making this PR. Also, once you open a PR, please _avoid using force pushes and rebasing_ since these make it difficult for reviewers to see what you've changed in response to their reviews. See [the 'If your pull request shows conflicts with master' section](https://github.com/apache/druid/blob/master/CONTRIBUTING.md#if-your-pull-request-shows-conflicts-with-master) for more details. -->
Fixes #XXXX.
<!-- Replace XXXX with the id of the issue fixed in this PR. Remove this section if there is no corresponding issue. Don't reference the issue in the title of this pull-request. -->
<!-- If you are a committer, follow the PR action item checklist for committers:
https://github.com/apache/druid/blob/master/dev/committer-instructions.md#pr-and-issue-action-item-checklist-for-committers. -->
### Description
<!-- Describe the goal of this PR, what problem are you fixing. If there is a corresponding issue (referenced above), it's not necessary to repeat the description here, however, you may choose to keep one summary sentence. -->
<!-- Describe your patch: what did you change in code? How did you fix the problem? -->
<!-- If there are several relatively logically separate changes in this PR, create a mini-section for each of them. For example: -->
#### Fixed the bug ...
#### Renamed the class ...
#### Added a forbidden-apis entry ...
<!--
In each section, please describe design decisions made, including:
- Choice of algorithms
- Behavioral aspects. What configuration values are acceptable? How are corner cases and error conditions handled, such as when there are insufficient resources?
- Class organization and design (how the logic is split between classes, inheritance, composition, design patterns)
- Method organization and design (how the logic is split between methods, parameters and return types)
- Naming (class, method, API, configuration, HTTP endpoint, names of emitted metrics)
-->
<!-- It's good to describe an alternative design (or mention an alternative name) for every design (or naming) decision point and compare the alternatives with the designs that you've implemented (or the names you've chosen) to highlight the advantages of the chosen designs and names. -->
<!-- If there was a discussion of the design of the feature implemented in this PR elsewhere (e. g. a "Proposal" issue, any other issue, or a thread in the development mailing list), link to that discussion from this PR description and explain what have changed in your final design compared to your original proposal or the consensus version in the end of the discussion. If something hasn't changed since the original discussion, you can omit a detailed discussion of those aspects of the design here, perhaps apart from brief mentioning for the sake of readability of this PR description. -->
<!-- Some of the aspects mentioned above may be omitted for simple and small changes. -->
#### Release note
<!-- Give your best effort to summarize your changes in a couple of sentences aimed toward Druid users.
If your change doesn't have end user impact, you can skip this section.
For tips about how to write a good release note, see [Release notes](https://github.com/apache/druid/blob/master/CONTRIBUTING.md#release-notes).
-->
<hr>
##### Key changed/added classes in this PR
* `MyFoo`
* `OurBar`
* `TheirBaz`
<hr>
<!-- Check the items by putting "x" in the brackets for the done things. Not all of these items apply to every PR. Remove the items which are not done or not relevant to the PR. None of the items from the checklist below are strictly necessary, but it would be very helpful if you at least self-review the PR. -->
This PR has:
- [ ] been self-reviewed.
- [ ] using the [concurrency checklist](https://github.com/apache/druid/blob/master/dev/code-review/concurrency.md) (Remove this item if the PR doesn't have any relation to concurrency.)
- [ ] added documentation for new or modified features or behaviors.
- [ ] a release note entry in the PR description.
- [ ] added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
- [ ] added or updated version, license, or notice information in [licenses.yaml](https://github.com/apache/druid/blob/master/dev/license.md)
- [ ] added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
- [ ] added unit tests or modified existing tests to cover new code paths, ensuring the threshold for [code coverage](https://github.com/apache/druid/blob/master/dev/code-review/code-coverage.md) is met.
- [ ] added integration tests.
- [ ] been tested in a test Druid cluster.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] introduce clear separation between proposed compaction, eligibility status, and compaction mode, along with improved task state tracking (druid)
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #18968: introduce clear separation between proposed compaction, eligibility status, and compaction mode, along with improved task state tracking
URL: https://github.com/apache/druid/pull/18968
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] introduce clear separation between proposed compaction, eligibility status, and compaction mode, along with improved task state tracking (druid)
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #18968:
URL: https://github.com/apache/druid/pull/18968#discussion_r2851645978
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionStatusTracker.java:
##########
@@ -80,43 +81,51 @@
* This method assumes that the given candidate is eligible for compaction
* based on the current compaction config/supervisor of the datasource.
*/
- public CompactionStatus computeCompactionStatus(
- CompactionCandidate candidate,
- CompactionCandidateSearchPolicy searchPolicy
- )
+ @Nullable
+ public TaskState computeCompactionTaskState(CompactionCandidate candidate)
{
// Skip intervals that already have a running task
final CompactionTaskStatus lastTaskStatus = getLatestTaskStatus(candidate);
if (lastTaskStatus != null && lastTaskStatus.getState() == TaskState.RUNNING) {
- return CompactionStatus.running("Task for interval is already running");
+ return TaskState.RUNNING;
}
// Skip intervals that have been recently compacted if segment timeline is not updated yet
final DateTime snapshotTime = segmentSnapshotTime.get();
if (lastTaskStatus != null
&& lastTaskStatus.getState() == TaskState.SUCCESS
&& snapshotTime != null && snapshotTime.isBefore(lastTaskStatus.getUpdatedTime())) {
- return CompactionStatus.skipped(
- "Segment timeline not updated since last compaction task succeeded"
- );
+ return TaskState.SUCCESS;
}
- // Skip intervals that have been filtered out by the policy
- final CompactionCandidateSearchPolicy.Eligibility eligibility
- = searchPolicy.checkEligibilityForCompaction(candidate, lastTaskStatus);
- if (eligibility.isEligible()) {
- return CompactionStatus.pending("Not compacted yet");
- } else {
- return CompactionStatus.skipped("Rejected by search policy: %s", eligibility.getReason());
- }
+ // if task has never been run or has failed recently, run the compaction candidate.
+ return null;
}
/**
* Tracks the latest compaction status of the given compaction candidates.
* Used only by the {@link CompactionRunSimulator}.
*/
- public void onCompactionStatusComputed(
- CompactionCandidate candidateSegments,
+ public void onSkippedCandidate(
+ CompactionCandidateAndStatus candidateSegments,
+ DataSourceCompactionConfig config,
Review Comment:
## Useless parameter
The parameter 'config' is never used.
[Show more details](https://github.com/apache/druid/security/code-scanning/10833)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] introduce clear separation between proposed compaction, eligibility status, and compaction mode, along with improved task state tracking (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18968:
URL: https://github.com/apache/druid/pull/18968#discussion_r2850861573
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionCandidate.java:
##########
@@ -37,76 +39,200 @@
*/
public class CompactionCandidate
{
- private final List<DataSegment> segments;
- private final Interval umbrellaInterval;
- private final Interval compactionInterval;
- private final String dataSource;
- private final long totalBytes;
- private final int numIntervals;
-
- private final CompactionStatus currentStatus;
-
- public static CompactionCandidate from(
- List<DataSegment> segments,
- @Nullable Granularity targetSegmentGranularity
- )
+ /**
+ * Non-empty list of segments of a datasource being proposed for compaction.
+ * A proposed compaction typically contains all the segments of a single time chunk.
+ */
+ public static class ProposedCompaction
Review Comment:
> don't we need the uncompacted & compacted stats in CompactionStatus to sort CompactionCandidate?
We do, that's why we shouldn't be touching the `CompactionStatus` class at all.
The `Eligibility` is its own thing. It does not need be put inside any other class.
In `CompactionJobQueue`, we will pick a `CompactionCandidate` from the queue and then ask the search policy on whether that candidate is eligible or not. Based on the eligibility, we will launch the job (info of job type i.e incremental or not) will also be included inside the eligibility object returned by the policy.
Similarly, `CompactSegments` can also check the eligibility of a candidate.
So, effectively, we are moving the call `searchPolicy.checkEligibility()` from `CompactionStatusTracker` to `CompactSegments` and `CompactionJobQueue`. We need to do this because we are getting rid of the method `CompactionStatusTracker.computeCompactionStatus()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Add compaction eligibility concept and incremental compaction support (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18968:
URL: https://github.com/apache/druid/pull/18968#discussion_r2762849696
##########
server/src/main/java/org/apache/druid/server/compaction/MostFragmentedIntervalFirstPolicy.java:
##########
@@ -185,8 +215,18 @@ public Eligibility checkEligibilityForCompaction(
"Average size[%,d] of uncompacted segments in interval must be at most [%,d]",
avgSegmentSize, maxAverageUncompactedBytesPerSegment.getBytes()
);
+ }
+
+ final double uncompactedBytesRatio = (double) uncompacted.getTotalBytes() /
+ (uncompacted.getTotalBytes() + candidate.getCompactedStats().getTotalBytes());
+ if (uncompactedBytesRatio < incrementalCompactionUncompactedBytesRatioThreshold) {
+ return Eligibility.incrementalCompaction(
Review Comment:
Let's not return `incrementalCompaction` here until incremental compaction is actually implemented.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] introduce clear separation between proposed compaction, eligibility status, and compaction mode, along with improved task state tracking (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18968:
URL: https://github.com/apache/druid/pull/18968#discussion_r2795235216
##########
server/src/main/java/org/apache/druid/server/compaction/DataSourceCompactibleSegmentIterator.java:
##########
@@ -121,9 +125,11 @@ private void populateQueue(SegmentTimeline timeline, List<Interval> skipInterval
if (!partialEternitySegments.isEmpty()) {
// Do not use the target segment granularity in the CompactionCandidate
// as Granularities.getIterable() will cause OOM due to the above issue
- CompactionCandidate candidatesWithStatus = CompactionCandidate
- .from(partialEternitySegments, null)
- .withCurrentStatus(CompactionStatus.skipped("Segments have partial-eternity intervals"));
+ CompactionCandidate candidatesWithStatus =
Review Comment:
This class probably doesn't need to change in this PR.
The `Eligibility` should be computed and used only by the `CompactionJobQueue`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] introduce clear separation between proposed compaction, eligibility status, and compaction mode, along with improved task state tracking (druid)
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #18968:
URL: https://github.com/apache/druid/pull/18968#issuecomment-4414082499
This pull request has been marked as stale due to 60 days of inactivity.
It will be closed in 4 weeks if no further activity occurs. If you think
that's incorrect or this pull request should instead be reviewed, please simply
write any comment. Even if closed, you can still revive the PR at any time or
discuss it on the dev@druid.apache.org list.
Thank you for your contributions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Add compaction eligibility concept and incremental compaction support (druid)
Posted by "cecemei (via GitHub)" <gi...@apache.org>.
cecemei commented on code in PR #18968:
URL: https://github.com/apache/druid/pull/18968#discussion_r2765199112
##########
server/src/main/java/org/apache/druid/server/compaction/DataSourceCompactibleSegmentIterator.java:
##########
@@ -329,17 +331,17 @@ private void findAndEnqueueSegmentsToCompact(CompactibleSegmentIterator compacti
continue;
}
- final CompactionCandidate candidates = CompactionCandidate.from(segments, config.getSegmentGranularity());
- final CompactionStatus compactionStatus = CompactionStatus.compute(candidates, config, fingerprintMapper);
- final CompactionCandidate candidatesWithStatus = candidates.withCurrentStatus(compactionStatus);
+ final CompactionCandidate candidatesWithStatus =
Review Comment:
why should it be jobQueue only? both jobQueue and not as a coordinator duty?
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionCandidateSearchPolicy.java:
##########
@@ -64,18 +62,27 @@ Eligibility checkEligibilityForCompaction(
*/
class Eligibility
{
- public static final Eligibility OK = new Eligibility(true, null);
+ public enum PolicyEligibility
Review Comment:
this is done
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionCandidateSearchPolicy.java:
##########
@@ -64,18 +62,27 @@ Eligibility checkEligibilityForCompaction(
*/
class Eligibility
{
- public static final Eligibility OK = new Eligibility(true, null);
+ public enum PolicyEligibility
+ {
+ FULL_COMPACTION,
+ INCREMENTAL_COMPACTION,
+ NOT_ELIGIBLE,
+ NOT_APPLICABLE
Review Comment:
they are tracked differently in `AutoCompactionSnapshot`, my original implementation only has three states and it failed: https://github.com/apache/druid/actions/runs/21509820801/job/61973864065. eventually i'd want evaluate() to only return eligibility and not status, so complete/pending status should have different eligibility
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionCandidateSearchPolicy.java:
##########
@@ -64,18 +62,27 @@ Eligibility checkEligibilityForCompaction(
*/
class Eligibility
Review Comment:
this is done
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionCandidate.java:
##########
@@ -160,21 +169,122 @@ public CompactionStatus getCurrentStatus()
return currentStatus;
}
+ @Nullable
+ public CompactionCandidateSearchPolicy.Eligibility getPolicyEligibility()
+ {
+ return policyEligiblity;
+ }
+
/**
* Creates a copy of this CompactionCandidate object with the given status.
*/
public CompactionCandidate withCurrentStatus(CompactionStatus status)
{
- return new CompactionCandidate(segments, umbrellaInterval, compactionInterval, numIntervals, status);
+ return new CompactionCandidate(
+ segments,
+ umbrellaInterval,
+ compactionInterval,
+ numIntervals,
+ policyEligiblity,
+ status
+ );
+ }
+
+ public CompactionCandidate withPolicyEligibility(CompactionCandidateSearchPolicy.Eligibility eligibility)
+ {
+ return new CompactionCandidate(
+ segments,
+ umbrellaInterval,
+ compactionInterval,
+ numIntervals,
+ eligibility,
+ currentStatus
+ );
+ }
+
+ /**
+ * Evaluates this candidate for compaction eligibility based on the provided
+ * compaction configuration and search policy.
+ * <p>
+ * This method first evaluates the candidate against the compaction configuration
+ * using a {@link CompactionStatus.Evaluator} to determine if any segments need
+ * compaction. If segments are pending compaction, the search policy is consulted
+ * to determine the type of compaction:
+ * <ul>
+ * <li><b>NOT_ELIGIBLE</b>: Returns a candidate with status SKIPPED, indicating
+ * the policy decided compaction should not occur at this time</li>
+ * <li><b>FULL_COMPACTION</b>: Returns this candidate with status PENDING,
+ * indicating all segments should be compacted</li>
+ * <li><b>INCREMENTAL_COMPACTION</b>: Returns a new candidate containing only
+ * the uncompacted segments (as determined by the evaluator), with status
+ * PENDING for incremental compaction</li>
+ * </ul>
+ *
+ * @param config the compaction configuration for the datasource
+ * @param searchPolicy the policy used to determine compaction eligibility
+ * @return a CompactionCandidate with updated status and potentially filtered segments
+ */
+ public CompactionCandidate evaluate(
Review Comment:
synced offline, DataSourceCompactibleSegmentIterator would be responsible for filtering out ineligible candidates
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Add compaction eligibility concept and incremental compaction support (druid)
Posted by "cecemei (via GitHub)" <gi...@apache.org>.
cecemei commented on code in PR #18968:
URL: https://github.com/apache/druid/pull/18968#discussion_r2771999278
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionStatusTracker.java:
##########
@@ -80,10 +80,7 @@ public Set<String> getSubmittedTaskIds()
* This method assumes that the given candidate is eligible for compaction
* based on the current compaction config/supervisor of the datasource.
*/
- public CompactionStatus computeCompactionStatus(
- CompactionCandidate candidate,
- CompactionCandidateSearchPolicy searchPolicy
- )
+ public CompactionStatus computeCompactionStatus(CompactionCandidate candidate)
Review Comment:
not sure i understand, dont we still track pending/running/completed status?
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionEligibility.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.compaction;
+
+import org.apache.druid.java.util.common.StringUtils;
+
+import java.util.Objects;
+
+/**
+ * Describes the eligibility of an interval for compaction.
+ */
+public class CompactionEligibility
Review Comment:
added a new evaluate static method
##########
server/src/main/java/org/apache/druid/client/indexing/ClientCompactionIntervalSpec.java:
##########
@@ -38,18 +42,37 @@ public class ClientCompactionIntervalSpec
private final Interval interval;
@Nullable
+ private final List<SegmentDescriptor> uncompactedSegments;
+ @Nullable
private final String sha256OfSortedSegmentIds;
@JsonCreator
public ClientCompactionIntervalSpec(
@JsonProperty("interval") Interval interval,
+ @JsonProperty("uncompactedSegments") @Nullable List<SegmentDescriptor> uncompactedSegments,
Review Comment:
SpecificSegmentsSpec is not supported by msq, and it somewhat felt a bit deprecated to me, maybe because of the segment lock stuff. i kinda like to specify an interval since it gives some certainty, also wonder maybe we should check non-null for interval here? i didnt see any instance with null interval.
##########
server/src/main/java/org/apache/druid/server/compaction/MostFragmentedIntervalFirstPolicy.java:
##########
@@ -47,13 +47,16 @@ public class MostFragmentedIntervalFirstPolicy extends BaseCandidateSearchPolicy
private final int minUncompactedCount;
private final HumanReadableBytes minUncompactedBytes;
private final HumanReadableBytes maxAverageUncompactedBytesPerSegment;
+ private final double incrementalCompactionUncompactedBytesRatioThreshold;
@JsonCreator
public MostFragmentedIntervalFirstPolicy(
@JsonProperty("minUncompactedCount") @Nullable Integer minUncompactedCount,
@JsonProperty("minUncompactedBytes") @Nullable HumanReadableBytes minUncompactedBytes,
@JsonProperty("maxAverageUncompactedBytesPerSegment") @Nullable
HumanReadableBytes maxAverageUncompactedBytesPerSegment,
+ @JsonProperty("incrementalCompactionUncompactedBytesRatioThreshold") @Nullable
Review Comment:
there's another pr that'd almost ready, i could also move this incremental part to a separate pr if you feel strong about this need to be separate from eligibility refactor.
##########
server/src/main/java/org/apache/druid/server/compaction/DataSourceCompactibleSegmentIterator.java:
##########
@@ -329,17 +331,17 @@ private void findAndEnqueueSegmentsToCompact(CompactibleSegmentIterator compacti
continue;
}
- final CompactionCandidate candidates = CompactionCandidate.from(segments, config.getSegmentGranularity());
- final CompactionStatus compactionStatus = CompactionStatus.compute(candidates, config, fingerprintMapper);
- final CompactionCandidate candidatesWithStatus = candidates.withCurrentStatus(compactionStatus);
+ final CompactionCandidate candidatesWithStatus =
Review Comment:
hmm still thinking about this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Add compaction eligibility concept and incremental compaction support (druid)
Posted by "capistrant (via GitHub)" <gi...@apache.org>.
capistrant commented on code in PR #18968:
URL: https://github.com/apache/druid/pull/18968#discussion_r2774684817
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionEligibility.java:
##########
@@ -0,0 +1,932 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.compaction;
+
+import com.google.common.base.Strings;
+import org.apache.commons.lang3.ArrayUtils;
+import org.apache.druid.client.indexing.ClientCompactionTaskQueryTuningConfig;
+import org.apache.druid.common.config.Configs;
+import org.apache.druid.data.input.impl.DimensionSchema;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.error.InvalidInput;
+import org.apache.druid.indexer.partitions.DimensionRangePartitionsSpec;
+import org.apache.druid.indexer.partitions.DynamicPartitionsSpec;
+import org.apache.druid.indexer.partitions.HashedPartitionsSpec;
+import org.apache.druid.indexer.partitions.PartitionsSpec;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.granularity.Granularity;
+import org.apache.druid.java.util.common.granularity.GranularityType;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.query.aggregation.AggregatorFactory;
+import org.apache.druid.segment.IndexSpec;
+import org.apache.druid.segment.metadata.IndexingStateFingerprintMapper;
+import org.apache.druid.segment.transform.CompactionTransformSpec;
+import org.apache.druid.server.coordinator.DataSourceCompactionConfig;
+import org.apache.druid.server.coordinator.UserCompactionTaskGranularityConfig;
+import org.apache.druid.timeline.CompactionState;
+import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.utils.CollectionUtils;
+import org.joda.time.Interval;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/**
+ * Describes the eligibility of an interval for compaction.
+ */
+public class CompactionEligibility
+{
+ public enum State
+ {
+ FULL_COMPACTION,
+ INCREMENTAL_COMPACTION,
+ NOT_ELIGIBLE,
+ NOT_APPLICABLE
+ }
+
+ static class CompactionEligibilityBuilder
+ {
+ private State state;
+ private CompactionStatistics compacted;
+ private CompactionStatistics uncompacted;
+ private List<DataSegment> uncompactedSegments;
+ private String reason;
+
+ CompactionEligibilityBuilder(State state, String reason)
+ {
+ this.state = state;
+ this.reason = reason;
+ }
+
+ CompactionEligibilityBuilder compacted(CompactionStatistics compacted)
+ {
+ this.compacted = compacted;
+ return this;
+ }
+
+ CompactionEligibilityBuilder uncompacted(CompactionStatistics uncompacted)
+ {
+ this.uncompacted = uncompacted;
+ return this;
+ }
+
+ CompactionEligibilityBuilder uncompactedSegments(List<DataSegment> uncompactedSegments)
+ {
+ this.uncompactedSegments = uncompactedSegments;
+ return this;
+ }
+
+ CompactionEligibility build()
+ {
+ return new CompactionEligibility(state, reason, compacted, uncompacted, uncompactedSegments);
+ }
+ }
+
+ public static final CompactionEligibility NOT_APPLICABLE = builder(State.NOT_APPLICABLE, "").build();
+
+ public static CompactionEligibility fail(String messageFormat, Object... args)
+ {
+ return builder(State.NOT_ELIGIBLE, StringUtils.format(messageFormat, args)).build();
+ }
+
+ private final State state;
+ private final String reason;
+
+ @Nullable
+ private final CompactionStatistics compacted;
+ @Nullable
+ private final CompactionStatistics uncompacted;
+ @Nullable
+ private final List<DataSegment> uncompactedSegments;
+
+ private CompactionEligibility(
+ State state,
+ String reason,
+ @Nullable CompactionStatistics compacted,
+ @Nullable CompactionStatistics uncompacted,
+ @Nullable List<DataSegment> uncompactedSegments
+ )
+ {
+ this.state = state;
+ this.reason = reason;
+ switch (state) {
+ case NOT_APPLICABLE:
+ break;
+ case NOT_ELIGIBLE:
+ InvalidInput.conditionalException(!Strings.isNullOrEmpty(reason), "must provide a reason");
+ break;
+ case FULL_COMPACTION:
+ case INCREMENTAL_COMPACTION:
+ InvalidInput.conditionalException(compacted != null, "must provide compacted stats");
+ InvalidInput.conditionalException(uncompacted != null, "must provide uncompacted stats");
+ InvalidInput.conditionalException(uncompactedSegments != null, "must provide uncompactedSegments");
+ break;
+ default:
+ throw DruidException.defensive("unexpected eligibility state[%s]", state);
+ }
+ this.compacted = compacted;
+ this.uncompacted = uncompacted;
+ this.uncompactedSegments = uncompactedSegments;
+ }
+
+ static CompactionEligibilityBuilder builder(State state, String reason)
+ {
+ return new CompactionEligibilityBuilder(state, reason);
+ }
+
+ public State getState()
+ {
+ return state;
+ }
+
+ public String getReason()
+ {
+ return reason;
+ }
+
+ @Nullable
+ public CompactionStatistics getUncompactedStats()
+ {
+ return uncompacted;
+ }
+
+ @Nullable
+ public CompactionStatistics getCompactedStats()
+ {
+ return compacted;
+ }
+
+ @Nullable
+ public List<DataSegment> getUncompactedSegments()
+ {
+ return uncompactedSegments;
+ }
+
+ public CompactionCandidate createCandidate(CompactionCandidate.ProposedCompaction proposedCompaction)
+ {
+ switch (state) {
+ case NOT_APPLICABLE:
+ return new CompactionCandidate(proposedCompaction, this, CompactionStatus.COMPLETE);
+ case NOT_ELIGIBLE:
+ return new CompactionCandidate(proposedCompaction, this, CompactionStatus.skipped(reason));
+ case FULL_COMPACTION:
+ return new CompactionCandidate(
+ proposedCompaction,
+ this,
+ CompactionStatus.pending(reason)
+ );
+ case INCREMENTAL_COMPACTION:
+ CompactionCandidate.ProposedCompaction newProposed = new CompactionCandidate.ProposedCompaction(
+ uncompactedSegments,
+ proposedCompaction.getUmbrellaInterval(),
+ proposedCompaction.getCompactionInterval(),
+ Math.toIntExact(uncompactedSegments.stream().map(DataSegment::getInterval).distinct().count())
+ );
+ return new CompactionCandidate(newProposed, this, CompactionStatus.pending(reason));
+ default:
+ throw DruidException.defensive("Unexpected eligibility state[%s]", state);
+ }
+ }
+
+ /**
+ * Evaluates a compaction candidate to determine its eligibility and compaction status.
+ * <p>
+ * This method performs a two-stage evaluation:
+ * <ol>
+ * <li>First, uses {@link Evaluator} to check if the candidate needs compaction
Review Comment:
did we make any changes to the Evaluator logic in moving it to this file or is it just lift and place in new location?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Add compaction eligibility concept and incremental compaction support (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18968:
URL: https://github.com/apache/druid/pull/18968#discussion_r2767229688
##########
server/src/main/java/org/apache/druid/server/coordinator/duty/CompactSegments.java:
##########
@@ -366,6 +364,10 @@ public static ClientCompactionTaskQuery createCompactionTask(
}
final CompactionEngine compactionEngine = config.getEngine() == null ? defaultEngine : config.getEngine();
+ if (CompactionEngine.MSQ.equals(compactionEngine) && !Boolean.TRUE.equals(dropExisting)) {
Review Comment:
Seems like this should be included in #18950 instead.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Add compaction eligibility concept and incremental compaction support (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on PR #18968:
URL: https://github.com/apache/druid/pull/18968#issuecomment-3861841202
Had a brief discussion with @cecemei on this. I think there has been some confusion regarding the roles of `CompactionStatus`, `CompactionEligibility` and the task status. It is in part due to the fact that `CompactionStatus` was originally serving all of these roles but not perfectly.
In short, we shouldn't need to make any major modification to any class in this PR except `CompactionEligibility`.
I have shared some details offline. We should be able to get this resolved by Monday.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Add compaction eligibility concept and incremental compaction support (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18968:
URL: https://github.com/apache/druid/pull/18968#discussion_r2772499291
##########
server/src/main/java/org/apache/druid/client/indexing/ClientCompactionIntervalSpec.java:
##########
@@ -38,18 +42,37 @@ public class ClientCompactionIntervalSpec
private final Interval interval;
@Nullable
+ private final List<SegmentDescriptor> uncompactedSegments;
+ @Nullable
private final String sha256OfSortedSegmentIds;
@JsonCreator
public ClientCompactionIntervalSpec(
@JsonProperty("interval") Interval interval,
+ @JsonProperty("uncompactedSegments") @Nullable List<SegmentDescriptor> uncompactedSegments,
Review Comment:
> i kinda like to specify an interval since it gives some certainty
Yes, I agree, in the new scheme of things, it is always nice to specify the interval.
That's why I advise adding a new implementation of `CompactionInputSpec` which will be used only for incremental compaction so that it is easy to distinguish the two on the task side as well. We need not piggyback on the existing `CompactionIntervalSpec`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] introduce clear separation between proposed compaction, eligibility status, and compaction mode, along with improved task state tracking (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18968:
URL: https://github.com/apache/druid/pull/18968#discussion_r2795265220
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionStatus.java:
##########
@@ -346,8 +361,10 @@ static DimensionRangePartitionsSpec getEffectiveRangePartitionsSpec(DimensionRan
*/
private static class Evaluator
Review Comment:
Yeah, I don't like the null returns much myself.
We should update each CHECK to return a `CheckResult` object.
The rest of this class should probably remain unchanged as it seems to be bloating up the PR without much value added.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] introduce clear separation between proposed compaction, eligibility status, and compaction mode, along with improved task state tracking (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18968:
URL: https://github.com/apache/druid/pull/18968#discussion_r2850846762
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionCandidateSearchPolicy.java:
##########
@@ -48,74 +45,11 @@ public interface CompactionCandidateSearchPolicy
int compareCandidates(CompactionCandidate candidateA, CompactionCandidate candidateB);
/**
- * Checks if the given {@link CompactionCandidate} is eligible for compaction
- * in the current iteration. A policy may implement this method to skip
- * compacting intervals or segments that do not fulfil some required criteria.
+ * Creates a {@link CompactionCandidate} after applying policy-specific checks to the proposed compaction candidate.
*
- * @return {@link Eligibility#OK} only if eligible.
- */
- Eligibility checkEligibilityForCompaction(
- CompactionCandidate candidate,
- CompactionTaskStatus latestTaskStatus
- );
-
- /**
- * Describes the eligibility of an interval for compaction.
+ * @param candidate the proposed compaction
+ * @param eligibility initial eligibility from compaction config checks
+ * @return final compaction candidate
*/
- class Eligibility
- {
- public static final Eligibility OK = new Eligibility(true, null);
-
- private final boolean eligible;
- private final String reason;
-
- private Eligibility(boolean eligible, String reason)
- {
- this.eligible = eligible;
- this.reason = reason;
- }
-
- public boolean isEligible()
- {
- return eligible;
- }
-
- public String getReason()
- {
- return reason;
- }
-
- public static Eligibility fail(String messageFormat, Object... args)
- {
- return new Eligibility(false, StringUtils.format(messageFormat, args));
- }
-
- @Override
- public boolean equals(Object object)
- {
- if (this == object) {
- return true;
- }
- if (object == null || getClass() != object.getClass()) {
- return false;
- }
- Eligibility that = (Eligibility) object;
- return eligible == that.eligible && Objects.equals(reason, that.reason);
- }
-
- @Override
- public int hashCode()
- {
- return Objects.hash(eligible, reason);
- }
-
- @Override
- public String toString()
- {
- return "Eligibility{" +
- "eligible=" + eligible +
- ", reason='" + reason + '\'' +
- '}';
- }
- }
+ CompactionCandidate createCandidate(CompactionCandidate.ProposedCompaction candidate, CompactionStatus eligibility);
Review Comment:
No we should continue to use the existing method in `CompactionCandidateSearchPolicy` i.e. `checkEligibility()`.
Just add a new enum inside the `Eligibility` class.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] introduce clear separation between proposed compaction, eligibility status, and compaction mode, along with improved task state tracking (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18968:
URL: https://github.com/apache/druid/pull/18968#discussion_r2851961212
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionStatusTracker.java:
##########
@@ -80,43 +80,48 @@ public Set<String> getSubmittedTaskIds()
* This method assumes that the given candidate is eligible for compaction
* based on the current compaction config/supervisor of the datasource.
*/
- public CompactionStatus computeCompactionStatus(
Review Comment:
No, we were snapshot time earlier when we didn't have compactedStats and uncompactedStats.
Now, since we have the stats, we can already make better decisions in the policy itself.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] introduce clear separation between proposed compaction, eligibility status, and compaction mode, along with improved task state tracking (druid)
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #18968:
URL: https://github.com/apache/druid/pull/18968#issuecomment-4640966789
This pull request/issue has been closed due to lack of activity. If you think that
is incorrect, or the pull request requires review, you can revive the PR at any time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Add compaction eligibility concept and incremental compaction support (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18968:
URL: https://github.com/apache/druid/pull/18968#discussion_r2762688447
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionCandidate.java:
##########
@@ -160,21 +169,122 @@ public CompactionStatus getCurrentStatus()
return currentStatus;
}
+ @Nullable
+ public CompactionCandidateSearchPolicy.Eligibility getPolicyEligibility()
+ {
+ return policyEligiblity;
+ }
+
/**
* Creates a copy of this CompactionCandidate object with the given status.
*/
public CompactionCandidate withCurrentStatus(CompactionStatus status)
{
- return new CompactionCandidate(segments, umbrellaInterval, compactionInterval, numIntervals, status);
+ return new CompactionCandidate(
+ segments,
+ umbrellaInterval,
+ compactionInterval,
+ numIntervals,
+ policyEligiblity,
+ status
+ );
+ }
+
+ public CompactionCandidate withPolicyEligibility(CompactionCandidateSearchPolicy.Eligibility eligibility)
+ {
+ return new CompactionCandidate(
+ segments,
+ umbrellaInterval,
+ compactionInterval,
+ numIntervals,
+ eligibility,
+ currentStatus
+ );
+ }
+
+ /**
+ * Evaluates this candidate for compaction eligibility based on the provided
+ * compaction configuration and search policy.
+ * <p>
+ * This method first evaluates the candidate against the compaction configuration
+ * using a {@link CompactionStatus.Evaluator} to determine if any segments need
+ * compaction. If segments are pending compaction, the search policy is consulted
+ * to determine the type of compaction:
+ * <ul>
+ * <li><b>NOT_ELIGIBLE</b>: Returns a candidate with status SKIPPED, indicating
+ * the policy decided compaction should not occur at this time</li>
+ * <li><b>FULL_COMPACTION</b>: Returns this candidate with status PENDING,
+ * indicating all segments should be compacted</li>
+ * <li><b>INCREMENTAL_COMPACTION</b>: Returns a new candidate containing only
+ * the uncompacted segments (as determined by the evaluator), with status
+ * PENDING for incremental compaction</li>
+ * </ul>
+ *
+ * @param config the compaction configuration for the datasource
+ * @param searchPolicy the policy used to determine compaction eligibility
+ * @return a CompactionCandidate with updated status and potentially filtered segments
+ */
+ public CompactionCandidate evaluate(
Review Comment:
This method should not live here. The computation of the eligiblity should be done in `CompactionStatusTracker`.
The overall workflow would be like this:
- identify a `CompactionCandidate`
- determine its `CompactionStatus`
- determine the eligiblity via `CompactionStatusTracker.computeEligibility()`
##########
server/src/main/java/org/apache/druid/client/indexing/ClientCompactionIntervalSpec.java:
##########
@@ -38,18 +42,37 @@ public class ClientCompactionIntervalSpec
private final Interval interval;
@Nullable
+ private final List<SegmentDescriptor> uncompactedSegments;
+ @Nullable
private final String sha256OfSortedSegmentIds;
@JsonCreator
public ClientCompactionIntervalSpec(
@JsonProperty("interval") Interval interval,
+ @JsonProperty("uncompactedSegments") @Nullable List<SegmentDescriptor> uncompactedSegments,
Review Comment:
It is a little confusing to add segments here since there is already a `SpecificSegmentsSpec` which specifies the set of segments that should be compacted.
I feel we should add a new type of `CompactionInputSpec` which will be used for incremental compaction only. It would have both a non-null interval as well as the set of segment IDs to compact incrementally.
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionCandidateSearchPolicy.java:
##########
@@ -64,18 +62,27 @@ Eligibility checkEligibilityForCompaction(
*/
class Eligibility
{
- public static final Eligibility OK = new Eligibility(true, null);
+ public enum PolicyEligibility
Review Comment:
Maybe we should call this enum `Status` (since this represents the eligibility status) to avoid confusion with `Eligibility` class itself.
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionCandidateSearchPolicy.java:
##########
@@ -64,18 +62,27 @@ Eligibility checkEligibilityForCompaction(
*/
class Eligibility
Review Comment:
Since this class is going to have a wider usage now, let's move it out into a separate file of its own.
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionStatus.java:
##########
@@ -357,29 +348,52 @@ private static class Evaluator
private final Map<CompactionState, List<DataSegment>> unknownStateToSegments = new HashMap<>();
@Nullable
- private final String targetFingerprint;
private final IndexingStateFingerprintMapper fingerprintMapper;
+ @Nullable
+ private final String targetFingerprint;
- private Evaluator(
+ Evaluator(
CompactionCandidate candidateSegments,
DataSourceCompactionConfig compactionConfig,
- @Nullable String targetFingerprint,
@Nullable IndexingStateFingerprintMapper fingerprintMapper
)
{
this.candidateSegments = candidateSegments;
this.compactionConfig = compactionConfig;
this.tuningConfig = ClientCompactionTaskQueryTuningConfig.from(compactionConfig);
this.configuredGranularitySpec = compactionConfig.getGranularitySpec();
- this.targetFingerprint = targetFingerprint;
this.fingerprintMapper = fingerprintMapper;
+ if (fingerprintMapper == null) {
+ targetFingerprint = null;
+ } else {
+ targetFingerprint = fingerprintMapper.generateFingerprint(
+ compactionConfig.getDataSource(),
+ compactionConfig.toCompactionState()
+ );
+ }
+ }
+
+ List<DataSegment> getUncompactedSegments()
+ {
+ return uncompactedSegments;
}
- private CompactionStatus evaluate()
+ /**
+ * Evaluates the compaction status of candidate segments through a multi-step process:
+ * <ol>
+ * <li>Validates input bytes are within limits</li>
+ * <li>Categorizes segments by compaction state (fingerprinted, uncompacted, or unknown)</li>
+ * <li>Performs fingerprint-based validation if available (fast path)</li>
+ * <li>Runs detailed checks against unknown states via {@link #CHECKS}</li>
+ * </ol>
+ *
+ * @return Pair of eligibility status and compaction status with reason for first failed check
+ */
+ Pair<CompactionCandidateSearchPolicy.Eligibility, CompactionStatus> evaluate()
{
- final CompactionStatus inputBytesCheck = inputBytesAreWithinLimit();
- if (inputBytesCheck.isSkipped()) {
- return inputBytesCheck;
+ final CompactionCandidateSearchPolicy.Eligibility inputBytesCheck = inputBytesAreWithinLimit();
+ if (inputBytesCheck != null) {
Review Comment:
I think the input bytes check should now be moved out of this class and used in `CompactionStatusTracker.computeEligibility()` instead.
##########
server/src/main/java/org/apache/druid/server/compaction/DataSourceCompactibleSegmentIterator.java:
##########
@@ -329,17 +331,17 @@ private void findAndEnqueueSegmentsToCompact(CompactibleSegmentIterator compacti
continue;
}
- final CompactionCandidate candidates = CompactionCandidate.from(segments, config.getSegmentGranularity());
- final CompactionStatus compactionStatus = CompactionStatus.compute(candidates, config, fingerprintMapper);
- final CompactionCandidate candidatesWithStatus = candidates.withCurrentStatus(compactionStatus);
+ final CompactionCandidate candidatesWithStatus =
Review Comment:
I feel this class should remain untouched since we are only computing the `CompactionStatus` here. The eligiblity computation should happen later.
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionCandidateSearchPolicy.java:
##########
@@ -64,18 +62,27 @@ Eligibility checkEligibilityForCompaction(
*/
class Eligibility
{
- public static final Eligibility OK = new Eligibility(true, null);
+ public enum PolicyEligibility
+ {
+ FULL_COMPACTION,
+ INCREMENTAL_COMPACTION,
+ NOT_ELIGIBLE,
+ NOT_APPLICABLE
Review Comment:
How is "not applicable" different from "not eligible"?
I think eligibility should have only 3 possible states - not eligible, incremental and full.
##########
server/src/main/java/org/apache/druid/server/compaction/MostFragmentedIntervalFirstPolicy.java:
##########
@@ -185,8 +215,18 @@ public Eligibility checkEligibilityForCompaction(
"Average size[%,d] of uncompacted segments in interval must be at most [%,d]",
avgSegmentSize, maxAverageUncompactedBytesPerSegment.getBytes()
);
+ }
+
+ final double uncompactedBytesRatio = (double) uncompacted.getTotalBytes() /
+ (uncompacted.getTotalBytes() + candidate.getCompactedStats().getTotalBytes());
+ if (uncompactedBytesRatio < incrementalCompactionUncompactedBytesRatioThreshold) {
+ return Eligibility.incrementalCompaction(
Review Comment:
Let's not add this config or return incremental compaction here since incremental compaction is not implemented yet. If a user configures this field by mistake, compaction might misbehave.
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionStatus.java:
##########
@@ -357,29 +348,52 @@ private static class Evaluator
private final Map<CompactionState, List<DataSegment>> unknownStateToSegments = new HashMap<>();
@Nullable
- private final String targetFingerprint;
private final IndexingStateFingerprintMapper fingerprintMapper;
+ @Nullable
+ private final String targetFingerprint;
- private Evaluator(
+ Evaluator(
CompactionCandidate candidateSegments,
DataSourceCompactionConfig compactionConfig,
- @Nullable String targetFingerprint,
@Nullable IndexingStateFingerprintMapper fingerprintMapper
)
{
this.candidateSegments = candidateSegments;
this.compactionConfig = compactionConfig;
this.tuningConfig = ClientCompactionTaskQueryTuningConfig.from(compactionConfig);
this.configuredGranularitySpec = compactionConfig.getGranularitySpec();
- this.targetFingerprint = targetFingerprint;
this.fingerprintMapper = fingerprintMapper;
+ if (fingerprintMapper == null) {
+ targetFingerprint = null;
+ } else {
+ targetFingerprint = fingerprintMapper.generateFingerprint(
+ compactionConfig.getDataSource(),
+ compactionConfig.toCompactionState()
+ );
+ }
+ }
+
+ List<DataSegment> getUncompactedSegments()
+ {
+ return uncompactedSegments;
}
- private CompactionStatus evaluate()
+ /**
+ * Evaluates the compaction status of candidate segments through a multi-step process:
+ * <ol>
+ * <li>Validates input bytes are within limits</li>
+ * <li>Categorizes segments by compaction state (fingerprinted, uncompacted, or unknown)</li>
+ * <li>Performs fingerprint-based validation if available (fast path)</li>
+ * <li>Runs detailed checks against unknown states via {@link #CHECKS}</li>
+ * </ol>
+ *
+ * @return Pair of eligibility status and compaction status with reason for first failed check
+ */
+ Pair<CompactionCandidateSearchPolicy.Eligibility, CompactionStatus> evaluate()
Review Comment:
Since the `CompactionStatus` and eligibility are going to be two separate things now, this class should not deal with eligibility at all. We can retain the old code in this class.
Once the `CompactionStatus` has been computed, that should be passed to `CompactionStatusTracker.computeEligiblity()` to determine the eligibility.
##########
server/src/main/java/org/apache/druid/server/compaction/MostFragmentedIntervalFirstPolicy.java:
##########
@@ -47,13 +47,16 @@ public class MostFragmentedIntervalFirstPolicy extends BaseCandidateSearchPolicy
private final int minUncompactedCount;
private final HumanReadableBytes minUncompactedBytes;
private final HumanReadableBytes maxAverageUncompactedBytesPerSegment;
+ private final double incrementalCompactionUncompactedBytesRatioThreshold;
@JsonCreator
public MostFragmentedIntervalFirstPolicy(
@JsonProperty("minUncompactedCount") @Nullable Integer minUncompactedCount,
@JsonProperty("minUncompactedBytes") @Nullable HumanReadableBytes minUncompactedBytes,
@JsonProperty("maxAverageUncompactedBytesPerSegment") @Nullable
HumanReadableBytes maxAverageUncompactedBytesPerSegment,
+ @JsonProperty("incrementalCompactionUncompactedBytesRatioThreshold") @Nullable
Review Comment:
Let's not add this user-facing config until incremental compaction has been implemented.
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionStatus.java:
##########
@@ -344,7 +335,7 @@ static DimensionRangePartitionsSpec getEffectiveRangePartitionsSpec(DimensionRan
* Evaluates {@link #CHECKS} to determine the compaction status of a
* {@link CompactionCandidate}.
*/
- private static class Evaluator
+ static class Evaluator
Review Comment:
Let's keep the evaluator private as before.
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionStatusTracker.java:
##########
@@ -80,10 +80,7 @@ public Set<String> getSubmittedTaskIds()
* This method assumes that the given candidate is eligible for compaction
* based on the current compaction config/supervisor of the datasource.
*/
- public CompactionStatus computeCompactionStatus(
- CompactionCandidate candidate,
- CompactionCandidateSearchPolicy searchPolicy
- )
+ public CompactionStatus computeCompactionStatus(CompactionCandidate candidate)
Review Comment:
Since we are now decoupling eligibility (i.e. whether or not to run compaction) from the current compaction status (i.e. current degree of compaction in the interval), it would make sense to do the following:
- Repurpose this method to compute the eligibility since the eligibility can be affected by the `lastTaskStatus` but the `CompactionStatus` should not depend on it.
- `CompactionStatus` need not have the `RUNNING` state anymore since it is really associated with the running state of a compaction task.
##########
server/src/main/java/org/apache/druid/server/coordinator/duty/CompactSegments.java:
##########
@@ -460,14 +462,27 @@ private static ClientCompactionTaskQuery compactSegments(
context.put("priority", compactionTaskPriority);
final String taskId = IdUtils.newTaskId(TASK_ID_PREFIX, ClientCompactionTaskQuery.TYPE, dataSource, null);
+ final ClientCompactionIntervalSpec clientCompactionIntervalSpec;
+ Preconditions.checkArgument(entry.getPolicyEligibility() != null, "Must have a policy eligibility");
+ switch (entry.getPolicyEligibility().getPolicyEligibility()) {
+ case FULL_COMPACTION:
+ clientCompactionIntervalSpec = new ClientCompactionIntervalSpec(entry.getCompactionInterval(), null, null);
+ break;
+ case INCREMENTAL_COMPACTION:
Review Comment:
Let's update this part only when incremental compaction has been implemented on the task side. Also, we should support incremental compaction with `CompactionJobQueue` (i.e. Overlord-based compaction) only.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] introduce clear separation between proposed compaction, eligibility status, and compaction mode, along with improved task state tracking (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18968:
URL: https://github.com/apache/druid/pull/18968#discussion_r2795122751
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionCandidate.java:
##########
@@ -37,76 +39,200 @@
*/
public class CompactionCandidate
{
- private final List<DataSegment> segments;
- private final Interval umbrellaInterval;
- private final Interval compactionInterval;
- private final String dataSource;
- private final long totalBytes;
- private final int numIntervals;
-
- private final CompactionStatus currentStatus;
-
- public static CompactionCandidate from(
- List<DataSegment> segments,
- @Nullable Granularity targetSegmentGranularity
- )
+ /**
+ * Non-empty list of segments of a datasource being proposed for compaction.
+ * A proposed compaction typically contains all the segments of a single time chunk.
+ */
+ public static class ProposedCompaction
Review Comment:
What we need is a way to distinguish between a candidate before and after the status has been computed, so that we can avoid the null checks on `currentStatus`. I don't think this change fully achieves that.
Also, the name `CompactionCandidate` captures the intent better as it defines a potential "candidate" for compaction. So let's stick with it.
Instead of this change, you may consider adding a new class `CompactionCandidateAndStatus` which will be used by both the methods of the search policy. The new class will contain only a `candidate` field and a `status` field.
I don't think we will need this class to contain the `Eligibility` or the `CompactionMode` since it will be computed and used only by the `CompactionJobQueue`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] introduce clear separation between proposed compaction, eligibility status, and compaction mode, along with improved task state tracking (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18968:
URL: https://github.com/apache/druid/pull/18968#discussion_r2795265220
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionStatus.java:
##########
@@ -346,8 +361,10 @@ static DimensionRangePartitionsSpec getEffectiveRangePartitionsSpec(DimensionRan
*/
private static class Evaluator
Review Comment:
Yeah, I don't like the null returns much myself.
We should update each CHECK to return a `CheckResult` object instead of a `CompactionStatus`.
The rest of this class should probably remain unchanged as it seems to be bloating up the PR without much value added.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] introduce clear separation between proposed compaction, eligibility status, and compaction mode, along with improved task state tracking (druid)
Posted by "cecemei (via GitHub)" <gi...@apache.org>.
cecemei commented on PR #18968:
URL: https://github.com/apache/druid/pull/18968#issuecomment-3866553773
Hi @kfaraz @capistrant , I made some major changes to this pr, and hopefully it's in a much better state, please take a look when you're available, thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Add compaction eligibility concept and incremental compaction support (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18968:
URL: https://github.com/apache/druid/pull/18968#discussion_r2762567010
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionCandidateSearchPolicy.java:
##########
@@ -64,18 +62,27 @@ Eligibility checkEligibilityForCompaction(
*/
class Eligibility
{
- public static final Eligibility OK = new Eligibility(true, null);
+ public enum PolicyEligibility
Review Comment:
Maybe we should call this enum `State` (since this represents the eligibility status) to avoid confusion with `Eligibility` class itself.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] introduce clear separation between proposed compaction, eligibility status, and compaction mode, along with improved task state tracking (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18968:
URL: https://github.com/apache/druid/pull/18968#discussion_r2795122751
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionCandidate.java:
##########
@@ -37,76 +39,200 @@
*/
public class CompactionCandidate
{
- private final List<DataSegment> segments;
- private final Interval umbrellaInterval;
- private final Interval compactionInterval;
- private final String dataSource;
- private final long totalBytes;
- private final int numIntervals;
-
- private final CompactionStatus currentStatus;
-
- public static CompactionCandidate from(
- List<DataSegment> segments,
- @Nullable Granularity targetSegmentGranularity
- )
+ /**
+ * Non-empty list of segments of a datasource being proposed for compaction.
+ * A proposed compaction typically contains all the segments of a single time chunk.
+ */
+ public static class ProposedCompaction
Review Comment:
What we need is a way to distinguish between a candidate before and after the status has been computed, so that we can avoid the null checks on `currentStatus`. I don't think this change fully achieves that.
Also, the name `CompactionCandidate` captures the intent better as it defines a potential "candidate" for compaction. So let's stick with it.
Instead of this change, you may consider adding a new class `CompactionCandidateAndStatus` which will be used by both the methods of the search policy. The new class will contain only a `candidate` field and a `status` field.
I don't think we will need this class to contain the `Eligibility` (or the `CompactionMode`) since it will be computed and used only by the `CompactionJobQueue`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Add compaction eligibility concept and incremental compaction support (druid)
Posted by "cecemei (via GitHub)" <gi...@apache.org>.
cecemei commented on PR #18968:
URL: https://github.com/apache/druid/pull/18968#issuecomment-3857643126
> Thanks for the PR, @cecemei !
>
> I like the idea of separating the `CompactionStatus` (i.e. the current degree of compaction of an interval) from the `Eligibility` (i.e. whether an interval should be picked for compaction or not).
>
> I have left some suggestions to aid with the separation. Let me know if they make sense.
I addressed most comments except for the coordinator based compaction and incremental compaction feature change.
The major change is i moved a lot of CompactionStatus stuff to CompactionEligibility, and update the CHECK to return String (null for previous CompactionStatus.COMPLETE, and non-null for previous CompactionStatus.pending...), it seems simpler since after all we're just looking for a reason to do compaction.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] introduce clear separation between proposed compaction, eligibility status, and compaction mode, along with improved task state tracking (druid)
Posted by "cecemei (via GitHub)" <gi...@apache.org>.
cecemei commented on code in PR #18968:
URL: https://github.com/apache/druid/pull/18968#discussion_r2851843520
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionStatus.java:
##########
@@ -346,8 +361,10 @@ static DimensionRangePartitionsSpec getEffectiveRangePartitionsSpec(DimensionRan
*/
private static class Evaluator
Review Comment:
added a new CheckResult class
##########
server/src/main/java/org/apache/druid/server/coordinator/duty/CompactSegments.java:
##########
@@ -371,8 +369,11 @@ public static ClientCompactionTaskQuery createCompactionTask(
}
final Map<String, Object> autoCompactionContext = newAutoCompactionContext(config.getTaskContext());
- if (candidate.getCurrentStatus() != null) {
- autoCompactionContext.put(COMPACTION_REASON_KEY, candidate.getCurrentStatus().getReason());
+ if (CompactionMode.NOT_APPLICABLE.equals(candidate.getMode())) {
Review Comment:
true updated
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionStatusTracker.java:
##########
@@ -80,43 +80,48 @@ public Set<String> getSubmittedTaskIds()
* This method assumes that the given candidate is eligible for compaction
* based on the current compaction config/supervisor of the datasource.
*/
- public CompactionStatus computeCompactionStatus(
Review Comment:
dont we still want to check `final DateTime snapshotTime = segmentSnapshotTime.get()`, and trigger compaction after new segment?
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionCandidate.java:
##########
@@ -37,76 +39,200 @@
*/
public class CompactionCandidate
{
- private final List<DataSegment> segments;
- private final Interval umbrellaInterval;
- private final Interval compactionInterval;
- private final String dataSource;
- private final long totalBytes;
- private final int numIntervals;
-
- private final CompactionStatus currentStatus;
-
- public static CompactionCandidate from(
- List<DataSegment> segments,
- @Nullable Granularity targetSegmentGranularity
- )
+ /**
+ * Non-empty list of segments of a datasource being proposed for compaction.
+ * A proposed compaction typically contains all the segments of a single time chunk.
+ */
+ public static class ProposedCompaction
Review Comment:
moved search policy check outside of data source iterator to job queue
##########
indexing-service/src/main/java/org/apache/druid/indexing/compact/CompactionJobQueue.java:
##########
@@ -282,18 +281,17 @@ private boolean startJobIfPendingAndReady(
}
// Check if the job is already running, completed or skipped
- final CompactionStatus compactionStatus = getCurrentStatusForJob(job, policy);
- switch (compactionStatus.getState()) {
- case RUNNING:
+ final CompactionCandidate.TaskState candidateState = getCurrentTaskStateForJob(job);
Review Comment:
updated to use option 1, and add queuedIntervals to CompactionSimulateResult
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] introduce clear separation between proposed compaction, eligibility status, and compaction mode, along with improved task state tracking (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18968:
URL: https://github.com/apache/druid/pull/18968#discussion_r2850843186
##########
indexing-service/src/main/java/org/apache/druid/indexing/compact/CompactionJobQueue.java:
##########
@@ -282,18 +281,17 @@ private boolean startJobIfPendingAndReady(
}
// Check if the job is already running, completed or skipped
- final CompactionStatus compactionStatus = getCurrentStatusForJob(job, policy);
- switch (compactionStatus.getState()) {
- case RUNNING:
+ final CompactionCandidate.TaskState candidateState = getCurrentTaskStateForJob(job);
Review Comment:
Why do you need a ready state? Everything that is currently in the `CompactionJobQueue` is ready anyway.
Having multiple differing states for task makes the definitions very confusing.
The compaction simulator is broken and has never actually been used.
We will either remove it completely and put up an improved version that can work with cascading reindexers too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] introduce clear separation between proposed compaction, eligibility status, and compaction mode, along with improved task state tracking (druid)
Posted by "cecemei (via GitHub)" <gi...@apache.org>.
cecemei commented on code in PR #18968:
URL: https://github.com/apache/druid/pull/18968#discussion_r2850138200
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionCandidateSearchPolicy.java:
##########
@@ -48,74 +45,11 @@ public interface CompactionCandidateSearchPolicy
int compareCandidates(CompactionCandidate candidateA, CompactionCandidate candidateB);
/**
- * Checks if the given {@link CompactionCandidate} is eligible for compaction
- * in the current iteration. A policy may implement this method to skip
- * compacting intervals or segments that do not fulfil some required criteria.
+ * Creates a {@link CompactionCandidate} after applying policy-specific checks to the proposed compaction candidate.
*
- * @return {@link Eligibility#OK} only if eligible.
- */
- Eligibility checkEligibilityForCompaction(
- CompactionCandidate candidate,
- CompactionTaskStatus latestTaskStatus
- );
-
- /**
- * Describes the eligibility of an interval for compaction.
+ * @param candidate the proposed compaction
+ * @param eligibility initial eligibility from compaction config checks
+ * @return final compaction candidate
*/
- class Eligibility
- {
- public static final Eligibility OK = new Eligibility(true, null);
-
- private final boolean eligible;
- private final String reason;
-
- private Eligibility(boolean eligible, String reason)
- {
- this.eligible = eligible;
- this.reason = reason;
- }
-
- public boolean isEligible()
- {
- return eligible;
- }
-
- public String getReason()
- {
- return reason;
- }
-
- public static Eligibility fail(String messageFormat, Object... args)
- {
- return new Eligibility(false, StringUtils.format(messageFormat, args));
- }
-
- @Override
- public boolean equals(Object object)
- {
- if (this == object) {
- return true;
- }
- if (object == null || getClass() != object.getClass()) {
- return false;
- }
- Eligibility that = (Eligibility) object;
- return eligible == that.eligible && Objects.equals(reason, that.reason);
- }
-
- @Override
- public int hashCode()
- {
- return Objects.hash(eligible, reason);
- }
-
- @Override
- public String toString()
- {
- return "Eligibility{" +
- "eligible=" + eligible +
- ", reason='" + reason + '\'' +
- '}';
- }
- }
+ CompactionCandidate createCandidate(CompactionCandidate.ProposedCompaction candidate, CompactionStatus eligibility);
Review Comment:
gotcha - can just return `CompactionMode` here and let callsite handle creation of candidate.
##########
server/src/test/java/org/apache/druid/server/coordinator/duty/CompactSegmentsTest.java:
##########
@@ -1545,17 +1550,20 @@ private void verifySnapshot(
{
Map<String, AutoCompactionSnapshot> autoCompactionSnapshots = compactSegments.getAutoCompactionSnapshot();
AutoCompactionSnapshot snapshot = autoCompactionSnapshots.get(dataSourceName);
- Assert.assertEquals(dataSourceName, snapshot.getDataSource());
- Assert.assertEquals(scheduleStatus, snapshot.getScheduleStatus());
- Assert.assertEquals(expectedByteCountAwaitingCompaction, snapshot.getBytesAwaitingCompaction());
- Assert.assertEquals(expectedByteCountCompressed, snapshot.getBytesCompacted());
- Assert.assertEquals(expectedByteCountSkipped, snapshot.getBytesSkipped());
- Assert.assertEquals(expectedIntervalCountAwaitingCompaction, snapshot.getIntervalCountAwaitingCompaction());
- Assert.assertEquals(expectedIntervalCountCompressed, snapshot.getIntervalCountCompacted());
- Assert.assertEquals(expectedIntervalCountSkipped, snapshot.getIntervalCountSkipped());
- Assert.assertEquals(expectedSegmentCountAwaitingCompaction, snapshot.getSegmentCountAwaitingCompaction());
- Assert.assertEquals(expectedSegmentCountCompressed, snapshot.getSegmentCountCompacted());
- Assert.assertEquals(expectedSegmentCountSkipped, snapshot.getSegmentCountSkipped());
+ Assert.assertEquals(new AutoCompactionSnapshot(
Review Comment:
i find it captures all the diff in the snapshot better, e.x. one less segment in skipped, one more segment in compressed, thus motivates me to make this change.
##########
indexing-service/src/main/java/org/apache/druid/indexing/compact/CompactionJobQueue.java:
##########
@@ -282,18 +281,17 @@ private boolean startJobIfPendingAndReady(
}
// Check if the job is already running, completed or skipped
- final CompactionStatus compactionStatus = getCurrentStatusForJob(job, policy);
- switch (compactionStatus.getState()) {
- case RUNNING:
+ final CompactionCandidate.TaskState candidateState = getCurrentTaskStateForJob(job);
Review Comment:
I've considered use `org.apache.druid.indexer.TaskState`(the one used in `CompactionTaskStatus` and not quite like it, because it doesn't have READY state (only running/failed/success), we could 1. use a null value for ready state, but then the simulator uses a map of TaskState; 2. or add a new READY/ NOT_RUN state, but then the existing `isRunnable` and `isComplete` method in `org.apache.druid.indexer.TaskState` is somewhat confusing, which we might need to update it if add a new enum value. Given that this is just an enum, i figured it might be ok to use a new class.
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionCandidate.java:
##########
@@ -37,76 +39,200 @@
*/
public class CompactionCandidate
{
- private final List<DataSegment> segments;
- private final Interval umbrellaInterval;
- private final Interval compactionInterval;
- private final String dataSource;
- private final long totalBytes;
- private final int numIntervals;
-
- private final CompactionStatus currentStatus;
-
- public static CompactionCandidate from(
- List<DataSegment> segments,
- @Nullable Granularity targetSegmentGranularity
- )
+ /**
+ * Non-empty list of segments of a datasource being proposed for compaction.
+ * A proposed compaction typically contains all the segments of a single time chunk.
+ */
+ public static class ProposedCompaction
Review Comment:
The motivation of this new `ProposedCompaction` is exactly to avoid null `CompactionStatus eligibility`, after this PR, the `CompactionCandidate` would always have an eligibility.
don't we need the uncompacted & compacted stats in `CompactionStatus` to sort `CompactionCandidate`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Add compaction eligibility concept and incremental compaction support (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18968:
URL: https://github.com/apache/druid/pull/18968#discussion_r2767199319
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionStatusTracker.java:
##########
@@ -80,10 +80,7 @@ public Set<String> getSubmittedTaskIds()
* This method assumes that the given candidate is eligible for compaction
* based on the current compaction config/supervisor of the datasource.
*/
- public CompactionStatus computeCompactionStatus(
- CompactionCandidate candidate,
- CompactionCandidateSearchPolicy searchPolicy
- )
+ public CompactionStatus computeCompactionStatus(CompactionCandidate candidate)
Review Comment:
I suppose we can now remove this method altogether, since we do not want to use the `lastTaskStatus` anyway.
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionStatus.java:
##########
@@ -357,29 +348,52 @@ private static class Evaluator
private final Map<CompactionState, List<DataSegment>> unknownStateToSegments = new HashMap<>();
@Nullable
- private final String targetFingerprint;
private final IndexingStateFingerprintMapper fingerprintMapper;
+ @Nullable
+ private final String targetFingerprint;
- private Evaluator(
+ Evaluator(
CompactionCandidate candidateSegments,
DataSourceCompactionConfig compactionConfig,
- @Nullable String targetFingerprint,
@Nullable IndexingStateFingerprintMapper fingerprintMapper
)
{
this.candidateSegments = candidateSegments;
this.compactionConfig = compactionConfig;
this.tuningConfig = ClientCompactionTaskQueryTuningConfig.from(compactionConfig);
this.configuredGranularitySpec = compactionConfig.getGranularitySpec();
- this.targetFingerprint = targetFingerprint;
this.fingerprintMapper = fingerprintMapper;
+ if (fingerprintMapper == null) {
+ targetFingerprint = null;
+ } else {
+ targetFingerprint = fingerprintMapper.generateFingerprint(
+ compactionConfig.getDataSource(),
+ compactionConfig.toCompactionState()
+ );
+ }
+ }
+
+ List<DataSegment> getUncompactedSegments()
+ {
+ return uncompactedSegments;
}
- private CompactionStatus evaluate()
+ /**
+ * Evaluates the compaction status of candidate segments through a multi-step process:
+ * <ol>
+ * <li>Validates input bytes are within limits</li>
+ * <li>Categorizes segments by compaction state (fingerprinted, uncompacted, or unknown)</li>
+ * <li>Performs fingerprint-based validation if available (fast path)</li>
+ * <li>Runs detailed checks against unknown states via {@link #CHECKS}</li>
+ * </ol>
+ *
+ * @return Pair of eligibility status and compaction status with reason for first failed check
+ */
+ Pair<CompactionCandidateSearchPolicy.Eligibility, CompactionStatus> evaluate()
{
- final CompactionStatus inputBytesCheck = inputBytesAreWithinLimit();
- if (inputBytesCheck.isSkipped()) {
- return inputBytesCheck;
+ final CompactionCandidateSearchPolicy.Eligibility inputBytesCheck = inputBytesAreWithinLimit();
+ if (inputBytesCheck != null) {
Review Comment:
@cecemei , I think this class should remain untouched so that `CompactionStatus` can remain completely agnostic of `CompactionEligibility`.
##########
server/src/main/java/org/apache/druid/server/compaction/DataSourceCompactibleSegmentIterator.java:
##########
@@ -329,17 +331,17 @@ private void findAndEnqueueSegmentsToCompact(CompactibleSegmentIterator compacti
continue;
}
- final CompactionCandidate candidates = CompactionCandidate.from(segments, config.getSegmentGranularity());
- final CompactionStatus compactionStatus = CompactionStatus.compute(candidates, config, fingerprintMapper);
- final CompactionCandidate candidatesWithStatus = candidates.withCurrentStatus(compactionStatus);
+ final CompactionCandidate candidatesWithStatus =
Review Comment:
Coordinator-based compaction will be deprecated soon.
We want to support incremental compaction on Overlord-based compaction only so that it becomes an incentive for users to migrate to compaction supervisors. We followed the same approach with indexing fingerprints as well.
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionEligibility.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.compaction;
+
+import org.apache.druid.java.util.common.StringUtils;
+
+import java.util.Objects;
+
+/**
+ * Describes the eligibility of an interval for compaction.
+ */
+public class CompactionEligibility
Review Comment:
We should add a method to this class:
```
public static CompactionEligibility compute(CompactionStatus status, CompactionCandidateSearchPolicy policy) {
// determine the eligibility here
}
```
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionCandidate.java:
##########
@@ -160,21 +169,122 @@ public CompactionStatus getCurrentStatus()
return currentStatus;
}
+ @Nullable
+ public CompactionCandidateSearchPolicy.Eligibility getPolicyEligibility()
+ {
+ return policyEligiblity;
+ }
+
/**
* Creates a copy of this CompactionCandidate object with the given status.
*/
public CompactionCandidate withCurrentStatus(CompactionStatus status)
{
- return new CompactionCandidate(segments, umbrellaInterval, compactionInterval, numIntervals, status);
+ return new CompactionCandidate(
+ segments,
+ umbrellaInterval,
+ compactionInterval,
+ numIntervals,
+ policyEligiblity,
+ status
+ );
+ }
+
+ public CompactionCandidate withPolicyEligibility(CompactionCandidateSearchPolicy.Eligibility eligibility)
+ {
+ return new CompactionCandidate(
+ segments,
+ umbrellaInterval,
+ compactionInterval,
+ numIntervals,
+ eligibility,
+ currentStatus
+ );
+ }
+
+ /**
+ * Evaluates this candidate for compaction eligibility based on the provided
+ * compaction configuration and search policy.
+ * <p>
+ * This method first evaluates the candidate against the compaction configuration
+ * using a {@link CompactionStatus.Evaluator} to determine if any segments need
+ * compaction. If segments are pending compaction, the search policy is consulted
+ * to determine the type of compaction:
+ * <ul>
+ * <li><b>NOT_ELIGIBLE</b>: Returns a candidate with status SKIPPED, indicating
+ * the policy decided compaction should not occur at this time</li>
+ * <li><b>FULL_COMPACTION</b>: Returns this candidate with status PENDING,
+ * indicating all segments should be compacted</li>
+ * <li><b>INCREMENTAL_COMPACTION</b>: Returns a new candidate containing only
+ * the uncompacted segments (as determined by the evaluator), with status
+ * PENDING for incremental compaction</li>
+ * </ul>
+ *
+ * @param config the compaction configuration for the datasource
+ * @param searchPolicy the policy used to determine compaction eligibility
+ * @return a CompactionCandidate with updated status and potentially filtered segments
+ */
+ public CompactionCandidate evaluate(
Review Comment:
+1
But this method should still not live here.
We should just have a `withEligibility()` method so that this class remains a bean for the most part.
This method can be moved to `CompactionEligibility.compute()` similar to `CompactionStatus.compute()`.
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionStatusTracker.java:
##########
@@ -80,10 +80,7 @@ public Set<String> getSubmittedTaskIds()
* This method assumes that the given candidate is eligible for compaction
* based on the current compaction config/supervisor of the datasource.
*/
- public CompactionStatus computeCompactionStatus(
- CompactionCandidate candidate,
- CompactionCandidateSearchPolicy searchPolicy
- )
+ public CompactionStatus computeCompactionStatus(CompactionCandidate candidate)
Review Comment:
Not needed anymore.
##########
server/src/main/java/org/apache/druid/server/compaction/DataSourceCompactibleSegmentIterator.java:
##########
@@ -329,17 +331,17 @@ private void findAndEnqueueSegmentsToCompact(CompactibleSegmentIterator compacti
continue;
}
- final CompactionCandidate candidates = CompactionCandidate.from(segments, config.getSegmentGranularity());
- final CompactionStatus compactionStatus = CompactionStatus.compute(candidates, config, fingerprintMapper);
- final CompactionCandidate candidatesWithStatus = candidates.withCurrentStatus(compactionStatus);
+ final CompactionCandidate candidatesWithStatus =
+ CompactionCandidate.from(segments, config.getSegmentGranularity())
+ .evaluate(config, searchPolicy, fingerprintMapper);
Review Comment:
This part should probably look more like the following:
```suggestion
final CompactionCandidate candidates = CompactionCandidate.from(segments, config.getSegmentGranularity());
final CompactionStatus compactionStatus = CompactionStatus.compute(candidates, config, fingerprintMapper);
final CompactionEligibility eligibility = CompactionEligibility.compute(compactionStatus, searchPolicy);
final CompactionCandidate candidatesWithStatus = candidates.withCurrentStatus(compactionStatus).withEligibility(eligibility);
```
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionCandidate.java:
##########
@@ -160,21 +169,123 @@ public CompactionStatus getCurrentStatus()
return currentStatus;
}
+ @Nullable
+ public CompactionEligibility getPolicyEligibility()
+ {
+ return policyEligiblity;
+ }
+
/**
* Creates a copy of this CompactionCandidate object with the given status.
*/
public CompactionCandidate withCurrentStatus(CompactionStatus status)
{
- return new CompactionCandidate(segments, umbrellaInterval, compactionInterval, numIntervals, status);
+ return new CompactionCandidate(
+ segments,
+ umbrellaInterval,
+ compactionInterval,
+ numIntervals,
+ policyEligiblity,
+ status
+ );
+ }
+
+ public CompactionCandidate withPolicyEligibility(CompactionEligibility eligibility)
Review Comment:
Nit: Thinking about this, I think it would be best to have a single method (e.g. `withStatusAndEligibility()`) that creates a copy of the `CompactionCandidate` with both the status and eligibility populated.
Otherwise, we would always create an intermediate copy unnecessarily.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Add compaction eligibility concept and incremental compaction support (druid)
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #18968:
URL: https://github.com/apache/druid/pull/18968#discussion_r2771821884
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionEligibility.java:
##########
@@ -0,0 +1,932 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.compaction;
+
+import com.google.common.base.Strings;
+import org.apache.commons.lang3.ArrayUtils;
+import org.apache.druid.client.indexing.ClientCompactionTaskQueryTuningConfig;
+import org.apache.druid.common.config.Configs;
+import org.apache.druid.data.input.impl.DimensionSchema;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.error.InvalidInput;
+import org.apache.druid.indexer.partitions.DimensionRangePartitionsSpec;
+import org.apache.druid.indexer.partitions.DynamicPartitionsSpec;
+import org.apache.druid.indexer.partitions.HashedPartitionsSpec;
+import org.apache.druid.indexer.partitions.PartitionsSpec;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.granularity.Granularity;
+import org.apache.druid.java.util.common.granularity.GranularityType;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.query.aggregation.AggregatorFactory;
+import org.apache.druid.segment.IndexSpec;
+import org.apache.druid.segment.metadata.IndexingStateFingerprintMapper;
+import org.apache.druid.segment.transform.CompactionTransformSpec;
+import org.apache.druid.server.coordinator.DataSourceCompactionConfig;
+import org.apache.druid.server.coordinator.UserCompactionTaskGranularityConfig;
+import org.apache.druid.timeline.CompactionState;
+import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.utils.CollectionUtils;
+import org.joda.time.Interval;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/**
+ * Describes the eligibility of an interval for compaction.
+ */
+public class CompactionEligibility
+{
+ public enum State
+ {
+ FULL_COMPACTION,
+ INCREMENTAL_COMPACTION,
+ NOT_ELIGIBLE,
+ NOT_APPLICABLE
+ }
+
+ static class CompactionEligibilityBuilder
+ {
+ private State state;
+ private CompactionStatistics compacted;
+ private CompactionStatistics uncompacted;
+ private List<DataSegment> uncompactedSegments;
+ private String reason;
+
+ CompactionEligibilityBuilder(State state, String reason)
+ {
+ this.state = state;
+ this.reason = reason;
+ }
+
+ CompactionEligibilityBuilder compacted(CompactionStatistics compacted)
+ {
+ this.compacted = compacted;
+ return this;
+ }
+
+ CompactionEligibilityBuilder uncompacted(CompactionStatistics uncompacted)
+ {
+ this.uncompacted = uncompacted;
+ return this;
+ }
+
+ CompactionEligibilityBuilder uncompactedSegments(List<DataSegment> uncompactedSegments)
+ {
+ this.uncompactedSegments = uncompactedSegments;
+ return this;
+ }
+
+ CompactionEligibility build()
+ {
+ return new CompactionEligibility(state, reason, compacted, uncompacted, uncompactedSegments);
+ }
+ }
+
+ public static final CompactionEligibility NOT_APPLICABLE = builder(State.NOT_APPLICABLE, "").build();
+
+ public static CompactionEligibility fail(String messageFormat, Object... args)
+ {
+ return builder(State.NOT_ELIGIBLE, StringUtils.format(messageFormat, args)).build();
+ }
+
+ private final State state;
+ private final String reason;
+
+ @Nullable
+ private final CompactionStatistics compacted;
+ @Nullable
+ private final CompactionStatistics uncompacted;
+ @Nullable
+ private final List<DataSegment> uncompactedSegments;
+
+ private CompactionEligibility(
+ State state,
+ String reason,
+ @Nullable CompactionStatistics compacted,
+ @Nullable CompactionStatistics uncompacted,
+ @Nullable List<DataSegment> uncompactedSegments
+ )
+ {
+ this.state = state;
+ this.reason = reason;
+ switch (state) {
+ case NOT_APPLICABLE:
+ break;
+ case NOT_ELIGIBLE:
+ InvalidInput.conditionalException(!Strings.isNullOrEmpty(reason), "must provide a reason");
+ break;
+ case FULL_COMPACTION:
+ case INCREMENTAL_COMPACTION:
+ InvalidInput.conditionalException(compacted != null, "must provide compacted stats");
+ InvalidInput.conditionalException(uncompacted != null, "must provide uncompacted stats");
+ InvalidInput.conditionalException(uncompactedSegments != null, "must provide uncompactedSegments");
+ break;
+ default:
+ throw DruidException.defensive("unexpected eligibility state[%s]", state);
+ }
+ this.compacted = compacted;
+ this.uncompacted = uncompacted;
+ this.uncompactedSegments = uncompactedSegments;
+ }
+
+ static CompactionEligibilityBuilder builder(State state, String reason)
+ {
+ return new CompactionEligibilityBuilder(state, reason);
+ }
+
+ public State getState()
+ {
+ return state;
+ }
+
+ public String getReason()
+ {
+ return reason;
+ }
+
+ @Nullable
+ public CompactionStatistics getUncompactedStats()
+ {
+ return uncompacted;
+ }
+
+ @Nullable
+ public CompactionStatistics getCompactedStats()
+ {
+ return compacted;
+ }
+
+ @Nullable
+ public List<DataSegment> getUncompactedSegments()
+ {
+ return uncompactedSegments;
+ }
+
+ public CompactionCandidate createCandidate(CompactionCandidate.ProposedCompaction proposedCompaction)
+ {
+ switch (state) {
+ case NOT_APPLICABLE:
+ return new CompactionCandidate(proposedCompaction, this, CompactionStatus.COMPLETE);
+ case NOT_ELIGIBLE:
+ return new CompactionCandidate(proposedCompaction, this, CompactionStatus.skipped(reason));
+ case FULL_COMPACTION:
+ return new CompactionCandidate(
+ proposedCompaction,
+ this,
+ CompactionStatus.pending(reason)
+ );
+ case INCREMENTAL_COMPACTION:
+ CompactionCandidate.ProposedCompaction newProposed = new CompactionCandidate.ProposedCompaction(
+ uncompactedSegments,
+ proposedCompaction.getUmbrellaInterval(),
+ proposedCompaction.getCompactionInterval(),
+ Math.toIntExact(uncompactedSegments.stream().map(DataSegment::getInterval).distinct().count())
+ );
+ return new CompactionCandidate(newProposed, this, CompactionStatus.pending(reason));
+ default:
+ throw DruidException.defensive("Unexpected eligibility state[%s]", state);
+ }
+ }
+
+ /**
+ * Evaluates a compaction candidate to determine its eligibility and compaction status.
+ * <p>
+ * This method performs a two-stage evaluation:
+ * <ol>
+ * <li>First, uses {@link Evaluator} to check if the candidate needs compaction
+ * based on the compaction config (e.g., checking segment granularity, partitions spec, etc.)</li>
+ * <li>Then, applies the search policy to determine if this candidate should be compacted in the
+ * current run (e.g., checking minimum segment count, bytes, or other policy criteria)</li>
+ * </ol>
+ *
+ * @param proposedCompaction the compaction candidate to evaluate
+ * @param config the compaction configuration for the datasource
+ * @param searchPolicy the policy that determines candidate ordering and eligibility
+ * @param fingerprintMapper mapper for indexing state fingerprints
+ * @return a new {@link CompactionCandidate} with updated eligibility and status. For incremental
+ * compaction, returns a candidate containing only the uncompacted segments.
+ */
+ public static CompactionEligibility evaluate(
+ CompactionCandidate.ProposedCompaction proposedCompaction,
+ DataSourceCompactionConfig config,
+ CompactionCandidateSearchPolicy searchPolicy,
+ IndexingStateFingerprintMapper fingerprintMapper
+ )
+ {
+ // ideally we should let this class only decides CompactionEligibility, and the callsite should handle recreation of candidate.
+ CompactionEligibility evaluatedCandidate = new Evaluator(proposedCompaction, config, fingerprintMapper).evaluate();
+ switch (Objects.requireNonNull(evaluatedCandidate).getState()) {
+ case NOT_APPLICABLE:
+ case NOT_ELIGIBLE:
+ return evaluatedCandidate;
+ case FULL_COMPACTION: // evaluator has decided compaction is needed, policy needs to further check
+ return searchPolicy.checkEligibilityForCompaction(proposedCompaction, evaluatedCandidate);
+ case INCREMENTAL_COMPACTION: // evaluator cant decide when to perform an incremental compaction
+ default:
+ throw DruidException.defensive("Unexpected eligibility[%s]", evaluatedCandidate);
+ }
+ }
+
+ @Override
+ public boolean equals(Object object)
+ {
+ if (this == object) {
+ return true;
+ }
+ if (object == null || getClass() != object.getClass()) {
+ return false;
+ }
+ CompactionEligibility that = (CompactionEligibility) object;
+ return state == that.state
+ && Objects.equals(reason, that.reason)
+ && Objects.equals(compacted, that.compacted)
+ && Objects.equals(uncompacted, that.uncompacted)
+ && Objects.equals(uncompactedSegments, that.uncompactedSegments);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(state, reason, compacted, uncompacted, uncompactedSegments);
+ }
+
+ @Override
+ public String toString()
+ {
+ return "CompactionEligibility{"
+ + "state=" + state
+ + ", reason='" + reason + '\''
+ + ", compacted=" + compacted
+ + ", uncompacted=" + uncompacted
+ + ", uncompactedSegments=" + uncompactedSegments
+ + '}';
+ }
+
+ /**
+ * List of checks performed to determine if compaction is already complete based on indexing state fingerprints.
+ */
+ static final List<Function<Evaluator, String>> FINGERPRINT_CHECKS = List.of(
+ Evaluator::allFingerprintedCandidatesHaveExpectedFingerprint
+ );
+
+ /**
+ * List of checks performed to determine if compaction is already complete.
+ * <p>
+ * The order of the checks must be honored while evaluating them.
+ */
+ static final List<Function<Evaluator, String>> CHECKS = Arrays.asList(
+ Evaluator::partitionsSpecIsUpToDate,
+ Evaluator::indexSpecIsUpToDate,
+ Evaluator::segmentGranularityIsUpToDate,
+ Evaluator::queryGranularityIsUpToDate,
+ Evaluator::rollupIsUpToDate,
+ Evaluator::dimensionsSpecIsUpToDate,
+ Evaluator::metricsSpecIsUpToDate,
+ Evaluator::transformSpecFilterIsUpToDate,
+ Evaluator::projectionsAreUpToDate
+ );
+
+ /**
+ * Evaluates checks to determine the compaction status of a
+ * {@link CompactionCandidate}.
+ */
+ private static class Evaluator
+ {
+ private static final Logger log = new Logger(Evaluator.class);
+
+ private final DataSourceCompactionConfig compactionConfig;
+ private final CompactionCandidate.ProposedCompaction proposedCompaction;
+ private final ClientCompactionTaskQueryTuningConfig tuningConfig;
+ private final UserCompactionTaskGranularityConfig configuredGranularitySpec;
+
+ private final List<DataSegment> fingerprintedSegments = new ArrayList<>();
+ private final List<DataSegment> compactedSegments = new ArrayList<>();
+ private final List<DataSegment> uncompactedSegments = new ArrayList<>();
+ private final Map<CompactionState, List<DataSegment>> unknownStateToSegments = new HashMap<>();
+
+ @Nullable
+ private final IndexingStateFingerprintMapper fingerprintMapper;
+ @Nullable
+ private final String targetFingerprint;
+
+ private Evaluator(
+ CompactionCandidate.ProposedCompaction proposedCompaction,
+ DataSourceCompactionConfig compactionConfig,
+ @Nullable IndexingStateFingerprintMapper fingerprintMapper
+ )
+ {
+ this.proposedCompaction = proposedCompaction;
+ this.compactionConfig = compactionConfig;
+ this.tuningConfig = ClientCompactionTaskQueryTuningConfig.from(compactionConfig);
+ this.configuredGranularitySpec = compactionConfig.getGranularitySpec();
+ this.fingerprintMapper = fingerprintMapper;
+ if (fingerprintMapper == null) {
+ targetFingerprint = null;
+ } else {
+ targetFingerprint = fingerprintMapper.generateFingerprint(
+ compactionConfig.getDataSource(),
+ compactionConfig.toCompactionState()
+ );
+ }
+ }
+
+ /**
+ * Evaluates the compaction status of candidate segments through a multi-step process:
+ * <ol>
+ * <li>Validates input bytes are within limits</li>
+ * <li>Categorizes segments by compaction state (fingerprinted, uncompacted, or unknown)</li>
+ * <li>Performs fingerprint-based validation if available (fast path)</li>
+ * <li>Runs detailed checks against unknown states via {@link CompactionEligibility#CHECKS}</li>
+ * </ol>
+ *
+ * @return Pair of eligibility status and compaction status with reason for first failed check
+ */
+ private CompactionEligibility evaluate()
+ {
+ final String inputBytesCheck = inputBytesAreWithinLimit();
+ if (inputBytesCheck != null) {
+ return CompactionEligibility.fail(inputBytesCheck);
+ }
+
+ List<String> reasonsForCompaction = new ArrayList<>();
+ String compactedOnceCheck = segmentsHaveBeenCompactedAtLeastOnce();
+ if (compactedOnceCheck != null) {
+ reasonsForCompaction.add(compactedOnceCheck);
+ }
+
+ if (fingerprintMapper != null && targetFingerprint != null) {
+ // First try fingerprint-based evaluation (fast path)
+ FINGERPRINT_CHECKS.stream()
+ .map(f -> f.apply(this))
+ .filter(Objects::nonNull)
+ .findFirst()
+ .ifPresent(reasonsForCompaction::add);
+
+ }
+
+ if (!unknownStateToSegments.isEmpty()) {
+ // Run CHECKS against any states with uknown compaction status
+ reasonsForCompaction.addAll(
+ CHECKS.stream()
+ .map(f -> f.apply(this))
+ .filter(Objects::nonNull)
+ .collect(Collectors.toList())
+ );
+
+ // Any segments left in unknownStateToSegments passed all checks and are considered compacted
+ compactedSegments.addAll(
+ unknownStateToSegments
+ .values()
+ .stream()
+ .flatMap(List::stream)
+ .collect(Collectors.toList())
+ );
+ }
+
+ if (reasonsForCompaction.isEmpty()) {
+ return CompactionEligibility.NOT_APPLICABLE;
+ } else {
+ return builder(State.FULL_COMPACTION, reasonsForCompaction.get(0)).compacted(createStats(compactedSegments))
+ .uncompacted(createStats(uncompactedSegments))
+ .uncompactedSegments(uncompactedSegments)
+ .build();
+ }
+ }
+
+ /**
+ * Evaluates the fingerprints of all fingerprinted candidate segments against the expected fingerprint.
+ * <p>
+ * If all fingerprinted segments have the expected fingerprint, the check can quickly pass as COMPLETE. However,
+ * if any fingerprinted segment has a mismatched fingerprint, we need to investigate further by adding them to
+ * {@link #unknownStateToSegments} where their indexing states will be analyzed.
+ * </p>
+ */
+ private String allFingerprintedCandidatesHaveExpectedFingerprint()
+ {
+ Map<String, List<DataSegment>> mismatchedFingerprintToSegmentMap = new HashMap<>();
+ for (DataSegment segment : fingerprintedSegments) {
+ String fingerprint = segment.getIndexingStateFingerprint();
+ if (fingerprint == null) {
+ // Should not happen since we are iterating over fingerprintedSegments
+ } else if (fingerprint.equals(targetFingerprint)) {
+ compactedSegments.add(segment);
+ } else {
+ mismatchedFingerprintToSegmentMap
+ .computeIfAbsent(fingerprint, k -> new ArrayList<>())
+ .add(segment);
+ }
+ }
+
+ if (mismatchedFingerprintToSegmentMap.isEmpty()) {
+ // All fingerprinted segments have the expected fingerprint - compaction is complete
+ return null;
+ }
+
+ if (fingerprintMapper == null) {
+ // Cannot evaluate further without a fingerprint mapper
+ uncompactedSegments.addAll(
+ mismatchedFingerprintToSegmentMap.values()
+ .stream()
+ .flatMap(List::stream)
+ .collect(Collectors.toList())
+ );
+ return "Segments have a mismatched fingerprint and no fingerprint mapper is available";
+ }
+
+ boolean fingerprintedSegmentWithoutCachedStateFound = false;
+
+ for (Map.Entry<String, List<DataSegment>> e : mismatchedFingerprintToSegmentMap.entrySet()) {
+ String fingerprint = e.getKey();
+ CompactionState stateToValidate = fingerprintMapper.getStateForFingerprint(fingerprint).orElse(null);
+ if (stateToValidate == null) {
+ log.warn("No indexing state found for fingerprint[%s]", fingerprint);
+ fingerprintedSegmentWithoutCachedStateFound = true;
+ uncompactedSegments.addAll(e.getValue());
+ } else {
+ // Note that this does not mean we need compaction yet - we need to validate the state further to determine this
+ unknownStateToSegments.compute(
+ stateToValidate,
+ (state, segments) -> {
+ if (segments == null) {
+ segments = new ArrayList<>();
+ }
+ segments.addAll(e.getValue());
+ return segments;
+ }
+ );
+ }
+ }
+
+ if (fingerprintedSegmentWithoutCachedStateFound) {
+ return "One or more fingerprinted segments do not have a cached indexing state";
+ } else {
+ return null;
+ }
+ }
+
+ /**
+ * Checks if all the segments have been compacted at least once and groups them into uncompacted, fingerprinted, or
+ * non-fingerprinted.
+ */
+ private String segmentsHaveBeenCompactedAtLeastOnce()
+ {
+ for (DataSegment segment : proposedCompaction.getSegments()) {
+ final String fingerprint = segment.getIndexingStateFingerprint();
+ final CompactionState segmentState = segment.getLastCompactionState();
+ if (fingerprint != null) {
+ fingerprintedSegments.add(segment);
+ } else if (segmentState == null) {
+ uncompactedSegments.add(segment);
+ } else {
+ unknownStateToSegments.computeIfAbsent(segmentState, s -> new ArrayList<>()).add(segment);
+ }
+ }
+
+ if (uncompactedSegments.isEmpty()) {
+ return null;
+ } else {
+ return "not compacted yet";
+ }
+ }
+
+ private String partitionsSpecIsUpToDate()
+ {
+ return evaluateForAllCompactionStates(this::partitionsSpecIsUpToDate);
+ }
+
+ private String indexSpecIsUpToDate()
+ {
+ return evaluateForAllCompactionStates(this::indexSpecIsUpToDate);
+ }
+
+ private String projectionsAreUpToDate()
+ {
+ return evaluateForAllCompactionStates(this::projectionsAreUpToDate);
+ }
+
+ private String segmentGranularityIsUpToDate()
+ {
+ return evaluateForAllCompactionStates(this::segmentGranularityIsUpToDate);
+ }
+
+ private String rollupIsUpToDate()
+ {
+ return evaluateForAllCompactionStates(this::rollupIsUpToDate);
+ }
+
+ private String queryGranularityIsUpToDate()
+ {
+ return evaluateForAllCompactionStates(this::queryGranularityIsUpToDate);
+ }
+
+ private String dimensionsSpecIsUpToDate()
+ {
+ return evaluateForAllCompactionStates(this::dimensionsSpecIsUpToDate);
+ }
+
+ private String metricsSpecIsUpToDate()
+ {
+ return evaluateForAllCompactionStates(this::metricsSpecIsUpToDate);
+ }
+
+ private String transformSpecFilterIsUpToDate()
+ {
+ return evaluateForAllCompactionStates(this::transformSpecFilterIsUpToDate);
+ }
+
+ private String partitionsSpecIsUpToDate(CompactionState lastCompactionState)
+ {
+ PartitionsSpec existingPartionsSpec = lastCompactionState.getPartitionsSpec();
+ if (existingPartionsSpec instanceof DimensionRangePartitionsSpec) {
+ existingPartionsSpec = getEffectiveRangePartitionsSpec((DimensionRangePartitionsSpec) existingPartionsSpec);
+ } else if (existingPartionsSpec instanceof DynamicPartitionsSpec) {
+ existingPartionsSpec = new DynamicPartitionsSpec(
+ existingPartionsSpec.getMaxRowsPerSegment(),
+ ((DynamicPartitionsSpec) existingPartionsSpec).getMaxTotalRowsOr(Long.MAX_VALUE)
+ );
+ }
+ return completeIfNullOrEqual(
+ "partitionsSpec",
+ findPartitionsSpecFromConfig(tuningConfig),
+ existingPartionsSpec,
+ CompactionEligibility::asString
+ );
+ }
+
+ private String indexSpecIsUpToDate(CompactionState lastCompactionState)
+ {
+ return completeIfNullOrEqual(
+ "indexSpec",
+ Configs.valueOrDefault(tuningConfig.getIndexSpec(), IndexSpec.getDefault()).getEffectiveSpec(),
+ lastCompactionState.getIndexSpec().getEffectiveSpec(),
+ String::valueOf
+ );
+ }
+
+ private String projectionsAreUpToDate(CompactionState lastCompactionState)
+ {
+ return completeIfNullOrEqual(
+ "projections",
+ compactionConfig.getProjections(),
+ lastCompactionState.getProjections(),
+ String::valueOf
+ );
+ }
+
+ @Nullable
+ private String inputBytesAreWithinLimit()
+ {
+ final long inputSegmentSize = compactionConfig.getInputSegmentSizeBytes();
+ if (proposedCompaction.getTotalBytes() > inputSegmentSize) {
+ return StringUtils.format(
+ "'inputSegmentSize' exceeded: Total segment size[%d] is larger than allowed inputSegmentSize[%d]",
+ proposedCompaction.getTotalBytes(), inputSegmentSize
+ );
+ }
+ return null;
+ }
+
+ private String segmentGranularityIsUpToDate(CompactionState lastCompactionState)
+ {
+ if (configuredGranularitySpec == null
+ || configuredGranularitySpec.getSegmentGranularity() == null) {
+ return null;
+ }
+
+ final Granularity configuredSegmentGranularity = configuredGranularitySpec.getSegmentGranularity();
+ final UserCompactionTaskGranularityConfig existingGranularitySpec = getGranularitySpec(lastCompactionState);
+ final Granularity existingSegmentGranularity
+ = existingGranularitySpec == null ? null : existingGranularitySpec.getSegmentGranularity();
+
+ if (configuredSegmentGranularity.equals(existingSegmentGranularity)) {
+ return null;
+ } else if (existingSegmentGranularity == null) {
+ // Candidate segments were compacted without segment granularity specified
+ // Check if the segments already have the desired segment granularity
+ final List<DataSegment> segmentsForState = unknownStateToSegments.get(lastCompactionState);
+ boolean needsCompaction = segmentsForState.stream().anyMatch(
+ segment -> !configuredSegmentGranularity.isAligned(segment.getInterval())
+ );
+ if (needsCompaction) {
+ return StringUtils.format(
+ "segmentGranularity: segments do not align with target[%s]",
+ CompactionEligibility.asString(configuredSegmentGranularity)
+ );
+ }
+ } else {
+ return configChanged(
+ "segmentGranularity",
+ configuredSegmentGranularity,
+ existingSegmentGranularity,
+ CompactionEligibility::asString
+ );
+ }
+
+ return null;
+ }
+
+ private String rollupIsUpToDate(CompactionState lastCompactionState)
+ {
+ if (configuredGranularitySpec == null) {
+ return null;
+ } else {
+ final UserCompactionTaskGranularityConfig existingGranularitySpec
+ = getGranularitySpec(lastCompactionState);
+ return completeIfNullOrEqual(
+ "rollup",
+ configuredGranularitySpec.isRollup(),
+ existingGranularitySpec == null ? null : existingGranularitySpec.isRollup(),
+ String::valueOf
+ );
+ }
+ }
+
+ private String queryGranularityIsUpToDate(CompactionState lastCompactionState)
+ {
+ if (configuredGranularitySpec == null) {
+ return null;
+ } else {
+ final UserCompactionTaskGranularityConfig existingGranularitySpec
+ = getGranularitySpec(lastCompactionState);
+ return completeIfNullOrEqual(
+ "queryGranularity",
+ configuredGranularitySpec.getQueryGranularity(),
+ existingGranularitySpec == null ? null : existingGranularitySpec.getQueryGranularity(),
+ CompactionEligibility::asString
+ );
+ }
+ }
+
+ /**
+ * Removes partition dimensions before comparison, since they are placed in front of the sort order --
+ * which can create a mismatch between expected and actual order of dimensions. Partition dimensions are separately
+ * covered in {@link Evaluator#partitionsSpecIsUpToDate()} check.
+ */
+ private String dimensionsSpecIsUpToDate(CompactionState lastCompactionState)
+ {
+ if (compactionConfig.getDimensionsSpec() == null) {
+ return null;
+ } else {
+ List<DimensionSchema> existingDimensions = getNonPartitioningDimensions(
+ lastCompactionState.getDimensionsSpec() == null
+ ? null
+ : lastCompactionState.getDimensionsSpec().getDimensions(),
+ lastCompactionState.getPartitionsSpec(),
+ lastCompactionState.getIndexSpec()
+ );
+ List<DimensionSchema> configuredDimensions = getNonPartitioningDimensions(
+ compactionConfig.getDimensionsSpec().getDimensions(),
+ compactionConfig.getTuningConfig() == null ? null : compactionConfig.getTuningConfig().getPartitionsSpec(),
+ compactionConfig.getTuningConfig() == null
+ ? IndexSpec.getDefault()
+ : compactionConfig.getTuningConfig().getIndexSpec()
+ );
+ return completeIfNullOrEqual(
+ "dimensionsSpec",
+ configuredDimensions,
+ existingDimensions,
+ String::valueOf
+ );
+ }
+ }
+
+ private String metricsSpecIsUpToDate(CompactionState lastCompactionState)
+ {
+ final AggregatorFactory[] configuredMetricsSpec = compactionConfig.getMetricsSpec();
+ if (ArrayUtils.isEmpty(configuredMetricsSpec)) {
+ return null;
+ }
+
+ final List<AggregatorFactory> metricSpecList = lastCompactionState.getMetricsSpec();
+ final AggregatorFactory[] existingMetricsSpec
+ = CollectionUtils.isNullOrEmpty(metricSpecList)
+ ? null : metricSpecList.toArray(new AggregatorFactory[0]);
+
+ if (existingMetricsSpec == null || !Arrays.deepEquals(configuredMetricsSpec, existingMetricsSpec)) {
+ return configChanged(
+ "metricsSpec",
+ configuredMetricsSpec,
+ existingMetricsSpec,
+ Arrays::toString
+ );
+ } else {
+ return null;
+ }
+ }
+
+ private String transformSpecFilterIsUpToDate(CompactionState lastCompactionState)
+ {
+ if (compactionConfig.getTransformSpec() == null) {
+ return null;
+ }
+
+ CompactionTransformSpec existingTransformSpec = lastCompactionState.getTransformSpec();
+ return completeIfNullOrEqual(
+ "transformSpec filter",
+ compactionConfig.getTransformSpec().getFilter(),
+ existingTransformSpec == null ? null : existingTransformSpec.getFilter(),
+ String::valueOf
+ );
+ }
+
+ /**
+ * Evaluates the given check for each entry in the {@link #unknownStateToSegments}.
+ * If any entry fails the given check by returning a status which is not
+ * COMPLETE, all the segments with that state are moved to {@link #uncompactedSegments}.
+ *
+ * @return The first status which is not COMPLETE.
+ */
+ private String evaluateForAllCompactionStates(Function<CompactionState, String> check)
+ {
+ String firstIncomplete = null;
+ for (CompactionState state : List.copyOf(unknownStateToSegments.keySet())) {
+ final String eligibleReason = check.apply(state);
+ if (eligibleReason != null) {
+ uncompactedSegments.addAll(unknownStateToSegments.remove(state));
+ if (firstIncomplete == null) {
+ firstIncomplete = eligibleReason;
+ }
+ }
+ }
+
+ return firstIncomplete;
+ }
+
+ private static UserCompactionTaskGranularityConfig getGranularitySpec(
+ CompactionState compactionState
+ )
+ {
+ return UserCompactionTaskGranularityConfig.from(compactionState.getGranularitySpec());
+ }
+
+ private static CompactionStatistics createStats(List<DataSegment> segments)
+ {
+ final Set<Interval> segmentIntervals =
+ segments.stream().map(DataSegment::getInterval).collect(Collectors.toSet());
+ final long totalBytes = segments.stream().mapToLong(DataSegment::getSize).sum();
+ return CompactionStatistics.create(totalBytes, segments.size(), segmentIntervals.size());
+ }
+ }
+
+
+ /**
+ * Computes compaction status for the given field. The status is assumed to be
+ * COMPLETE (i.e. no further compaction is required) if the configured value
+ * of the field is null or equal to the current value.
+ */
+ private static <T> String completeIfNullOrEqual(
+ String field,
+ T configured,
+ T current,
+ Function<T, String> stringFunction
+ )
+ {
+ if (configured == null || configured.equals(current)) {
+ return null;
+ } else {
+ return configChanged(field, configured, current, stringFunction);
+ }
+ }
+
+ private static <T> String configChanged(
+ String field,
+ T target,
+ T current,
+ Function<T, String> stringFunction
+ )
+ {
+ return StringUtils.format(
+ "'%s' mismatch: required[%s], current[%s]",
+ field,
+ target == null ? null : stringFunction.apply(target),
+ current == null ? null : stringFunction.apply(current)
+ );
+ }
+
+ private static String asString(Granularity granularity)
+ {
+ if (granularity == null) {
+ return null;
+ }
+ for (GranularityType type : GranularityType.values()) {
+ if (type.getDefaultGranularity().equals(granularity)) {
+ return type.toString();
+ }
+ }
+ return granularity.toString();
+ }
+
+ private static String asString(PartitionsSpec partitionsSpec)
+ {
+ if (partitionsSpec instanceof DimensionRangePartitionsSpec) {
+ DimensionRangePartitionsSpec rangeSpec = (DimensionRangePartitionsSpec) partitionsSpec;
+ return StringUtils.format(
+ "'range' on %s with %,d rows",
+ rangeSpec.getPartitionDimensions(), rangeSpec.getTargetRowsPerSegment()
+ );
+ } else if (partitionsSpec instanceof HashedPartitionsSpec) {
+ HashedPartitionsSpec hashedSpec = (HashedPartitionsSpec) partitionsSpec;
+ return StringUtils.format(
+ "'hashed' on %s with %,d rows",
+ hashedSpec.getPartitionDimensions(), hashedSpec.getTargetRowsPerSegment()
+ );
+ } else if (partitionsSpec instanceof DynamicPartitionsSpec) {
+ DynamicPartitionsSpec dynamicSpec = (DynamicPartitionsSpec) partitionsSpec;
+ return StringUtils.format(
+ "'dynamic' with %,d rows",
+ dynamicSpec.getMaxRowsPerSegment()
+ );
+ } else {
+ return partitionsSpec.toString();
+ }
+ }
+
+ @Nullable
+ public static PartitionsSpec findPartitionsSpecFromConfig(ClientCompactionTaskQueryTuningConfig tuningConfig)
+ {
+ final PartitionsSpec partitionsSpecFromTuningConfig = tuningConfig.getPartitionsSpec();
+ if (partitionsSpecFromTuningConfig == null) {
+ final Long maxTotalRows = tuningConfig.getMaxTotalRows();
+ final Integer maxRowsPerSegment = tuningConfig.getMaxRowsPerSegment();
Review Comment:
## Deprecated method or constructor invocation
Invoking [ClientCompactionTaskQueryTuningConfig.getMaxRowsPerSegment](1) should be avoided because it has been deprecated.
[Show more details](https://github.com/apache/druid/security/code-scanning/10806)
##########
server/src/test/java/org/apache/druid/server/compaction/CompactionEligibilityEvaluateTest.java:
##########
@@ -0,0 +1,895 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.compaction;
+
+import org.apache.druid.client.indexing.ClientCompactionTaskQueryTuningConfig;
+import org.apache.druid.data.input.impl.AggregateProjectionSpec;
+import org.apache.druid.data.input.impl.DimensionsSpec;
+import org.apache.druid.data.input.impl.LongDimensionSchema;
+import org.apache.druid.data.input.impl.StringDimensionSchema;
+import org.apache.druid.indexer.granularity.GranularitySpec;
+import org.apache.druid.indexer.granularity.UniformGranularitySpec;
+import org.apache.druid.indexer.partitions.DimensionRangePartitionsSpec;
+import org.apache.druid.indexer.partitions.DynamicPartitionsSpec;
+import org.apache.druid.indexer.partitions.HashedPartitionsSpec;
+import org.apache.druid.indexer.partitions.PartitionsSpec;
+import org.apache.druid.jackson.DefaultObjectMapper;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.java.util.common.granularity.Granularity;
+import org.apache.druid.query.aggregation.LongSumAggregatorFactory;
+import org.apache.druid.segment.AutoTypeColumnSchema;
+import org.apache.druid.segment.IndexSpec;
+import org.apache.druid.segment.TestDataSource;
+import org.apache.druid.segment.data.CompressionStrategy;
+import org.apache.druid.segment.metadata.DefaultIndexingStateFingerprintMapper;
+import org.apache.druid.segment.metadata.HeapMemoryIndexingStateStorage;
+import org.apache.druid.segment.metadata.IndexingStateCache;
+import org.apache.druid.segment.metadata.IndexingStateFingerprintMapper;
+import org.apache.druid.segment.nested.NestedCommonFormatColumnFormatSpec;
+import org.apache.druid.server.coordinator.DataSourceCompactionConfig;
+import org.apache.druid.server.coordinator.InlineSchemaDataSourceCompactionConfig;
+import org.apache.druid.server.coordinator.UserCompactionTaskDimensionsConfig;
+import org.apache.druid.server.coordinator.UserCompactionTaskGranularityConfig;
+import org.apache.druid.server.coordinator.UserCompactionTaskQueryTuningConfig;
+import org.apache.druid.timeline.CompactionState;
+import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.SegmentId;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.List;
+
+public class CompactionEligibilityEvaluateTest
+{
+ private static final DataSegment WIKI_SEGMENT
+ = DataSegment.builder(SegmentId.of(TestDataSource.WIKI, Intervals.of("2013-01-01/PT1H"), "v1", 0))
+ .size(100_000_000L)
+ .build();
+ private static final DataSegment WIKI_SEGMENT_2
+ = DataSegment.builder(SegmentId.of(TestDataSource.WIKI, Intervals.of("2013-01-01/PT1H"), "v1", 1))
+ .size(100_000_000L)
+ .build();
+
+ private HeapMemoryIndexingStateStorage indexingStateStorage;
+ private IndexingStateCache indexingStateCache;
+ private IndexingStateFingerprintMapper fingerprintMapper;
+
+ @Before
+ public void setUp()
+ {
+ indexingStateStorage = new HeapMemoryIndexingStateStorage();
+ indexingStateCache = new IndexingStateCache();
+ fingerprintMapper = new DefaultIndexingStateFingerprintMapper(
+ indexingStateCache,
+ new DefaultObjectMapper()
+ );
+ }
+
+ /**
+ * Helper to sync the cache with states stored in the manager (for tests that persist states).
+ */
+ private void syncCacheFromManager()
+ {
+ indexingStateCache.resetIndexingStatesForPublishedSegments(indexingStateStorage.getAllStoredStates());
+ }
+
+ @Test
+ public void testFindPartitionsSpecWhenGivenIsNull()
+ {
+ final ClientCompactionTaskQueryTuningConfig tuningConfig
+ = ClientCompactionTaskQueryTuningConfig.from(null);
+ Assert.assertNull(CompactionEligibility.findPartitionsSpecFromConfig(tuningConfig));
+ }
+
+ @Test
+ public void testFindPartitionsSpecWhenGivenIsDynamicWithNullMaxTotalRows()
+ {
+ final PartitionsSpec partitionsSpec = new DynamicPartitionsSpec(null, null);
+ final ClientCompactionTaskQueryTuningConfig tuningConfig
+ = ClientCompactionTaskQueryTuningConfig.from(createCompactionConfig(partitionsSpec));
+ Assert.assertEquals(
+ new DynamicPartitionsSpec(null, Long.MAX_VALUE),
+ CompactionEligibility.findPartitionsSpecFromConfig(tuningConfig)
+ );
+ }
+
+ @Test
+ public void testFindPartitionsSpecWhenGivenIsDynamicWithMaxTotalRows()
+ {
+ final PartitionsSpec partitionsSpec = new DynamicPartitionsSpec(null, 1000L);
+ final ClientCompactionTaskQueryTuningConfig tuningConfig
+ = ClientCompactionTaskQueryTuningConfig.from(createCompactionConfig(partitionsSpec));
+ Assert.assertEquals(
+ partitionsSpec,
+ CompactionEligibility.findPartitionsSpecFromConfig(tuningConfig)
+ );
+ }
+
+ @Test
+ public void testFindPartitionsSpecWhenGivenIsDynamicWithMaxRowsPerSegment()
+ {
+ final PartitionsSpec partitionsSpec = new DynamicPartitionsSpec(100, 1000L);
+ final ClientCompactionTaskQueryTuningConfig tuningConfig
+ = ClientCompactionTaskQueryTuningConfig.from(createCompactionConfig(partitionsSpec));
+ Assert.assertEquals(
+ partitionsSpec,
+ CompactionEligibility.findPartitionsSpecFromConfig(tuningConfig)
+ );
+ }
+
+ @Test
+ public void testFindPartitionsSpecFromConfigWithDeprecatedMaxRowsPerSegmentAndMaxTotalRowsReturnGivenValues()
+ {
+ final DataSourceCompactionConfig config =
+ InlineSchemaDataSourceCompactionConfig.builder()
+ .forDataSource("datasource")
+ .withMaxRowsPerSegment(100)
Review Comment:
## Deprecated method or constructor invocation
Invoking [Builder.withMaxRowsPerSegment](1) should be avoided because it has been deprecated.
[Show more details](https://github.com/apache/druid/security/code-scanning/10807)
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionEligibility.java:
##########
@@ -0,0 +1,932 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.compaction;
+
+import com.google.common.base.Strings;
+import org.apache.commons.lang3.ArrayUtils;
+import org.apache.druid.client.indexing.ClientCompactionTaskQueryTuningConfig;
+import org.apache.druid.common.config.Configs;
+import org.apache.druid.data.input.impl.DimensionSchema;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.error.InvalidInput;
+import org.apache.druid.indexer.partitions.DimensionRangePartitionsSpec;
+import org.apache.druid.indexer.partitions.DynamicPartitionsSpec;
+import org.apache.druid.indexer.partitions.HashedPartitionsSpec;
+import org.apache.druid.indexer.partitions.PartitionsSpec;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.granularity.Granularity;
+import org.apache.druid.java.util.common.granularity.GranularityType;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.query.aggregation.AggregatorFactory;
+import org.apache.druid.segment.IndexSpec;
+import org.apache.druid.segment.metadata.IndexingStateFingerprintMapper;
+import org.apache.druid.segment.transform.CompactionTransformSpec;
+import org.apache.druid.server.coordinator.DataSourceCompactionConfig;
+import org.apache.druid.server.coordinator.UserCompactionTaskGranularityConfig;
+import org.apache.druid.timeline.CompactionState;
+import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.utils.CollectionUtils;
+import org.joda.time.Interval;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/**
+ * Describes the eligibility of an interval for compaction.
+ */
+public class CompactionEligibility
+{
+ public enum State
+ {
+ FULL_COMPACTION,
+ INCREMENTAL_COMPACTION,
+ NOT_ELIGIBLE,
+ NOT_APPLICABLE
+ }
+
+ static class CompactionEligibilityBuilder
+ {
+ private State state;
+ private CompactionStatistics compacted;
+ private CompactionStatistics uncompacted;
+ private List<DataSegment> uncompactedSegments;
+ private String reason;
+
+ CompactionEligibilityBuilder(State state, String reason)
+ {
+ this.state = state;
+ this.reason = reason;
+ }
+
+ CompactionEligibilityBuilder compacted(CompactionStatistics compacted)
+ {
+ this.compacted = compacted;
+ return this;
+ }
+
+ CompactionEligibilityBuilder uncompacted(CompactionStatistics uncompacted)
+ {
+ this.uncompacted = uncompacted;
+ return this;
+ }
+
+ CompactionEligibilityBuilder uncompactedSegments(List<DataSegment> uncompactedSegments)
+ {
+ this.uncompactedSegments = uncompactedSegments;
+ return this;
+ }
+
+ CompactionEligibility build()
+ {
+ return new CompactionEligibility(state, reason, compacted, uncompacted, uncompactedSegments);
+ }
+ }
+
+ public static final CompactionEligibility NOT_APPLICABLE = builder(State.NOT_APPLICABLE, "").build();
+
+ public static CompactionEligibility fail(String messageFormat, Object... args)
+ {
+ return builder(State.NOT_ELIGIBLE, StringUtils.format(messageFormat, args)).build();
+ }
+
+ private final State state;
+ private final String reason;
+
+ @Nullable
+ private final CompactionStatistics compacted;
+ @Nullable
+ private final CompactionStatistics uncompacted;
+ @Nullable
+ private final List<DataSegment> uncompactedSegments;
+
+ private CompactionEligibility(
+ State state,
+ String reason,
+ @Nullable CompactionStatistics compacted,
+ @Nullable CompactionStatistics uncompacted,
+ @Nullable List<DataSegment> uncompactedSegments
+ )
+ {
+ this.state = state;
+ this.reason = reason;
+ switch (state) {
+ case NOT_APPLICABLE:
+ break;
+ case NOT_ELIGIBLE:
+ InvalidInput.conditionalException(!Strings.isNullOrEmpty(reason), "must provide a reason");
+ break;
+ case FULL_COMPACTION:
+ case INCREMENTAL_COMPACTION:
+ InvalidInput.conditionalException(compacted != null, "must provide compacted stats");
+ InvalidInput.conditionalException(uncompacted != null, "must provide uncompacted stats");
+ InvalidInput.conditionalException(uncompactedSegments != null, "must provide uncompactedSegments");
+ break;
+ default:
+ throw DruidException.defensive("unexpected eligibility state[%s]", state);
+ }
+ this.compacted = compacted;
+ this.uncompacted = uncompacted;
+ this.uncompactedSegments = uncompactedSegments;
+ }
+
+ static CompactionEligibilityBuilder builder(State state, String reason)
+ {
+ return new CompactionEligibilityBuilder(state, reason);
+ }
+
+ public State getState()
+ {
+ return state;
+ }
+
+ public String getReason()
+ {
+ return reason;
+ }
+
+ @Nullable
+ public CompactionStatistics getUncompactedStats()
+ {
+ return uncompacted;
+ }
+
+ @Nullable
+ public CompactionStatistics getCompactedStats()
+ {
+ return compacted;
+ }
+
+ @Nullable
+ public List<DataSegment> getUncompactedSegments()
+ {
+ return uncompactedSegments;
+ }
+
+ public CompactionCandidate createCandidate(CompactionCandidate.ProposedCompaction proposedCompaction)
+ {
+ switch (state) {
+ case NOT_APPLICABLE:
+ return new CompactionCandidate(proposedCompaction, this, CompactionStatus.COMPLETE);
+ case NOT_ELIGIBLE:
+ return new CompactionCandidate(proposedCompaction, this, CompactionStatus.skipped(reason));
+ case FULL_COMPACTION:
+ return new CompactionCandidate(
+ proposedCompaction,
+ this,
+ CompactionStatus.pending(reason)
+ );
+ case INCREMENTAL_COMPACTION:
+ CompactionCandidate.ProposedCompaction newProposed = new CompactionCandidate.ProposedCompaction(
+ uncompactedSegments,
+ proposedCompaction.getUmbrellaInterval(),
+ proposedCompaction.getCompactionInterval(),
+ Math.toIntExact(uncompactedSegments.stream().map(DataSegment::getInterval).distinct().count())
+ );
+ return new CompactionCandidate(newProposed, this, CompactionStatus.pending(reason));
+ default:
+ throw DruidException.defensive("Unexpected eligibility state[%s]", state);
+ }
+ }
+
+ /**
+ * Evaluates a compaction candidate to determine its eligibility and compaction status.
+ * <p>
+ * This method performs a two-stage evaluation:
+ * <ol>
+ * <li>First, uses {@link Evaluator} to check if the candidate needs compaction
+ * based on the compaction config (e.g., checking segment granularity, partitions spec, etc.)</li>
+ * <li>Then, applies the search policy to determine if this candidate should be compacted in the
+ * current run (e.g., checking minimum segment count, bytes, or other policy criteria)</li>
+ * </ol>
+ *
+ * @param proposedCompaction the compaction candidate to evaluate
+ * @param config the compaction configuration for the datasource
+ * @param searchPolicy the policy that determines candidate ordering and eligibility
+ * @param fingerprintMapper mapper for indexing state fingerprints
+ * @return a new {@link CompactionCandidate} with updated eligibility and status. For incremental
+ * compaction, returns a candidate containing only the uncompacted segments.
+ */
+ public static CompactionEligibility evaluate(
+ CompactionCandidate.ProposedCompaction proposedCompaction,
+ DataSourceCompactionConfig config,
+ CompactionCandidateSearchPolicy searchPolicy,
+ IndexingStateFingerprintMapper fingerprintMapper
+ )
+ {
+ // ideally we should let this class only decides CompactionEligibility, and the callsite should handle recreation of candidate.
+ CompactionEligibility evaluatedCandidate = new Evaluator(proposedCompaction, config, fingerprintMapper).evaluate();
+ switch (Objects.requireNonNull(evaluatedCandidate).getState()) {
+ case NOT_APPLICABLE:
+ case NOT_ELIGIBLE:
+ return evaluatedCandidate;
+ case FULL_COMPACTION: // evaluator has decided compaction is needed, policy needs to further check
+ return searchPolicy.checkEligibilityForCompaction(proposedCompaction, evaluatedCandidate);
+ case INCREMENTAL_COMPACTION: // evaluator cant decide when to perform an incremental compaction
+ default:
+ throw DruidException.defensive("Unexpected eligibility[%s]", evaluatedCandidate);
+ }
+ }
+
+ @Override
+ public boolean equals(Object object)
+ {
+ if (this == object) {
+ return true;
+ }
+ if (object == null || getClass() != object.getClass()) {
+ return false;
+ }
+ CompactionEligibility that = (CompactionEligibility) object;
+ return state == that.state
+ && Objects.equals(reason, that.reason)
+ && Objects.equals(compacted, that.compacted)
+ && Objects.equals(uncompacted, that.uncompacted)
+ && Objects.equals(uncompactedSegments, that.uncompactedSegments);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(state, reason, compacted, uncompacted, uncompactedSegments);
+ }
+
+ @Override
+ public String toString()
+ {
+ return "CompactionEligibility{"
+ + "state=" + state
+ + ", reason='" + reason + '\''
+ + ", compacted=" + compacted
+ + ", uncompacted=" + uncompacted
+ + ", uncompactedSegments=" + uncompactedSegments
+ + '}';
+ }
+
+ /**
+ * List of checks performed to determine if compaction is already complete based on indexing state fingerprints.
+ */
+ static final List<Function<Evaluator, String>> FINGERPRINT_CHECKS = List.of(
+ Evaluator::allFingerprintedCandidatesHaveExpectedFingerprint
+ );
+
+ /**
+ * List of checks performed to determine if compaction is already complete.
+ * <p>
+ * The order of the checks must be honored while evaluating them.
+ */
+ static final List<Function<Evaluator, String>> CHECKS = Arrays.asList(
+ Evaluator::partitionsSpecIsUpToDate,
+ Evaluator::indexSpecIsUpToDate,
+ Evaluator::segmentGranularityIsUpToDate,
+ Evaluator::queryGranularityIsUpToDate,
+ Evaluator::rollupIsUpToDate,
+ Evaluator::dimensionsSpecIsUpToDate,
+ Evaluator::metricsSpecIsUpToDate,
+ Evaluator::transformSpecFilterIsUpToDate,
+ Evaluator::projectionsAreUpToDate
+ );
+
+ /**
+ * Evaluates checks to determine the compaction status of a
+ * {@link CompactionCandidate}.
+ */
+ private static class Evaluator
+ {
+ private static final Logger log = new Logger(Evaluator.class);
+
+ private final DataSourceCompactionConfig compactionConfig;
+ private final CompactionCandidate.ProposedCompaction proposedCompaction;
+ private final ClientCompactionTaskQueryTuningConfig tuningConfig;
+ private final UserCompactionTaskGranularityConfig configuredGranularitySpec;
+
+ private final List<DataSegment> fingerprintedSegments = new ArrayList<>();
+ private final List<DataSegment> compactedSegments = new ArrayList<>();
+ private final List<DataSegment> uncompactedSegments = new ArrayList<>();
+ private final Map<CompactionState, List<DataSegment>> unknownStateToSegments = new HashMap<>();
+
+ @Nullable
+ private final IndexingStateFingerprintMapper fingerprintMapper;
+ @Nullable
+ private final String targetFingerprint;
+
+ private Evaluator(
+ CompactionCandidate.ProposedCompaction proposedCompaction,
+ DataSourceCompactionConfig compactionConfig,
+ @Nullable IndexingStateFingerprintMapper fingerprintMapper
+ )
+ {
+ this.proposedCompaction = proposedCompaction;
+ this.compactionConfig = compactionConfig;
+ this.tuningConfig = ClientCompactionTaskQueryTuningConfig.from(compactionConfig);
+ this.configuredGranularitySpec = compactionConfig.getGranularitySpec();
+ this.fingerprintMapper = fingerprintMapper;
+ if (fingerprintMapper == null) {
+ targetFingerprint = null;
+ } else {
+ targetFingerprint = fingerprintMapper.generateFingerprint(
+ compactionConfig.getDataSource(),
+ compactionConfig.toCompactionState()
+ );
+ }
+ }
+
+ /**
+ * Evaluates the compaction status of candidate segments through a multi-step process:
+ * <ol>
+ * <li>Validates input bytes are within limits</li>
+ * <li>Categorizes segments by compaction state (fingerprinted, uncompacted, or unknown)</li>
+ * <li>Performs fingerprint-based validation if available (fast path)</li>
+ * <li>Runs detailed checks against unknown states via {@link CompactionEligibility#CHECKS}</li>
+ * </ol>
+ *
+ * @return Pair of eligibility status and compaction status with reason for first failed check
+ */
+ private CompactionEligibility evaluate()
+ {
+ final String inputBytesCheck = inputBytesAreWithinLimit();
+ if (inputBytesCheck != null) {
+ return CompactionEligibility.fail(inputBytesCheck);
+ }
+
+ List<String> reasonsForCompaction = new ArrayList<>();
+ String compactedOnceCheck = segmentsHaveBeenCompactedAtLeastOnce();
+ if (compactedOnceCheck != null) {
+ reasonsForCompaction.add(compactedOnceCheck);
+ }
+
+ if (fingerprintMapper != null && targetFingerprint != null) {
+ // First try fingerprint-based evaluation (fast path)
+ FINGERPRINT_CHECKS.stream()
+ .map(f -> f.apply(this))
+ .filter(Objects::nonNull)
+ .findFirst()
+ .ifPresent(reasonsForCompaction::add);
+
+ }
+
+ if (!unknownStateToSegments.isEmpty()) {
+ // Run CHECKS against any states with uknown compaction status
+ reasonsForCompaction.addAll(
+ CHECKS.stream()
+ .map(f -> f.apply(this))
+ .filter(Objects::nonNull)
+ .collect(Collectors.toList())
+ );
+
+ // Any segments left in unknownStateToSegments passed all checks and are considered compacted
+ compactedSegments.addAll(
+ unknownStateToSegments
+ .values()
+ .stream()
+ .flatMap(List::stream)
+ .collect(Collectors.toList())
+ );
+ }
+
+ if (reasonsForCompaction.isEmpty()) {
+ return CompactionEligibility.NOT_APPLICABLE;
+ } else {
+ return builder(State.FULL_COMPACTION, reasonsForCompaction.get(0)).compacted(createStats(compactedSegments))
+ .uncompacted(createStats(uncompactedSegments))
+ .uncompactedSegments(uncompactedSegments)
+ .build();
+ }
+ }
+
+ /**
+ * Evaluates the fingerprints of all fingerprinted candidate segments against the expected fingerprint.
+ * <p>
+ * If all fingerprinted segments have the expected fingerprint, the check can quickly pass as COMPLETE. However,
+ * if any fingerprinted segment has a mismatched fingerprint, we need to investigate further by adding them to
+ * {@link #unknownStateToSegments} where their indexing states will be analyzed.
+ * </p>
+ */
+ private String allFingerprintedCandidatesHaveExpectedFingerprint()
+ {
+ Map<String, List<DataSegment>> mismatchedFingerprintToSegmentMap = new HashMap<>();
+ for (DataSegment segment : fingerprintedSegments) {
+ String fingerprint = segment.getIndexingStateFingerprint();
+ if (fingerprint == null) {
+ // Should not happen since we are iterating over fingerprintedSegments
+ } else if (fingerprint.equals(targetFingerprint)) {
+ compactedSegments.add(segment);
+ } else {
+ mismatchedFingerprintToSegmentMap
+ .computeIfAbsent(fingerprint, k -> new ArrayList<>())
+ .add(segment);
+ }
+ }
+
+ if (mismatchedFingerprintToSegmentMap.isEmpty()) {
+ // All fingerprinted segments have the expected fingerprint - compaction is complete
+ return null;
+ }
+
+ if (fingerprintMapper == null) {
+ // Cannot evaluate further without a fingerprint mapper
+ uncompactedSegments.addAll(
+ mismatchedFingerprintToSegmentMap.values()
+ .stream()
+ .flatMap(List::stream)
+ .collect(Collectors.toList())
+ );
+ return "Segments have a mismatched fingerprint and no fingerprint mapper is available";
+ }
+
+ boolean fingerprintedSegmentWithoutCachedStateFound = false;
+
+ for (Map.Entry<String, List<DataSegment>> e : mismatchedFingerprintToSegmentMap.entrySet()) {
+ String fingerprint = e.getKey();
+ CompactionState stateToValidate = fingerprintMapper.getStateForFingerprint(fingerprint).orElse(null);
+ if (stateToValidate == null) {
+ log.warn("No indexing state found for fingerprint[%s]", fingerprint);
+ fingerprintedSegmentWithoutCachedStateFound = true;
+ uncompactedSegments.addAll(e.getValue());
+ } else {
+ // Note that this does not mean we need compaction yet - we need to validate the state further to determine this
+ unknownStateToSegments.compute(
+ stateToValidate,
+ (state, segments) -> {
+ if (segments == null) {
+ segments = new ArrayList<>();
+ }
+ segments.addAll(e.getValue());
+ return segments;
+ }
+ );
+ }
+ }
+
+ if (fingerprintedSegmentWithoutCachedStateFound) {
+ return "One or more fingerprinted segments do not have a cached indexing state";
+ } else {
+ return null;
+ }
+ }
+
+ /**
+ * Checks if all the segments have been compacted at least once and groups them into uncompacted, fingerprinted, or
+ * non-fingerprinted.
+ */
+ private String segmentsHaveBeenCompactedAtLeastOnce()
+ {
+ for (DataSegment segment : proposedCompaction.getSegments()) {
+ final String fingerprint = segment.getIndexingStateFingerprint();
+ final CompactionState segmentState = segment.getLastCompactionState();
+ if (fingerprint != null) {
+ fingerprintedSegments.add(segment);
+ } else if (segmentState == null) {
+ uncompactedSegments.add(segment);
+ } else {
+ unknownStateToSegments.computeIfAbsent(segmentState, s -> new ArrayList<>()).add(segment);
+ }
+ }
+
+ if (uncompactedSegments.isEmpty()) {
+ return null;
+ } else {
+ return "not compacted yet";
+ }
+ }
+
+ private String partitionsSpecIsUpToDate()
+ {
+ return evaluateForAllCompactionStates(this::partitionsSpecIsUpToDate);
+ }
+
+ private String indexSpecIsUpToDate()
+ {
+ return evaluateForAllCompactionStates(this::indexSpecIsUpToDate);
+ }
+
+ private String projectionsAreUpToDate()
+ {
+ return evaluateForAllCompactionStates(this::projectionsAreUpToDate);
+ }
+
+ private String segmentGranularityIsUpToDate()
+ {
+ return evaluateForAllCompactionStates(this::segmentGranularityIsUpToDate);
+ }
+
+ private String rollupIsUpToDate()
+ {
+ return evaluateForAllCompactionStates(this::rollupIsUpToDate);
+ }
+
+ private String queryGranularityIsUpToDate()
+ {
+ return evaluateForAllCompactionStates(this::queryGranularityIsUpToDate);
+ }
+
+ private String dimensionsSpecIsUpToDate()
+ {
+ return evaluateForAllCompactionStates(this::dimensionsSpecIsUpToDate);
+ }
+
+ private String metricsSpecIsUpToDate()
+ {
+ return evaluateForAllCompactionStates(this::metricsSpecIsUpToDate);
+ }
+
+ private String transformSpecFilterIsUpToDate()
+ {
+ return evaluateForAllCompactionStates(this::transformSpecFilterIsUpToDate);
+ }
+
+ private String partitionsSpecIsUpToDate(CompactionState lastCompactionState)
+ {
+ PartitionsSpec existingPartionsSpec = lastCompactionState.getPartitionsSpec();
+ if (existingPartionsSpec instanceof DimensionRangePartitionsSpec) {
+ existingPartionsSpec = getEffectiveRangePartitionsSpec((DimensionRangePartitionsSpec) existingPartionsSpec);
+ } else if (existingPartionsSpec instanceof DynamicPartitionsSpec) {
+ existingPartionsSpec = new DynamicPartitionsSpec(
+ existingPartionsSpec.getMaxRowsPerSegment(),
+ ((DynamicPartitionsSpec) existingPartionsSpec).getMaxTotalRowsOr(Long.MAX_VALUE)
+ );
+ }
+ return completeIfNullOrEqual(
+ "partitionsSpec",
+ findPartitionsSpecFromConfig(tuningConfig),
+ existingPartionsSpec,
+ CompactionEligibility::asString
+ );
+ }
+
+ private String indexSpecIsUpToDate(CompactionState lastCompactionState)
+ {
+ return completeIfNullOrEqual(
+ "indexSpec",
+ Configs.valueOrDefault(tuningConfig.getIndexSpec(), IndexSpec.getDefault()).getEffectiveSpec(),
+ lastCompactionState.getIndexSpec().getEffectiveSpec(),
+ String::valueOf
+ );
+ }
+
+ private String projectionsAreUpToDate(CompactionState lastCompactionState)
+ {
+ return completeIfNullOrEqual(
+ "projections",
+ compactionConfig.getProjections(),
+ lastCompactionState.getProjections(),
+ String::valueOf
+ );
+ }
+
+ @Nullable
+ private String inputBytesAreWithinLimit()
+ {
+ final long inputSegmentSize = compactionConfig.getInputSegmentSizeBytes();
+ if (proposedCompaction.getTotalBytes() > inputSegmentSize) {
+ return StringUtils.format(
+ "'inputSegmentSize' exceeded: Total segment size[%d] is larger than allowed inputSegmentSize[%d]",
+ proposedCompaction.getTotalBytes(), inputSegmentSize
+ );
+ }
+ return null;
+ }
+
+ private String segmentGranularityIsUpToDate(CompactionState lastCompactionState)
+ {
+ if (configuredGranularitySpec == null
+ || configuredGranularitySpec.getSegmentGranularity() == null) {
+ return null;
+ }
+
+ final Granularity configuredSegmentGranularity = configuredGranularitySpec.getSegmentGranularity();
+ final UserCompactionTaskGranularityConfig existingGranularitySpec = getGranularitySpec(lastCompactionState);
+ final Granularity existingSegmentGranularity
+ = existingGranularitySpec == null ? null : existingGranularitySpec.getSegmentGranularity();
+
+ if (configuredSegmentGranularity.equals(existingSegmentGranularity)) {
+ return null;
+ } else if (existingSegmentGranularity == null) {
+ // Candidate segments were compacted without segment granularity specified
+ // Check if the segments already have the desired segment granularity
+ final List<DataSegment> segmentsForState = unknownStateToSegments.get(lastCompactionState);
+ boolean needsCompaction = segmentsForState.stream().anyMatch(
+ segment -> !configuredSegmentGranularity.isAligned(segment.getInterval())
+ );
+ if (needsCompaction) {
+ return StringUtils.format(
+ "segmentGranularity: segments do not align with target[%s]",
+ CompactionEligibility.asString(configuredSegmentGranularity)
+ );
+ }
+ } else {
+ return configChanged(
+ "segmentGranularity",
+ configuredSegmentGranularity,
+ existingSegmentGranularity,
+ CompactionEligibility::asString
+ );
+ }
+
+ return null;
+ }
+
+ private String rollupIsUpToDate(CompactionState lastCompactionState)
+ {
+ if (configuredGranularitySpec == null) {
+ return null;
+ } else {
+ final UserCompactionTaskGranularityConfig existingGranularitySpec
+ = getGranularitySpec(lastCompactionState);
+ return completeIfNullOrEqual(
+ "rollup",
+ configuredGranularitySpec.isRollup(),
+ existingGranularitySpec == null ? null : existingGranularitySpec.isRollup(),
+ String::valueOf
+ );
+ }
+ }
+
+ private String queryGranularityIsUpToDate(CompactionState lastCompactionState)
+ {
+ if (configuredGranularitySpec == null) {
+ return null;
+ } else {
+ final UserCompactionTaskGranularityConfig existingGranularitySpec
+ = getGranularitySpec(lastCompactionState);
+ return completeIfNullOrEqual(
+ "queryGranularity",
+ configuredGranularitySpec.getQueryGranularity(),
+ existingGranularitySpec == null ? null : existingGranularitySpec.getQueryGranularity(),
+ CompactionEligibility::asString
+ );
+ }
+ }
+
+ /**
+ * Removes partition dimensions before comparison, since they are placed in front of the sort order --
+ * which can create a mismatch between expected and actual order of dimensions. Partition dimensions are separately
+ * covered in {@link Evaluator#partitionsSpecIsUpToDate()} check.
+ */
+ private String dimensionsSpecIsUpToDate(CompactionState lastCompactionState)
+ {
+ if (compactionConfig.getDimensionsSpec() == null) {
+ return null;
+ } else {
+ List<DimensionSchema> existingDimensions = getNonPartitioningDimensions(
+ lastCompactionState.getDimensionsSpec() == null
+ ? null
+ : lastCompactionState.getDimensionsSpec().getDimensions(),
+ lastCompactionState.getPartitionsSpec(),
+ lastCompactionState.getIndexSpec()
+ );
+ List<DimensionSchema> configuredDimensions = getNonPartitioningDimensions(
+ compactionConfig.getDimensionsSpec().getDimensions(),
+ compactionConfig.getTuningConfig() == null ? null : compactionConfig.getTuningConfig().getPartitionsSpec(),
+ compactionConfig.getTuningConfig() == null
+ ? IndexSpec.getDefault()
+ : compactionConfig.getTuningConfig().getIndexSpec()
+ );
+ return completeIfNullOrEqual(
+ "dimensionsSpec",
+ configuredDimensions,
+ existingDimensions,
+ String::valueOf
+ );
+ }
+ }
+
+ private String metricsSpecIsUpToDate(CompactionState lastCompactionState)
+ {
+ final AggregatorFactory[] configuredMetricsSpec = compactionConfig.getMetricsSpec();
+ if (ArrayUtils.isEmpty(configuredMetricsSpec)) {
+ return null;
+ }
+
+ final List<AggregatorFactory> metricSpecList = lastCompactionState.getMetricsSpec();
+ final AggregatorFactory[] existingMetricsSpec
+ = CollectionUtils.isNullOrEmpty(metricSpecList)
+ ? null : metricSpecList.toArray(new AggregatorFactory[0]);
+
+ if (existingMetricsSpec == null || !Arrays.deepEquals(configuredMetricsSpec, existingMetricsSpec)) {
+ return configChanged(
+ "metricsSpec",
+ configuredMetricsSpec,
+ existingMetricsSpec,
+ Arrays::toString
+ );
+ } else {
+ return null;
+ }
+ }
+
+ private String transformSpecFilterIsUpToDate(CompactionState lastCompactionState)
+ {
+ if (compactionConfig.getTransformSpec() == null) {
+ return null;
+ }
+
+ CompactionTransformSpec existingTransformSpec = lastCompactionState.getTransformSpec();
+ return completeIfNullOrEqual(
+ "transformSpec filter",
+ compactionConfig.getTransformSpec().getFilter(),
+ existingTransformSpec == null ? null : existingTransformSpec.getFilter(),
+ String::valueOf
+ );
+ }
+
+ /**
+ * Evaluates the given check for each entry in the {@link #unknownStateToSegments}.
+ * If any entry fails the given check by returning a status which is not
+ * COMPLETE, all the segments with that state are moved to {@link #uncompactedSegments}.
+ *
+ * @return The first status which is not COMPLETE.
+ */
+ private String evaluateForAllCompactionStates(Function<CompactionState, String> check)
+ {
+ String firstIncomplete = null;
+ for (CompactionState state : List.copyOf(unknownStateToSegments.keySet())) {
+ final String eligibleReason = check.apply(state);
+ if (eligibleReason != null) {
+ uncompactedSegments.addAll(unknownStateToSegments.remove(state));
+ if (firstIncomplete == null) {
+ firstIncomplete = eligibleReason;
+ }
+ }
+ }
+
+ return firstIncomplete;
+ }
+
+ private static UserCompactionTaskGranularityConfig getGranularitySpec(
+ CompactionState compactionState
+ )
+ {
+ return UserCompactionTaskGranularityConfig.from(compactionState.getGranularitySpec());
+ }
+
+ private static CompactionStatistics createStats(List<DataSegment> segments)
+ {
+ final Set<Interval> segmentIntervals =
+ segments.stream().map(DataSegment::getInterval).collect(Collectors.toSet());
+ final long totalBytes = segments.stream().mapToLong(DataSegment::getSize).sum();
+ return CompactionStatistics.create(totalBytes, segments.size(), segmentIntervals.size());
+ }
+ }
+
+
+ /**
+ * Computes compaction status for the given field. The status is assumed to be
+ * COMPLETE (i.e. no further compaction is required) if the configured value
+ * of the field is null or equal to the current value.
+ */
+ private static <T> String completeIfNullOrEqual(
+ String field,
+ T configured,
+ T current,
+ Function<T, String> stringFunction
+ )
+ {
+ if (configured == null || configured.equals(current)) {
+ return null;
+ } else {
+ return configChanged(field, configured, current, stringFunction);
+ }
+ }
+
+ private static <T> String configChanged(
+ String field,
+ T target,
+ T current,
+ Function<T, String> stringFunction
+ )
+ {
+ return StringUtils.format(
+ "'%s' mismatch: required[%s], current[%s]",
+ field,
+ target == null ? null : stringFunction.apply(target),
+ current == null ? null : stringFunction.apply(current)
+ );
+ }
+
+ private static String asString(Granularity granularity)
+ {
+ if (granularity == null) {
+ return null;
+ }
+ for (GranularityType type : GranularityType.values()) {
+ if (type.getDefaultGranularity().equals(granularity)) {
+ return type.toString();
+ }
+ }
+ return granularity.toString();
+ }
+
+ private static String asString(PartitionsSpec partitionsSpec)
+ {
+ if (partitionsSpec instanceof DimensionRangePartitionsSpec) {
+ DimensionRangePartitionsSpec rangeSpec = (DimensionRangePartitionsSpec) partitionsSpec;
+ return StringUtils.format(
+ "'range' on %s with %,d rows",
+ rangeSpec.getPartitionDimensions(), rangeSpec.getTargetRowsPerSegment()
+ );
+ } else if (partitionsSpec instanceof HashedPartitionsSpec) {
+ HashedPartitionsSpec hashedSpec = (HashedPartitionsSpec) partitionsSpec;
+ return StringUtils.format(
+ "'hashed' on %s with %,d rows",
+ hashedSpec.getPartitionDimensions(), hashedSpec.getTargetRowsPerSegment()
+ );
+ } else if (partitionsSpec instanceof DynamicPartitionsSpec) {
+ DynamicPartitionsSpec dynamicSpec = (DynamicPartitionsSpec) partitionsSpec;
+ return StringUtils.format(
+ "'dynamic' with %,d rows",
+ dynamicSpec.getMaxRowsPerSegment()
+ );
+ } else {
+ return partitionsSpec.toString();
+ }
+ }
+
+ @Nullable
+ public static PartitionsSpec findPartitionsSpecFromConfig(ClientCompactionTaskQueryTuningConfig tuningConfig)
+ {
+ final PartitionsSpec partitionsSpecFromTuningConfig = tuningConfig.getPartitionsSpec();
+ if (partitionsSpecFromTuningConfig == null) {
+ final Long maxTotalRows = tuningConfig.getMaxTotalRows();
Review Comment:
## Deprecated method or constructor invocation
Invoking [ClientCompactionTaskQueryTuningConfig.getMaxTotalRows](1) should be avoided because it has been deprecated.
[Show more details](https://github.com/apache/druid/security/code-scanning/10805)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] introduce clear separation between proposed compaction, eligibility status, and compaction mode, along with improved task state tracking (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18968:
URL: https://github.com/apache/druid/pull/18968#discussion_r2795244365
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionStatusTracker.java:
##########
@@ -80,43 +80,48 @@ public Set<String> getSubmittedTaskIds()
* This method assumes that the given candidate is eligible for compaction
* based on the current compaction config/supervisor of the datasource.
*/
- public CompactionStatus computeCompactionStatus(
Review Comment:
The preferable approach here is to completely remove this method and use the existing method `getLatestTaskStatus()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Add compaction eligibility concept and incremental compaction support (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18968:
URL: https://github.com/apache/druid/pull/18968#discussion_r2763298885
##########
server/src/main/java/org/apache/druid/server/compaction/DataSourceCompactibleSegmentIterator.java:
##########
@@ -329,17 +331,17 @@ private void findAndEnqueueSegmentsToCompact(CompactibleSegmentIterator compacti
continue;
}
- final CompactionCandidate candidates = CompactionCandidate.from(segments, config.getSegmentGranularity());
- final CompactionStatus compactionStatus = CompactionStatus.compute(candidates, config, fingerprintMapper);
- final CompactionCandidate candidatesWithStatus = candidates.withCurrentStatus(compactionStatus);
+ final CompactionCandidate candidatesWithStatus =
Review Comment:
I feel this class should remain untouched since we are only computing the `CompactionStatus` here. The eligiblity computation should happen later in the `CompactionJobQueue` only.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Add compaction eligibility concept and incremental compaction support (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on PR #18968:
URL: https://github.com/apache/druid/pull/18968#issuecomment-3846703954
> Currently, CompactionStatus conflates two concerns:
Status: What's the task status of a CompactionCandidate?
Eligibility: Should we compact this interval now, and how?
@cecemei , just so that we are on the same page, the `CompactionStatus` does not exactly contain information of the compaction "task" (except for the RUNNING state, which we can now remove in this PR). It contains info of the current state of compaction in the interval, i.e. number of compacted segments, number of uncompacted segments and overall state (COMPLETE/PENDING/SKIPPED).
It would be nice to get rid of the SKIPPED state too since it is more of an eligibility thing, but I guess we can leave it for now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Add compaction eligibility concept and incremental compaction support (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18968:
URL: https://github.com/apache/druid/pull/18968#discussion_r2772501203
##########
server/src/main/java/org/apache/druid/server/compaction/MostFragmentedIntervalFirstPolicy.java:
##########
@@ -47,13 +47,16 @@ public class MostFragmentedIntervalFirstPolicy extends BaseCandidateSearchPolicy
private final int minUncompactedCount;
private final HumanReadableBytes minUncompactedBytes;
private final HumanReadableBytes maxAverageUncompactedBytesPerSegment;
+ private final double incrementalCompactionUncompactedBytesRatioThreshold;
@JsonCreator
public MostFragmentedIntervalFirstPolicy(
@JsonProperty("minUncompactedCount") @Nullable Integer minUncompactedCount,
@JsonProperty("minUncompactedBytes") @Nullable HumanReadableBytes minUncompactedBytes,
@JsonProperty("maxAverageUncompactedBytesPerSegment") @Nullable
HumanReadableBytes maxAverageUncompactedBytesPerSegment,
+ @JsonProperty("incrementalCompactionUncompactedBytesRatioThreshold") @Nullable
Review Comment:
Yes, it is best to add this config in the same PR that adds the incremental compaction feature.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Add compaction eligibility concept and incremental compaction support (druid)
Posted by "capistrant (via GitHub)" <gi...@apache.org>.
capistrant commented on code in PR #18968:
URL: https://github.com/apache/druid/pull/18968#discussion_r2774705747
##########
server/src/main/java/org/apache/druid/server/compaction/DataSourceCompactibleSegmentIterator.java:
##########
@@ -329,17 +331,17 @@ private void findAndEnqueueSegmentsToCompact(CompactibleSegmentIterator compacti
continue;
}
- final CompactionCandidate candidates = CompactionCandidate.from(segments, config.getSegmentGranularity());
- final CompactionStatus compactionStatus = CompactionStatus.compute(candidates, config, fingerprintMapper);
- final CompactionCandidate candidatesWithStatus = candidates.withCurrentStatus(compactionStatus);
+ final CompactionCandidate candidatesWithStatus =
Review Comment:
For reference, Cece. Changes I have been working on have also been only adding functionality to the supervisor. The IndexingState fingerprinting that compaction uses is only available for supervisors and rule based compaction/reindexing will also be supervisor only. Maybe it is time we officially do a separate PR and mark this duty as deprecated and update docs as such since so much is happening in this space these days
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionEligibility.java:
##########
@@ -0,0 +1,932 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.compaction;
+
+import com.google.common.base.Strings;
+import org.apache.commons.lang3.ArrayUtils;
+import org.apache.druid.client.indexing.ClientCompactionTaskQueryTuningConfig;
+import org.apache.druid.common.config.Configs;
+import org.apache.druid.data.input.impl.DimensionSchema;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.error.InvalidInput;
+import org.apache.druid.indexer.partitions.DimensionRangePartitionsSpec;
+import org.apache.druid.indexer.partitions.DynamicPartitionsSpec;
+import org.apache.druid.indexer.partitions.HashedPartitionsSpec;
+import org.apache.druid.indexer.partitions.PartitionsSpec;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.granularity.Granularity;
+import org.apache.druid.java.util.common.granularity.GranularityType;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.query.aggregation.AggregatorFactory;
+import org.apache.druid.segment.IndexSpec;
+import org.apache.druid.segment.metadata.IndexingStateFingerprintMapper;
+import org.apache.druid.segment.transform.CompactionTransformSpec;
+import org.apache.druid.server.coordinator.DataSourceCompactionConfig;
+import org.apache.druid.server.coordinator.UserCompactionTaskGranularityConfig;
+import org.apache.druid.timeline.CompactionState;
+import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.utils.CollectionUtils;
+import org.joda.time.Interval;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/**
+ * Describes the eligibility of an interval for compaction.
+ */
+public class CompactionEligibility
+{
+ public enum State
+ {
+ FULL_COMPACTION,
+ INCREMENTAL_COMPACTION,
+ NOT_ELIGIBLE,
+ NOT_APPLICABLE
+ }
+
+ static class CompactionEligibilityBuilder
+ {
+ private State state;
+ private CompactionStatistics compacted;
+ private CompactionStatistics uncompacted;
+ private List<DataSegment> uncompactedSegments;
+ private String reason;
+
+ CompactionEligibilityBuilder(State state, String reason)
+ {
+ this.state = state;
+ this.reason = reason;
+ }
+
+ CompactionEligibilityBuilder compacted(CompactionStatistics compacted)
+ {
+ this.compacted = compacted;
+ return this;
+ }
+
+ CompactionEligibilityBuilder uncompacted(CompactionStatistics uncompacted)
+ {
+ this.uncompacted = uncompacted;
+ return this;
+ }
+
+ CompactionEligibilityBuilder uncompactedSegments(List<DataSegment> uncompactedSegments)
+ {
+ this.uncompactedSegments = uncompactedSegments;
+ return this;
+ }
+
+ CompactionEligibility build()
+ {
+ return new CompactionEligibility(state, reason, compacted, uncompacted, uncompactedSegments);
+ }
+ }
+
+ public static final CompactionEligibility NOT_APPLICABLE = builder(State.NOT_APPLICABLE, "").build();
+
+ public static CompactionEligibility fail(String messageFormat, Object... args)
+ {
+ return builder(State.NOT_ELIGIBLE, StringUtils.format(messageFormat, args)).build();
+ }
+
+ private final State state;
+ private final String reason;
+
+ @Nullable
+ private final CompactionStatistics compacted;
+ @Nullable
+ private final CompactionStatistics uncompacted;
+ @Nullable
+ private final List<DataSegment> uncompactedSegments;
+
+ private CompactionEligibility(
+ State state,
+ String reason,
+ @Nullable CompactionStatistics compacted,
+ @Nullable CompactionStatistics uncompacted,
+ @Nullable List<DataSegment> uncompactedSegments
+ )
+ {
+ this.state = state;
+ this.reason = reason;
+ switch (state) {
+ case NOT_APPLICABLE:
+ break;
+ case NOT_ELIGIBLE:
+ InvalidInput.conditionalException(!Strings.isNullOrEmpty(reason), "must provide a reason");
+ break;
+ case FULL_COMPACTION:
+ case INCREMENTAL_COMPACTION:
+ InvalidInput.conditionalException(compacted != null, "must provide compacted stats");
+ InvalidInput.conditionalException(uncompacted != null, "must provide uncompacted stats");
+ InvalidInput.conditionalException(uncompactedSegments != null, "must provide uncompactedSegments");
+ break;
+ default:
+ throw DruidException.defensive("unexpected eligibility state[%s]", state);
+ }
+ this.compacted = compacted;
+ this.uncompacted = uncompacted;
+ this.uncompactedSegments = uncompactedSegments;
+ }
+
+ static CompactionEligibilityBuilder builder(State state, String reason)
+ {
+ return new CompactionEligibilityBuilder(state, reason);
+ }
+
+ public State getState()
+ {
+ return state;
+ }
+
+ public String getReason()
+ {
+ return reason;
+ }
+
+ @Nullable
+ public CompactionStatistics getUncompactedStats()
+ {
+ return uncompacted;
+ }
+
+ @Nullable
+ public CompactionStatistics getCompactedStats()
+ {
+ return compacted;
+ }
+
+ @Nullable
+ public List<DataSegment> getUncompactedSegments()
+ {
+ return uncompactedSegments;
+ }
+
+ public CompactionCandidate createCandidate(CompactionCandidate.ProposedCompaction proposedCompaction)
+ {
+ switch (state) {
+ case NOT_APPLICABLE:
+ return new CompactionCandidate(proposedCompaction, this, CompactionStatus.COMPLETE);
+ case NOT_ELIGIBLE:
+ return new CompactionCandidate(proposedCompaction, this, CompactionStatus.skipped(reason));
+ case FULL_COMPACTION:
+ return new CompactionCandidate(
+ proposedCompaction,
+ this,
+ CompactionStatus.pending(reason)
+ );
+ case INCREMENTAL_COMPACTION:
+ CompactionCandidate.ProposedCompaction newProposed = new CompactionCandidate.ProposedCompaction(
+ uncompactedSegments,
+ proposedCompaction.getUmbrellaInterval(),
+ proposedCompaction.getCompactionInterval(),
+ Math.toIntExact(uncompactedSegments.stream().map(DataSegment::getInterval).distinct().count())
+ );
+ return new CompactionCandidate(newProposed, this, CompactionStatus.pending(reason));
+ default:
+ throw DruidException.defensive("Unexpected eligibility state[%s]", state);
+ }
+ }
+
+ /**
+ * Evaluates a compaction candidate to determine its eligibility and compaction status.
+ * <p>
+ * This method performs a two-stage evaluation:
+ * <ol>
+ * <li>First, uses {@link Evaluator} to check if the candidate needs compaction
Review Comment:
any changes to the Evaluator logic in moving it to this file?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] introduce clear separation between proposed compaction, eligibility status, and compaction mode, along with improved task state tracking (druid)
Posted by "capistrant (via GitHub)" <gi...@apache.org>.
capistrant commented on code in PR #18968:
URL: https://github.com/apache/druid/pull/18968#discussion_r2788628771
##########
server/src/main/java/org/apache/druid/server/coordinator/duty/CompactSegments.java:
##########
@@ -371,8 +369,11 @@ public static ClientCompactionTaskQuery createCompactionTask(
}
final Map<String, Object> autoCompactionContext = newAutoCompactionContext(config.getTaskContext());
- if (candidate.getCurrentStatus() != null) {
- autoCompactionContext.put(COMPACTION_REASON_KEY, candidate.getCurrentStatus().getReason());
+ if (CompactionMode.NOT_APPLICABLE.equals(candidate.getMode())) {
Review Comment:
Is this supposed to be `if (!CompactionMode.NOT_APPLICABLE.equals(candidate.getMode()))`?
Seems to me like for compaction jobs the reason will be missing now, and for NOT_APPLICABLE a task wouldn't even be started anyways, as an exception would get thrown by `compactSegments`
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionMode.java:
##########
Review Comment:
I think some javadocs in this class would be helpful. A class level one + at least a brief sentence of each mode, even if they are a bit self explanatory
##########
server/src/test/java/org/apache/druid/server/coordinator/duty/CompactSegmentsTest.java:
##########
@@ -1545,17 +1550,20 @@ private void verifySnapshot(
{
Map<String, AutoCompactionSnapshot> autoCompactionSnapshots = compactSegments.getAutoCompactionSnapshot();
AutoCompactionSnapshot snapshot = autoCompactionSnapshots.get(dataSourceName);
- Assert.assertEquals(dataSourceName, snapshot.getDataSource());
- Assert.assertEquals(scheduleStatus, snapshot.getScheduleStatus());
- Assert.assertEquals(expectedByteCountAwaitingCompaction, snapshot.getBytesAwaitingCompaction());
- Assert.assertEquals(expectedByteCountCompressed, snapshot.getBytesCompacted());
- Assert.assertEquals(expectedByteCountSkipped, snapshot.getBytesSkipped());
- Assert.assertEquals(expectedIntervalCountAwaitingCompaction, snapshot.getIntervalCountAwaitingCompaction());
- Assert.assertEquals(expectedIntervalCountCompressed, snapshot.getIntervalCountCompacted());
- Assert.assertEquals(expectedIntervalCountSkipped, snapshot.getIntervalCountSkipped());
- Assert.assertEquals(expectedSegmentCountAwaitingCompaction, snapshot.getSegmentCountAwaitingCompaction());
- Assert.assertEquals(expectedSegmentCountCompressed, snapshot.getSegmentCountCompacted());
- Assert.assertEquals(expectedSegmentCountSkipped, snapshot.getSegmentCountSkipped());
+ Assert.assertEquals(new AutoCompactionSnapshot(
Review Comment:
will this make it less clear what is causing the difference than having all the individual assertions? your way looks better on the eyes, but I wonder if it makes it more annoying for a dev to identify what exactly they broke if it starts failing?
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionStatus.java:
##########
@@ -346,8 +361,10 @@ static DimensionRangePartitionsSpec getEffectiveRangePartitionsSpec(DimensionRan
*/
private static class Evaluator
Review Comment:
I'm not sure how much I like moving away from a more descriptive return type for the individual checks. It feels like a nitpick since functionally everything is the same. And I guess I understand why you wanted to not be using `CompactionStatus` all over the place in the Evaluator like we were. My only alternative suggestion would be yet another simple record like object or something signifying a check with a result and reason (populated if the check failed). The "return null if check passed" just feels a little unintuitive. Maybe a compromise would be explicitly stating how the checks work in javadoc for the CHECKS and FINGERPRINT_CHECKS lists (or in the evalaute method javadoc)
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionStatus.java:
##########
@@ -357,29 +348,52 @@ private static class Evaluator
private final Map<CompactionState, List<DataSegment>> unknownStateToSegments = new HashMap<>();
@Nullable
- private final String targetFingerprint;
private final IndexingStateFingerprintMapper fingerprintMapper;
+ @Nullable
+ private final String targetFingerprint;
- private Evaluator(
+ Evaluator(
CompactionCandidate candidateSegments,
DataSourceCompactionConfig compactionConfig,
- @Nullable String targetFingerprint,
@Nullable IndexingStateFingerprintMapper fingerprintMapper
)
{
this.candidateSegments = candidateSegments;
this.compactionConfig = compactionConfig;
this.tuningConfig = ClientCompactionTaskQueryTuningConfig.from(compactionConfig);
this.configuredGranularitySpec = compactionConfig.getGranularitySpec();
- this.targetFingerprint = targetFingerprint;
this.fingerprintMapper = fingerprintMapper;
+ if (fingerprintMapper == null) {
+ targetFingerprint = null;
+ } else {
+ targetFingerprint = fingerprintMapper.generateFingerprint(
+ compactionConfig.getDataSource(),
+ compactionConfig.toCompactionState()
+ );
+ }
+ }
+
+ List<DataSegment> getUncompactedSegments()
+ {
+ return uncompactedSegments;
}
- private CompactionStatus evaluate()
+ /**
+ * Evaluates the compaction status of candidate segments through a multi-step process:
+ * <ol>
+ * <li>Validates input bytes are within limits</li>
+ * <li>Categorizes segments by compaction state (fingerprinted, uncompacted, or unknown)</li>
+ * <li>Performs fingerprint-based validation if available (fast path)</li>
+ * <li>Runs detailed checks against unknown states via {@link #CHECKS}</li>
+ * </ol>
+ *
+ * @return Pair of eligibility status and compaction status with reason for first failed check
+ */
+ Pair<CompactionCandidateSearchPolicy.Eligibility, CompactionStatus> evaluate()
{
- final CompactionStatus inputBytesCheck = inputBytesAreWithinLimit();
- if (inputBytesCheck.isSkipped()) {
- return inputBytesCheck;
+ final CompactionCandidateSearchPolicy.Eligibility inputBytesCheck = inputBytesAreWithinLimit();
+ if (inputBytesCheck != null) {
Review Comment:
I guess this conversation is out of date after latest refactorings?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Add compaction eligibility concept and incremental compaction support (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18968:
URL: https://github.com/apache/druid/pull/18968#discussion_r2767211242
##########
server/src/main/java/org/apache/druid/server/coordinator/duty/CompactSegments.java:
##########
@@ -460,14 +462,27 @@ private static ClientCompactionTaskQuery compactSegments(
context.put("priority", compactionTaskPriority);
final String taskId = IdUtils.newTaskId(TASK_ID_PREFIX, ClientCompactionTaskQuery.TYPE, dataSource, null);
+ final ClientCompactionIntervalSpec clientCompactionIntervalSpec;
+ Preconditions.checkArgument(entry.getPolicyEligibility() != null, "Must have a policy eligibility");
+ switch (entry.getPolicyEligibility().getPolicyEligibility()) {
+ case FULL_COMPACTION:
+ clientCompactionIntervalSpec = new ClientCompactionIntervalSpec(entry.getCompactionInterval(), null, null);
+ break;
+ case INCREMENTAL_COMPACTION:
Review Comment:
Please revert this change since we need not support incremental compaction with Coordinator-based compaction duty.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Add compaction eligibility concept and incremental compaction support (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18968:
URL: https://github.com/apache/druid/pull/18968#discussion_r2762596742
##########
server/src/main/java/org/apache/druid/client/indexing/ClientCompactionIntervalSpec.java:
##########
@@ -38,18 +42,37 @@ public class ClientCompactionIntervalSpec
private final Interval interval;
@Nullable
+ private final List<SegmentDescriptor> uncompactedSegments;
+ @Nullable
private final String sha256OfSortedSegmentIds;
@JsonCreator
public ClientCompactionIntervalSpec(
@JsonProperty("interval") Interval interval,
+ @JsonProperty("uncompactedSegments") @Nullable List<SegmentDescriptor> uncompactedSegments,
Review Comment:
It is a little confusing to add segments here since there is already a `SpecificSegmentsSpec` which specifies the set of segments that should be compacted.
I feel we should add a new type of `CompactionInputSpec` which will be used for incremental compaction only. It would have both a non-null interval as well as a non-empty set of segment IDs to compact incrementally. The `CompactionTask` would handle the different input specs accordingly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] introduce clear separation between proposed compaction, eligibility status, and compaction mode, along with improved task state tracking (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18968:
URL: https://github.com/apache/druid/pull/18968#discussion_r2795122751
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionCandidate.java:
##########
@@ -37,76 +39,200 @@
*/
public class CompactionCandidate
{
- private final List<DataSegment> segments;
- private final Interval umbrellaInterval;
- private final Interval compactionInterval;
- private final String dataSource;
- private final long totalBytes;
- private final int numIntervals;
-
- private final CompactionStatus currentStatus;
-
- public static CompactionCandidate from(
- List<DataSegment> segments,
- @Nullable Granularity targetSegmentGranularity
- )
+ /**
+ * Non-empty list of segments of a datasource being proposed for compaction.
+ * A proposed compaction typically contains all the segments of a single time chunk.
+ */
+ public static class ProposedCompaction
Review Comment:
What we need is a way to distinguish between a candidate before and after the status has been computed, so that we can avoid the null checks on `currentStatus`. I don't think this change fully achieves that.
Also, the name `CompactionCandidate` captures the intent better as it defines a potential "candidate" for compaction. So let's stick with it.
Instead of this change, you may consider adding a new class `CompactionCandidateAndStatus` which will be used by both the methods of the search policy. This class will contain only a `candidate` field and a `status` field.
I don't think we will need this class to contain the `Eligibility` or the `CompactionMode` since it will be computed and used only by the `CompactionJobQueue`.
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionMode.java:
##########
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.server.compaction;
+
+import org.apache.druid.error.DruidException;
+import org.apache.druid.java.util.common.StringUtils;
+
+import javax.annotation.Nullable;
+
+public enum CompactionMode
+{
+ FULL_COMPACTION {
+ @Override
+ public CompactionCandidate createCandidate(
+ CompactionCandidate.ProposedCompaction proposedCompaction,
+ CompactionStatus eligibility,
+ @Nullable String policyNote
+ )
+ {
+ return new CompactionCandidate(proposedCompaction, eligibility, policyNote, this);
+ }
+ },
+ NOT_APPLICABLE;
+
+ public CompactionCandidate createCandidate(
+ CompactionCandidate.ProposedCompaction proposedCompaction,
+ CompactionStatus eligibility
+ )
+ {
+ return createCandidate(proposedCompaction, eligibility, null);
+ }
+
+ public CompactionCandidate createCandidate(
Review Comment:
We shouldn't really have this method here. The `CompactionMode` is an attribute that we associate to a `CompactionCandidate`. One should not be responsible for creating the other.
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionCandidateSearchPolicy.java:
##########
@@ -48,74 +45,11 @@ public interface CompactionCandidateSearchPolicy
int compareCandidates(CompactionCandidate candidateA, CompactionCandidate candidateB);
/**
- * Checks if the given {@link CompactionCandidate} is eligible for compaction
- * in the current iteration. A policy may implement this method to skip
- * compacting intervals or segments that do not fulfil some required criteria.
+ * Creates a {@link CompactionCandidate} after applying policy-specific checks to the proposed compaction candidate.
*
- * @return {@link Eligibility#OK} only if eligible.
- */
- Eligibility checkEligibilityForCompaction(
- CompactionCandidate candidate,
- CompactionTaskStatus latestTaskStatus
- );
-
- /**
- * Describes the eligibility of an interval for compaction.
+ * @param candidate the proposed compaction
+ * @param eligibility initial eligibility from compaction config checks
+ * @return final compaction candidate
*/
- class Eligibility
- {
- public static final Eligibility OK = new Eligibility(true, null);
-
- private final boolean eligible;
- private final String reason;
-
- private Eligibility(boolean eligible, String reason)
- {
- this.eligible = eligible;
- this.reason = reason;
- }
-
- public boolean isEligible()
- {
- return eligible;
- }
-
- public String getReason()
- {
- return reason;
- }
-
- public static Eligibility fail(String messageFormat, Object... args)
- {
- return new Eligibility(false, StringUtils.format(messageFormat, args));
- }
-
- @Override
- public boolean equals(Object object)
- {
- if (this == object) {
- return true;
- }
- if (object == null || getClass() != object.getClass()) {
- return false;
- }
- Eligibility that = (Eligibility) object;
- return eligible == that.eligible && Objects.equals(reason, that.reason);
- }
-
- @Override
- public int hashCode()
- {
- return Objects.hash(eligible, reason);
- }
-
- @Override
- public String toString()
- {
- return "Eligibility{" +
- "eligible=" + eligible +
- ", reason='" + reason + '\'' +
- '}';
- }
- }
+ CompactionCandidate createCandidate(CompactionCandidate.ProposedCompaction candidate, CompactionStatus eligibility);
Review Comment:
It doesn't make sense for the policy to be creating candidates since the candidates are identified much earlier by the `DatasourceCompactibleSegmentIterator` itself. The policy can only check the eligibility of a candidate for compaction. (I suppose the name `CompactionCandidateSearchPolicy` might be misleading here, since it is really the `CompactionCandidatePickAndPriorizePolicy` but it can be a mouthful, so we can stick with what we have.)
We should retain the original method here. The caller (`CompactionJobQueue`) should just check the eligibility and then decide whether to launch a job or not. If yes, then whether full or incremental.
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionCandidateSearchPolicy.java:
##########
@@ -48,74 +45,11 @@ public interface CompactionCandidateSearchPolicy
int compareCandidates(CompactionCandidate candidateA, CompactionCandidate candidateB);
/**
- * Checks if the given {@link CompactionCandidate} is eligible for compaction
- * in the current iteration. A policy may implement this method to skip
- * compacting intervals or segments that do not fulfil some required criteria.
+ * Creates a {@link CompactionCandidate} after applying policy-specific checks to the proposed compaction candidate.
*
- * @return {@link Eligibility#OK} only if eligible.
- */
- Eligibility checkEligibilityForCompaction(
- CompactionCandidate candidate,
- CompactionTaskStatus latestTaskStatus
- );
-
- /**
- * Describes the eligibility of an interval for compaction.
+ * @param candidate the proposed compaction
+ * @param eligibility initial eligibility from compaction config checks
+ * @return final compaction candidate
*/
- class Eligibility
Review Comment:
Let's retain the `Eligibility` class since it answers the question:
_"Does the search policy consider this `CompactionCandidate` eligible for compaction?"_
The `Eligibility` class should now also contain an enum field for `CompactionMode`. The rest of the class can remain unchanged.
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionCandidate.java:
##########
@@ -37,76 +39,200 @@
*/
public class CompactionCandidate
{
- private final List<DataSegment> segments;
- private final Interval umbrellaInterval;
- private final Interval compactionInterval;
- private final String dataSource;
- private final long totalBytes;
- private final int numIntervals;
-
- private final CompactionStatus currentStatus;
-
- public static CompactionCandidate from(
- List<DataSegment> segments,
- @Nullable Granularity targetSegmentGranularity
- )
+ /**
+ * Non-empty list of segments of a datasource being proposed for compaction.
+ * A proposed compaction typically contains all the segments of a single time chunk.
+ */
+ public static class ProposedCompaction
{
- if (segments == null || segments.isEmpty()) {
- throw InvalidInput.exception("Segments to compact must be non-empty");
+ private final List<DataSegment> segments;
+ private final Interval umbrellaInterval;
+ private final Interval compactionInterval;
+ private final String dataSource;
+ private final long totalBytes;
+ private final int numIntervals;
+
+ public static ProposedCompaction from(
+ List<DataSegment> segments,
+ @Nullable Granularity targetSegmentGranularity
+ )
+ {
+ if (segments == null || segments.isEmpty()) {
+ throw InvalidInput.exception("Segments to compact must be non-empty");
+ }
+
+ final Set<Interval> segmentIntervals =
+ segments.stream().map(DataSegment::getInterval).collect(Collectors.toSet());
+ final Interval umbrellaInterval = JodaUtils.umbrellaInterval(segmentIntervals);
+ final Interval compactionInterval =
+ targetSegmentGranularity == null
+ ? umbrellaInterval
+ : JodaUtils.umbrellaInterval(targetSegmentGranularity.getIterable(umbrellaInterval));
+
+ return new ProposedCompaction(
+ segments,
+ umbrellaInterval,
+ compactionInterval,
+ segmentIntervals.size()
+ );
}
- final Set<Interval> segmentIntervals =
- segments.stream().map(DataSegment::getInterval).collect(Collectors.toSet());
- final Interval umbrellaInterval = JodaUtils.umbrellaInterval(segmentIntervals);
- final Interval compactionInterval =
- targetSegmentGranularity == null
- ? umbrellaInterval
- : JodaUtils.umbrellaInterval(targetSegmentGranularity.getIterable(umbrellaInterval));
-
- return new CompactionCandidate(
- segments,
- umbrellaInterval,
- compactionInterval,
- segmentIntervals.size(),
- null
- );
+ ProposedCompaction(
+ List<DataSegment> segments,
+ Interval umbrellaInterval,
+ Interval compactionInterval,
+ int numDistinctSegmentIntervals
+ )
+ {
+ this.segments = segments;
+ this.totalBytes = segments.stream().mapToLong(DataSegment::getSize).sum();
+
+ this.umbrellaInterval = umbrellaInterval;
+ this.compactionInterval = compactionInterval;
+
+ this.numIntervals = numDistinctSegmentIntervals;
+ this.dataSource = segments.get(0).getDataSource();
+ }
+
+ /**
+ * @return Non-empty list of segments that make up this proposed compaction.
+ */
+ public List<DataSegment> getSegments()
+ {
+ return segments;
+ }
+
+ public long getTotalBytes()
+ {
+ return totalBytes;
+ }
+
+ public int numSegments()
+ {
+ return segments.size();
+ }
+
+ /**
+ * Umbrella interval of all the segments in this proposed compaction. This typically
+ * corresponds to a single time chunk in the segment timeline.
+ */
+ public Interval getUmbrellaInterval()
+ {
+ return umbrellaInterval;
+ }
+
+ /**
+ * Interval aligned to the target segment granularity used for the compaction
+ * task. This interval completely contains the {@link #umbrellaInterval}.
+ */
+ public Interval getCompactionInterval()
+ {
+ return compactionInterval;
+ }
+
+ public String getDataSource()
+ {
+ return dataSource;
+ }
+
+ public CompactionStatistics getStats()
+ {
+ return CompactionStatistics.create(totalBytes, numSegments(), numIntervals);
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ ProposedCompaction that = (ProposedCompaction) o;
+ return totalBytes == that.totalBytes
+ && numIntervals == that.numIntervals
+ && segments.equals(that.segments)
+ && umbrellaInterval.equals(that.umbrellaInterval)
+ && compactionInterval.equals(that.compactionInterval)
+ && dataSource.equals(that.dataSource);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(segments, umbrellaInterval, compactionInterval, dataSource, totalBytes, numIntervals);
+ }
+
+ @Override
+ public String toString()
+ {
+ return "ProposedCompaction{" +
+ "datasource=" + dataSource +
+ ", umbrellaInterval=" + umbrellaInterval +
+ ", compactionInterval=" + compactionInterval +
+ ", numIntervals=" + numIntervals +
+ ", segments=" + SegmentUtils.commaSeparatedIdentifiers(segments) +
+ ", totalSize=" + totalBytes +
+ '}';
+ }
}
- private CompactionCandidate(
- List<DataSegment> segments,
- Interval umbrellaInterval,
- Interval compactionInterval,
- int numDistinctSegmentIntervals,
- @Nullable CompactionStatus currentStatus
- )
+ /**
+ * Used by {@link CompactionStatusTracker#computeCompactionTaskState(CompactionCandidate)}.
+ * The callsite then determines whether to launch compaction task or not.
+ */
+ public enum TaskState
Review Comment:
Please do not add a new `TaskState`, use `CompactionTaskStatus` instead. If it doesn't contain any info, let's add it there.
##########
indexing-service/src/main/java/org/apache/druid/indexing/compact/CompactionJobQueue.java:
##########
@@ -282,18 +281,17 @@ private boolean startJobIfPendingAndReady(
}
// Check if the job is already running, completed or skipped
- final CompactionStatus compactionStatus = getCurrentStatusForJob(job, policy);
- switch (compactionStatus.getState()) {
- case RUNNING:
+ final CompactionCandidate.TaskState candidateState = getCurrentTaskStateForJob(job);
Review Comment:
Please do not add a new task state. Use the existing `CompactionTaskStatus` class.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] introduce clear separation between proposed compaction, eligibility status, and compaction mode, along with improved task state tracking (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18968:
URL: https://github.com/apache/druid/pull/18968#discussion_r2795235216
##########
server/src/main/java/org/apache/druid/server/compaction/DataSourceCompactibleSegmentIterator.java:
##########
@@ -121,9 +125,11 @@ private void populateQueue(SegmentTimeline timeline, List<Interval> skipInterval
if (!partialEternitySegments.isEmpty()) {
// Do not use the target segment granularity in the CompactionCandidate
// as Granularities.getIterable() will cause OOM due to the above issue
- CompactionCandidate candidatesWithStatus = CompactionCandidate
- .from(partialEternitySegments, null)
- .withCurrentStatus(CompactionStatus.skipped("Segments have partial-eternity intervals"));
+ CompactionCandidate candidatesWithStatus =
Review Comment:
This class probably doesn't need to change in this PR.
The `Eligibility` should be computed and used only by the `CompactionJobQueue` (and maybe `CompactSegments` for backward compatibility).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Add compaction eligibility concept and incremental compaction support (druid)
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #18968:
URL: https://github.com/apache/druid/pull/18968#discussion_r2777438863
##########
server/src/main/java/org/apache/druid/server/compaction/CompactionStatusTracker.java:
##########
@@ -80,43 +80,48 @@
* This method assumes that the given candidate is eligible for compaction
* based on the current compaction config/supervisor of the datasource.
*/
- public CompactionStatus computeCompactionStatus(
- CompactionCandidate candidate,
- CompactionCandidateSearchPolicy searchPolicy
- )
+ public CompactionCandidate.TaskState computeCompactionTaskState(CompactionCandidate candidate)
{
// Skip intervals that already have a running task
final CompactionTaskStatus lastTaskStatus = getLatestTaskStatus(candidate);
if (lastTaskStatus != null && lastTaskStatus.getState() == TaskState.RUNNING) {
- return CompactionStatus.running("Task for interval is already running");
+ return CompactionCandidate.TaskState.TASK_IN_PROGRESS;
}
// Skip intervals that have been recently compacted if segment timeline is not updated yet
final DateTime snapshotTime = segmentSnapshotTime.get();
if (lastTaskStatus != null
&& lastTaskStatus.getState() == TaskState.SUCCESS
&& snapshotTime != null && snapshotTime.isBefore(lastTaskStatus.getUpdatedTime())) {
- return CompactionStatus.skipped(
- "Segment timeline not updated since last compaction task succeeded"
- );
+ return CompactionCandidate.TaskState.RECENTLY_COMPLETED;
}
- // Skip intervals that have been filtered out by the policy
- final CompactionCandidateSearchPolicy.Eligibility eligibility
- = searchPolicy.checkEligibilityForCompaction(candidate, lastTaskStatus);
- if (eligibility.isEligible()) {
- return CompactionStatus.pending("Not compacted yet");
- } else {
- return CompactionStatus.skipped("Rejected by search policy: %s", eligibility.getReason());
- }
+ return CompactionCandidate.TaskState.READY;
}
/**
* Tracks the latest compaction status of the given compaction candidates.
* Used only by the {@link CompactionRunSimulator}.
*/
- public void onCompactionStatusComputed(
+ public void onSkippedCandidate(
+ CompactionCandidate candidateSegments,
+ DataSourceCompactionConfig config
Review Comment:
## Useless parameter
The parameter 'config' is never used.
[Show more details](https://github.com/apache/druid/security/code-scanning/10812)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org