Jackie-Jiang commented on code in PR #19776:
URL: https://github.com/apache/pinot/pull/19776#discussion_r4449360692
##########
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:
**[P1] Require successful sidecar invalidation before replacing the
segment.** If an earlier attempt published its segment and sidecar but failed
the ZK commit, another replica can retry with different segment bytes. Here a
failed existence check or deletion is logged and swallowed, so
`moveSegmentFile()` still replaces the full segment. If the retry has no
successfully published replacement sidecar, the previous metadata remains
beside different segment bytes. `SegmentPushUtils` trusts an existing sidecar
during metadata push without checking it against the full segment. Please
propagate/retry invalidation failures before replacing the segment; optional
creation can remain best effort. Add coverage for failed `exists()` and failed
deletion with an existing sidecar.
##########
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:
**[P2] Invalidate sidecars on controller-upload refreshes as well.** This
only covers the LLC commit/repair move. Upsert compaction can replace a
REALTIME segment under the same name through
`BaseSingleSegmentConversionExecutor`'s TAR upload and
`ZKOperator.copySegmentToDeepStore()`, which overwrites the full segment
without removing its sidecar. A later metadata push can therefore consume the
old CRC/document counts, and deletion preserves a mismatched pair. This
weakness already existed for some OFFLINE producers; this PR newly creates the
durable sidecar for normal REALTIME ingestion. The PR acknowledges the
limitation, but please apply invalidation to the common controller replacement
path too; regeneration is unnecessary when the upload contains only the full
tar.
##########
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:
**Cleanup: distinguish invalidation failures from publication failures.**
`REALTIME_METADATA_TAR_MOVE_FAILURE` counts failed deletion of previous
metadata as well as failed sidecar moves. Separate invalidation and publication
counters would make diagnosis and alerting clearer: failed invalidation can
leave stale metadata, while failed publication after successful invalidation
normally means falling back to the full segment.
##########
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:
**[P2] Keep the metadata directory outside the segment archive namespace.**
With preservation enabled, archive another segment's sidecar first, then delete
a valid OFFLINE segment named `metadata` using default/null retention. Its
full-segment archive destination is the existing
`Deleted_Segments/<table>/metadata` directory. `BasePinotFS.move(..., true)`
recursively deletes that destination before replacing it, removing all retained
sidecars for the table before their retention expires. On directory-based
stores, archiving the `metadata` segment first instead prevents later sidecar
moves beneath that file. Full segment archives remain intact. This is the
acknowledged reserved-name limitation, distinct from the now-fixed
`foo.metadata` collision. Please use a collision-free layout or safely handle
the collision before overwriting, with tests for both deletion orders.
##########
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:
**Cleanup: document the flag's supported modes beside the config.** Please
explain that metadata upload requires the direct server-to-store uploader and
non-pauseless consumption; controller-upload commits and pauseless consumption
do not create sidecars. Also point out that preserving sidecars after deletion
requires `controller.deleted.segments.preserveMetadata=true` and positive
retention. The timeout independently bounds the additional metadata upload
wait. These details are useful at the config/accessor declaration so operators
and callers do not need the PR description or committer implementation to
discover the restrictions.
##########
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:
**Optional cleanup: share the upload bookkeeping.** This largely repeats
`uploadSegment()`'s copy, timing, future wait, and outcome recording. A small
private helper could share those mechanics and reduce future drift between the
two upload paths. Keep the segment-specific and metadata-specific
cancellation/cleanup policies explicit when doing so; this suggestion is
non-blocking.
##########
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:
**Cleanup: share metadata archive construction with OFFLINE.** This
staging/copy/tar sequence duplicates
`SegmentPushUtils.generateSegmentMetadataFile(File, File, String)`. Consider
extracting an overload/helper accepting the metadata files while keeping each
caller's temporary-directory ownership explicit. That gives OFFLINE and
REALTIME one implementation of the top-level directory layout and byte-for-byte
`creation.meta` handling, reducing the chance that the archive formats diverge.
--
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]
