This is an automated email from the ASF dual-hosted git repository.
bamaer pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/main by this push:
new 810ed58ec5 processFiles performance regression, fixes #9162 (#9164)
810ed58ec5 is described below
commit 810ed58ec5bfc688511900f100b12f7b51a0fc96
Author: Hans Van Akelyen <[email protected]>
AuthorDate: Sat Oct 31 08:41:42 2026 +0100
processFiles performance regression, fixes #9162 (#9164)
* processFiles performance regression, fixes #9162
* fix logging remark
---
.../transforms/processfiles/ProcessFiles.java | 179 ++++++++++++++-------
.../transforms/processfiles/ProcessFilesData.java | 8 +
.../messages/messages_en_US.properties | 4 +
.../transforms/processfiles/ProcessFilesTest.java | 81 ++++++++++
4 files changed, 212 insertions(+), 60 deletions(-)
diff --git
a/plugins/transforms/processfiles/src/main/java/org/apache/hop/pipeline/transforms/processfiles/ProcessFiles.java
b/plugins/transforms/processfiles/src/main/java/org/apache/hop/pipeline/transforms/processfiles/ProcessFiles.java
index 96ad4a2090..eadabe942e 100644
---
a/plugins/transforms/processfiles/src/main/java/org/apache/hop/pipeline/transforms/processfiles/ProcessFiles.java
+++
b/plugins/transforms/processfiles/src/main/java/org/apache/hop/pipeline/transforms/processfiles/ProcessFiles.java
@@ -33,6 +33,7 @@ import org.apache.hop.core.util.Utils;
import org.apache.hop.core.vfs.HopVfs;
import org.apache.hop.i18n.BaseMessages;
import org.apache.hop.lineage.LineageFileIoEmitter;
+import org.apache.hop.lineage.hub.LineageHub;
import org.apache.hop.lineage.model.FileIoOperation;
import org.apache.hop.pipeline.Pipeline;
import org.apache.hop.pipeline.PipelineMeta;
@@ -44,6 +45,12 @@ public class ProcessFiles extends
BaseTransform<ProcessFilesMeta, ProcessFilesDa
private static final int COPY_BUFFER_SIZE = 8192;
+ /**
+ * Number of files logged one by one at Basic level. Past this, per-file
lines only appear at
+ * Detailed level and a summary is logged when the transform finishes.
+ */
+ static final int MAX_BASIC_FILE_LOG_LINES = 100;
+
public ProcessFiles(
TransformMeta transformMeta,
ProcessFilesMeta meta,
@@ -113,13 +120,14 @@ public class ProcessFiles extends
BaseTransform<ProcessFilesMeta, ProcessFilesDa
}
data.sourceFile = HopVfs.getFileObject(sourceFilename, variables);
- if (!data.sourceFile.exists()) {
+ FileType sourceType = data.sourceFile.getType();
+ if (sourceType == FileType.IMAGINARY) {
logError(
BaseMessages.getString(PKG,
"ProcessFiles.Error.SourceFileNotExist", sourceFilename));
throw new HopException(
BaseMessages.getString(PKG,
"ProcessFiles.Error.SourceFileNotExist", sourceFilename));
}
- if (data.sourceFile.getType() != FileType.FILE) {
+ if (sourceType != FileType.FILE) {
logError(
BaseMessages.getString(PKG,
"ProcessFiles.Error.SourceFileNotFile", sourceFilename));
throw new HopException(
@@ -135,7 +143,10 @@ public class ProcessFiles extends
BaseTransform<ProcessFilesMeta, ProcessFilesDa
throw new HopException(BaseMessages.getString(PKG,
"ProcessFiles.Error.TargetFileEmpty"));
}
data.targetFile = HopVfs.getFileObject(targetFilename, variables);
- if (data.targetFile.exists()) {
+ // Look the target up once: the copy and move operations reuse this
answer
+ FileType targetType = data.targetFile.getType();
+ data.targetFileExists = targetType != FileType.IMAGINARY;
+ if (data.targetFileExists) {
if (isDetailed()) {
logDetailed(
BaseMessages.getString(
@@ -143,7 +154,7 @@ public class ProcessFiles extends
BaseTransform<ProcessFilesMeta, ProcessFilesDa
}
// check if target is really a file otherwise it could overwrite a
complete folder by copy
// or move operations
- if (data.targetFile.getType() != FileType.FILE) {
+ if (targetType != FileType.FILE) {
logError(
BaseMessages.getString(
PKG, "ProcessFiles.Error.TargetFileNotFile",
targetFilename));
@@ -260,6 +271,27 @@ public class ProcessFiles extends
BaseTransform<ProcessFilesMeta, ProcessFilesDa
return true;
}
+ @Override
+ public void dispose() {
+ // One line per transform copy, also when it stopped early; nothing ran if
first is still set
+ if (!first && !meta.simulate && isBasic()) {
+ String summaryKey =
+ switch (meta.getOperationType()) {
+ case ProcessFilesMeta.OPERATION_TYPE_MOVE ->
"ProcessFiles.Log.Basic.SummaryMove";
+ case ProcessFilesMeta.OPERATION_TYPE_DELETE ->
"ProcessFiles.Log.Basic.SummaryDelete";
+ default -> "ProcessFiles.Log.Basic.SummaryCopy";
+ };
+ logBasic(
+ BaseMessages.getString(
+ PKG,
+ summaryKey,
+ Long.toString(data.filesProcessed),
+ Long.toString(data.filesSkipped),
+ Long.toString(data.filesOverwritten)));
+ }
+ super.dispose();
+ }
+
private String operationTypeLabel() {
switch (meta.getOperationType()) {
case ProcessFilesMeta.OPERATION_TYPE_COPY:
@@ -285,11 +317,34 @@ public class ProcessFiles extends
BaseTransform<ProcessFilesMeta, ProcessFilesDa
}
}
+ /**
+ * Logs one processed file at Basic level, up to {@link
#MAX_BASIC_FILE_LOG_LINES} files. The
+ * Basic-level overwrite warnings share this limit. The Detailed-level line
for the same file is
+ * logged separately and is not capped.
+ */
+ private void logFileProcessed(String messageKey, String... filenames) {
+ if (!isBasic()) {
+ return;
+ }
+ data.basicFileLogLines++;
+ if (data.basicFileLogLines <= MAX_BASIC_FILE_LOG_LINES) {
+ String[] friendlyNames = new String[filenames.length];
+ for (int i = 0; i < filenames.length; i++) {
+ friendlyNames[i] = friendlyPath(filenames[i]);
+ }
+ logBasic(BaseMessages.getString(PKG, messageKey, friendlyNames));
+ } else if (data.basicFileLogLines == MAX_BASIC_FILE_LOG_LINES + 1) {
+ logBasic(
+ BaseMessages.getString(
+ PKG,
+ "ProcessFiles.Log.Basic.FileLogLimitReached",
+ Integer.toString(MAX_BASIC_FILE_LOG_LINES)));
+ }
+ }
+
private void processCopyOperation(String sourceFilename, String
targetFilename) throws Exception {
- boolean targetFileExists =
- data.targetFile.exists() && data.targetFile.getType() == FileType.FILE;
- boolean shouldPerform =
- !data.targetFile.exists() || (meta.isOverwriteTargetFile() &&
targetFileExists);
+ // processRow() already refused a target that exists but is not a file
+ boolean shouldPerform = !data.targetFileExists ||
meta.isOverwriteTargetFile();
if (meta.simulate) {
if (shouldPerform && isDetailed()) {
@@ -312,10 +367,15 @@ public class ProcessFiles extends
BaseTransform<ProcessFilesMeta, ProcessFilesDa
}
if (shouldPerform) {
- if (meta.isOverwriteTargetFile() && data.targetFile.exists() &&
isBasic()) {
- logBasic(
- BaseMessages.getString(
- PKG, "ProcessFiles.Log.Basic.OverwriteWarningCopy",
friendlyPath(targetFilename)));
+ if (data.targetFileExists) {
+ data.filesOverwritten++;
+ if (isBasic() && data.basicFileLogLines < MAX_BASIC_FILE_LOG_LINES) {
+ logBasic(
+ BaseMessages.getString(
+ PKG,
+ "ProcessFiles.Log.Basic.OverwriteWarningCopy",
+ friendlyPath(targetFilename)));
+ }
}
if (isDetailed()) {
logDetailed(
@@ -326,44 +386,41 @@ public class ProcessFiles extends
BaseTransform<ProcessFilesMeta, ProcessFilesDa
friendlyPath(targetFilename)));
}
// Delete target first; some providers do not truncate correctly on
overwrite.
- data.targetFile.delete();
+ if (data.targetFileExists) {
+ data.targetFile.delete();
+ }
if (isDataVolumeMetricEnabled()) {
copyFileTrackingVolume(data.sourceFile, data.targetFile);
} else {
data.targetFile.copyFrom(data.sourceFile, new
TextOneToOneFileSelector());
}
- if (isBasic()) {
- logBasic(
+ data.filesProcessed++;
+ logFileProcessed("ProcessFiles.Log.Basic.FileCopied", sourceFilename,
targetFilename);
+ if (isDetailed()) {
+ logDetailed(
BaseMessages.getString(
PKG,
- "ProcessFiles.Log.Basic.FileCopied",
+ "ProcessFiles.Log.Detail.FileCopied",
friendlyPath(sourceFilename),
friendlyPath(targetFilename)));
}
+ emitCopyLineage();
+ } else {
+ data.filesSkipped++;
if (isDetailed()) {
logDetailed(
BaseMessages.getString(
PKG,
- "ProcessFiles.Log.Detail.FileCopied",
- friendlyPath(sourceFilename),
- friendlyPath(targetFilename)));
+ "ProcessFiles.Log.Detail.SkippedCopyExistsNoOverwrite",
+ friendlyPath(targetFilename),
+ friendlyPath(sourceFilename)));
}
- emitCopyLineage();
- } else if (isDetailed()) {
- logDetailed(
- BaseMessages.getString(
- PKG,
- "ProcessFiles.Log.Detail.SkippedCopyExistsNoOverwrite",
- friendlyPath(targetFilename),
- friendlyPath(sourceFilename)));
}
}
private void processMoveOperation(String sourceFilename, String
targetFilename) throws Exception {
- boolean targetFileExists =
- data.targetFile.exists() && data.targetFile.getType() == FileType.FILE;
- boolean shouldPerform =
- !data.targetFile.exists() || (meta.isOverwriteTargetFile() &&
targetFileExists);
+ // processRow() already refused a target that exists but is not a file
+ boolean shouldPerform = !data.targetFileExists ||
meta.isOverwriteTargetFile();
if (meta.simulate) {
if (shouldPerform && isDetailed()) {
@@ -386,10 +443,15 @@ public class ProcessFiles extends
BaseTransform<ProcessFilesMeta, ProcessFilesDa
}
if (shouldPerform) {
- if (meta.isOverwriteTargetFile() && data.targetFile.exists() &&
isBasic()) {
- logBasic(
- BaseMessages.getString(
- PKG, "ProcessFiles.Log.Basic.OverwriteWarningMove",
friendlyPath(targetFilename)));
+ if (data.targetFileExists) {
+ data.filesOverwritten++;
+ if (isBasic() && data.basicFileLogLines < MAX_BASIC_FILE_LOG_LINES) {
+ logBasic(
+ BaseMessages.getString(
+ PKG,
+ "ProcessFiles.Log.Basic.OverwriteWarningMove",
+ friendlyPath(targetFilename)));
+ }
}
if (isDetailed()) {
logDetailed(
@@ -399,9 +461,11 @@ public class ProcessFiles extends
BaseTransform<ProcessFilesMeta, ProcessFilesDa
friendlyPath(sourceFilename),
friendlyPath(targetFilename)));
}
- String sourceUri = vfsUriString(data.sourceFile);
- String targetUri = vfsUriString(data.targetFile);
- Long lineageBytes = nullableContentSize(data.sourceFile);
+ // The source is gone after the move, so collect what lineage needs first
+ boolean lineageEnabled = LineageHub.getInstance().isEnabled();
+ String sourceUri = lineageEnabled ? vfsUriString(data.sourceFile) : null;
+ String targetUri = lineageEnabled ? vfsUriString(data.targetFile) : null;
+ Long lineageBytes = lineageEnabled ?
nullableContentSize(data.sourceFile) : null;
long movedBytes = 0;
if (isDataVolumeMetricEnabled()) {
movedBytes = safeContentSize(data.sourceFile);
@@ -411,30 +475,27 @@ public class ProcessFiles extends
BaseTransform<ProcessFilesMeta, ProcessFilesDa
dataVolumeIn = (dataVolumeIn != null ? dataVolumeIn : 0L) + movedBytes;
dataVolumeOut = (dataVolumeOut != null ? dataVolumeOut : 0L) +
movedBytes;
}
- if (isBasic()) {
- logBasic(
+ data.filesProcessed++;
+ logFileProcessed("ProcessFiles.Log.Basic.FileMoved", sourceFilename,
targetFilename);
+ if (isDetailed()) {
+ logDetailed(
BaseMessages.getString(
PKG,
- "ProcessFiles.Log.Basic.FileMoved",
+ "ProcessFiles.Log.Detail.FileMoved",
friendlyPath(sourceFilename),
friendlyPath(targetFilename)));
}
+ emitMoveLineageAfterSuccess(sourceUri, targetUri, lineageBytes);
+ } else {
+ data.filesSkipped++;
if (isDetailed()) {
logDetailed(
BaseMessages.getString(
PKG,
- "ProcessFiles.Log.Detail.FileMoved",
- friendlyPath(sourceFilename),
- friendlyPath(targetFilename)));
+ "ProcessFiles.Log.Detail.SkippedMoveExistsNoOverwrite",
+ friendlyPath(targetFilename),
+ friendlyPath(sourceFilename)));
}
- emitMoveLineageAfterSuccess(sourceUri, targetUri, lineageBytes);
- } else if (isDetailed()) {
- logDetailed(
- BaseMessages.getString(
- PKG,
- "ProcessFiles.Log.Detail.SkippedMoveExistsNoOverwrite",
- friendlyPath(targetFilename),
- friendlyPath(sourceFilename)));
}
}
@@ -448,19 +509,17 @@ public class ProcessFiles extends
BaseTransform<ProcessFilesMeta, ProcessFilesDa
}
return;
}
- String sourceUri = vfsUriString(data.sourceFile);
- Long deleteBytes = nullableContentSize(data.sourceFile);
+ boolean lineageEnabled = LineageHub.getInstance().isEnabled();
+ String sourceUri = lineageEnabled ? vfsUriString(data.sourceFile) : null;
+ Long deleteBytes = lineageEnabled ? nullableContentSize(data.sourceFile) :
null;
if (!data.sourceFile.delete()) {
throw new HopException(
BaseMessages.getString(
PKG, "ProcessFiles.Error.CanNotDeleteFile",
data.sourceFile.toString()));
}
emitDeleteLineage(sourceUri, deleteBytes);
- if (isBasic()) {
- logBasic(
- BaseMessages.getString(
- PKG, "ProcessFiles.Log.Basic.FileDeleted",
friendlyPath(sourceFilename)));
- }
+ data.filesProcessed++;
+ logFileProcessed("ProcessFiles.Log.Basic.FileDeleted", sourceFilename);
if (isDetailed()) {
logDetailed(
BaseMessages.getString(
@@ -469,7 +528,7 @@ public class ProcessFiles extends
BaseTransform<ProcessFilesMeta, ProcessFilesDa
}
private void emitCopyLineage() {
- if (meta.simulate) {
+ if (meta.simulate || !LineageHub.getInstance().isEnabled()) {
return;
}
Long bytes = nullableContentSize(data.targetFile);
diff --git
a/plugins/transforms/processfiles/src/main/java/org/apache/hop/pipeline/transforms/processfiles/ProcessFilesData.java
b/plugins/transforms/processfiles/src/main/java/org/apache/hop/pipeline/transforms/processfiles/ProcessFilesData.java
index a624c06fa4..987ae1edba 100644
---
a/plugins/transforms/processfiles/src/main/java/org/apache/hop/pipeline/transforms/processfiles/ProcessFilesData.java
+++
b/plugins/transforms/processfiles/src/main/java/org/apache/hop/pipeline/transforms/processfiles/ProcessFilesData.java
@@ -28,6 +28,14 @@ public class ProcessFilesData extends BaseTransformData
implements ITransformDat
public FileObject sourceFile;
public FileObject targetFile;
+ /** Whether the target of the current row existed before the operation. */
+ public boolean targetFileExists;
+
+ public long basicFileLogLines;
+ public long filesProcessed;
+ public long filesSkipped;
+ public long filesOverwritten;
+
public ProcessFilesData() {
super();
indexOfSourceFilename = -1;
diff --git
a/plugins/transforms/processfiles/src/main/resources/org/apache/hop/pipeline/transforms/processfiles/messages/messages_en_US.properties
b/plugins/transforms/processfiles/src/main/resources/org/apache/hop/pipeline/transforms/processfiles/messages/messages_en_US.properties
index 8b606addab..72df5bd194 100644
---
a/plugins/transforms/processfiles/src/main/resources/org/apache/hop/pipeline/transforms/processfiles/messages/messages_en_US.properties
+++
b/plugins/transforms/processfiles/src/main/resources/org/apache/hop/pipeline/transforms/processfiles/messages/messages_en_US.properties
@@ -37,9 +37,13 @@ ProcessFiles.Injection.TargetFilenameField=Target filename
fieldname
ProcessFiles.LineNumber=Process files: processed {0} input rows
ProcessFiles.Log.Basic.FileCopied=Copied file {0} to {1}
ProcessFiles.Log.Basic.FileDeleted=Deleted file {0}
+ProcessFiles.Log.Basic.FileLogLimitReached=More than {0} files processed: the
remaining files are only logged one by one at Detailed log level
ProcessFiles.Log.Basic.FileMoved=Moved file {0} to {1}
ProcessFiles.Log.Basic.OverwriteWarningCopy=Copy will replace existing target
file {0}
ProcessFiles.Log.Basic.OverwriteWarningMove=Move will replace existing target
file {0}
+ProcessFiles.Log.Basic.SummaryCopy=Copied {0} files ({1} skipped because the
target exists, {2} replaced an existing target)
+ProcessFiles.Log.Basic.SummaryDelete=Deleted {0} files
+ProcessFiles.Log.Basic.SummaryMove=Moved {0} files ({1} skipped because the
target exists, {2} replaced an existing target)
ProcessFiles.Log.Detail.CopyStarted=Starting copy: {0} -> {1}
ProcessFiles.Log.Detail.FileCopied=Finished copy: {0} -> {1}
ProcessFiles.Log.Detail.FileDeleted=Finished delete: {0}
diff --git
a/plugins/transforms/processfiles/src/test/java/org/apache/hop/pipeline/transforms/processfiles/ProcessFilesTest.java
b/plugins/transforms/processfiles/src/test/java/org/apache/hop/pipeline/transforms/processfiles/ProcessFilesTest.java
index 8c659f98c5..4be23947a0 100644
---
a/plugins/transforms/processfiles/src/test/java/org/apache/hop/pipeline/transforms/processfiles/ProcessFilesTest.java
+++
b/plugins/transforms/processfiles/src/test/java/org/apache/hop/pipeline/transforms/processfiles/ProcessFilesTest.java
@@ -25,6 +25,7 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.nullable;
+import static org.mockito.Mockito.atLeastOnce;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
@@ -57,6 +58,7 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
import org.junit.jupiter.api.io.TempDir;
+import org.mockito.ArgumentCaptor;
class ProcessFilesTest {
@@ -392,4 +394,83 @@ class ProcessFilesTest {
assertEquals("1", Files.readString(t1, StandardCharsets.UTF_8));
assertEquals("2", Files.readString(t2, StandardCharsets.UTF_8));
}
+
+ @Test
+ void basicLog_capsPerFileLinesAndLogsSummary(@TempDir Path tempDir) throws
Exception {
+ int fileCount = ProcessFiles.MAX_BASIC_FILE_LOG_LINES + 5;
+ Object[][] rows = new Object[fileCount][];
+ for (int i = 0; i < fileCount; i++) {
+ Path source = tempDir.resolve("in" + i + ".txt");
+ Files.writeString(source, Integer.toString(i), StandardCharsets.UTF_8);
+ rows[i] = new Object[] {vfsPath(source), vfsPath(tempDir.resolve("out" +
i + ".txt"))};
+ }
+ when(helper.iLogChannel.isBasic()).thenReturn(true);
+
+ ProcessFilesMeta meta = baseMeta(ProcessFilesMeta.OPERATION_TYPE_COPY);
+ ProcessFiles transform = createTransform(meta, new ProcessFilesData());
+ wireInput(transform, rowMeta(), rows);
+ PipelineTestingUtil.execute(transform, fileCount, false);
+ transform.dispose();
+
+ ArgumentCaptor<String> messages = ArgumentCaptor.forClass(String.class);
+ verify(helper.iLogChannel, atLeastOnce()).logBasic(messages.capture());
+ List<String> logged = messages.getAllValues();
+ assertEquals(
+ ProcessFiles.MAX_BASIC_FILE_LOG_LINES,
+ logged.stream().filter(m -> m.startsWith("Copied file ")).count());
+ assertEquals(1, logged.stream().filter(m -> m.startsWith("More than 100
files")).count());
+ assertTrue(
+ logged.contains(
+ "Copied "
+ + fileCount
+ + " files (0 skipped because the target"
+ + " exists, 0 replaced an existing target)"));
+ assertTrue(Files.exists(tempDir.resolve("out" + (fileCount - 1) +
".txt")));
+ }
+
+ @Test
+ void summary_countsSkippedAndOverwrittenTargets(@TempDir Path tempDir)
throws Exception {
+ Path s1 = tempDir.resolve("a.txt");
+ Path t1 = tempDir.resolve("a.out");
+ Path s2 = tempDir.resolve("b.txt");
+ Path t2 = tempDir.resolve("b.out");
+ Files.writeString(s1, "1", StandardCharsets.UTF_8);
+ Files.writeString(s2, "2", StandardCharsets.UTF_8);
+ Files.writeString(t2, "old", StandardCharsets.UTF_8);
+ when(helper.iLogChannel.isBasic()).thenReturn(true);
+
+ ProcessFilesMeta meta = baseMeta(ProcessFilesMeta.OPERATION_TYPE_COPY);
+ meta.setOverwriteTargetFile(true);
+ ProcessFilesData data = new ProcessFilesData();
+ ProcessFiles transform = createTransform(meta, data);
+ wireInput(
+ transform,
+ rowMeta(),
+ new Object[] {vfsPath(s1), vfsPath(t1)},
+ new Object[] {vfsPath(s2), vfsPath(t2)});
+ PipelineTestingUtil.execute(transform, 2, false);
+
+ assertEquals(2, data.filesProcessed);
+ assertEquals(0, data.filesSkipped);
+ assertEquals(1, data.filesOverwritten);
+ assertEquals("2", Files.readString(t2, StandardCharsets.UTF_8));
+ }
+
+ @Test
+ void summary_countsSkippedTargetsWithoutOverwrite(@TempDir Path tempDir)
throws Exception {
+ Path source = tempDir.resolve("in.txt");
+ Path target = tempDir.resolve("out.txt");
+ Files.writeString(source, "new", StandardCharsets.UTF_8);
+ Files.writeString(target, "old", StandardCharsets.UTF_8);
+
+ ProcessFilesMeta meta = baseMeta(ProcessFilesMeta.OPERATION_TYPE_COPY);
+ ProcessFilesData data = new ProcessFilesData();
+ ProcessFiles transform = createTransform(meta, data);
+ wireInput(transform, rowMeta(), new Object[] {vfsPath(source),
vfsPath(target)});
+ PipelineTestingUtil.execute(transform, 1, false);
+
+ assertEquals(0, data.filesProcessed);
+ assertEquals(1, data.filesSkipped);
+ assertEquals("old", Files.readString(target, StandardCharsets.UTF_8));
+ }
}
