AmatyaAvadhanula commented on code in PR #16144:
URL: https://github.com/apache/druid/pull/16144#discussion_r1566969884
##########
server/src/main/java/org/apache/druid/metadata/IndexerSQLMetadataStorageCoordinator.java:
##########
@@ -1473,183 +1427,6 @@ private void insertPendingSegmentIntoMetastore(
.execute();
}
- /**
- * Creates new IDs for the given append segments if a REPLACE task started
and
- * finished after these append segments had already been allocated. The newly
- * created IDs belong to the same interval and version as the segments
committed
- * by the REPLACE task.
- */
- private Set<DataSegment> createNewIdsForAppendSegments(
- Handle handle,
- String dataSource,
- Set<DataSegment> segmentsToAppend
- )
- {
- if (segmentsToAppend.isEmpty()) {
- return Collections.emptySet();
- }
-
- final Set<Interval> appendIntervals = new HashSet<>();
- final TreeMap<String, Set<DataSegment>> appendVersionToSegments = new
TreeMap<>();
- for (DataSegment segment : segmentsToAppend) {
- appendIntervals.add(segment.getInterval());
- appendVersionToSegments.computeIfAbsent(segment.getVersion(), v -> new
HashSet<>())
- .add(segment);
- }
-
- // Fetch all used segments that overlap with any of the append intervals
- final Collection<DataSegment> overlappingSegments =
retrieveUsedSegmentsForIntervals(
- dataSource,
- new ArrayList<>(appendIntervals),
- Segments.INCLUDING_OVERSHADOWED
- );
-
- final Map<String, Set<Interval>> overlappingVersionToIntervals = new
HashMap<>();
- final Map<Interval, Set<DataSegment>> overlappingIntervalToSegments = new
HashMap<>();
- for (DataSegment segment : overlappingSegments) {
- overlappingVersionToIntervals.computeIfAbsent(segment.getVersion(), v ->
new HashSet<>())
- .add(segment.getInterval());
- overlappingIntervalToSegments.computeIfAbsent(segment.getInterval(), i
-> new HashSet<>())
- .add(segment);
- }
-
- final Set<DataSegment> upgradedSegments = new HashSet<>();
- for (Map.Entry<String, Set<Interval>> entry :
overlappingVersionToIntervals.entrySet()) {
- final String upgradeVersion = entry.getKey();
- Map<Interval, Set<DataSegment>> segmentsToUpgrade =
getSegmentsWithVersionLowerThan(
- upgradeVersion,
- entry.getValue(),
- appendVersionToSegments
- );
- for (Map.Entry<Interval, Set<DataSegment>> upgradeEntry :
segmentsToUpgrade.entrySet()) {
- final Interval upgradeInterval = upgradeEntry.getKey();
- final Set<DataSegment> segmentsAlreadyOnVersion
- = overlappingIntervalToSegments.getOrDefault(upgradeInterval,
Collections.emptySet())
- .stream()
- .filter(s ->
s.getVersion().equals(upgradeVersion))
- .collect(Collectors.toSet());
- Set<DataSegment> segmentsUpgradedToVersion =
createNewIdsForAppendSegmentsWithVersion(
- handle,
- upgradeVersion,
- upgradeInterval,
- upgradeEntry.getValue(),
- segmentsAlreadyOnVersion
- );
- log.info("Upgraded [%d] segments to version[%s].",
segmentsUpgradedToVersion.size(), upgradeVersion);
- upgradedSegments.addAll(segmentsUpgradedToVersion);
- }
- }
-
- return upgradedSegments;
- }
-
- /**
- * Creates a Map from eligible interval to Set of segments that are fully
- * contained in that interval and have a version strictly lower than {@code
#cutoffVersion}.
- */
- private Map<Interval, Set<DataSegment>> getSegmentsWithVersionLowerThan(
Review Comment:
Yes, they aren't
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
