Vamsi-klu commented on code in PR #19082:
URL: https://github.com/apache/pinot/pull/19082#discussion_r3755546337
##########
pinot-plugins/pinot-batch-ingestion/pinot-batch-ingestion-common/src/main/java/org/apache/pinot/plugin/ingestion/batch/common/SegmentGenerationJobUtils.java:
##########
@@ -108,27 +119,155 @@ 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 {@link #DEFAULT_STAGING_COPY_PARALLELISM} worker threads.
+ * <p>
+ * The shared {@link PinotFS} instance must support concurrent {@code move}
of distinct paths (true for
+ * {@code LocalPinotFS} and typical remote implementations).
*
- * @param fs
- * @param sourceDir
- * @param destDir
- * @param overwrite
- * @throws IOException
- * @throws URISyntaxException
+ * @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 {
+ throws IOException, URISyntaxException {
+ moveFiles(fs, sourceDir, destDir, overwrite,
DEFAULT_STAGING_COPY_PARALLELISM);
+ }
+
+ /**
+ * Move all files from the <sourceDir> to the <destDir> using up to {@code
parallelism} threads.
+ * Directories in the source listing are skipped; parent directories on the
destination are created by
+ * {@link PinotFS#move}. Relative path layout under {@code sourceDir} is
preserved.
+ * <p>
+ * On partial failure, remaining in-flight moves are allowed to finish, then
an {@link IOException} is thrown
+ * with any additional failures attached as suppressed exceptions.
+ *
+ * @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(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;
+ }
+
+ 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;
+ boolean interrupted = false;
+ for (Future<Void> future : futures) {
+ try {
+ future.get();
+ } catch (InterruptedException e) {
+ interrupted = true;
+ future.cancel(true);
+ if (firstFailure == null) {
+ firstFailure = new IOException("Interrupted while moving files
from " + sourceDir + " to " + destDir, e);
+ } else {
+ firstFailure.addSuppressed(e);
+ }
+ } 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 (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ if (firstFailure != null) {
+ throw firstFailure;
+ }
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ /**
+ * Resolve staging-copy parallelism from job {@code
executionFrameworkSpec.extraConfigs}.
+ * Missing/invalid/non-positive values fall back to {@link
#DEFAULT_STAGING_COPY_PARALLELISM}.
+ * Values above {@link #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<>();
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);
-
- if (!overwrite && fs.exists(destFileUri)) {
- LOGGER.warn("Can't overwrite existing output segment tar file: {}",
destFileUri);
- } else {
- fs.move(sourceFileUri, destFileUri, true);
+ if (fs.isDirectory(sourceFileUri)) {
+ continue;
}
+ 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);
+ } else {
+ fs.move(sourceFileUri, destFileUri, true);
}
Review Comment:
Fixed in `a07aa82`. `moveOneFile` passes the caller's `overwrite` through to
`fs.move` now instead of a hardcoded true. The `exists()` pre-check stays as a
cheap warn-and-skip short circuit, and a false return from `move` is logged as
a skip so the parallel path behaves like the serial one.
##########
pinot-plugins/pinot-batch-ingestion/pinot-batch-ingestion-common/src/main/java/org/apache/pinot/plugin/ingestion/batch/common/SegmentGenerationJobUtils.java:
##########
@@ -108,27 +119,155 @@ 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 {@link #DEFAULT_STAGING_COPY_PARALLELISM} worker threads.
+ * <p>
+ * The shared {@link PinotFS} instance must support concurrent {@code move}
of distinct paths (true for
+ * {@code LocalPinotFS} and typical remote implementations).
*
- * @param fs
- * @param sourceDir
- * @param destDir
- * @param overwrite
- * @throws IOException
- * @throws URISyntaxException
+ * @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 {
+ throws IOException, URISyntaxException {
+ moveFiles(fs, sourceDir, destDir, overwrite,
DEFAULT_STAGING_COPY_PARALLELISM);
+ }
+
+ /**
+ * Move all files from the <sourceDir> to the <destDir> using up to {@code
parallelism} threads.
+ * Directories in the source listing are skipped; parent directories on the
destination are created by
+ * {@link PinotFS#move}. Relative path layout under {@code sourceDir} is
preserved.
+ * <p>
+ * On partial failure, remaining in-flight moves are allowed to finish, then
an {@link IOException} is thrown
+ * with any additional failures attached as suppressed exceptions.
+ *
+ * @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(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;
+ }
+
+ 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;
+ boolean interrupted = false;
+ for (Future<Void> future : futures) {
+ try {
+ future.get();
+ } catch (InterruptedException e) {
+ interrupted = true;
+ future.cancel(true);
+ if (firstFailure == null) {
+ firstFailure = new IOException("Interrupted while moving files
from " + sourceDir + " to " + destDir, e);
+ } else {
+ firstFailure.addSuppressed(e);
+ }
+ } 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 (interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ if (firstFailure != null) {
+ throw firstFailure;
+ }
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ /**
+ * Resolve staging-copy parallelism from job {@code
executionFrameworkSpec.extraConfigs}.
+ * Missing/invalid/non-positive values fall back to {@link
#DEFAULT_STAGING_COPY_PARALLELISM}.
+ * Values above {@link #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<>();
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);
-
- if (!overwrite && fs.exists(destFileUri)) {
- LOGGER.warn("Can't overwrite existing output segment tar file: {}",
destFileUri);
- } else {
- fs.move(sourceFileUri, destFileUri, true);
+ if (fs.isDirectory(sourceFileUri)) {
+ continue;
}
+ sourceFileUris.add(sourceFileUri);
+ }
+ return sourceFileUris;
Review Comment:
Fixed in `a07aa82`. `listSourceFiles` tries
`fs.listFilesWithMetadata(sourceDir, true)` and filters on `isDirectory()`,
falling back to the old `listFiles` plus per-entry stat only on
`UnsupportedOperationException`, which is what the SPI default throws.
`IOException` still propagates rather than silently degrading to the slow path.
##########
pinot-plugins/pinot-batch-ingestion/pinot-batch-ingestion-common/src/main/java/org/apache/pinot/plugin/ingestion/batch/common/SegmentGenerationJobUtils.java:
##########
@@ -108,27 +119,155 @@ 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 {@link #DEFAULT_STAGING_COPY_PARALLELISM} worker threads.
+ * <p>
+ * The shared {@link PinotFS} instance must support concurrent {@code move}
of distinct paths (true for
+ * {@code LocalPinotFS} and typical remote implementations).
*
- * @param fs
- * @param sourceDir
- * @param destDir
- * @param overwrite
- * @throws IOException
- * @throws URISyntaxException
+ * @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 {
+ throws IOException, URISyntaxException {
+ moveFiles(fs, sourceDir, destDir, overwrite,
DEFAULT_STAGING_COPY_PARALLELISM);
+ }
+
+ /**
+ * Move all files from the <sourceDir> to the <destDir> using up to {@code
parallelism} threads.
+ * Directories in the source listing are skipped; parent directories on the
destination are created by
+ * {@link PinotFS#move}. Relative path layout under {@code sourceDir} is
preserved.
+ * <p>
+ * On partial failure, remaining in-flight moves are allowed to finish, then
an {@link IOException} is thrown
+ * with any additional failures attached as suppressed exceptions.
+ *
+ * @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(parallelism,
sourceFileUris.size()));
+ LOGGER.info("Moving {} files from [{}] to [{}] with parallelism {}",
sourceFileUris.size(), sourceDir, destDir,
+ effectiveParallelism);
Review Comment:
Fixed in `a07aa82`: `effectiveParallelism` is clamped against
`MAX_STAGING_COPY_PARALLELISM` inside `moveFiles` as well, so a direct caller
of that overload cannot exceed the cap `getStagingCopyParallelism` enforces.
##########
pinot-plugins/pinot-batch-ingestion/pinot-batch-ingestion-common/src/main/java/org/apache/pinot/plugin/ingestion/batch/common/SegmentGenerationJobUtils.java:
##########
@@ -108,27 +119,155 @@ 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 {@link #DEFAULT_STAGING_COPY_PARALLELISM} worker threads.
+ * <p>
+ * The shared {@link PinotFS} instance must support concurrent {@code move}
of distinct paths (true for
+ * {@code LocalPinotFS} and typical remote implementations).
*
- * @param fs
- * @param sourceDir
- * @param destDir
- * @param overwrite
- * @throws IOException
- * @throws URISyntaxException
+ * @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 {
+ throws IOException, URISyntaxException {
+ moveFiles(fs, sourceDir, destDir, overwrite,
DEFAULT_STAGING_COPY_PARALLELISM);
+ }
+
+ /**
+ * Move all files from the <sourceDir> to the <destDir> using up to {@code
parallelism} threads.
+ * Directories in the source listing are skipped; parent directories on the
destination are created by
+ * {@link PinotFS#move}. Relative path layout under {@code sourceDir} is
preserved.
+ * <p>
+ * On partial failure, remaining in-flight moves are allowed to finish, then
an {@link IOException} is thrown
+ * with any additional failures attached as suppressed exceptions.
+ *
+ * @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(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;
+ }
+
+ 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;
+ boolean interrupted = false;
+ for (Future<Void> future : futures) {
+ try {
+ future.get();
+ } catch (InterruptedException e) {
+ interrupted = true;
+ future.cancel(true);
+ if (firstFailure == null) {
+ firstFailure = new IOException("Interrupted while moving files
from " + sourceDir + " to " + destDir, e);
+ } else {
+ firstFailure.addSuppressed(e);
+ }
+ } catch (Exception e) {
Review Comment:
Fixed in `a07aa82`: on interrupt all outstanding futures are cancelled and
the loop breaks immediately rather than continuing to block on the remaining
`get()` calls, the interrupt status is restored, and the aggregated
`IOException` is thrown. `executor.shutdownNow()` stays in the finally block as
a backstop.
--
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]
