xiangfu0 commented on code in PR #19947:
URL: https://github.com/apache/pinot/pull/19947#discussion_r4364889605
##########
pinot-plugins/pinot-batch-ingestion/pinot-batch-ingestion-hadoop/src/main/java/org/apache/pinot/plugin/ingestion/batch/hadoop/HadoopSegmentTarPushJobRunner.java:
##########
@@ -80,17 +79,12 @@ public void run() {
throw new RuntimeException("Unable to list all files under outputDirURI
- '" + outputDirURI + "'");
}
- List<String> segmentsToPush = new ArrayList<>();
- for (String file : files) {
- if (file.endsWith(Constants.TAR_GZ_FILE_EXT)) {
- segmentsToPush.add(file);
- }
+ List<String> segmentsToPush = new ArrayList<>(
+ SegmentPushUtils.getSegmentUriToTarPathMap(outputDirURI,
_spec.getPushJobSpec(), files).values());
Review Comment:
Clarified the selection rules in the PR description: Hadoop/Spark honor
pushFileNamePattern and skip .metadata.tar.gz sidecars. After rebasing, this
shared selection behavior is inherited from merged #19898; #19947 adds
duplicate validation after filtering.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/SegmentPushUtils.java:
##########
@@ -684,17 +702,20 @@ public static Map<String, String>
getSegmentUriToTarPathMap(URI outputDirURI, Pu
}
}
- URI uri = URI.create(file);
+ URI uri = outputDirURI != null &&
"file".equals(outputDirURI.getScheme()) && new File(file).isAbsolute()
+ ? new File(file).toURI() : URI.create(file);
if (uri.getPath().endsWith(Constants.METADATA_TAR_GZ_FILE_EXT)) {
// Skip segment metadata tar gz files
continue;
}
if (uri.getPath().endsWith(Constants.TAR_GZ_FILE_EXT)) {
+ segmentPaths.add(file);
URI updatedURI = SegmentPushUtils.generateSegmentTarURI(outputDirURI,
uri, pushSpec.getSegmentUriPrefix(),
pushSpec.getSegmentUriSuffix());
segmentUriToTarPathMap.put(updatedURI.toString(), file);
}
}
+ validateUniqueSegmentNames(segmentPaths);
Review Comment:
Documented that selected logical segment names now pass the existing
illegal-character validation before replacement/upload. This intentionally
fails early for invalid names instead of letting them proceed through the URI
map.
##########
pinot-plugins/pinot-batch-ingestion/pinot-batch-ingestion-hadoop/src/main/java/org/apache/pinot/plugin/ingestion/batch/hadoop/HadoopSegmentTarPushJobRunner.java:
##########
@@ -80,17 +79,12 @@ public void run() {
throw new RuntimeException("Unable to list all files under outputDirURI
- '" + outputDirURI + "'");
}
- List<String> segmentsToPush = new ArrayList<>();
- for (String file : files) {
- if (file.endsWith(Constants.TAR_GZ_FILE_EXT)) {
- segmentsToPush.add(file);
- }
+ List<String> segmentsToPush = new ArrayList<>(
+ SegmentPushUtils.getSegmentUriToTarPathMap(outputDirURI,
_spec.getPushJobSpec(), files).values());
Review Comment:
Updated the compatibility note with the Hadoop/Spark pattern and
metadata-sidecar selection rules, plus an example for selecting one output
cohort. The rebase now inherits those shared selection rules from merged #19898.
--
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]
