This is an automated email from the ASF dual-hosted git repository.
gianm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 04d643f5fb6 perf: Limit details returned by
RetrieveUsedSegmentsAction. (#20006)
04d643f5fb6 is described below
commit 04d643f5fb6afb81032d17e0b7b85393883fbde3
Author: Gian Merlino <[email protected]>
AuthorDate: Fri Aug 14 15:53:38 2026 -0700
perf: Limit details returned by RetrieveUsedSegmentsAction. (#20006)
DataSegment objects can have a variety of different details. Most of them
are not necessary for the tasks that use this action. The new "details"
parameter allows tasks to limit what is returned, saving bandwidth and
serde time.
---
.../ActionBasedPublishedSegmentRetriever.java | 12 +-
.../common/actions/RetrieveUsedSegmentsAction.java | 96 ++++++++--
.../common/task/AbstractBatchIndexTask.java | 9 +-
.../druid/indexing/common/task/CompactionTask.java | 10 +-
.../druid/indexing/common/task/IndexTask.java | 4 +-
.../common/task/KillUnusedSegmentsTask.java | 5 +-
.../parallel/ParallelIndexSupervisorTask.java | 4 +-
.../task/batch/parallel/SinglePhaseSubTask.java | 4 +-
.../task/batch/parallel/TombstoneHelper.java | 4 +-
.../druid/indexing/input/DruidInputSource.java | 5 +-
.../ActionBasedPublishedSegmentRetrieverTest.java | 5 +-
.../actions/RetrieveSegmentsActionsTest.java | 3 +-
.../RetrieveUsedSegmentsActionSerdeTest.java | 36 +++-
.../concurrent/ConcurrentReplaceAndAppendTest.java | 7 +-
.../ConcurrentReplaceAndStreamingAppendTest.java | 7 +-
.../msq/indexing/IndexerTableInputSpecSlicer.java | 10 +-
.../org/apache/druid/msq/exec/MSQReplaceTest.java | 67 +++++--
.../org/apache/druid/timeline/DataSegment.java | 27 +++
.../org/apache/druid/timeline/SegmentDetail.java | 132 ++++++++++++++
.../apache/druid/timeline/SegmentDetailTest.java | 193 +++++++++++++++++++++
.../appenderator/BaseAppenderatorDriver.java | 25 ++-
21 files changed, 610 insertions(+), 55 deletions(-)
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/appenderator/ActionBasedPublishedSegmentRetriever.java
b/indexing-service/src/main/java/org/apache/druid/indexing/appenderator/ActionBasedPublishedSegmentRetriever.java
index bb349cc9790..74ea1cbebde 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/appenderator/ActionBasedPublishedSegmentRetriever.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/appenderator/ActionBasedPublishedSegmentRetriever.java
@@ -29,12 +29,14 @@ import org.apache.druid.java.util.common.JodaUtils;
import org.apache.druid.java.util.common.logger.Logger;
import
org.apache.druid.segment.realtime.appenderator.PublishedSegmentRetriever;
import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.SegmentDetail;
import org.apache.druid.timeline.SegmentId;
import org.joda.time.Interval;
import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
+import java.util.EnumSet;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
@@ -92,7 +94,15 @@ public class ActionBasedPublishedSegmentRetriever implements
PublishedSegmentRet
Iterables.transform(segmentIds, SegmentId::getInterval)
);
final Collection<DataSegment> foundUsedSegments = taskActionClient.submit(
- new RetrieveUsedSegmentsAction(dataSource, usedSearchIntervals,
Segments.INCLUDING_OVERSHADOWED)
+ new RetrieveUsedSegmentsAction(
+ dataSource,
+ usedSearchIntervals,
+ Segments.INCLUDING_OVERSHADOWED,
+ // LOAD_SPEC is required: callers compare the load specs of the
returned segments against the ones they
+ // pushed themselves, to decide whether the pushed copies are safe
to delete from deep storage. See
+ // BaseAppenderatorDriver#publishInBackground.
+ EnumSet.of(SegmentDetail.LOAD_SPEC)
+ )
);
for (DataSegment segment : foundUsedSegments) {
if (segmentIds.contains(segment.getId())) {
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/RetrieveUsedSegmentsAction.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/RetrieveUsedSegmentsAction.java
index 51bec50a600..4ba2a137f85 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/RetrieveUsedSegmentsAction.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/RetrieveUsedSegmentsAction.java
@@ -20,6 +20,7 @@
package org.apache.druid.indexing.common.actions;
import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.core.type.TypeReference;
import org.apache.druid.common.config.Configs;
@@ -34,12 +35,16 @@ import org.apache.druid.java.util.common.logger.Logger;
import org.apache.druid.metadata.ReplaceTaskLock;
import org.apache.druid.timeline.DataSegment;
import org.apache.druid.timeline.Partitions;
+import org.apache.druid.timeline.SegmentDetail;
import org.apache.druid.timeline.SegmentTimeline;
import org.apache.druid.utils.CollectionUtils;
import org.joda.time.Interval;
import javax.annotation.Nullable;
+import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
+import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -56,6 +61,9 @@ import java.util.stream.Collectors;
* only segments that were created before the REPLACE lock was acquired are
* returned for an interval. This ensures that the input set of segments for
this
* replace task remains consistent even when new data is appended by other
concurrent tasks.
+ * <p>
+ * Callers declare which optional segment details they need through {@code
details}. Details that
+ * are not requested are nulled out of the returned segments.
*/
public class RetrieveUsedSegmentsAction implements
TaskAction<Collection<DataSegment>>
{
@@ -65,11 +73,18 @@ public class RetrieveUsedSegmentsAction implements
TaskAction<Collection<DataSeg
private final List<Interval> intervals;
private final Segments visibility;
- @JsonCreator
+ /**
+ * Optional segment details to include in the returned segments; see {@link
SegmentDetail}. Null means "include
+ * everything", which is what a client older than this parameter expects.
+ */
+ @Nullable
+ private final Set<SegmentDetail> details;
+
public RetrieveUsedSegmentsAction(
- @JsonProperty("dataSource") String dataSource,
- @JsonProperty("intervals") Collection<Interval> intervals,
- @JsonProperty("visibility") @Nullable Segments visibility
+ String dataSource,
+ Collection<Interval> intervals,
+ @Nullable Segments visibility,
+ @Nullable EnumSet<SegmentDetail> details
)
{
if (CollectionUtils.isNullOrEmpty(intervals)) {
@@ -79,11 +94,31 @@ public class RetrieveUsedSegmentsAction implements
TaskAction<Collection<DataSeg
this.dataSource = dataSource;
this.intervals = JodaUtils.condenseIntervals(intervals);
this.visibility = Configs.valueOrDefault(visibility,
Segments.ONLY_VISIBLE);
+ this.details = details == null ? null :
Collections.unmodifiableSet(EnumSet.copyOf(details));
+ }
+
+ public RetrieveUsedSegmentsAction(
+ String dataSource,
+ Collection<Interval> intervals,
+ EnumSet<SegmentDetail> details
+ )
+ {
+ this(dataSource, intervals, Segments.ONLY_VISIBLE, details);
}
- public RetrieveUsedSegmentsAction(String dataSource, Collection<Interval>
intervals)
+ /**
+ * Factory for deserialization. Takes the details as raw names rather than
as {@link SegmentDetail} so that a name
+ * this version of Druid does not know about is skipped instead of failing
the whole request.
+ */
+ @JsonCreator
+ static RetrieveUsedSegmentsAction fromJson(
+ @JsonProperty("dataSource") String dataSource,
+ @JsonProperty("intervals") Collection<Interval> intervals,
+ @JsonProperty("visibility") @Nullable Segments visibility,
+ @JsonProperty("details") @Nullable Collection<String> details
+ )
{
- this(dataSource, intervals, Segments.ONLY_VISIBLE);
+ return new RetrieveUsedSegmentsAction(dataSource, intervals, visibility,
SegmentDetail.fromNamesLenient(details));
}
@JsonProperty
@@ -104,6 +139,19 @@ public class RetrieveUsedSegmentsAction implements
TaskAction<Collection<DataSeg
return visibility;
}
+ /**
+ * Segment details to include. Null means include all details, empty means
include no details.
+ *
+ * @see DataSegment#retainOnlyDetails(Set)
+ */
+ @Nullable
+ @JsonProperty
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ public Set<SegmentDetail> getDetails()
+ {
+ return details;
+ }
+
@Override
public TypeReference<Collection<DataSegment>> getReturnTypeReference()
{
@@ -112,6 +160,11 @@ public class RetrieveUsedSegmentsAction implements
TaskAction<Collection<DataSeg
@Override
public Collection<DataSegment> perform(Task task, TaskActionToolbox toolbox)
+ {
+ return retainRequestedDetails(retrieveSegments(task, toolbox));
+ }
+
+ private Collection<DataSegment> retrieveSegments(Task task,
TaskActionToolbox toolbox)
{
// When fetching segments for a datasource other than the one this task is
writing to,
// just return all segments with the needed visibility.
@@ -186,6 +239,23 @@ public class RetrieveUsedSegmentsAction implements
TaskAction<Collection<DataSeg
.retrieveUsedSegmentsForIntervals(dataSource, intervals,
visibility);
}
+ /**
+ * Strips the optional details that the caller did not ask for out of {@code
segments}. A null {@link #details} means
+ * the caller is older than this parameter and expects every detail, so the
segments pass through untouched.
+ */
+ private Collection<DataSegment> retainRequestedDetails(final
Collection<DataSegment> segments)
+ {
+ if (details == null) {
+ return segments;
+ }
+
+ final List<DataSegment> retVal = new ArrayList<>(segments.size());
+ for (final DataSegment segment : segments) {
+ retVal.add(segment.retainOnlyDetails(details));
+ }
+ return retVal;
+ }
+
@Override
public boolean equals(Object o)
{
@@ -198,19 +268,16 @@ public class RetrieveUsedSegmentsAction implements
TaskAction<Collection<DataSeg
RetrieveUsedSegmentsAction that = (RetrieveUsedSegmentsAction) o;
- if (!dataSource.equals(that.dataSource)) {
- return false;
- }
- if (!intervals.equals(that.intervals)) {
- return false;
- }
- return visibility.equals(that.visibility);
+ return dataSource.equals(that.dataSource)
+ && intervals.equals(that.intervals)
+ && visibility.equals(that.visibility)
+ && Objects.equals(details, that.details);
}
@Override
public int hashCode()
{
- return Objects.hash(dataSource, intervals, visibility);
+ return Objects.hash(dataSource, intervals, visibility, details);
}
@Override
@@ -220,6 +287,7 @@ public class RetrieveUsedSegmentsAction implements
TaskAction<Collection<DataSeg
"dataSource='" + dataSource + '\'' +
", intervals=" + intervals +
", visibility=" + visibility +
+ ", details=" + details +
'}';
}
}
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/AbstractBatchIndexTask.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/AbstractBatchIndexTask.java
index a555056c995..bb1f90d1299 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/AbstractBatchIndexTask.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/AbstractBatchIndexTask.java
@@ -81,6 +81,7 @@ import
org.apache.druid.segment.transform.CompactionTransformSpec;
import org.apache.druid.timeline.CompactionState;
import org.apache.druid.timeline.DataSegment;
import org.apache.druid.timeline.Partitions;
+import org.apache.druid.timeline.SegmentDetail;
import org.apache.druid.timeline.SegmentTimeline;
import org.apache.druid.timeline.partition.HashBasedNumberedShardSpec;
import org.apache.druid.timeline.partition.TombstoneShardSpec;
@@ -94,6 +95,7 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.EnumSet;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
@@ -697,13 +699,12 @@ public abstract class AbstractBatchIndexTask extends
AbstractTask
protected static List<DataSegment> findInputSegments(
String dataSource,
TaskActionClient actionClient,
- List<Interval> intervalsToRead
+ List<Interval> intervalsToRead,
+ EnumSet<SegmentDetail> details
) throws IOException
{
return ImmutableList.copyOf(
- actionClient.submit(
- new RetrieveUsedSegmentsAction(dataSource, intervalsToRead)
- )
+ actionClient.submit(new RetrieveUsedSegmentsAction(dataSource,
intervalsToRead, details))
);
}
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionTask.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionTask.java
index d4a8b68d26c..0e857e52786 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionTask.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/CompactionTask.java
@@ -110,6 +110,7 @@ import
org.apache.druid.server.coordinator.CompactionConfigValidationResult;
import org.apache.druid.server.lookup.cache.LookupLoadingSpec;
import org.apache.druid.server.security.ResourceAction;
import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.SegmentDetail;
import org.apache.druid.timeline.SegmentId;
import org.apache.druid.timeline.SegmentTimeline;
import org.apache.druid.timeline.TimelineObjectHolder;
@@ -125,6 +126,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
+import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -493,7 +495,7 @@ public class CompactionTask extends AbstractBatchIndexTask
implements PendingSeg
throws IOException
{
return ImmutableList.copyOf(
- taskActionClient.submit(new
RetrieveUsedSegmentsAction(getDataSource(), intervals))
+ taskActionClient.submit(new
RetrieveUsedSegmentsAction(getDataSource(), intervals, SegmentDetail.none()))
);
}
@@ -1385,7 +1387,11 @@ public class CompactionTask extends
AbstractBatchIndexTask implements PendingSeg
{
return new ArrayList<>(
actionClient.submit(
- new RetrieveUsedSegmentsAction(dataSource,
ImmutableList.of(interval))
+ new RetrieveUsedSegmentsAction(
+ dataSource,
+ ImmutableList.of(interval),
+ EnumSet.of(SegmentDetail.LOAD_SPEC)
+ )
)
);
}
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/IndexTask.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/IndexTask.java
index b29899a77cf..75b3c25e1e9 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/IndexTask.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/IndexTask.java
@@ -94,6 +94,7 @@ import org.apache.druid.server.security.AuthorizationUtils;
import org.apache.druid.server.security.AuthorizerMapper;
import org.apache.druid.server.security.ResourceAction;
import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.SegmentDetail;
import org.apache.druid.timeline.partition.HashBasedNumberedShardSpec;
import org.apache.druid.timeline.partition.NumberedShardSpec;
import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
@@ -269,7 +270,8 @@ public class IndexTask extends AbstractBatchIndexTask
implements ChatHandler, Pe
return findInputSegments(
getDataSource(),
taskActionClient,
- intervals
+ intervals,
+ SegmentDetail.none()
);
}
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/KillUnusedSegmentsTask.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/KillUnusedSegmentsTask.java
index 5c74770c230..3e8ca76b39a 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/KillUnusedSegmentsTask.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/KillUnusedSegmentsTask.java
@@ -53,6 +53,7 @@ import org.apache.druid.server.http.DataSegmentPlus;
import org.apache.druid.server.lookup.cache.LookupLoadingSpec;
import org.apache.druid.server.security.ResourceAction;
import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.SegmentDetail;
import org.apache.druid.utils.CollectionUtils;
import org.joda.time.DateTime;
import org.joda.time.Interval;
@@ -61,6 +62,7 @@ import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -235,7 +237,8 @@ public class KillUnusedSegmentsTask extends
AbstractFixedIntervalTask
RetrieveUsedSegmentsAction retrieveUsedSegmentsAction = new
RetrieveUsedSegmentsAction(
getDataSource(),
ImmutableList.of(getInterval()),
- Segments.INCLUDING_OVERSHADOWED
+ Segments.INCLUDING_OVERSHADOWED,
+ EnumSet.of(SegmentDetail.LOAD_SPEC)
);
// Fetch the load specs of all segments overlapping with the unused
segment intervals
final Set<Map<String, Object>> usedSegmentLoadSpecs =
taskActionClient.submit(retrieveUsedSegmentsAction)
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTask.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTask.java
index 2764c26c6af..4793b77ac44 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTask.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTask.java
@@ -85,6 +85,7 @@ import org.apache.druid.server.security.AuthorizationUtils;
import org.apache.druid.server.security.AuthorizerMapper;
import org.apache.druid.server.security.ResourceAction;
import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.SegmentDetail;
import org.apache.druid.timeline.partition.BuildingShardSpec;
import org.apache.druid.timeline.partition.NumberedShardSpec;
import org.apache.druid.timeline.partition.PartitionBoundaries;
@@ -460,7 +461,8 @@ public class ParallelIndexSupervisorTask extends
AbstractBatchIndexTask
return findInputSegments(
getDataSource(),
taskActionClient,
- intervals
+ intervals,
+ SegmentDetail.none()
);
}
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/SinglePhaseSubTask.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/SinglePhaseSubTask.java
index 53a480b66cc..2b7d3043127 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/SinglePhaseSubTask.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/SinglePhaseSubTask.java
@@ -68,6 +68,7 @@ import org.apache.druid.server.security.AuthorizationUtils;
import org.apache.druid.server.security.AuthorizerMapper;
import org.apache.druid.server.security.ResourceAction;
import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.SegmentDetail;
import org.apache.druid.timeline.SegmentTimeline;
import org.apache.druid.timeline.TimelineObjectHolder;
import org.apache.druid.timeline.partition.PartitionChunk;
@@ -310,7 +311,8 @@ public class SinglePhaseSubTask extends
AbstractBatchSubtask implements ChatHand
return findInputSegments(
getDataSource(),
taskActionClient,
- intervals
+ intervals,
+ SegmentDetail.none()
);
}
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/TombstoneHelper.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/TombstoneHelper.java
index 7d739dd02e9..9f0f017c864 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/TombstoneHelper.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/TombstoneHelper.java
@@ -35,6 +35,7 @@ import
org.apache.druid.java.util.common.granularity.IntervalsByGranularity;
import org.apache.druid.segment.indexing.DataSchema;
import org.apache.druid.segment.realtime.appenderator.SegmentIdWithShardSpec;
import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.SegmentDetail;
import org.apache.druid.timeline.partition.ShardSpec;
import org.apache.druid.timeline.partition.TombstoneShardSpec;
import org.joda.time.Interval;
@@ -310,7 +311,8 @@ public class TombstoneHelper
Collection<DataSegment> usedSegmentsInInputInterval =
taskActionClient.submit(new RetrieveUsedSegmentsAction(
dataSource,
- condensedInputIntervals
+ condensedInputIntervals,
+ SegmentDetail.none()
));
for (DataSegment usedSegment : usedSegmentsInInputInterval) {
for (Interval condensedInputInterval : condensedInputIntervals) {
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/input/DruidInputSource.java
b/indexing-service/src/main/java/org/apache/druid/indexing/input/DruidInputSource.java
index 917936c03c5..c1178a421fa 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/input/DruidInputSource.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/input/DruidInputSource.java
@@ -61,6 +61,7 @@ import org.apache.druid.segment.IndexIO;
import org.apache.druid.segment.column.ColumnHolder;
import org.apache.druid.segment.loading.SegmentCacheManager;
import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.SegmentDetail;
import org.apache.druid.timeline.SegmentId;
import org.apache.druid.timeline.SegmentTimeline;
import org.apache.druid.timeline.TimelineObjectHolder;
@@ -77,6 +78,7 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
+import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
@@ -589,7 +591,8 @@ public class DruidInputSource extends AbstractInputSource
implements SplittableI
usedSegments = toolbox.getTaskActionClient()
.submit(new RetrieveUsedSegmentsAction(
dataSource,
- Collections.singletonList(interval)
+ Collections.singletonList(interval),
+ EnumSet.of(SegmentDetail.LOAD_SPEC)
));
}
catch (IOException e) {
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/appenderator/ActionBasedPublishedSegmentRetrieverTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/appenderator/ActionBasedPublishedSegmentRetrieverTest.java
index 14b51b55782..c300e66cc19 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/appenderator/ActionBasedPublishedSegmentRetrieverTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/appenderator/ActionBasedPublishedSegmentRetrieverTest.java
@@ -31,6 +31,7 @@ import
org.apache.druid.java.util.common.granularity.Granularities;
import org.apache.druid.segment.TestDataSource;
import org.apache.druid.server.coordinator.CreateDataSegments;
import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.SegmentDetail;
import org.apache.druid.timeline.SegmentId;
import org.easymock.EasyMock;
import org.junit.jupiter.api.Assertions;
@@ -39,6 +40,7 @@ import org.junit.jupiter.api.Test;
import java.io.IOException;
import java.util.Collections;
+import java.util.EnumSet;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
@@ -105,7 +107,8 @@ public class ActionBasedPublishedSegmentRetrieverTest
new RetrieveUsedSegmentsAction(
TestDataSource.WIKI,
Collections.singletonList(Intervals.of("2013-01-01/P3D")),
- Segments.INCLUDING_OVERSHADOWED
+ Segments.INCLUDING_OVERSHADOWED,
+ EnumSet.of(SegmentDetail.LOAD_SPEC)
)
)
).andReturn(segments).once();
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/actions/RetrieveSegmentsActionsTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/actions/RetrieveSegmentsActionsTest.java
index d2edd148733..7b4858d20ea 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/common/actions/RetrieveSegmentsActionsTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/actions/RetrieveSegmentsActionsTest.java
@@ -26,6 +26,7 @@ import org.apache.druid.indexing.common.task.Task;
import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.Intervals;
import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.SegmentDetail;
import org.apache.druid.timeline.partition.NoneShardSpec;
import org.joda.time.Interval;
import org.junit.jupiter.api.Assertions;
@@ -100,7 +101,7 @@ public class RetrieveSegmentsActionsTest
public void testRetrieveUsedSegmentsAction()
{
final RetrieveUsedSegmentsAction action =
- new RetrieveUsedSegmentsAction(task.getDataSource(),
ImmutableList.of(INTERVAL));
+ new RetrieveUsedSegmentsAction(task.getDataSource(),
ImmutableList.of(INTERVAL), SegmentDetail.all());
final Set<DataSegment> observedUsedSegments = new
HashSet<>(action.perform(task, actionTestKit.getTaskActionToolbox()));
Assertions.assertEquals(expectedUsedSegments, observedUsedSegments);
}
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/actions/RetrieveUsedSegmentsActionSerdeTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/actions/RetrieveUsedSegmentsActionSerdeTest.java
index 6a1a20ed6c9..f58377d922d 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/common/actions/RetrieveUsedSegmentsActionSerdeTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/actions/RetrieveUsedSegmentsActionSerdeTest.java
@@ -24,11 +24,13 @@ import com.google.common.collect.ImmutableList;
import org.apache.druid.indexing.overlord.Segments;
import org.apache.druid.java.util.common.Intervals;
import org.apache.druid.segment.TestHelper;
+import org.apache.druid.timeline.SegmentDetail;
import org.joda.time.Interval;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import java.util.Collections;
+import java.util.EnumSet;
import java.util.List;
/**
@@ -42,8 +44,12 @@ public class RetrieveUsedSegmentsActionSerdeTest
{
Interval interval = Intervals.of("2014/2015");
- RetrieveUsedSegmentsAction expected =
- new RetrieveUsedSegmentsAction("dataSource",
Collections.singletonList(interval), Segments.ONLY_VISIBLE);
+ RetrieveUsedSegmentsAction expected = new RetrieveUsedSegmentsAction(
+ "dataSource",
+ Collections.singletonList(interval),
+ Segments.ONLY_VISIBLE,
+ SegmentDetail.all()
+ );
RetrieveUsedSegmentsAction actual =
MAPPER.readValue(MAPPER.writeValueAsString(expected),
RetrieveUsedSegmentsAction.class);
@@ -57,7 +63,8 @@ public class RetrieveUsedSegmentsActionSerdeTest
List<Interval> intervals = ImmutableList.of(Intervals.of("2014/2015"),
Intervals.of("2016/2017"));
RetrieveUsedSegmentsAction expected = new RetrieveUsedSegmentsAction(
"dataSource",
- intervals
+ intervals,
+ SegmentDetail.none()
);
RetrieveUsedSegmentsAction actual =
@@ -66,6 +73,25 @@ public class RetrieveUsedSegmentsActionSerdeTest
Assertions.assertEquals(expected, actual);
}
+ @Test
+ public void testPartialDetailsSerde() throws Exception
+ {
+ RetrieveUsedSegmentsAction expected = new RetrieveUsedSegmentsAction(
+ "dataSource",
+ Collections.singletonList(Intervals.of("2014/2015")),
+ Segments.INCLUDING_OVERSHADOWED,
+ EnumSet.of(SegmentDetail.LOAD_SPEC, SegmentDetail.DIMENSIONS)
+ );
+
+ final String json = MAPPER.writeValueAsString(expected);
+ RetrieveUsedSegmentsAction actual = MAPPER.readValue(json,
RetrieveUsedSegmentsAction.class);
+ Assertions.assertEquals(
+ EnumSet.of(SegmentDetail.DIMENSIONS, SegmentDetail.LOAD_SPEC),
+ actual.getDetails()
+ );
+ Assertions.assertEquals(expected, actual);
+ }
+
@Test
public void testOldJsonDeserialization() throws Exception
{
@@ -76,9 +102,11 @@ public class RetrieveUsedSegmentsActionSerdeTest
new RetrieveUsedSegmentsAction(
"test",
Collections.singletonList(Intervals.of("2014/2015")),
- Segments.ONLY_VISIBLE
+ Segments.ONLY_VISIBLE,
+ null
),
actual
);
+ Assertions.assertNull(actual.getDetails());
}
}
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/concurrent/ConcurrentReplaceAndAppendTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/concurrent/ConcurrentReplaceAndAppendTest.java
index 6a11cc30ce4..a96c650dd31 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/concurrent/ConcurrentReplaceAndAppendTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/concurrent/ConcurrentReplaceAndAppendTest.java
@@ -63,6 +63,7 @@ import org.apache.druid.server.DruidNode;
import org.apache.druid.server.metrics.NoopServiceEmitter;
import org.apache.druid.tasklogs.NoopTaskLogs;
import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.SegmentDetail;
import org.apache.druid.timeline.SegmentId;
import org.apache.druid.timeline.partition.NumberedShardSpec;
import org.joda.time.Interval;
@@ -1298,7 +1299,8 @@ public class ConcurrentReplaceAndAppendTest extends
IngestionTestBase
new RetrieveUsedSegmentsAction(
TestDataSource.WIKI,
ImmutableList.of(interval),
- visibility
+ visibility,
+ SegmentDetail.all()
)
);
Assertions.assertEquals(Sets.newHashSet(expectedSegments),
Sets.newHashSet(allUsedSegments));
@@ -1315,7 +1317,8 @@ public class ConcurrentReplaceAndAppendTest extends
IngestionTestBase
Collection<DataSegment> allUsedSegments = taskActionClient.submit(
new RetrieveUsedSegmentsAction(
TestDataSource.WIKI,
- Collections.singletonList(interval)
+ Collections.singletonList(interval),
+ SegmentDetail.all()
)
);
Assertions.assertEquals(Sets.newHashSet(expectedSegments),
Sets.newHashSet(allUsedSegments));
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/concurrent/ConcurrentReplaceAndStreamingAppendTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/concurrent/ConcurrentReplaceAndStreamingAppendTest.java
index 57c25a65060..6f31c07002d 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/concurrent/ConcurrentReplaceAndStreamingAppendTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/concurrent/ConcurrentReplaceAndStreamingAppendTest.java
@@ -64,6 +64,7 @@ import org.apache.druid.server.DruidNode;
import org.apache.druid.server.metrics.NoopServiceEmitter;
import org.apache.druid.tasklogs.NoopTaskLogs;
import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.SegmentDetail;
import org.apache.druid.timeline.SegmentId;
import org.apache.druid.timeline.partition.NumberedShardSpec;
import org.easymock.Capture;
@@ -704,7 +705,8 @@ public class ConcurrentReplaceAndStreamingAppendTest
extends IngestionTestBase
new RetrieveUsedSegmentsAction(
TestDataSource.WIKI,
ImmutableList.of(interval),
- visibility
+ visibility,
+ SegmentDetail.all()
)
);
Assertions.assertEquals(Sets.newHashSet(expectedSegments),
Sets.newHashSet(allUsedSegments));
@@ -823,7 +825,8 @@ public class ConcurrentReplaceAndStreamingAppendTest
extends IngestionTestBase
new RetrieveUsedSegmentsAction(
TestDataSource.WIKI,
ImmutableList.of(Intervals.ETERNITY),
- Segments.INCLUDING_OVERSHADOWED
+ Segments.INCLUDING_OVERSHADOWED,
+ SegmentDetail.all()
)
);
}
diff --git
a/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/IndexerTableInputSpecSlicer.java
b/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/IndexerTableInputSpecSlicer.java
index ddbfdde9854..825becde940 100644
---
a/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/IndexerTableInputSpecSlicer.java
+++
b/multi-stage-query/src/main/java/org/apache/druid/msq/indexing/IndexerTableInputSpecSlicer.java
@@ -48,6 +48,7 @@ import org.apache.druid.query.SegmentDescriptor;
import org.apache.druid.query.filter.SegmentPruner;
import org.apache.druid.server.coordination.DruidServerMetadata;
import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.SegmentDetail;
import org.apache.druid.timeline.SegmentTimeline;
import org.apache.druid.timeline.TimelineLookup;
import org.apache.druid.timeline.VersionedIntervalTimeline;
@@ -221,8 +222,13 @@ public class IndexerTableInputSpecSlicer implements
InputSpecSlicer
if (intervals.isEmpty()) {
publishedUsedSegments = Collections.emptySet();
} else {
- publishedUsedSegments =
- taskActionClient.submit(new RetrieveUsedSegmentsAction(dataSource,
intervals));
+ publishedUsedSegments = taskActionClient.submit(
+ new RetrieveUsedSegmentsAction(
+ dataSource,
+ intervals,
+ SegmentDetail.none() // Even LoadSpec is not needed, because
workers fetch them from the Coordinator.
+ )
+ );
}
}
catch (IOException e) {
diff --git
a/multi-stage-query/src/test/java/org/apache/druid/msq/exec/MSQReplaceTest.java
b/multi-stage-query/src/test/java/org/apache/druid/msq/exec/MSQReplaceTest.java
index c51a7312c10..9acabcf1c8b 100644
---
a/multi-stage-query/src/test/java/org/apache/druid/msq/exec/MSQReplaceTest.java
+++
b/multi-stage-query/src/test/java/org/apache/druid/msq/exec/MSQReplaceTest.java
@@ -69,6 +69,7 @@ import
org.apache.druid.segment.virtual.ExpressionVirtualColumn;
import org.apache.druid.sql.calcite.util.CalciteTests;
import org.apache.druid.timeline.CompactionState;
import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.SegmentDetail;
import org.apache.druid.timeline.SegmentId;
import org.apache.druid.timeline.partition.DimensionRangeShardSpec;
import org.apache.druid.timeline.partition.NumberedShardSpec;
@@ -144,7 +145,11 @@ public class MSQReplaceTest extends MSQTestBase
Mockito.doCallRealMethod()
.doReturn(ImmutableSet.of(existingDataSegment0,
existingDataSegment1))
.when(testTaskActionClient)
- .submit(new RetrieveUsedSegmentsAction("foo",
ImmutableList.of(Intervals.ETERNITY)));
+ .submit(new RetrieveUsedSegmentsAction(
+ "foo",
+ ImmutableList.of(Intervals.ETERNITY),
+ SegmentDetail.none()
+ ));
testIngestQuery().setSql(" REPLACE INTO foo OVERWRITE ALL "
+ "SELECT __time, m1 "
@@ -235,7 +240,11 @@ public class MSQReplaceTest extends MSQTestBase
Mockito.doCallRealMethod()
.doReturn(ImmutableSet.of(existingDataSegment0,
existingDataSegment1))
.when(testTaskActionClient)
- .submit(new RetrieveUsedSegmentsAction("foo",
ImmutableList.of(Intervals.ETERNITY)));
+ .submit(new RetrieveUsedSegmentsAction(
+ "foo",
+ ImmutableList.of(Intervals.ETERNITY),
+ SegmentDetail.none()
+ ));
testIngestQuery().setSql(" REPLACE INTO foo OVERWRITE ALL "
+ "SELECT __time, dim1, m1 "
@@ -315,7 +324,11 @@ public class MSQReplaceTest extends MSQTestBase
Mockito.doCallRealMethod()
.doReturn(ImmutableSet.of(existingDataSegment0,
existingDataSegment1))
.when(testTaskActionClient)
- .submit(new RetrieveUsedSegmentsAction("foo",
ImmutableList.of(Intervals.ETERNITY)));
+ .submit(new RetrieveUsedSegmentsAction(
+ "foo",
+ ImmutableList.of(Intervals.ETERNITY),
+ SegmentDetail.none()
+ ));
testIngestQuery().setSql(" REPLACE INTO foo OVERWRITE ALL "
+ "SELECT __time, dim1, m1 "
@@ -408,7 +421,11 @@ public class MSQReplaceTest extends MSQTestBase
Mockito.doCallRealMethod()
.doReturn(ImmutableSet.of(existingDataSegment0,
existingDataSegment1))
.when(testTaskActionClient)
- .submit(new RetrieveUsedSegmentsAction("foo",
ImmutableList.of(Intervals.ETERNITY)));
+ .submit(new RetrieveUsedSegmentsAction(
+ "foo",
+ ImmutableList.of(Intervals.ETERNITY),
+ SegmentDetail.none()
+ ));
testIngestQuery().setSql(" REPLACE INTO foo OVERWRITE ALL "
+ "SELECT __time, dim1, m1 "
@@ -494,7 +511,11 @@ public class MSQReplaceTest extends MSQTestBase
Mockito.doCallRealMethod()
.doReturn(ImmutableSet.of(existingDataSegment0,
existingDataSegment1))
.when(testTaskActionClient)
- .submit(new RetrieveUsedSegmentsAction("foo",
ImmutableList.of(Intervals.ETERNITY)));
+ .submit(new RetrieveUsedSegmentsAction(
+ "foo",
+ ImmutableList.of(Intervals.ETERNITY),
+ SegmentDetail.none()
+ ));
testIngestQuery().setSql(" REPLACE INTO foo OVERWRITE ALL "
+ "SELECT __time, dim1, m1 "
@@ -620,7 +641,11 @@ public class MSQReplaceTest extends MSQTestBase
Mockito.doCallRealMethod()
.doReturn(ImmutableSet.of(existingDataSegment0,
existingDataSegment1))
.when(testTaskActionClient)
- .submit(new RetrieveUsedSegmentsAction("foo",
ImmutableList.of(Intervals.ETERNITY)));
+ .submit(new RetrieveUsedSegmentsAction(
+ "foo",
+ ImmutableList.of(Intervals.ETERNITY),
+ SegmentDetail.none()
+ ));
testIngestQuery().setSql(" REPLACE INTO foo OVERWRITE ALL "
+ "SELECT __time, dim1, m1 "
@@ -1147,7 +1172,11 @@ public class MSQReplaceTest extends MSQTestBase
Mockito.doCallRealMethod()
.doReturn(ImmutableSet.of(existingDataSegment0,
existingDataSegment1))
.when(testTaskActionClient)
- .submit(new RetrieveUsedSegmentsAction("foo",
ImmutableList.of(Intervals.ETERNITY)));
+ .submit(new RetrieveUsedSegmentsAction(
+ "foo",
+ ImmutableList.of(Intervals.ETERNITY),
+ SegmentDetail.none()
+ ));
testIngestQuery().setSql(" REPLACE INTO foo "
@@ -1230,7 +1259,11 @@ public class MSQReplaceTest extends MSQTestBase
Mockito.doReturn(ImmutableSet.of(existingDataSegment0))
.when(testTaskActionClient)
- .submit(new RetrieveUsedSegmentsAction("foo",
ImmutableList.of(Intervals.of("2000-01-01/2000-03-01"))));
+ .submit(new RetrieveUsedSegmentsAction(
+ "foo",
+ ImmutableList.of(Intervals.of("2000-01-01/2000-03-01")),
+ SegmentDetail.none()
+ ));
testIngestQuery().setSql(" REPLACE INTO foo "
+ "OVERWRITE WHERE __time >= TIMESTAMP
'2000-01-01' AND __time < TIMESTAMP '2000-03-01' "
@@ -1319,7 +1352,8 @@ public class MSQReplaceTest extends MSQTestBase
.when(testTaskActionClient)
.submit(new RetrieveUsedSegmentsAction(
EasyMock.eq("foo"),
-
EasyMock.eq(ImmutableList.of(Intervals.of("2000-01-01/2002-01-01")))
+
EasyMock.eq(ImmutableList.of(Intervals.of("2000-01-01/2002-01-01"))),
+ SegmentDetail.none()
));
@@ -1499,7 +1533,8 @@ public class MSQReplaceTest extends MSQTestBase
.when(testTaskActionClient)
.submit(new RetrieveUsedSegmentsAction(
EasyMock.eq("foo"),
-
EasyMock.eq(ImmutableList.of(Intervals.of("2000-01-01/2000-03-01")))
+
EasyMock.eq(ImmutableList.of(Intervals.of("2000-01-01/2000-03-01"))),
+ SegmentDetail.none()
));
testIngestQuery().setSql(" REPLACE INTO foo "
@@ -1613,7 +1648,8 @@ public class MSQReplaceTest extends MSQTestBase
.when(testTaskActionClient)
.submit(new RetrieveUsedSegmentsAction(
EasyMock.eq("foo"),
- EasyMock.eq(ImmutableList.of(Intervals.of("2000/2002")))
+ EasyMock.eq(ImmutableList.of(Intervals.of("2000/2002"))),
+ SegmentDetail.none()
));
testIngestQuery().setSql(" REPLACE INTO foo "
@@ -1672,7 +1708,8 @@ public class MSQReplaceTest extends MSQTestBase
.when(testTaskActionClient)
.submit(new RetrieveUsedSegmentsAction(
EasyMock.eq("foo"),
- EasyMock.eq(ImmutableList.of(Intervals.ETERNITY))
+ EasyMock.eq(ImmutableList.of(Intervals.ETERNITY)),
+ SegmentDetail.none()
));
testIngestQuery().setSql(" REPLACE INTO foo "
@@ -2100,7 +2137,8 @@ public class MSQReplaceTest extends MSQTestBase
.when(testTaskActionClient)
.submit(new RetrieveUsedSegmentsAction(
EasyMock.eq("foo"),
- EasyMock.eq(ImmutableList.of(Intervals.of("1999/2002")))
+ EasyMock.eq(ImmutableList.of(Intervals.of("1999/2002"))),
+ SegmentDetail.none()
));
testIngestQuery().setSql(" REPLACE INTO foo "
@@ -2179,7 +2217,8 @@ public class MSQReplaceTest extends MSQTestBase
.when(testTaskActionClient)
.submit(new RetrieveUsedSegmentsAction(
EasyMock.eq("foo1"),
- EasyMock.eq(ImmutableList.of(Intervals.of("2000/2002")))
+ EasyMock.eq(ImmutableList.of(Intervals.of("2000/2002"))),
+ SegmentDetail.none()
));
List<Object[]> expectedResults = ImmutableList.of(
diff --git
a/processing/src/main/java/org/apache/druid/timeline/DataSegment.java
b/processing/src/main/java/org/apache/druid/timeline/DataSegment.java
index 0a82f8575a8..9e9208bfcd2 100644
--- a/processing/src/main/java/org/apache/druid/timeline/DataSegment.java
+++ b/processing/src/main/java/org/apache/druid/timeline/DataSegment.java
@@ -47,6 +47,7 @@ import org.joda.time.Interval;
import javax.annotation.Nullable;
import java.util.List;
import java.util.Map;
+import java.util.Set;
/**
@@ -532,6 +533,32 @@ public class DataSegment implements
Comparable<DataSegment>, Overshadowable<Data
return
builder(this).indexingStateFingerprint(indexingStateFingerprint).build();
}
+ /**
+ * Returns a copy of this segment with every optional top-level field that
is not listed in {@code details} removed.
+ * The fields that are not optional are always retained; see {@link
SegmentDetail} for the full split.
+ *
+ * @param details the optional details to retain; null retains all details
and an empty set retains none
+ */
+ public DataSegment retainOnlyDetails(@Nullable final Set<SegmentDetail>
details)
+ {
+ if (details == null) {
+ return this;
+ } else {
+ return builder(this)
+ .dimensions(details.contains(SegmentDetail.DIMENSIONS) ? dimensions
: null)
+ .metrics(details.contains(SegmentDetail.METRICS) ? metrics : null)
+ .projections(details.contains(SegmentDetail.PROJECTIONS) ?
projections : null)
+ .clusterGroups(details.contains(SegmentDetail.CLUSTER_GROUPS) ?
clusterGroups : null)
+
.lastCompactionState(details.contains(SegmentDetail.COMPACTION_STATE) ?
lastCompactionState : null)
+ .loadSpec(details.contains(SegmentDetail.LOAD_SPEC) ? loadSpec :
null)
+ .totalRows(details.contains(SegmentDetail.ROW_COUNT) ? totalRows :
null)
+ .indexingStateFingerprint(
+ details.contains(SegmentDetail.INDEXING_STATE_FINGERPRINT) ?
indexingStateFingerprint : null
+ )
+ .build();
+ }
+ }
+
public DataSegment.Builder toBuilder()
{
return builder(this);
diff --git
a/processing/src/main/java/org/apache/druid/timeline/SegmentDetail.java
b/processing/src/main/java/org/apache/druid/timeline/SegmentDetail.java
new file mode 100644
index 00000000000..713340e08cb
--- /dev/null
+++ b/processing/src/main/java/org/apache/druid/timeline/SegmentDetail.java
@@ -0,0 +1,132 @@
+/*
+ * 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.timeline;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonValue;
+import org.apache.druid.java.util.common.IAE;
+
+import javax.annotation.Nullable;
+import java.util.Collection;
+import java.util.EnumSet;
+
+/**
+ * The optional top-level fields of a {@link DataSegment}: the ones that a
segment can be returned without, so that a
+ * caller that does not need them can avoid paying for them in serialization
size and heap.
+ * <p>
+ * The rest of {@link DataSegment} is never optional: {@link
DataSegment#getId()} (and therefore the dataSource,
+ * interval, and version), {@link DataSegment#getShardSpec()}, {@link
DataSegment#getBinaryVersion()}, and
+ * {@link DataSegment#getSize()} are always populated.
+ * <p>
+ * Use {@link DataSegment#retainOnlyDetails} to drop the details that are not
wanted.
+ *
+ * @see org.apache.druid.indexing.common.actions.RetrieveUsedSegmentsAction
+ */
+public enum SegmentDetail
+{
+ DIMENSIONS("dimensions"),
+ METRICS("metrics"),
+ PROJECTIONS("projections"),
+ CLUSTER_GROUPS("clusterGroups"),
+ COMPACTION_STATE("lastCompactionState"),
+ LOAD_SPEC("loadSpec"),
+ ROW_COUNT("totalRows"),
+ INDEXING_STATE_FINGERPRINT("indexingStateFingerprint");
+
+ private final String jsonName;
+
+ SegmentDetail(final String jsonName)
+ {
+ this.jsonName = jsonName;
+ }
+
+ /**
+ * All details.
+ */
+ public static EnumSet<SegmentDetail> all()
+ {
+ return EnumSet.allOf(SegmentDetail.class);
+ }
+
+ /**
+ * No details.
+ */
+ public static EnumSet<SegmentDetail> none()
+ {
+ return EnumSet.noneOf(SegmentDetail.class);
+ }
+
+ /**
+ * Parses a collection of names, skipping any name that this version of
Druid does not recognize.
+ */
+ @Nullable
+ public static EnumSet<SegmentDetail> fromNamesLenient(@Nullable final
Collection<String> names)
+ {
+ if (names == null) {
+ return null;
+ }
+
+ final EnumSet<SegmentDetail> retVal = none();
+ for (final String name : names) {
+ final SegmentDetail detail = fromNameLenient(name);
+ if (detail != null) {
+ retVal.add(detail);
+ }
+ }
+ return retVal;
+ }
+
+ /**
+ * Returns the {@link SegmentDetail} for a name, or throws {@link
IllegalArgumentException} if none exists.
+ */
+ @JsonCreator
+ public static SegmentDetail fromName(final String name)
+ {
+ final SegmentDetail detail = fromNameLenient(name);
+ if (detail == null) {
+ throw new IAE("No such SegmentDetail[%s]", name);
+ }
+ return detail;
+ }
+
+ /**
+ * Returns the {@link SegmentDetail} for a name, or null if none exists.
+ */
+ @Nullable
+ public static SegmentDetail fromNameLenient(@Nullable final String name)
+ {
+ if (name == null) {
+ return null;
+ }
+ for (final SegmentDetail detail : values()) {
+ if (detail.jsonName.equalsIgnoreCase(name)) {
+ return detail;
+ }
+ }
+ return null;
+ }
+
+ @Override
+ @JsonValue
+ public String toString()
+ {
+ return jsonName;
+ }
+}
diff --git
a/processing/src/test/java/org/apache/druid/timeline/SegmentDetailTest.java
b/processing/src/test/java/org/apache/druid/timeline/SegmentDetailTest.java
new file mode 100644
index 00000000000..2bea51c8778
--- /dev/null
+++ b/processing/src/test/java/org/apache/druid/timeline/SegmentDetailTest.java
@@ -0,0 +1,193 @@
+/*
+ * 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.timeline;
+
+import com.fasterxml.jackson.databind.InjectableValues;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import org.apache.druid.indexer.partitions.HashedPartitionsSpec;
+import org.apache.druid.jackson.DefaultObjectMapper;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.segment.column.ColumnType;
+import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.timeline.partition.NumberedShardSpec;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.EnumSet;
+
+public class SegmentDetailTest
+{
+ private static final ObjectMapper MAPPER = new DefaultObjectMapper();
+ private static final DataSegment FULL_SEGMENT =
+ DataSegment
+ .builder(SegmentId.of("wiki", Intervals.of("2011/2012"), "v1", new
NumberedShardSpec(3, 5)))
+ .loadSpec(ImmutableMap.of("type", "local", "path", "/tmp/wiki"))
+ .dimensions(Arrays.asList("dim1", "dim2"))
+ .metrics(Arrays.asList("met1", "met2"))
+ .projections(Arrays.asList("proj1", "proj2"))
+ .clusterGroups(
+ new ClusterGroupTuples(
+ RowSignature.builder().add("dim1",
ColumnType.STRING).build(),
+ ImmutableList.of(ImmutableList.of("a"))
+ )
+ )
+ .lastCompactionState(
+ CompactionState.builder()
+ .partitionsSpec(new HashedPartitionsSpec(100,
null, ImmutableList.of("dim1")))
+ .build()
+ )
+ .binaryVersion(9)
+ .size(1234L)
+ .totalRows(42)
+ .indexingStateFingerprint("abcdef")
+ .build();
+
+ @BeforeAll
+ public static void setUpClass()
+ {
+ final InjectableValues.Std injectableValues = new InjectableValues.Std();
+ injectableValues.addValue(DataSegment.PruneSpecsHolder.class,
DataSegment.PruneSpecsHolder.DEFAULT);
+ MAPPER.setInjectableValues(injectableValues);
+ }
+
+ @Test
+ public void test_retainOnlyDetails_none()
+ {
+ final DataSegment retained =
FULL_SEGMENT.retainOnlyDetails(SegmentDetail.none());
+
+ // Mandatory fields survive.
+ Assertions.assertEquals(FULL_SEGMENT.getId(), retained.getId());
+ Assertions.assertEquals(FULL_SEGMENT.getShardSpec(),
retained.getShardSpec());
+ Assertions.assertEquals(FULL_SEGMENT.getBinaryVersion(),
retained.getBinaryVersion());
+ Assertions.assertEquals(FULL_SEGMENT.getSize(), retained.getSize());
+
+ // Optional details do not.
+ Assertions.assertNull(retained.getLoadSpec());
+ Assertions.assertEquals(Collections.emptyList(), retained.getDimensions());
+ Assertions.assertEquals(Collections.emptyList(), retained.getMetrics());
+ Assertions.assertNull(retained.getProjections());
+ Assertions.assertNull(retained.getClusterGroups());
+ Assertions.assertNull(retained.getLastCompactionState());
+ Assertions.assertNull(retained.getTotalRows());
+ Assertions.assertNull(retained.getIndexingStateFingerprint());
+ }
+
+ @Test
+ public void test_retainOnlyDetails_all()
+ {
+ final DataSegment retained =
FULL_SEGMENT.retainOnlyDetails(SegmentDetail.all());
+ assertAllFieldsEqual(FULL_SEGMENT, retained);
+ }
+
+ @Test
+ public void test_retainOnlyDetails_null()
+ {
+ // Equivalent to "retain all".
+ final DataSegment retained = FULL_SEGMENT.retainOnlyDetails(null);
+ assertAllFieldsEqual(FULL_SEGMENT, retained);
+ }
+
+ @Test
+ public void test_retainOnlyDetails_some()
+ {
+ final DataSegment retained =
+ FULL_SEGMENT.retainOnlyDetails(EnumSet.of(SegmentDetail.LOAD_SPEC,
SegmentDetail.ROW_COUNT));
+
+ Assertions.assertEquals(FULL_SEGMENT.getLoadSpec(),
retained.getLoadSpec());
+ Assertions.assertEquals(FULL_SEGMENT.getTotalRows(),
retained.getTotalRows());
+
+ Assertions.assertEquals(Collections.emptyList(), retained.getDimensions());
+ Assertions.assertEquals(Collections.emptyList(), retained.getMetrics());
+ Assertions.assertNull(retained.getProjections());
+ Assertions.assertNull(retained.getClusterGroups());
+ Assertions.assertNull(retained.getLastCompactionState());
+ Assertions.assertNull(retained.getIndexingStateFingerprint());
+ }
+
+ @Test
+ public void test_all()
+ {
+ Assertions.assertEquals(SegmentDetail.values().length,
SegmentDetail.all().size());
+ }
+
+ @Test
+ public void test_none()
+ {
+ Assertions.assertEquals(0, SegmentDetail.none().size());
+ }
+
+ @Test
+ public void test_fromNamesLenient()
+ {
+ Assertions.assertNull(SegmentDetail.fromNamesLenient(null));
+ Assertions.assertEquals(SegmentDetail.none(),
SegmentDetail.fromNamesLenient(Collections.emptyList()));
+ Assertions.assertEquals(
+ EnumSet.of(SegmentDetail.LOAD_SPEC, SegmentDetail.ROW_COUNT),
+ SegmentDetail.fromNamesLenient(ImmutableList.of("loadSpec",
"a_detail_from_the_future", "totalRows"))
+ );
+ }
+
+ @Test
+ public void test_serde() throws Exception
+ {
+ for (final SegmentDetail detail : SegmentDetail.values()) {
+ final String json = MAPPER.writeValueAsString(detail);
+ Assertions.assertEquals("\"" + detail + "\"", json);
+ Assertions.assertEquals(detail, MAPPER.readValue(json,
SegmentDetail.class));
+ }
+
+ Assertions.assertEquals(SegmentDetail.LOAD_SPEC,
MAPPER.readValue("\"loadSpec\"", SegmentDetail.class));
+ Assertions.assertThrows(Exception.class, () ->
MAPPER.readValue("\"nonexistent\"", SegmentDetail.class));
+ }
+
+ @Test
+ public void test_serde_ofDataSegment() throws Exception
+ {
+ final DataSegment retained =
FULL_SEGMENT.retainOnlyDetails(EnumSet.of(SegmentDetail.LOAD_SPEC));
+ final DataSegment deserialized =
MAPPER.readValue(MAPPER.writeValueAsString(retained), DataSegment.class);
+ assertAllFieldsEqual(retained, deserialized);
+ Assertions.assertEquals(FULL_SEGMENT.getLoadSpec(),
deserialized.getLoadSpec());
+ }
+
+ private static void assertAllFieldsEqual(final DataSegment expected, final
DataSegment actual)
+ {
+ Assertions.assertEquals(expected.getId(), actual.getId(), "id");
+ Assertions.assertEquals(expected.getShardSpec(), actual.getShardSpec(),
"shardSpec");
+ Assertions.assertEquals(expected.getBinaryVersion(),
actual.getBinaryVersion(), "binaryVersion");
+ Assertions.assertEquals(expected.getSize(), actual.getSize(), "size");
+ Assertions.assertEquals(expected.getLoadSpec(), actual.getLoadSpec(),
"loadSpec");
+ Assertions.assertEquals(expected.getDimensions(), actual.getDimensions(),
"dimensions");
+ Assertions.assertEquals(expected.getMetrics(), actual.getMetrics(),
"metrics");
+ Assertions.assertEquals(expected.getProjections(),
actual.getProjections(), "projections");
+ Assertions.assertEquals(expected.getClusterGroups(),
actual.getClusterGroups(), "clusterGroups");
+ Assertions.assertEquals(expected.getLastCompactionState(),
actual.getLastCompactionState(), "lastCompactionState");
+ Assertions.assertEquals(expected.getTotalRows(), actual.getTotalRows(),
"totalRows");
+ Assertions.assertEquals(
+ expected.getIndexingStateFingerprint(),
+ actual.getIndexingStateFingerprint(),
+ "indexingStateFingerprint"
+ );
+ }
+}
diff --git
a/server/src/main/java/org/apache/druid/segment/realtime/appenderator/BaseAppenderatorDriver.java
b/server/src/main/java/org/apache/druid/segment/realtime/appenderator/BaseAppenderatorDriver.java
index b45c11bee7e..b0ec16c7f70 100644
---
a/server/src/main/java/org/apache/druid/segment/realtime/appenderator/BaseAppenderatorDriver.java
+++
b/server/src/main/java/org/apache/druid/segment/realtime/appenderator/BaseAppenderatorDriver.java
@@ -36,6 +36,7 @@ import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.SettableFuture;
import org.apache.druid.data.input.Committer;
import org.apache.druid.data.input.InputRow;
+import org.apache.druid.error.DruidException;
import org.apache.druid.indexing.overlord.SegmentPublishResult;
import org.apache.druid.java.util.common.ISE;
import org.apache.druid.java.util.common.Intervals;
@@ -687,8 +688,8 @@ public abstract class BaseAppenderatorDriver implements
Closeable
// Clean up pushed segments if they are physically
disjoint from the published ones (this means
// they were probably pushed by a replica, and with the
unique paths option).
final boolean physicallyDisjoint = Sets.intersection(
-
publishedSegments.stream().map(DataSegment::getLoadSpec).collect(Collectors.toSet()),
-
ourSegments.stream().map(DataSegment::getLoadSpec).collect(Collectors.toSet())
+ getLoadSpecs("published", publishedSegments),
+ getLoadSpecs("ours", ourSegments)
).isEmpty();
if (physicallyDisjoint) {
@@ -763,6 +764,26 @@ public abstract class BaseAppenderatorDriver implements
Closeable
executor.shutdownNow();
}
+ /**
+ * Returns a Set of {@link DataSegment#getLoadSpec()} from the provided
segments. Throws if any loadspecs
+ * are null, which may indicate that they were incorrectly pruned by {@link
DataSegment#retainOnlyDetails(Set)}.
+ *
+ * @param label label for the error message, if it fires
+ * @param dataSegments the segments
+ */
+ private static Set<Map<String, Object>> getLoadSpecs(String label,
Iterable<DataSegment> dataSegments)
+ {
+ final Set<Map<String, Object>> loadSpecs = new HashSet<>();
+ for (final DataSegment segment : dataSegments) {
+ final Map<String, Object> loadSpec = segment.getLoadSpec();
+ if (loadSpec == null) {
+ throw DruidException.defensive("Segment[%s] (%s) missing loadSpec",
segment.getId(), label);
+ }
+ loadSpecs.add(loadSpec);
+ }
+ return loadSpecs;
+ }
+
/**
* Wrapped committer for BaseAppenderatorDriver. Used in only {@link
StreamAppenderatorDriver} because batch ingestion
* doesn't need committing intermediate states.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
