tarun11Mavani commented on code in PR #19776:
URL: https://github.com/apache/pinot/pull/19776#discussion_r4466096250
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -621,6 +627,47 @@ private boolean isPeerUrl(String segmentLocation) {
CommonConstants.Segment.PEER_SEGMENT_DOWNLOAD_SCHEME.length());
}
+ /// Moves the metadata sidecar the server uploaded under a tmp name to
`<segment>.metadata.tar.gz`, like
+ /// [#moveSegmentFile] does for the segment file. Best effort: a failure is
logged and counted, and never fails the
+ /// commit.
+ private void moveMetadataTarFile(String realtimeTableName, String
rawTableName, String segmentName,
+ @Nullable String metadataTarLocation, PinotFS pinotFS) {
+ if (metadataTarLocation == null) {
+ return;
+ }
+ URI uriToMoveTo = createSegmentPath(rawTableName, segmentName +
METADATA_TAR_GZ_FILE_EXT);
+ try {
+ if (pinotFS.move(URIUtils.getUri(metadataTarLocation), uriToMoveTo,
true)) {
+ LOGGER.info("Moved metadata tar of segment: {} from: {} to: {}",
segmentName, metadataTarLocation,
+ uriToMoveTo);
+ return;
+ }
+ LOGGER.warn("Failed to move metadata tar of segment: {} from: {} to:
{}", segmentName, metadataTarLocation,
+ uriToMoveTo);
+ } catch (Exception e) {
+ LOGGER.warn("Caught exception while moving metadata tar of segment: {}
from: {} to: {}", segmentName,
+ metadataTarLocation, uriToMoveTo, e);
+ }
+ _controllerMetrics.addMeteredTableValue(realtimeTableName,
ControllerMeter.REALTIME_METADATA_TAR_MOVE_FAILURE,
+ 1L);
+ }
+
+ /// Best effort, like [#moveMetadataTarFile]: a failure is logged and
counted, and never fails the commit.
+ private void deleteMetadataTarFile(String rawTableName, String segmentName,
PinotFS pinotFS) {
+ URI metadataTarUri = createSegmentPath(rawTableName, segmentName +
METADATA_TAR_GZ_FILE_EXT);
+ try {
+ if (!pinotFS.exists(metadataTarUri) || pinotFS.delete(metadataTarUri,
true)) {
+ return;
+ }
+ LOGGER.warn("Failed to delete previous metadata tar of segment: {} at:
{}", segmentName, metadataTarUri);
+ } catch (Exception e) {
+ LOGGER.warn("Caught exception while deleting previous metadata tar of
segment: {} at: {}", segmentName,
+ metadataTarUri, e);
Review Comment:
Fixed in e2581e5: `deleteMetadataTarFile` now throws if the existence check
or delete fails, so `moveSegmentFile` never replaces the segment without
invalidating the old sidecar. Counted by the new
`REALTIME_METADATA_TAR_INVALIDATE_FAILURE`.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -3115,6 +3162,10 @@ String moveSegmentFile(String rawTableName, String
segmentName, String segmentLo
throws IOException {
URI segmentFileURI = URIUtils.getUri(segmentLocation);
URI uriToMoveTo = createSegmentPath(rawTableName, segmentName);
+ // An earlier attempt for this segment (e.g. one that published its files
and then failed the ZK metadata commit)
+ // may have left a sidecar describing a different build. Drop it before
replacing the segment file, on both the
+ // commit and the deep store repair path, so a sidecar only ever describes
the segment file next to it.
+ deleteMetadataTarFile(rawTableName, segmentName, pinotFS);
Review Comment:
Fixed in e2581e5 (guard refined in d933833): `ZKOperator` deletes the stale
sidecar before `copySegmentToDeepStore` on a CRC-changed refresh, except a
METADATA push whose own sidecar is the one at the final location. A failed
delete fails the refresh.
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/SplitSegmentCommitter.java:
##########
@@ -100,4 +126,53 @@ protected String uploadSegment(File segmentTarFile,
SegmentUploader segmentUploa
}
return null;
}
+
+ // A peer:// location means the segment is NOT in deep store, so there is
nothing for a sidecar to describe.
+ // Mirrors the check in PinotLLCRealtimeSegmentManager.commitSegmentFile().
+ private static boolean isPeerSegmentLocation(String segmentLocation) {
+ return segmentLocation.regionMatches(true, 0,
CommonConstants.Segment.PEER_SEGMENT_DOWNLOAD_SCHEME, 0,
+ CommonConstants.Segment.PEER_SEGMENT_DOWNLOAD_SCHEME.length());
+ }
+
+ /// Best-effort: builds a tar containing one directory with
metadata.properties and creation.meta, and uploads it
+ /// under a tmp name next to the segment. Returns the tmp location for the
controller to move at commit end, or
+ /// null on any problem. Never fails or throws; a sidecar problem must not
affect the segment commit.
+ @Nullable
+ private String uploadMetadataTarQuietly(@Nullable Map<String, File>
metadataFiles, String segmentName) {
+ if (metadataFiles == null || metadataFiles.isEmpty()) {
+ return null;
+ }
+ String rawTableName = null;
+ File stagingDir = null;
+ File metadataTarFile = null;
+ try {
+ LLCSegmentName llcSegmentName = new LLCSegmentName(segmentName);
+ rawTableName =
TableNameBuilder.extractRawTableName(llcSegmentName.getTableName());
+ stagingDir = new File(FileUtils.getTempDirectory(), segmentName +
"_meta_" + UUID.randomUUID());
+ FileUtils.forceMkdir(stagingDir);
+ for (Map.Entry<String, File> entry : metadataFiles.entrySet()) {
+ // Byte-for-byte copy: creation.meta is binary.
+ FileUtils.copyFile(entry.getValue(), new File(stagingDir,
entry.getKey()));
+ }
+ metadataTarFile = new File(FileUtils.getTempDirectory(),
+ segmentName + "_" + UUID.randomUUID() +
Constants.METADATA_TAR_GZ_FILE_EXT);
+ // Tar the DIRECTORY so that the first archive entry is a directory, as
DefaultMetadataExtractor expects.
+ TarCompressionUtils.createCompressedTarFile(stagingDir, metadataTarFile);
Review Comment:
Done in e2581e5: added `SegmentPushUtils.generateSegmentMetadataFile(Map,
File, String)`. The `File` overload delegates to it, and
`SplitSegmentCommitter` owns and cleans up its `outputDir`.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java:
##########
@@ -621,6 +627,47 @@ private boolean isPeerUrl(String segmentLocation) {
CommonConstants.Segment.PEER_SEGMENT_DOWNLOAD_SCHEME.length());
}
+ /// Moves the metadata sidecar the server uploaded under a tmp name to
`<segment>.metadata.tar.gz`, like
+ /// [#moveSegmentFile] does for the segment file. Best effort: a failure is
logged and counted, and never fails the
+ /// commit.
+ private void moveMetadataTarFile(String realtimeTableName, String
rawTableName, String segmentName,
+ @Nullable String metadataTarLocation, PinotFS pinotFS) {
+ if (metadataTarLocation == null) {
+ return;
+ }
+ URI uriToMoveTo = createSegmentPath(rawTableName, segmentName +
METADATA_TAR_GZ_FILE_EXT);
+ try {
+ if (pinotFS.move(URIUtils.getUri(metadataTarLocation), uriToMoveTo,
true)) {
+ LOGGER.info("Moved metadata tar of segment: {} from: {} to: {}",
segmentName, metadataTarLocation,
+ uriToMoveTo);
+ return;
+ }
+ LOGGER.warn("Failed to move metadata tar of segment: {} from: {} to:
{}", segmentName, metadataTarLocation,
+ uriToMoveTo);
+ } catch (Exception e) {
+ LOGGER.warn("Caught exception while moving metadata tar of segment: {}
from: {} to: {}", segmentName,
+ metadataTarLocation, uriToMoveTo, e);
+ }
+ _controllerMetrics.addMeteredTableValue(realtimeTableName,
ControllerMeter.REALTIME_METADATA_TAR_MOVE_FAILURE,
+ 1L);
+ }
+
+ /// Best effort, like [#moveMetadataTarFile]: a failure is logged and
counted, and never fails the commit.
+ private void deleteMetadataTarFile(String rawTableName, String segmentName,
PinotFS pinotFS) {
+ URI metadataTarUri = createSegmentPath(rawTableName, segmentName +
METADATA_TAR_GZ_FILE_EXT);
+ try {
+ if (!pinotFS.exists(metadataTarUri) || pinotFS.delete(metadataTarUri,
true)) {
+ return;
+ }
+ LOGGER.warn("Failed to delete previous metadata tar of segment: {} at:
{}", segmentName, metadataTarUri);
+ } catch (Exception e) {
+ LOGGER.warn("Caught exception while deleting previous metadata tar of
segment: {} at: {}", segmentName,
+ metadataTarUri, e);
+ }
+
_controllerMetrics.addMeteredTableValue(TableNameBuilder.REALTIME.tableNameWithType(rawTableName),
+ ControllerMeter.REALTIME_METADATA_TAR_MOVE_FAILURE, 1L);
Review Comment:
Done in e2581e5: invalidation failures now use the new
`REALTIME_METADATA_TAR_INVALIDATE_FAILURE`;
`REALTIME_METADATA_TAR_MOVE_FAILURE` counts only publication failures.
##########
pinot-server/src/main/java/org/apache/pinot/server/starter/helix/HelixInstanceDataManagerConfig.java:
##########
@@ -121,6 +121,10 @@ public class HelixInstanceDataManagerConfig implements
InstanceDataManagerConfig
public static final String UPLOAD_SEGMENT_TO_DEEP_STORE =
"segment.upload.to.deep.store";
public static final boolean DEFAULT_UPLOAD_SEGMENT_TO_DEEP_STORE = false;
+ public static final String UPLOAD_SEGMENT_METADATA_TAR_GZ =
"segment.upload.metadata.tar.gz";
+ public static final boolean DEFAULT_UPLOAD_SEGMENT_METADATA_TAR_GZ = false;
+ public static final String UPLOAD_SEGMENT_METADATA_TAR_GZ_TIMEOUT_MS =
"segment.upload.metadata.tar.gz.timeout.ms";
Review Comment:
Done in e2581e5: documented on `HelixInstanceDataManagerConfig` and the
`InstanceDataManagerConfig` accessor (direct uploader, non-pauseless,
`preserveMetadata` with positive retention, independent timeout).
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/PinotFSSegmentUploader.java:
##########
@@ -112,4 +116,81 @@ public URI uploadSegment(File segmentFile, LLCSegmentName
segmentName, int timeo
return null;
}
+
+ @Override
+ public boolean isMetadataTarUploadSupported() {
+ return true;
+ }
+
+ @Override
+ public URI uploadMetadataTar(File metadataTarFile, LLCSegmentName
segmentName, int timeoutInMillis) {
+ if (_segmentStoreUriStr == null || _segmentStoreUriStr.isEmpty()) {
+ LOGGER.error("Missing segment store uri. Failed to upload metadata tar
{} for {}.", metadataTarFile.getName(),
+ segmentName.getSegmentName());
+ return null;
+ }
+ final String rawTableName =
TableNameBuilder.extractRawTableName(segmentName.getTableName());
+ // Like the segment file, upload to a unique tmp name. The controller
moves it to <segment>.metadata.tar.gz at
+ // commit end. The final object is never written or deleted here, so a
failed or partial upload is never visible
+ // under the final name.
+ final URI destUri;
+ try {
+ destUri = new URI(StringUtil.join(File.separator, _segmentStoreUriStr,
segmentName.getTableName(),
+ SegmentCompletionUtils.generateTmpSegmentFileName(
+ segmentName.getSegmentName() +
Constants.METADATA_TAR_GZ_FILE_EXT)));
+ } catch (URISyntaxException e) {
+ LOGGER.warn("Invalid segment store uri {} for metadata tar of segment
{}", _segmentStoreUriStr, segmentName, e);
+ _serverMetrics.addMeteredTableValue(rawTableName,
ServerMeter.METADATA_TAR_UPLOAD_FAILURE, 1);
+ return null;
+ }
+ Callable<URI> uploadTask = () -> {
+ long startTime = System.currentTimeMillis();
+ try {
+ PinotFS pinotFS = PinotFSFactory.create(new
URI(_segmentStoreUriStr).getScheme());
+ pinotFS.copyFromLocalFile(metadataTarFile, destUri);
Review Comment:
Not changing this: the two paths differ in pre-delete, cancellation, tmp
cleanup and meters, so a shared helper would mostly be flags. Happy to revisit
if you prefer it.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/SegmentDeletionManager.java:
##########
@@ -77,29 +78,36 @@ public class SegmentDeletionManager {
// file will be permanently deleted after Feb 2nd 2022 12PM.
public static final String DELETED_SEGMENTS = "Deleted_Segments";
private static final String RETENTION_UNTIL_SEPARATOR =
"__RETENTION_UNTIL__";
- private static final String RETENTION_DATE_FORMAT_STR = "yyyyMMddHHmm";
- private static final SimpleDateFormat RETENTION_DATE_FORMAT;
+ // Subdirectory of Deleted_Segments/<table>/ holding retained metadata
sidecars, named like the archived segment.
+ // '~' is never produced by URIUtils.encode (URLEncoder), so no archived
segment name can equal this directory name.
+ public static final String DELETED_SEGMENT_METADATA_DIR = "~metadata";
Review Comment:
Good catch, reproduced the decode on LocalPinotFS. Fixed in d933833:
sidecars now go to a separate root, `Deleted_Segments_Metadata/<table>/`, so no
segment name can collide. A test covers `metadata`, `~metadata` and the root
name in both deletion orders.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/upload/ZKOperator.java:
##########
@@ -678,4 +682,22 @@ private void copyFromSegmentURIToDeepStore(URI
sourceDownloadURI, URI finalSegme
PinotFSFactory.create(finalSegmentLocationURI.getScheme()).copy(sourceDownloadURI,
finalSegmentLocationURI);
}
}
+
+ /// A refresh replaces the full segment, so a sidecar left by the previous
build would describe the wrong bytes (a
+ /// metadata push trusts an existing sidecar). Not needed when a METADATA
push already points at the final location:
+ /// nothing is copied and the sidecar there belongs to that push. Failing to
invalidate fails the refresh.
+ private void invalidateStaleMetadataTar(String segmentName, FileUploadType
uploadType, String sourceDownloadURIStr,
+ URI finalSegmentLocationURI)
+ throws Exception {
+ if (uploadType == FileUploadType.METADATA && new
URI(sourceDownloadURIStr).equals(finalSegmentLocationURI)) {
Review Comment:
Right, the guard never matched. Fixed in d933833: it now compares the
sidecar next to the source tar with the one at the final location. The new test
uses realistic URIs and failed before the fix.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/SegmentDeletionManager.java:
##########
@@ -369,6 +382,36 @@ private void moveSegmentsToDeletedDir(String segmentId,
Long deletedSegmentsRete
LOGGER.warn("Could not move segment {} from {} to {}", segmentId,
fileToDeleteURI.toString(),
deletedSegmentMoveDestURI.toString(), e);
}
+ if (_preserveDeletedSegmentMetadata) {
+ // Even if the segment object vanished after it was located, so no
orphan sidecar is left in the live directory.
+ moveSegmentMetadataToDeletedDir(pinotFS, fileToDeleteURI, segmentId,
rawTableName, retentionSuffix);
+ }
+ }
+
+ private void moveSegmentMetadataToDeletedDir(PinotFS pinotFS, URI
segmentFileUri, String segmentId,
+ String rawTableName, String retentionSuffix) {
+ // Retained sidecars live in their own subdirectory, so they can never
take the archive name of a segment (e.g. the
+ // sidecar of "foo" vs. a segment named "foo.metadata"). The name matches
the archived segment, so both carry the
+ // same retention suffix and expire at the same time.
+ try {
+ URI segmentMetadataUri =
SegmentPushUtils.generateSegmentMetadataURI(segmentFileUri.toString(),
segmentId);
+ if (!pinotFS.exists(segmentMetadataUri)) {
+ return;
+ }
+ URI deletedMetadataUri = URIUtils.getUri(_dataDir, DELETED_SEGMENTS,
rawTableName, DELETED_SEGMENT_METADATA_DIR,
+ URIUtils.encode(segmentId) + retentionSuffix);
+ if (pinotFS.move(segmentMetadataUri, deletedMetadataUri, true)) {
Review Comment:
Fixed in e2581e5 and tightened in d933833: retained sidecars now live in a
separate root, `Deleted_Segments_Metadata/<table>/`, so no archived segment
name (e.g. `metadata`) can share their directory.
--
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]
