Skip to content

Build METADATA push metadata from the local segment instead of re-downloading the staged tar #19490

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
KKcorps wants to merge 2 commits into
base: master
Choose a base branch
from
Draft

Build METADATA push metadata from the local segment instead of re-downloading the staged tar #19490

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c0a33a7
Add an opt-in controller-copy METADATA push for same-name segment ref…
Sep 7, 2026
3f20c40
Build METADATA push metadata from the local segment instead of re-dow…
Sep 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
67 changes: 34 additions & 33 deletions ...ain/java/org/apache/pinot/plugin/minion/tasks/BaseMultipleSegmentsConversionExecutor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.stream.Collectors;
import javax.annotation.Nullable;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;
Expand All @@ -57,7 +58,6 @@
import org.apache.pinot.segment.spi.index.metadata.SegmentMetadataImpl;
import org.apache.pinot.spi.auth.AuthProvider;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.filesystem.PinotFS;
import org.apache.pinot.spi.ingestion.batch.BatchConfigProperties;
import org.apache.pinot.spi.ingestion.batch.spec.PushJobSpec;
import org.apache.pinot.spi.ingestion.batch.spec.SegmentGenerationJobSpec;
Expand Down Expand Up @@ -214,6 +214,10 @@ public List<SegmentConversionResult> executeTask(PinotTaskConfig pinotTaskConfig

int numOutputSegments = segmentConversionResults.size();
List<File> tarredSegmentFiles = new ArrayList<>(numOutputSegments);
// METADATA pushes send only metadata.properties and creation.meta, built here from the local converted segment
// so the staged tar is never downloaded back. Null entries for TAR push.
List<File> segmentMetadataTarFiles = new ArrayList<>(numOutputSegments);
BatchConfigProperties.SegmentPushType pushType = getSegmentPushType(taskConfigs);
int count = 1;
for (SegmentConversionResult segmentConversionResult : segmentConversionResults) {
File convertedSegmentDir = segmentConversionResult.getFile();
Expand All @@ -226,6 +230,9 @@ public List<SegmentConversionResult> executeTask(PinotTaskConfig pinotTaskConfig
segmentConversionResult.getSegmentName() + TarCompressionUtils.TAR_GZ_FILE_EXTENSION);
TarCompressionUtils.createCompressedTarFile(convertedSegmentDir, convertedSegmentTarFile);
tarredSegmentFiles.add(convertedSegmentTarFile);
segmentMetadataTarFiles.add(pushType == BatchConfigProperties.SegmentPushType.TAR ? null
: SegmentPushUtils.generateSegmentMetadataFile(convertedSegmentDir, convertedTarredSegmentDir,
segmentConversionResult.getSegmentName()));
if (!FileUtils.deleteQuietly(convertedSegmentDir)) {
LOGGER.warn("Failed to delete converted segment: {}", convertedSegmentDir.getAbsolutePath());
}
Expand All @@ -249,13 +256,14 @@ public List<SegmentConversionResult> executeTask(PinotTaskConfig pinotTaskConfig

SegmentUploadContext segmentUploadContext = new SegmentUploadContext(pinotTaskConfig, segmentConversionResults);
preUploadSegments(segmentUploadContext);
Map<String, String> segmentUriToTarPathMap = new HashMap<>();
Map<String, File> segmentUriToMetadataFileMap = new HashMap<>();
PushJobSpec pushJobSpec = getPushJobSpec(taskConfigs);
boolean batchSegmentUpload = pushJobSpec.isBatchSegmentUpload();

// Upload the tarred segments
for (int i = 0; i < numOutputSegments; i++) {
File convertedTarredSegmentFile = tarredSegmentFiles.get(i);
File segmentMetadataTarFile = segmentMetadataTarFiles.get(i);
SegmentConversionResult segmentConversionResult = segmentConversionResults.get(i);
String resultSegmentName = segmentConversionResult.getSegmentName();
_eventObserver.notifyProgress(_pinotTaskConfig, "Uploading segment: " + resultSegmentName + " (" + (i + 1)
Expand Down Expand Up @@ -294,20 +302,24 @@ public List<SegmentConversionResult> executeTask(PinotTaskConfig pinotTaskConfig
}

if (batchSegmentUpload) {
updateSegmentUriToTarPathMap(taskConfigs, outputSegmentTarURI, segmentConversionResult,
segmentUriToTarPathMap, pushJobSpec);
for (String segmentUri : getSegmentUris(taskConfigs, outputSegmentTarURI, segmentConversionResult,
pushJobSpec)) {
segmentUriToMetadataFileMap.put(segmentUri, segmentMetadataTarFile);
}
} else {
String rawTableName = TableNameBuilder.extractRawTableName(tableNameWithType);
pushSegment(rawTableName, taskConfigs, outputSegmentTarURI, httpHeaders, parameters, segmentConversionResult);
pushSegment(rawTableName, taskConfigs, outputSegmentTarURI, httpHeaders, parameters, segmentConversionResult,
segmentMetadataTarFile);
if (!FileUtils.deleteQuietly(convertedTarredSegmentFile)) {
LOGGER.warn("Failed to delete tarred converted segment: {}", convertedTarredSegmentFile.getAbsolutePath());
}
FileUtils.deleteQuietly(segmentMetadataTarFile);
}
}

if (batchSegmentUpload) {
try {
pushSegments(tableNameWithType, taskConfigs, pinotTaskConfig, segmentUriToTarPathMap, pushJobSpec,
pushSegments(tableNameWithType, taskConfigs, pinotTaskConfig, segmentUriToMetadataFileMap, pushJobSpec,
authProvider, segmentConversionResults);
} finally {
for (File convertedTarredSegmentFile : tarredSegmentFiles) {
Expand All @@ -316,6 +328,7 @@ public List<SegmentConversionResult> executeTask(PinotTaskConfig pinotTaskConfig
convertedTarredSegmentFile.getAbsolutePath());
}
}
segmentMetadataTarFiles.forEach(FileUtils::deleteQuietly);
}
}

Expand Down Expand Up @@ -410,21 +423,17 @@ private void downloadAndUntarSegment(String tableNameWithType, String taskType,
}
}

/// Download URIs the controller should register for a staged segment tar (prefix/suffix rules from the push spec).
@VisibleForTesting
void updateSegmentUriToTarPathMap(Map<String, String> taskConfigs, URI outputSegmentTarURI,
SegmentConversionResult segmentConversionResult, Map<String, String> segmentUriToTarPathMap,
PushJobSpec pushJobSpec) {
Set<String> getSegmentUris(Map<String, String> taskConfigs, URI outputSegmentTarURI,
SegmentConversionResult segmentConversionResult, PushJobSpec pushJobSpec) {
String segmentName = segmentConversionResult.getSegmentName();
if (!taskConfigs.containsKey(BatchConfigProperties.OUTPUT_SEGMENT_DIR_URI)) {
throw new RuntimeException("Output dir URI missing for metadata push while processing segment: " + segmentName);
}
URI outputSegmentDirURI = URI.create(taskConfigs.get(BatchConfigProperties.OUTPUT_SEGMENT_DIR_URI));
Map<String, String> localSegmentUriToTarPathMap =
SegmentPushUtils.getSegmentUriToTarPathMap(outputSegmentDirURI, pushJobSpec,
new String[]{outputSegmentTarURI.toString()});
if (!localSegmentUriToTarPathMap.isEmpty()) {
segmentUriToTarPathMap.putAll(localSegmentUriToTarPathMap);
}
return SegmentPushUtils.getSegmentUriToTarPathMap(outputSegmentDirURI, pushJobSpec,
new String[]{outputSegmentTarURI.toString()}).keySet();
}

@VisibleForTesting
Expand All @@ -436,23 +445,20 @@ List<Header> getSegmentPushCommonHeaders(PinotTaskConfig pinotTaskConfig, AuthPr
}

private void pushSegments(String tableNameWithType, Map<String, String> taskConfigs, PinotTaskConfig pinotTaskConfig,
Map<String, String> segmentUriToTarPathMap, PushJobSpec pushJobSpec,
AuthProvider authProvider, List<SegmentConversionResult> segmentConversionResults)
Map<String, File> segmentUriToMetadataFileMap, PushJobSpec pushJobSpec, AuthProvider authProvider,
List<SegmentConversionResult> segmentConversionResults)
throws Exception {
String tableName = TableNameBuilder.extractRawTableName(tableNameWithType);
SegmentGenerationJobSpec spec = generateSegmentGenerationJobSpec(tableName, taskConfigs, pushJobSpec);

List<Header> headers = getSegmentPushCommonHeaders(pinotTaskConfig, authProvider, segmentConversionResults);
List<NameValuePair> parameters = getSegmentPushCommonParams(tableNameWithType);

URI outputSegmentDirURI = URI.create(taskConfigs.get(BatchConfigProperties.OUTPUT_SEGMENT_DIR_URI));
try (PinotFS outputFileFS = MinionTaskUtils.getOutputPinotFS(taskConfigs, outputSegmentDirURI)) {
SegmentPushUtils.sendSegmentsUriAndMetadata(spec, outputFileFS, segmentUriToTarPathMap, headers, parameters);
}
SegmentPushUtils.sendSegmentsUriAndMetadata(spec, segmentUriToMetadataFileMap, headers, parameters);
}

private void pushSegment(String tableName, Map<String, String> taskConfigs, URI outputSegmentTarURI,
List<Header> headers, List<NameValuePair> parameters, SegmentConversionResult segmentConversionResult)
List<Header> headers, List<NameValuePair> parameters, SegmentConversionResult segmentConversionResult,
@Nullable File segmentMetadataTarFile)
throws Exception {
BatchConfigProperties.SegmentPushType pushType = getSegmentPushType(taskConfigs);
LOGGER.info("Trying to push Pinot segment with push mode {} from {}", pushType, outputSegmentTarURI);
Expand All @@ -470,17 +476,12 @@ private void pushSegment(String tableName, Map<String, String> taskConfigs, URI
uploadURL, tarFile);
break;
case METADATA:
if (taskConfigs.containsKey(BatchConfigProperties.OUTPUT_SEGMENT_DIR_URI)) {
URI outputSegmentDirURI = URI.create(taskConfigs.get(BatchConfigProperties.OUTPUT_SEGMENT_DIR_URI));
try (PinotFS outputFileFS = MinionTaskUtils.getOutputPinotFS(taskConfigs, outputSegmentDirURI)) {
Map<String, String> segmentUriToTarPathMap =
SegmentPushUtils.getSegmentUriToTarPathMap(outputSegmentDirURI, pushJobSpec,
new String[]{outputSegmentTarURI.toString()});
SegmentPushUtils.sendSegmentUriAndMetadata(spec, outputFileFS, segmentUriToTarPathMap, headers, parameters);
}
} else {
throw new RuntimeException("Output dir URI missing for metadata push");
Map<String, File> segmentUriToMetadataFileMap = new HashMap<>();
for (String segmentUri : getSegmentUris(taskConfigs, outputSegmentTarURI, segmentConversionResult,
pushJobSpec)) {
segmentUriToMetadataFileMap.put(segmentUri, segmentMetadataTarFile);
}
SegmentPushUtils.sendSegmentUriAndMetadata(spec, segmentUriToMetadataFileMap, headers, parameters);
break;
default:
throw new UnsupportedOperationException("Unrecognized push mode - " + pushType);
Expand Down
Loading
Loading