Copilot commented on code in PR #19082:
URL: https://github.com/apache/pinot/pull/19082#discussion_r3762688204
##########
pinot-plugins/pinot-batch-ingestion/pinot-batch-ingestion-common/src/main/java/org/apache/pinot/plugin/ingestion/batch/common/SegmentGenerationJobUtils.java:
##########
@@ -104,26 +114,173 @@ public static void moveLocalTarFileToRemote(File
localMetadataTarFile, URI outpu
/// If <overwrite> is true, and the source file exists in the destination
directory, then replace it, otherwise
/// log a warning and continue. We assume that source and destination
directories are on the same filesystem,
/// so that move() can be used.
+ /// Uses [#DEFAULT_STAGING_COPY_PARALLELISM] worker threads.
///
- /// @param fs
- /// @param sourceDir
- /// @param destDir
- /// @param overwrite
- /// @throws IOException
- /// @throws URISyntaxException
+ /// The shared [PinotFS] instance must support concurrent `move` of distinct
paths (true for
+ /// `LocalPinotFS` and typical remote implementations).
+ ///
+ /// @param fs filesystem used for both source and destination
+ /// @param sourceDir source directory URI
+ /// @param destDir destination directory URI
+ /// @param overwrite whether to overwrite existing destination files
+ /// @throws IOException on listing or move failure
+ /// @throws URISyntaxException on URI construction failure
public static void moveFiles(PinotFS fs, URI sourceDir, URI destDir, boolean
overwrite)
- throws IOException, URISyntaxException {
- for (String sourcePath : fs.listFiles(sourceDir, true)) {
- URI sourceFileUri = SegmentGenerationUtils.getFileURI(sourcePath,
sourceDir);
- String sourceFilename =
SegmentGenerationUtils.getFileName(sourceFileUri);
- URI destFileUri =
- SegmentGenerationUtils.getRelativeOutputPath(sourceDir,
sourceFileUri, destDir).resolve(sourceFilename);
+ throws IOException, URISyntaxException {
+ moveFiles(fs, sourceDir, destDir, overwrite,
DEFAULT_STAGING_COPY_PARALLELISM);
+ }
+
+ /// Move all files from the <sourceDir> to the <destDir> using up to
`parallelism` threads.
+ /// Directories in the source listing are skipped; parent directories on the
destination are created by
+ /// [PinotFS#move]. Relative path layout under `sourceDir` is preserved.
+ ///
+ /// `parallelism` is clamped to at most [#MAX_STAGING_COPY_PARALLELISM] and
to the number of files to
+ /// move, so direct callers of this overload cannot exceed the cap that
+ /// [#getStagingCopyParallelism] enforces.
+ ///
+ /// On partial failure, remaining in-flight moves are allowed to finish,
then an [IOException] is thrown
+ /// with any additional failures attached as suppressed exceptions. On
interruption, all outstanding
+ /// moves are cancelled, the interrupt status is restored, and an
[IOException] is thrown.
+ ///
+ /// @param fs filesystem used for both source and destination (must be safe
for concurrent move of
+ /// distinct paths)
+ /// @param sourceDir source directory URI
+ /// @param destDir destination directory URI
+ /// @param overwrite whether to overwrite existing destination files
+ /// @param parallelism number of concurrent move workers; values <= 1 run
serially
+ /// @throws IOException on listing or move failure
+ /// @throws URISyntaxException on URI construction failure
+ public static void moveFiles(PinotFS fs, URI sourceDir, URI destDir, boolean
overwrite, int parallelism)
+ throws IOException, URISyntaxException {
+ List<URI> sourceFileUris = listSourceFiles(fs, sourceDir);
+ if (sourceFileUris.isEmpty()) {
+ return;
+ }
+ int effectiveParallelism =
+ Math.max(1, Math.min(Math.min(parallelism,
MAX_STAGING_COPY_PARALLELISM), sourceFileUris.size()));
+ LOGGER.info("Moving {} files from [{}] to [{}] with parallelism {}",
sourceFileUris.size(), sourceDir, destDir,
+ effectiveParallelism);
+ if (effectiveParallelism == 1) {
+ for (URI sourceFileUri : sourceFileUris) {
+ moveOneFile(fs, sourceDir, sourceFileUri, destDir, overwrite);
+ }
+ return;
+ }
- if (!overwrite && fs.exists(destFileUri)) {
- LOGGER.warn("Can't overwrite existing output segment tar file: {}",
destFileUri);
- } else {
- fs.move(sourceFileUri, destFileUri, true);
+ ExecutorService executor =
Executors.newFixedThreadPool(effectiveParallelism, r -> {
+ Thread t = new Thread(r, "pinot-staging-copy");
+ t.setDaemon(true);
+ return t;
+ });
+ try {
+ List<Future<Void>> futures = new ArrayList<>(sourceFileUris.size());
+ for (URI sourceFileUri : sourceFileUris) {
+ futures.add(executor.submit(() -> {
+ moveOneFile(fs, sourceDir, sourceFileUri, destDir, overwrite);
+ return null;
+ }));
+ }
+ IOException firstFailure = null;
+ for (Future<Void> future : futures) {
+ try {
+ future.get();
+ } catch (InterruptedException e) {
+ // Cancel every outstanding move and stop waiting instead of
blocking on the remaining futures.
+ futures.forEach(f -> f.cancel(true));
+ Thread.currentThread().interrupt();
+ IOException interruptedFailure =
+ new IOException("Interrupted while moving files from " +
sourceDir + " to " + destDir, e);
+ if (firstFailure == null) {
+ firstFailure = interruptedFailure;
+ } else {
+ firstFailure.addSuppressed(interruptedFailure);
+ }
+ break;
+ } catch (Exception e) {
+ Throwable cause = e.getCause() != null ? e.getCause() : e;
+ if (firstFailure == null) {
+ firstFailure = cause instanceof IOException ? (IOException) cause
+ : new IOException("Failed to move files from " + sourceDir + "
to " + destDir, cause);
+ } else {
+ firstFailure.addSuppressed(cause);
+ }
+ }
}
+ if (firstFailure != null) {
+ throw firstFailure;
+ }
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ /// Resolve staging-copy parallelism from job
`executionFrameworkSpec.extraConfigs`.
+ /// Missing/invalid/non-positive values fall back to
[#DEFAULT_STAGING_COPY_PARALLELISM].
+ /// Values above [#MAX_STAGING_COPY_PARALLELISM] are capped.
+ public static int getStagingCopyParallelism(Map<String, String>
extraConfigs) {
+ if (extraConfigs == null) {
+ return DEFAULT_STAGING_COPY_PARALLELISM;
+ }
+ String value = extraConfigs.get(STAGING_COPY_PARALLELISM);
+ if (value == null || value.isEmpty()) {
+ return DEFAULT_STAGING_COPY_PARALLELISM;
+ }
+ try {
+ int parallelism = Integer.parseInt(value.trim());
+ if (parallelism < 1) {
+ LOGGER.warn("Invalid {}={}, using default {}",
STAGING_COPY_PARALLELISM, value,
+ DEFAULT_STAGING_COPY_PARALLELISM);
+ return DEFAULT_STAGING_COPY_PARALLELISM;
+ }
+ if (parallelism > MAX_STAGING_COPY_PARALLELISM) {
+ LOGGER.warn("Capping {}={} to max {}", STAGING_COPY_PARALLELISM,
parallelism, MAX_STAGING_COPY_PARALLELISM);
+ return MAX_STAGING_COPY_PARALLELISM;
+ }
+ return parallelism;
+ } catch (NumberFormatException e) {
+ LOGGER.warn("Invalid {}={}, using default {}", STAGING_COPY_PARALLELISM,
value, DEFAULT_STAGING_COPY_PARALLELISM);
+ return DEFAULT_STAGING_COPY_PARALLELISM;
+ }
+ }
+
+ private static List<URI> listSourceFiles(PinotFS fs, URI sourceDir)
+ throws IOException, URISyntaxException {
+ List<URI> sourceFileUris = new ArrayList<>();
+ try {
+ // Recursive listings include directory entries on several
implementations, so always filter them out.
+ for (FileMetadata fileMetadata : fs.listFilesWithMetadata(sourceDir,
true)) {
+ if (!fileMetadata.isDirectory()) {
+
sourceFileUris.add(SegmentGenerationUtils.getFileURI(fileMetadata.getFilePath(),
sourceDir));
+ }
+ }
+ } catch (UnsupportedOperationException e) {
+ // The PinotFS SPI default throws this when the implementation has no
metadata listing: fall back to
+ // listFiles() plus one isDirectory() stat per entry. IOExceptions are
left to propagate.
+ sourceFileUris.clear();
+ for (String sourcePath : fs.listFiles(sourceDir, true)) {
+ URI sourceFileUri = SegmentGenerationUtils.getFileURI(sourcePath,
sourceDir);
+ if (!fs.isDirectory(sourceFileUri)) {
+ sourceFileUris.add(sourceFileUri);
+ }
+ }
+ }
+ return sourceFileUris;
+ }
+
+ private static void moveOneFile(PinotFS fs, URI sourceDir, URI
sourceFileUri, URI destDir, boolean overwrite)
+ throws IOException, URISyntaxException {
+ String sourceFilename = SegmentGenerationUtils.getFileName(sourceFileUri);
+ URI destFileUri =
+ SegmentGenerationUtils.getRelativeOutputPath(sourceDir, sourceFileUri,
destDir).resolve(sourceFilename);
+ if (!overwrite && fs.exists(destFileUri)) {
+ LOGGER.warn("Can't overwrite existing output segment tar file: {}",
destFileUri);
+ return;
+ }
+ // The exists() check above is only a cheap short-circuit; passing the
flag down lets PinotFS.move reject a
+ // destination that showed up in between, which parallel moves make more
likely.
+ if (!fs.move(sourceFileUri, destFileUri, overwrite)) {
+ LOGGER.warn("Skipped moving {} to {}, move returned false (destination
may already exist)", sourceFileUri,
+ destFileUri);
}
Review Comment:
`PinotFS.move` defines `false` as an unsuccessful move, and implementations
such as `HadoopPinotFS.doMove` return `false` when the underlying rename fails;
S3/GCS can also return `false` for copy/delete failures. Treating every `false`
as a benign skip lets the job report success, after which both runners delete
the staging directory in `finally`, potentially deleting the only remaining
segment. Only treat this as a skip when `overwrite` is false and the
destination now exists; otherwise throw an `IOException` so the failure is
aggregated.
--
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]
