This is an automated email from the ASF dual-hosted git repository.
hansva 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 dac22e67e4 issue #3165: add Workflow from field to Workflow Executor
(#7809)
dac22e67e4 is described below
commit dac22e67e419f589e61e479823f0cee3f68867fd
Author: Matt Casters <[email protected]>
AuthorDate: Thu Aug 6 20:39:06 2026 +0200
issue #3165: add Workflow from field to Workflow Executor (#7809)
Bring feature parity with Pipeline Executor so the child workflow path can
come from an input stream field. Load/reload at runtime without mutating
shared transform meta, add dialog controls, unit tests, and extend the
main-0083 integration suite.
---
.../pipeline/transforms/workflow-executor.adoc | 7 +-
...> 0083-workflow-executor-fromfield-called1.hwf} | 67 ++----
...> 0083-workflow-executor-fromfield-called2.hwf} | 67 ++----
.../0083-workflow-executor-fromfield.hpl | 256 +++++++++++++++++++++
.../main-0083-workflow-executor-test.hwf | 44 +++-
.../workflowexecutor/WorkflowExecutor.java | 228 ++++++++++--------
.../workflowexecutor/WorkflowExecutorData.java | 9 +
.../workflowexecutor/WorkflowExecutorDialog.java | 143 ++++++++++--
.../workflowexecutor/WorkflowExecutorMeta.java | 40 +++-
.../messages/messages_en_US.properties | 3 +
.../workflowexecutor/WorkflowExecutorMetaTest.java | 23 ++
.../workflowexecutor/WorkflowExecutorTest.java | 151 ++++++++++++
.../transforms/workflowexecutor/minimal-child.hwf | 65 +-----
...l => workflow-executor-fromfield-transform.xml} | 2 +
.../test/resources/workflow-executor-transform.xml | 2 +
15 files changed, 839 insertions(+), 268 deletions(-)
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/workflow-executor.adoc
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/workflow-executor.adoc
index 33b1c7065c..2ca1dccfd6 100644
---
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/workflow-executor.adoc
+++
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/workflow-executor.adoc
@@ -69,7 +69,12 @@ Samples (samples project)
|===
|Option|Description
|Transform name|Name of the transform.
-|Workflow|Use this option to specify a workflow stored in a file (.hwf file)
+|Workflow|Use this option to specify a workflow stored in a file (.hwf file).
+The filename may contain variables (for example, you can use the built-in
Internal.Pipeline.Filename.Directory variable to construct a filename relative
to the current pipeline), or you can use the "Browse" button to select a file
using a file browser.
+|Workflow from field?|Enable to specify the workflow file name(s) in a field
in the input stream.
+|Workflow field|When the previous option is enabled, you can specify the field
that will contain the workflow filename(s) at runtime. The filename may contain
variables.
+
+*NOTE:* It must be considered that, by specifying the workflow file name using
this option, we can experience a little performance penalty because the
workflow that will be executed will be initialized at runtime and not during
the initialization phase as usual. On the other side this option gives greater
flexibility in specifying the workflow filename dynamically and this could be
useful for many use-cases.
|Run configuration|Specify the Workflow Run Configuration to be used for
execution.
|===
diff --git a/integration-tests/transforms/main-0083-workflow-executor-test.hwf
b/integration-tests/transforms/0083-workflow-executor-fromfield-called1.hwf
similarity index 58%
copy from integration-tests/transforms/main-0083-workflow-executor-test.hwf
copy to
integration-tests/transforms/0083-workflow-executor-fromfield-called1.hwf
index 895b54c5da..c5dc69b72c 100644
--- a/integration-tests/transforms/main-0083-workflow-executor-test.hwf
+++ b/integration-tests/transforms/0083-workflow-executor-fromfield-called1.hwf
@@ -18,15 +18,15 @@ limitations under the License.
-->
<workflow>
- <name>main-0083-workflow-executor-test</name>
+ <name>0083-workflow-executor-fromfield-called1</name>
<name_sync_with_filename>Y</name_sync_with_filename>
<description/>
<extended_description/>
<workflow_version/>
<created_user>-</created_user>
- <created_date>2020/12/07 13:03:35.562</created_date>
+ <created_date>2026/03/23 12:00:00.000</created_date>
<modified_user>-</modified_user>
- <modified_date>2020/12/07 13:03:35.562</modified_date>
+ <modified_date>2026/03/23 12:00:00.000</modified_date>
<parameters>
</parameters>
<actions>
@@ -45,80 +45,49 @@ limitations under the License.
<schedulerType>0</schedulerType>
<weekDay>1</weekDay>
<parallel>N</parallel>
- <xloc>160</xloc>
- <yloc>96</yloc>
+ <xloc>112</xloc>
+ <yloc>112</yloc>
<attributes_hac/>
</action>
<action>
- <name>Success</name>
- <description/>
- <type>SUCCESS</type>
- <attributes/>
- <parallel>N</parallel>
- <xloc>544</xloc>
- <yloc>96</yloc>
- <attributes_hac/>
- </action>
- <action>
- <name>Abort workflow</name>
+ <name>Write to log</name>
<description/>
- <type>ABORT</type>
+ <type>WRITE_TO_LOG</type>
<attributes/>
- <always_log_rows>N</always_log_rows>
+ <loglevel>Basic</loglevel>
+ <logmessage>fromfield called1</logmessage>
+ <logsubject/>
<parallel>N</parallel>
- <xloc>320</xloc>
- <yloc>256</yloc>
+ <xloc>304</xloc>
+ <yloc>112</yloc>
<attributes_hac/>
</action>
<action>
- <name>0083-workflow-executor-run-wf.hpl</name>
+ <name>Success</name>
<description/>
- <type>PIPELINE</type>
+ <type>SUCCESS</type>
<attributes/>
- <add_date>N</add_date>
- <add_time>N</add_time>
- <clear_files>N</clear_files>
- <clear_rows>N</clear_rows>
- <create_parent_folder>N</create_parent_folder>
- <exec_per_row>N</exec_per_row>
- <filename>${PROJECT_HOME}/0083-workflow-executor-run-wf.hpl</filename>
- <loglevel>Basic</loglevel>
- <parameters>
- <pass_all_parameters>Y</pass_all_parameters>
- </parameters>
- <params_from_previous>N</params_from_previous>
- <run_configuration>local</run_configuration>
- <set_append_logfile>N</set_append_logfile>
- <set_logfile>N</set_logfile>
- <wait_until_finished>Y</wait_until_finished>
<parallel>N</parallel>
- <xloc>320</xloc>
- <yloc>98</yloc>
+ <xloc>496</xloc>
+ <yloc>112</yloc>
<attributes_hac/>
</action>
</actions>
<hops>
<hop>
<from>START</from>
- <to>0083-workflow-executor-run-wf.hpl</to>
+ <to>Write to log</to>
<enabled>Y</enabled>
<evaluation>Y</evaluation>
<unconditional>Y</unconditional>
</hop>
<hop>
- <from>0083-workflow-executor-run-wf.hpl</from>
+ <from>Write to log</from>
<to>Success</to>
<enabled>Y</enabled>
<evaluation>Y</evaluation>
<unconditional>N</unconditional>
</hop>
- <hop>
- <from>0083-workflow-executor-run-wf.hpl</from>
- <to>Abort workflow</to>
- <enabled>Y</enabled>
- <evaluation>N</evaluation>
- <unconditional>N</unconditional>
- </hop>
</hops>
<notepads>
</notepads>
diff --git a/integration-tests/transforms/main-0083-workflow-executor-test.hwf
b/integration-tests/transforms/0083-workflow-executor-fromfield-called2.hwf
similarity index 58%
copy from integration-tests/transforms/main-0083-workflow-executor-test.hwf
copy to
integration-tests/transforms/0083-workflow-executor-fromfield-called2.hwf
index 895b54c5da..c948e52581 100644
--- a/integration-tests/transforms/main-0083-workflow-executor-test.hwf
+++ b/integration-tests/transforms/0083-workflow-executor-fromfield-called2.hwf
@@ -18,15 +18,15 @@ limitations under the License.
-->
<workflow>
- <name>main-0083-workflow-executor-test</name>
+ <name>0083-workflow-executor-fromfield-called2</name>
<name_sync_with_filename>Y</name_sync_with_filename>
<description/>
<extended_description/>
<workflow_version/>
<created_user>-</created_user>
- <created_date>2020/12/07 13:03:35.562</created_date>
+ <created_date>2026/03/23 12:00:00.000</created_date>
<modified_user>-</modified_user>
- <modified_date>2020/12/07 13:03:35.562</modified_date>
+ <modified_date>2026/03/23 12:00:00.000</modified_date>
<parameters>
</parameters>
<actions>
@@ -45,80 +45,49 @@ limitations under the License.
<schedulerType>0</schedulerType>
<weekDay>1</weekDay>
<parallel>N</parallel>
- <xloc>160</xloc>
- <yloc>96</yloc>
+ <xloc>112</xloc>
+ <yloc>112</yloc>
<attributes_hac/>
</action>
<action>
- <name>Success</name>
- <description/>
- <type>SUCCESS</type>
- <attributes/>
- <parallel>N</parallel>
- <xloc>544</xloc>
- <yloc>96</yloc>
- <attributes_hac/>
- </action>
- <action>
- <name>Abort workflow</name>
+ <name>Write to log</name>
<description/>
- <type>ABORT</type>
+ <type>WRITE_TO_LOG</type>
<attributes/>
- <always_log_rows>N</always_log_rows>
+ <loglevel>Basic</loglevel>
+ <logmessage>fromfield called2</logmessage>
+ <logsubject/>
<parallel>N</parallel>
- <xloc>320</xloc>
- <yloc>256</yloc>
+ <xloc>304</xloc>
+ <yloc>112</yloc>
<attributes_hac/>
</action>
<action>
- <name>0083-workflow-executor-run-wf.hpl</name>
+ <name>Success</name>
<description/>
- <type>PIPELINE</type>
+ <type>SUCCESS</type>
<attributes/>
- <add_date>N</add_date>
- <add_time>N</add_time>
- <clear_files>N</clear_files>
- <clear_rows>N</clear_rows>
- <create_parent_folder>N</create_parent_folder>
- <exec_per_row>N</exec_per_row>
- <filename>${PROJECT_HOME}/0083-workflow-executor-run-wf.hpl</filename>
- <loglevel>Basic</loglevel>
- <parameters>
- <pass_all_parameters>Y</pass_all_parameters>
- </parameters>
- <params_from_previous>N</params_from_previous>
- <run_configuration>local</run_configuration>
- <set_append_logfile>N</set_append_logfile>
- <set_logfile>N</set_logfile>
- <wait_until_finished>Y</wait_until_finished>
<parallel>N</parallel>
- <xloc>320</xloc>
- <yloc>98</yloc>
+ <xloc>496</xloc>
+ <yloc>112</yloc>
<attributes_hac/>
</action>
</actions>
<hops>
<hop>
<from>START</from>
- <to>0083-workflow-executor-run-wf.hpl</to>
+ <to>Write to log</to>
<enabled>Y</enabled>
<evaluation>Y</evaluation>
<unconditional>Y</unconditional>
</hop>
<hop>
- <from>0083-workflow-executor-run-wf.hpl</from>
+ <from>Write to log</from>
<to>Success</to>
<enabled>Y</enabled>
<evaluation>Y</evaluation>
<unconditional>N</unconditional>
</hop>
- <hop>
- <from>0083-workflow-executor-run-wf.hpl</from>
- <to>Abort workflow</to>
- <enabled>Y</enabled>
- <evaluation>N</evaluation>
- <unconditional>N</unconditional>
- </hop>
</hops>
<notepads>
</notepads>
diff --git a/integration-tests/transforms/0083-workflow-executor-fromfield.hpl
b/integration-tests/transforms/0083-workflow-executor-fromfield.hpl
new file mode 100644
index 0000000000..666c77ae78
--- /dev/null
+++ b/integration-tests/transforms/0083-workflow-executor-fromfield.hpl
@@ -0,0 +1,256 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+-->
+<pipeline>
+ <info>
+ <name>0083-workflow-executor-fromfield</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description/>
+ <extended_description/>
+ <pipeline_version/>
+ <pipeline_type>Normal</pipeline_type>
+ <parameters>
+ </parameters>
+ <capture_transform_performance>N</capture_transform_performance>
+
<transform_performance_capturing_delay>1000</transform_performance_capturing_delay>
+
<transform_performance_capturing_size_limit>100</transform_performance_capturing_size_limit>
+ <created_user>-</created_user>
+ <created_date>2026/03/23 12:00:00.000</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/03/23 12:00:00.000</modified_date>
+ </info>
+ <notepads>
+ </notepads>
+ <order>
+ <hop>
+ <from>Data grid</from>
+ <to>Workflow executor</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>Workflow executor</from>
+ <to>Group by</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>Group by</from>
+ <to>count = 2</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>count = 2</from>
+ <to>Abort</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>count = 2</from>
+ <to>Success</to>
+ <enabled>Y</enabled>
+ </hop>
+ </order>
+ <transform>
+ <name>Abort</name>
+ <type>Abort</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <abort_option>ABORT_WITH_ERROR</abort_option>
+ <always_log_rows>Y</always_log_rows>
+ <message>Workflow executor from field: expected 2 successful
executions</message>
+ <row_threshold>0</row_threshold>
+ <attributes/>
+ <GUI>
+ <xloc>704</xloc>
+ <yloc>224</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Data grid</name>
+ <type>DataGrid</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <fields>
+ <field>
+ <set_empty_string>N</set_empty_string>
+ <length>-1</length>
+ <name>wf_to_run</name>
+ <precision>-1</precision>
+ <type>String</type>
+ </field>
+ </fields>
+ <data>
+ <line>
+
<item>${PROJECT_HOME}/0083-workflow-executor-fromfield-called1.hwf</item>
+ </line>
+ <line>
+
<item>${PROJECT_HOME}/0083-workflow-executor-fromfield-called2.hwf</item>
+ </line>
+ </data>
+ <attributes/>
+ <GUI>
+ <xloc>112</xloc>
+ <yloc>112</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Group by</name>
+ <type>GroupBy</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <all_rows>N</all_rows>
+ <ignore_aggregate>N</ignore_aggregate>
+ <field_ignore/>
+ <directory>${java.io.tmpdir}</directory>
+ <prefix>grp</prefix>
+ <add_linenr>N</add_linenr>
+ <linenr_fieldname/>
+ <give_back_row>N</give_back_row>
+ <group>
+ </group>
+ <fields>
+ <field>
+ <aggregate>count</aggregate>
+ <subject>ExecutionResult</subject>
+ <type>COUNT_ALL</type>
+ <valuefield/>
+ </field>
+ </fields>
+ <attributes/>
+ <GUI>
+ <xloc>432</xloc>
+ <yloc>112</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Success</name>
+ <type>Dummy</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <attributes/>
+ <GUI>
+ <xloc>832</xloc>
+ <yloc>112</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Workflow executor</name>
+ <type>WorkflowExecutor</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <run_configuration>local</run_configuration>
+ <filename/>
+ <filenameInField>Y</filenameInField>
+ <filenameField>wf_to_run</filenameField>
+ <group_size>1</group_size>
+ <group_field/>
+ <group_time/>
+ <parameters>
+ </parameters>
+ <inherit_all_vars>Y</inherit_all_vars>
+ <execution_result_target_transform>Group
by</execution_result_target_transform>
+ <execution_time_field>ExecutionTime</execution_time_field>
+ <execution_result_field>ExecutionResult</execution_result_field>
+ <execution_errors_field>ExecutionNrErrors</execution_errors_field>
+ <execution_lines_read_field>ExecutionLinesRead</execution_lines_read_field>
+
<execution_lines_written_field>ExecutionLinesWritten</execution_lines_written_field>
+
<execution_lines_input_field>ExecutionLinesInput</execution_lines_input_field>
+
<execution_lines_output_field>ExecutionLinesOutput</execution_lines_output_field>
+
<execution_lines_rejected_field>ExecutionLinesRejected</execution_lines_rejected_field>
+
<execution_lines_updated_field>ExecutionLinesUpdated</execution_lines_updated_field>
+
<execution_lines_deleted_field>ExecutionLinesDeleted</execution_lines_deleted_field>
+
<execution_files_retrieved_field>ExecutionFilesRetrieved</execution_files_retrieved_field>
+
<execution_exit_status_field>ExecutionExitStatus</execution_exit_status_field>
+ <execution_log_text_field>ExecutionLogText</execution_log_text_field>
+
<execution_log_channelid_field>ExecutionLogChannelId</execution_log_channelid_field>
+ <result_files_file_name_field>FileName</result_files_file_name_field>
+ <attributes/>
+ <GUI>
+ <xloc>272</xloc>
+ <yloc>112</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>count = 2</name>
+ <type>FilterRows</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <send_true_to>Success</send_true_to>
+ <send_false_to>Abort</send_false_to>
+ <compare>
+ <condition>
+ <negated>N</negated>
+ <leftvalue>count</leftvalue>
+ <function>=</function>
+ <rightvalue/>
+ <value>
+ <name>constant</name>
+ <type>Integer</type>
+ <text>2</text>
+ <length>-1</length>
+ <precision>0</precision>
+ <isnull>N</isnull>
+ <mask>####0;-####0</mask>
+ </value>
+ </condition>
+ </compare>
+ <attributes/>
+ <GUI>
+ <xloc>592</xloc>
+ <yloc>112</yloc>
+ </GUI>
+ </transform>
+ <transform_error_handling>
+ </transform_error_handling>
+ <attributes/>
+</pipeline>
diff --git a/integration-tests/transforms/main-0083-workflow-executor-test.hwf
b/integration-tests/transforms/main-0083-workflow-executor-test.hwf
index 895b54c5da..941586f1dc 100644
--- a/integration-tests/transforms/main-0083-workflow-executor-test.hwf
+++ b/integration-tests/transforms/main-0083-workflow-executor-test.hwf
@@ -55,7 +55,7 @@ limitations under the License.
<type>SUCCESS</type>
<attributes/>
<parallel>N</parallel>
- <xloc>544</xloc>
+ <xloc>768</xloc>
<yloc>96</yloc>
<attributes_hac/>
</action>
@@ -96,6 +96,32 @@ limitations under the License.
<yloc>98</yloc>
<attributes_hac/>
</action>
+ <action>
+ <name>0083-workflow-executor-fromfield.hpl</name>
+ <description/>
+ <type>PIPELINE</type>
+ <attributes/>
+ <add_date>N</add_date>
+ <add_time>N</add_time>
+ <clear_files>N</clear_files>
+ <clear_rows>N</clear_rows>
+ <create_parent_folder>N</create_parent_folder>
+ <exec_per_row>N</exec_per_row>
+ <filename>${PROJECT_HOME}/0083-workflow-executor-fromfield.hpl</filename>
+ <loglevel>Basic</loglevel>
+ <parameters>
+ <pass_all_parameters>Y</pass_all_parameters>
+ </parameters>
+ <params_from_previous>N</params_from_previous>
+ <run_configuration>local</run_configuration>
+ <set_append_logfile>N</set_append_logfile>
+ <set_logfile>N</set_logfile>
+ <wait_until_finished>Y</wait_until_finished>
+ <parallel>N</parallel>
+ <xloc>544</xloc>
+ <yloc>96</yloc>
+ <attributes_hac/>
+ </action>
</actions>
<hops>
<hop>
@@ -107,7 +133,7 @@ limitations under the License.
</hop>
<hop>
<from>0083-workflow-executor-run-wf.hpl</from>
- <to>Success</to>
+ <to>0083-workflow-executor-fromfield.hpl</to>
<enabled>Y</enabled>
<evaluation>Y</evaluation>
<unconditional>N</unconditional>
@@ -119,6 +145,20 @@ limitations under the License.
<evaluation>N</evaluation>
<unconditional>N</unconditional>
</hop>
+ <hop>
+ <from>0083-workflow-executor-fromfield.hpl</from>
+ <to>Success</to>
+ <enabled>Y</enabled>
+ <evaluation>Y</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
+ <hop>
+ <from>0083-workflow-executor-fromfield.hpl</from>
+ <to>Abort workflow</to>
+ <enabled>Y</enabled>
+ <evaluation>N</evaluation>
+ <unconditional>N</unconditional>
+ </hop>
</hops>
<notepads>
</notepads>
diff --git
a/plugins/transforms/workflowexecutor/src/main/java/org/apache/hop/pipeline/transforms/workflowexecutor/WorkflowExecutor.java
b/plugins/transforms/workflowexecutor/src/main/java/org/apache/hop/pipeline/transforms/workflowexecutor/WorkflowExecutor.java
index 109e8d91fe..956115ba7f 100644
---
a/plugins/transforms/workflowexecutor/src/main/java/org/apache/hop/pipeline/transforms/workflowexecutor/WorkflowExecutor.java
+++
b/plugins/transforms/workflowexecutor/src/main/java/org/apache/hop/pipeline/transforms/workflowexecutor/WorkflowExecutor.java
@@ -71,7 +71,7 @@ public class WorkflowExecutor extends
BaseTransform<WorkflowExecutorMeta, Workfl
Object[] row = getRow();
if (row == null) {
- if (!data.groupBuffer.isEmpty()) {
+ if (data.groupBuffer != null && !data.groupBuffer.isEmpty()) {
executeWorkflow();
}
setOutputDone();
@@ -80,69 +80,40 @@ public class WorkflowExecutor extends
BaseTransform<WorkflowExecutorMeta, Workfl
if (first) {
first = false;
-
- // calculate the various output row layouts first...
- //
- data.inputRowMeta = getInputRowMeta();
- data.executionResultsOutputRowMeta = data.inputRowMeta.clone();
- data.resultRowsOutputRowMeta = data.inputRowMeta.clone();
- data.resultFilesOutputRowMeta = data.inputRowMeta.clone();
-
- if (meta.getExecutionResultTargetTransformMeta() != null) {
- meta.getFields(
- data.executionResultsOutputRowMeta,
- getTransformName(),
- null,
- meta.getExecutionResultTargetTransformMeta(),
- this,
- metadataProvider);
- data.executionResultRowSet =
-
findOutputRowSet(meta.getExecutionResultTargetTransformMeta().getName());
- }
- if (meta.getResultRowsTargetTransformMeta() != null) {
- meta.getFields(
- data.resultRowsOutputRowMeta,
- getTransformName(),
- null,
- meta.getResultRowsTargetTransformMeta(),
- this,
- metadataProvider);
- data.resultRowsRowSet =
-
findOutputRowSet(meta.getResultRowsTargetTransformMeta().getName());
- }
- if (meta.getResultFilesTargetTransformMeta() != null) {
- meta.getFields(
- data.resultFilesOutputRowMeta,
- getTransformName(),
- null,
- meta.getResultFilesTargetTransformMeta(),
- this,
- metadataProvider);
- data.resultFilesRowSet =
-
findOutputRowSet(meta.getResultFilesTargetTransformMeta().getName());
+ if (!meta.isFilenameInField()) {
+ initOnFirstProcessingIteration();
}
+ }
- // Remember which column to group on, if any...
- //
- data.groupFieldIndex = -1;
- if (!Utils.isEmpty(data.groupField)) {
- data.groupFieldIndex =
getInputRowMeta().indexOfValue(data.groupField);
- if (data.groupFieldIndex < 0) {
+ if (meta.isFilenameInField()) {
+ int pos = getInputRowMeta().indexOfValue(meta.getFilenameField());
+ if (pos < 0) {
+ throw new HopException(
+ BaseMessages.getString(
+ PKG, "WorkflowExecutor.Exception.UnableToFindField",
meta.getFilenameField()));
+ }
+ String filename = getInputRowMeta().getString(row, pos);
+ if (data.prevFilename == null || !data.prevFilename.equals(filename)) {
+ // When grouping by size, flush a partial group before switching
child workflow so each
+ // execution only receives rows for the previous path. Per-copy
buffer; shared meta is
+ // not touched (see runtimeWorkflowFilename).
+ if (data.prevFilename != null
+ && data.groupSize > 0
+ && data.groupBuffer != null
+ && !data.groupBuffer.isEmpty()) {
+ executeWorkflow();
+ }
+ if (isDetailed()) {
+ logDetailed("Identified a new workflow to execute: '" + filename +
"'");
+ }
+ data.runtimeWorkflowFilename = filename;
+ data.prevFilename = filename;
+ if (!initWorkflow()) {
throw new HopException(
BaseMessages.getString(
- PKG, "WorkflowExecutor.Exception.GroupFieldNotFound",
data.groupField));
+ PKG, "WorkflowExecutor.Exception.UnableToLoadWorkflow",
filename));
}
- data.groupFieldMeta =
getInputRowMeta().getValueMeta(data.groupFieldIndex);
- }
-
- // Edge case: no grouping information given at all...
- //
- if (data.groupSize < 0 && data.groupFieldIndex < 0 && data.groupTime
<= 0) {
- // We assume that we want to execute once per input row, not once
for all the input
- // rows...
- // This is the default but the case might come about anyway
- //
- data.groupSize = 1;
+ initOnFirstProcessingIteration();
}
}
@@ -181,6 +152,70 @@ public class WorkflowExecutor extends
BaseTransform<WorkflowExecutorMeta, Workfl
}
}
+ private void initOnFirstProcessingIteration() throws HopException {
+ // calculate the various output row layouts first...
+ //
+ data.inputRowMeta = getInputRowMeta();
+ data.executionResultsOutputRowMeta = data.inputRowMeta.clone();
+ data.resultRowsOutputRowMeta = data.inputRowMeta.clone();
+ data.resultFilesOutputRowMeta = data.inputRowMeta.clone();
+
+ if (meta.getExecutionResultTargetTransformMeta() != null) {
+ meta.getFields(
+ data.executionResultsOutputRowMeta,
+ getTransformName(),
+ null,
+ meta.getExecutionResultTargetTransformMeta(),
+ this,
+ metadataProvider);
+ data.executionResultRowSet =
+
findOutputRowSet(meta.getExecutionResultTargetTransformMeta().getName());
+ }
+ if (meta.getResultRowsTargetTransformMeta() != null) {
+ meta.getFields(
+ data.resultRowsOutputRowMeta,
+ getTransformName(),
+ null,
+ meta.getResultRowsTargetTransformMeta(),
+ this,
+ metadataProvider);
+ data.resultRowsRowSet =
findOutputRowSet(meta.getResultRowsTargetTransformMeta().getName());
+ }
+ if (meta.getResultFilesTargetTransformMeta() != null) {
+ meta.getFields(
+ data.resultFilesOutputRowMeta,
+ getTransformName(),
+ null,
+ meta.getResultFilesTargetTransformMeta(),
+ this,
+ metadataProvider);
+ data.resultFilesRowSet =
findOutputRowSet(meta.getResultFilesTargetTransformMeta().getName());
+ }
+
+ // Remember which column to group on, if any...
+ //
+ data.groupFieldIndex = -1;
+ if (!Utils.isEmpty(data.groupField)) {
+ data.groupFieldIndex = getInputRowMeta().indexOfValue(data.groupField);
+ if (data.groupFieldIndex < 0) {
+ throw new HopException(
+ BaseMessages.getString(
+ PKG, "WorkflowExecutor.Exception.GroupFieldNotFound",
data.groupField));
+ }
+ data.groupFieldMeta =
getInputRowMeta().getValueMeta(data.groupFieldIndex);
+ }
+
+ // Edge case: no grouping information given at all...
+ //
+ if (data.groupSize < 0 && data.groupFieldIndex < 0 && data.groupTime <= 0)
{
+ // We assume that we want to execute once per input row, not once for
all the input
+ // rows...
+ // This is the default but the case might come about anyway
+ //
+ data.groupSize = 1;
+ }
+ }
+
private void executeWorkflow() throws HopException {
// If we got 0 rows on input we don't really want to execute the workflow
@@ -427,45 +462,42 @@ public class WorkflowExecutor extends
BaseTransform<WorkflowExecutorMeta, Workfl
public boolean init() {
if (super.init()) {
- // First we need to load the mapping (pipeline)
try {
+ // Init commons parameters and data structures. groupBuffer is
initialized only once.
+ data.groupBuffer = new ArrayList<>();
- data.executorWorkflowMeta =
- WorkflowExecutorMeta.loadWorkflowMeta(meta, metadataProvider,
this);
-
- // Do we have a workflow at all?
- //
- if (data.executorWorkflowMeta != null) {
- data.groupBuffer = new ArrayList<>();
-
- // How many rows do we group together for the workflow?
- //
- data.groupSize = -1;
- if (!Utils.isEmpty(meta.getGroupSize())) {
- data.groupSize = Const.toIntExpanded(resolve(meta.getGroupSize()),
-1);
- }
+ // How many rows do we group together for the workflow?
+ data.groupSize = -1;
+ if (!Utils.isEmpty(meta.getGroupSize())) {
+ data.groupSize = Const.toIntExpanded(resolve(meta.getGroupSize()),
-1);
+ }
- // Is there a grouping time set?
- //
- data.groupTime = -1;
- if (!Utils.isEmpty(meta.getGroupTime())) {
- data.groupTime = Const.toInt(resolve(meta.getGroupTime()), -1);
- }
- data.groupTimeStart = System.currentTimeMillis();
+ // Is there a grouping time set?
+ data.groupTime = -1;
+ if (!Utils.isEmpty(meta.getGroupTime())) {
+ data.groupTime = Const.toInt(resolve(meta.getGroupTime()), -1);
+ }
+ data.groupTimeStart = System.currentTimeMillis();
- // Is there a grouping field set?
- //
- data.groupField = null;
- if (!Utils.isEmpty(meta.getGroupField())) {
- data.groupField = resolve(meta.getGroupField());
- }
+ // Is there a grouping field set?
+ data.groupField = null;
+ if (!Utils.isEmpty(meta.getGroupField())) {
+ data.groupField = resolve(meta.getGroupField());
+ }
- // That's all for now...
- return true;
- } else {
- logError("No valid workflow was specified nor loaded!");
+ // First we need to load the workflow (unless the filename comes from
a field)
+ if ((!meta.isFilenameInField() && Utils.isEmpty(meta.getFilename()))
+ || (meta.isFilenameInField() &&
Utils.isEmpty(meta.getFilenameField()))) {
+ logError("No workflow filename given either in path or in a field!");
return false;
}
+
+ if (!meta.isFilenameInField() && !Utils.isEmpty(meta.getFilename())) {
+ return initWorkflow();
+ }
+
+ // Filename comes from a field; load on first row in processRow()
+ return true;
} catch (Exception e) {
logError("Unable to load the executor workflow because of an error :
", e);
}
@@ -473,6 +505,20 @@ public class WorkflowExecutor extends
BaseTransform<WorkflowExecutorMeta, Workfl
return false;
}
+ private boolean initWorkflow() throws HopException {
+ String explicit = meta.isFilenameInField() ? data.runtimeWorkflowFilename
: null;
+ data.executorWorkflowMeta =
+ WorkflowExecutorMeta.loadWorkflowMeta(meta, explicit,
metadataProvider, this);
+
+ if (data.executorWorkflowMeta != null) {
+ data.groupTimeStart = System.currentTimeMillis();
+ return true;
+ } else {
+ logError("No valid workflow was specified nor loaded!");
+ return false;
+ }
+ }
+
@Override
public void dispose() {
data.groupBuffer = null;
diff --git
a/plugins/transforms/workflowexecutor/src/main/java/org/apache/hop/pipeline/transforms/workflowexecutor/WorkflowExecutorData.java
b/plugins/transforms/workflowexecutor/src/main/java/org/apache/hop/pipeline/transforms/workflowexecutor/WorkflowExecutorData.java
index 9ef5a36978..f5173e7f66 100644
---
a/plugins/transforms/workflowexecutor/src/main/java/org/apache/hop/pipeline/transforms/workflowexecutor/WorkflowExecutorData.java
+++
b/plugins/transforms/workflowexecutor/src/main/java/org/apache/hop/pipeline/transforms/workflowexecutor/WorkflowExecutorData.java
@@ -48,6 +48,15 @@ public class WorkflowExecutorData extends BaseTransformData
implements ITransfor
public IRowSet resultFilesRowSet;
public IRowSet executionResultRowSet;
+ public String prevFilename;
+
+ /**
+ * When the child workflow path comes from a field, the resolved path for
this transform copy
+ * only. Shared transform meta filename must not be updated at runtime so
multiple copies stay
+ * isolated.
+ */
+ public String runtimeWorkflowFilename;
+
public WorkflowExecutorData() {
super();
}
diff --git
a/plugins/transforms/workflowexecutor/src/main/java/org/apache/hop/pipeline/transforms/workflowexecutor/WorkflowExecutorDialog.java
b/plugins/transforms/workflowexecutor/src/main/java/org/apache/hop/pipeline/transforms/workflowexecutor/WorkflowExecutorDialog.java
index b6cb1402e9..82f27ceceb 100644
---
a/plugins/transforms/workflowexecutor/src/main/java/org/apache/hop/pipeline/transforms/workflowexecutor/WorkflowExecutorDialog.java
+++
b/plugins/transforms/workflowexecutor/src/main/java/org/apache/hop/pipeline/transforms/workflowexecutor/WorkflowExecutorDialog.java
@@ -54,6 +54,12 @@ import org.eclipse.swt.custom.CCombo;
import org.eclipse.swt.custom.CTabFolder;
import org.eclipse.swt.custom.CTabItem;
import org.eclipse.swt.custom.ScrolledComposite;
+import org.eclipse.swt.events.FocusEvent;
+import org.eclipse.swt.events.FocusListener;
+import org.eclipse.swt.events.ModifyListener;
+import org.eclipse.swt.events.SelectionAdapter;
+import org.eclipse.swt.events.SelectionEvent;
+import org.eclipse.swt.graphics.Cursor;
import org.eclipse.swt.graphics.Rectangle;
import org.eclipse.swt.layout.FillLayout;
import org.eclipse.swt.layout.FormAttachment;
@@ -77,7 +83,16 @@ public class WorkflowExecutorDialog extends
BaseTransformDialog {
private WorkflowExecutorMeta workflowExecutorMeta;
+ private Label wlPath;
private TextVar wPath;
+ private Button wbBrowse;
+
+ private Button wbWorkflowNameInField;
+
+ private Label wlWorkflowNameField;
+ private ComboVar wWorkflowNameField;
+
+ private boolean gotPreviousFields = false;
protected Label wlRunConfiguration;
protected ComboVar wRunConfiguration;
@@ -144,7 +159,9 @@ public class WorkflowExecutorDialog extends
BaseTransformDialog {
buildButtonBar().ok(e -> ok()).cancel(e -> cancel()).build();
- Label wlPath = new Label(shell, SWT.RIGHT);
+ ModifyListener lsMod = e -> workflowExecutorMeta.setChanged();
+
+ wlPath = new Label(shell, SWT.RIGHT);
PropsUi.setLook(wlPath);
wlPath.setText(BaseMessages.getString(PKG,
"WorkflowExecutorDialog.Workflow.Label"));
FormData fdlJobformation = new FormData();
@@ -153,7 +170,7 @@ public class WorkflowExecutorDialog extends
BaseTransformDialog {
fdlJobformation.right = new FormAttachment(middle, -margin);
wlPath.setLayoutData(fdlJobformation);
- Button wbBrowse = new Button(shell, SWT.PUSH);
+ wbBrowse = new Button(shell, SWT.PUSH);
PropsUi.setLook(wbBrowse);
wbBrowse.setText(BaseMessages.getString(PKG,
"WorkflowExecutorDialog.Browse.Label"));
FormData fdBrowse = new FormData();
@@ -170,13 +187,66 @@ public class WorkflowExecutorDialog extends
BaseTransformDialog {
fdPath.right = new FormAttachment(wbBrowse, -margin);
wPath.setLayoutData(fdPath);
+ wbWorkflowNameInField = new Button(shell, SWT.CHECK);
+ PropsUi.setLook(wbWorkflowNameInField);
+ wbWorkflowNameInField.setText(
+ BaseMessages.getString(PKG,
"WorkflowExecutorDialog.WorkflowNameInField.Label"));
+ FormData fdWorkflowNameInField = new FormData();
+ fdWorkflowNameInField.left = new FormAttachment(middle, 0);
+ fdWorkflowNameInField.top = new FormAttachment(wPath, margin);
+ wbWorkflowNameInField.setLayoutData(fdWorkflowNameInField);
+ wbWorkflowNameInField.addSelectionListener(
+ new SelectionAdapter() {
+ @Override
+ public void widgetSelected(SelectionEvent e) {
+ workflowExecutorMeta.setChanged();
+ activeWorkflowNameField();
+ }
+ });
+
+ wlWorkflowNameField = new Label(shell, SWT.RIGHT);
+ wlWorkflowNameField.setText(
+ BaseMessages.getString(PKG,
"WorkflowExecutorDialog.WorkflowNameField.Label"));
+ PropsUi.setLook(wlWorkflowNameField);
+ FormData fdlWorkflowNameField = new FormData();
+ fdlWorkflowNameField.left = new FormAttachment(0, 0);
+ fdlWorkflowNameField.right = new FormAttachment(middle, -margin);
+ fdlWorkflowNameField.top = new FormAttachment(wbWorkflowNameInField,
margin);
+ wlWorkflowNameField.setLayoutData(fdlWorkflowNameField);
+
+ wWorkflowNameField = new ComboVar(variables, shell, SWT.SINGLE | SWT.LEFT
| SWT.BORDER);
+ PropsUi.setLook(wWorkflowNameField);
+ wWorkflowNameField.addModifyListener(lsMod);
+ FormData fdWorkflowNameField = new FormData();
+ fdWorkflowNameField.left = new FormAttachment(middle, 0);
+ fdWorkflowNameField.top = new FormAttachment(wlWorkflowNameField, 0,
SWT.CENTER);
+ fdWorkflowNameField.right = new FormAttachment(100, 0);
+ wWorkflowNameField.setLayoutData(fdWorkflowNameField);
+ wWorkflowNameField.setEnabled(false);
+ wWorkflowNameField.addFocusListener(
+ new FocusListener() {
+ @Override
+ public void focusLost(FocusEvent e) {
+ // Do nothing
+ }
+
+ @Override
+ public void focusGained(FocusEvent e) {
+ Cursor busy = new Cursor(shell.getDisplay(), SWT.CURSOR_WAIT);
+ shell.setCursor(busy);
+ getFields();
+ shell.setCursor(null);
+ busy.dispose();
+ }
+ });
+
wlRunConfiguration = new Label(shell, SWT.RIGHT);
wlRunConfiguration.setText(
BaseMessages.getString(PKG,
"WorkflowExecutorDialog.RunConfiguration.Label"));
PropsUi.setLook(wlRunConfiguration);
FormData fdlRunConfiguration = new FormData();
fdlRunConfiguration.left = new FormAttachment(0, 0);
- fdlRunConfiguration.top = new FormAttachment(wPath, margin);
+ fdlRunConfiguration.top = new FormAttachment(wWorkflowNameField, margin);
fdlRunConfiguration.right = new FormAttachment(middle, -margin);
wlRunConfiguration.setLayoutData(fdlRunConfiguration);
@@ -221,6 +291,42 @@ public class WorkflowExecutorDialog extends
BaseTransformDialog {
return transformName;
}
+ private void getFields() {
+ if (!gotPreviousFields) {
+ try {
+ String field = wWorkflowNameField.getText();
+ IRowMeta r = pipelineMeta.getPrevTransformFields(variables,
transformName);
+ if (r != null) {
+ wWorkflowNameField.setItems(r.getFieldNames());
+ }
+ if (field != null) {
+ wWorkflowNameField.setText(field);
+ }
+ } catch (HopException ke) {
+ new ErrorDialog(
+ shell,
+ BaseMessages.getString(PKG,
"WorkflowExecutorDialog.ErrorLoadingWorkflow.DialogTitle"),
+ BaseMessages.getString(
+ PKG,
"WorkflowExecutorDialog.ErrorLoadingWorkflow.DialogMessage"),
+ ke);
+ }
+ gotPreviousFields = true;
+ }
+ }
+
+ private void activeWorkflowNameField() {
+ wlWorkflowNameField.setEnabled(wbWorkflowNameInField.getSelection());
+ wWorkflowNameField.setEnabled(wbWorkflowNameInField.getSelection());
+ wPath.setEnabled(!wbWorkflowNameInField.getSelection());
+ wlPath.setEnabled(!wbWorkflowNameInField.getSelection());
+ wbBrowse.setEnabled(!wbWorkflowNameInField.getSelection());
+ if (wbWorkflowNameInField.getSelection()) {
+ wPath.setText("");
+ } else {
+ wWorkflowNameField.setText("");
+ }
+ }
+
private void selectWorkflowFile() {
String curFile = variables.resolve(wPath.getText());
@@ -308,6 +414,12 @@ public class WorkflowExecutorDialog extends
BaseTransformDialog {
} else {
wRunConfiguration.setText(workflowExecutorMeta.getRunConfigurationName());
}
+
+
wbWorkflowNameInField.setSelection(workflowExecutorMeta.isFilenameInField());
+ if (workflowExecutorMeta.getFilenameField() != null) {
+ wWorkflowNameField.setText(workflowExecutorMeta.getFilenameField());
+ }
+ activeWorkflowNameField();
} catch (Exception e) {
LogChannel.UI.logError("Error getting workflow run configurations", e);
}
@@ -1038,19 +1150,24 @@ public class WorkflowExecutorDialog extends
BaseTransformDialog {
transformName = wTransformName.getText(); // return value
- try {
- loadWorkflow();
- } catch (HopException e) {
- new ErrorDialog(
- shell,
- BaseMessages.getString(
- PKG,
CONST_WORKFLOW_EXECUTOR_DIALOG_ERROR_LOADING_SPECIFIED_JOB_TITLE),
- BaseMessages.getString(
- PKG,
CONST_WORKFLOW_EXECUTOR_DIALOG_ERROR_LOADING_SPECIFIED_JOB_MESSAGE),
- e);
+ // No check if the workflow to be executed comes from the stream field.
+ if (!wbWorkflowNameInField.getSelection()) {
+ try {
+ loadWorkflow();
+ } catch (HopException e) {
+ new ErrorDialog(
+ shell,
+ BaseMessages.getString(
+ PKG,
CONST_WORKFLOW_EXECUTOR_DIALOG_ERROR_LOADING_SPECIFIED_JOB_TITLE),
+ BaseMessages.getString(
+ PKG,
CONST_WORKFLOW_EXECUTOR_DIALOG_ERROR_LOADING_SPECIFIED_JOB_MESSAGE),
+ e);
+ }
}
workflowExecutorMeta.setFilename(wPath.getText());
+
workflowExecutorMeta.setFilenameInField(wbWorkflowNameInField.getSelection());
+ workflowExecutorMeta.setFilenameField(wWorkflowNameField.getText());
workflowExecutorMeta.setRunConfigurationName(wRunConfiguration.getText());
// Load the information on the tabs, optionally do some
diff --git
a/plugins/transforms/workflowexecutor/src/main/java/org/apache/hop/pipeline/transforms/workflowexecutor/WorkflowExecutorMeta.java
b/plugins/transforms/workflowexecutor/src/main/java/org/apache/hop/pipeline/transforms/workflowexecutor/WorkflowExecutorMeta.java
index f6f3b330bb..d1ba9c7e2a 100644
---
a/plugins/transforms/workflowexecutor/src/main/java/org/apache/hop/pipeline/transforms/workflowexecutor/WorkflowExecutorMeta.java
+++
b/plugins/transforms/workflowexecutor/src/main/java/org/apache/hop/pipeline/transforms/workflowexecutor/WorkflowExecutorMeta.java
@@ -93,6 +93,14 @@ public class WorkflowExecutorMeta
hopMetadataPropertyType = HopMetadataPropertyType.WORKFLOW_FILE)
private String filename;
+ /** Flag that indicate that workflow name is specified in a stream's field */
+ @HopMetadataProperty(key = "filenameInField")
+ private boolean filenameInField;
+
+ /** Name of the field containing the workflow file's name */
+ @HopMetadataProperty(key = "filenameField")
+ private String filenameField;
+
/**
* The number of input rows that are sent as result rows to the workflow in
one go, defaults to
* "1"
@@ -253,6 +261,7 @@ public class WorkflowExecutorMeta
parameters = new ArrayList<>();
resultRowsField = new ArrayList<>();
inheritingAllVariables = true;
+ filenameInField = false;
groupSize = "1";
groupField = "";
@@ -432,20 +441,41 @@ public class WorkflowExecutorMeta
IHopMetadataProvider metadataProvider,
IVariables variables)
throws HopException {
- WorkflowMeta mappingWorkflowMeta = null;
+ return loadWorkflowMeta(executorMeta, null, metadataProvider, variables);
+ }
+
+ /**
+ * Loads child workflow metadata from a file.
+ *
+ * @param explicitWorkflowFilename when non-empty, this path is loaded and
resolved instead of the
+ * filename stored on {@code executorMeta}. Use when the workflow path
is taken from an
+ * incoming row at runtime so multiple transform copies must not mutate
shared meta.
+ */
+ public static final synchronized WorkflowMeta loadWorkflowMeta(
+ WorkflowExecutorMeta executorMeta,
+ String explicitWorkflowFilename,
+ IHopMetadataProvider metadataProvider,
+ IVariables variables)
+ throws HopException {
+ String filenameToUse =
+ !Utils.isEmpty(explicitWorkflowFilename)
+ ? explicitWorkflowFilename
+ : executorMeta.getFilename();
CurrentDirectoryResolver r = new CurrentDirectoryResolver();
IVariables tmpSpace =
- r.resolveCurrentDirectory(
- variables, executorMeta.getParentTransformMeta(),
executorMeta.getFilename());
+ r.resolveCurrentDirectory(variables,
executorMeta.getParentTransformMeta(), filenameToUse);
- String realFilename = tmpSpace.resolve(executorMeta.getFilename());
+ String realFilename = tmpSpace.resolve(filenameToUse);
+ if (variables != null) {
+ realFilename = variables.resolve(realFilename);
+ }
// OK, load the meta-data from file...
//
// Don't set internal variables: they belong to the parent thread!
//
- mappingWorkflowMeta = new WorkflowMeta(variables, realFilename,
metadataProvider);
+ WorkflowMeta mappingWorkflowMeta = new WorkflowMeta(variables,
realFilename, metadataProvider);
LogChannel.GENERAL.logDetailed(
"Loaded workflow", "Workflow was loaded from XML file [" +
realFilename + "]");
diff --git
a/plugins/transforms/workflowexecutor/src/main/resources/org/apache/hop/pipeline/transforms/workflowexecutor/messages/messages_en_US.properties
b/plugins/transforms/workflowexecutor/src/main/resources/org/apache/hop/pipeline/transforms/workflowexecutor/messages/messages_en_US.properties
index f48a49c196..d5d8aac442 100644
---
a/plugins/transforms/workflowexecutor/src/main/resources/org/apache/hop/pipeline/transforms/workflowexecutor/messages/messages_en_US.properties
+++
b/plugins/transforms/workflowexecutor/src/main/resources/org/apache/hop/pipeline/transforms/workflowexecutor/messages/messages_en_US.properties
@@ -18,6 +18,7 @@
WorkflowExecutor.Description=This transform executes a Hop workflow, sets
parameters and passes rows.
WorkflowExecutor.Exception.GroupFieldNotFound=Group field ''{0}'' could not be
found in the input stream
WorkflowExecutor.Exception.UnableToFindField=Unable to find field ''{0}'' in
the input stream.
+WorkflowExecutor.Exception.UnableToLoadWorkflow=Unable to load the specified
workflow\: {0}
WorkflowExecutor.IncorrectDataTypePassed=The ''{0}'' data types passed from
the workflows result rows does not correspond to the specified ''{1}'' data
types. Make sure you are passing rows with the expected layout.
WorkflowExecutor.Name=Workflow executor
WorkflowExecutor.UnexpectedError=There was an unexpected error\:
@@ -72,6 +73,8 @@ WorkflowExecutorDialog.RunConfiguration.Label=Run
configuration
WorkflowExecutorDialog.Shell.Title=Workflow executor
WorkflowExecutorDialog.TransformName.Label=Transform name
WorkflowExecutorDialog.Workflow.Label=Workflow
+WorkflowExecutorDialog.WorkflowNameField.Label=Workflow field
+WorkflowExecutorDialog.WorkflowNameInField.Label=Workflow from field
WorkflowExecutorMeta.CheckResult.NoInputReceived=No input is received from
previous transforms
WorkflowExecutorMeta.CheckResult.NotReceivingAnyFields=No input fields are
received from previous transforms
WorkflowExecutorMeta.CheckResult.TransformReceivingFields=This transform is
receiving fields from previous transforms.
diff --git
a/plugins/transforms/workflowexecutor/src/test/java/org/apache/hop/pipeline/transforms/workflowexecutor/WorkflowExecutorMetaTest.java
b/plugins/transforms/workflowexecutor/src/test/java/org/apache/hop/pipeline/transforms/workflowexecutor/WorkflowExecutorMetaTest.java
index c45a1053a8..f15100fe1a 100644
---
a/plugins/transforms/workflowexecutor/src/test/java/org/apache/hop/pipeline/transforms/workflowexecutor/WorkflowExecutorMetaTest.java
+++
b/plugins/transforms/workflowexecutor/src/test/java/org/apache/hop/pipeline/transforms/workflowexecutor/WorkflowExecutorMetaTest.java
@@ -18,6 +18,8 @@
package org.apache.hop.pipeline.transforms.workflowexecutor;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import org.apache.hop.pipeline.transform.TransformSerializationTestUtil;
import org.junit.jupiter.api.Test;
@@ -31,6 +33,8 @@ class WorkflowExecutorMetaTest {
"/workflow-executor-transform.xml", WorkflowExecutorMeta.class);
assertEquals("${PROJECT_HOME}/loops/child-workflow-executor.hwf",
meta.getFilename());
+ assertFalse(meta.isFilenameInField());
+ assertEquals("", meta.getFilenameField() == null ? "" :
meta.getFilenameField());
assertEquals("ExecutionTime", meta.getExecutionTimeField());
assertEquals("ExecutionResult", meta.getExecutionResultField());
assertEquals("ExecutionNrErrors", meta.getExecutionNrErrorsField());
@@ -61,6 +65,8 @@ class WorkflowExecutorMetaTest {
WorkflowExecutorMeta clone = (WorkflowExecutorMeta) meta.clone();
assertEquals(clone.getFilename(), meta.getFilename());
+ assertEquals(clone.isFilenameInField(), meta.isFilenameInField());
+ assertEquals(clone.getFilenameField(), meta.getFilenameField());
assertEquals(clone.getExecutionTimeField(), meta.getExecutionTimeField());
assertEquals(clone.getExecutionResultField(),
meta.getExecutionResultField());
assertEquals(clone.getExecutionNrErrorsField(),
meta.getExecutionNrErrorsField());
@@ -81,4 +87,21 @@ class WorkflowExecutorMetaTest {
assertEquals(clone.getResultRowsField().size(),
meta.getResultRowsField().size());
assertEquals(clone.getParameters().size(), meta.getParameters().size());
}
+
+ @Test
+ void setDefaultDisablesFilenameInField() {
+ WorkflowExecutorMeta meta = new WorkflowExecutorMeta();
+ meta.setDefault();
+ assertFalse(meta.isFilenameInField());
+ }
+
+ @Test
+ void testSerializationFilenameInField() throws Exception {
+ WorkflowExecutorMeta meta =
+ TransformSerializationTestUtil.testSerialization(
+ "/workflow-executor-fromfield-transform.xml",
WorkflowExecutorMeta.class);
+
+ assertTrue(meta.isFilenameInField());
+ assertEquals("wf_to_run", meta.getFilenameField());
+ }
}
diff --git
a/plugins/transforms/workflowexecutor/src/test/java/org/apache/hop/pipeline/transforms/workflowexecutor/WorkflowExecutorTest.java
b/plugins/transforms/workflowexecutor/src/test/java/org/apache/hop/pipeline/transforms/workflowexecutor/WorkflowExecutorTest.java
new file mode 100644
index 0000000000..4868c84cb1
--- /dev/null
+++
b/plugins/transforms/workflowexecutor/src/test/java/org/apache/hop/pipeline/transforms/workflowexecutor/WorkflowExecutorTest.java
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hop.pipeline.transforms.workflowexecutor;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+import java.net.URISyntaxException;
+import java.nio.file.Paths;
+import java.util.Objects;
+import org.apache.hop.core.HopEnvironment;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.variables.Variables;
+import org.apache.hop.junit.rules.RestoreHopEngineEnvironmentExtension;
+import org.apache.hop.metadata.serializer.memory.MemoryMetadataProvider;
+import org.apache.hop.pipeline.PipelineMeta;
+import org.apache.hop.pipeline.engines.local.LocalPipelineEngine;
+import org.apache.hop.pipeline.transform.TransformMeta;
+import org.apache.hop.workflow.WorkflowMeta;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+/** Unit tests for {@link WorkflowExecutor} runtime behaviour. */
+class WorkflowExecutorTest {
+
+ @RegisterExtension
+ static final RestoreHopEngineEnvironmentExtension env =
+ new RestoreHopEngineEnvironmentExtension();
+
+ @BeforeEach
+ void initHop() throws HopException {
+ HopEnvironment.init();
+ }
+
+ private WorkflowExecutor newExecutor(WorkflowExecutorMeta meta,
WorkflowExecutorData data) {
+ TransformMeta transformMeta = mock(TransformMeta.class);
+ when(transformMeta.getName()).thenReturn("workflow_executor");
+ when(transformMeta.isPartitioned()).thenReturn(false);
+ PipelineMeta pipelineMeta = mock(PipelineMeta.class);
+ when(pipelineMeta.findTransform(anyString())).thenReturn(transformMeta);
+ LocalPipelineEngine pipeline = spy(new LocalPipelineEngine());
+ WorkflowExecutor executor =
+ new WorkflowExecutor(transformMeta, meta, data, 0, pipelineMeta,
pipeline);
+ executor.setMetadataProvider(new MemoryMetadataProvider());
+ return executor;
+ }
+
+ @Test
+ void initFailsWhenStaticFilenameMissing() {
+ WorkflowExecutorMeta meta = new WorkflowExecutorMeta();
+ meta.setDefault();
+ meta.setFilenameInField(false);
+ meta.setFilename("");
+ meta.setRunConfigurationName("local");
+
+ WorkflowExecutor executor = newExecutor(meta, new WorkflowExecutorData());
+ assertFalse(executor.init());
+ }
+
+ @Test
+ void initSucceedsWhenFilenameComesFromField() {
+ WorkflowExecutorMeta meta = new WorkflowExecutorMeta();
+ meta.setDefault();
+ meta.setFilenameInField(true);
+ meta.setFilenameField("child_path");
+ meta.setFilename("/tmp/ignored-default.hwf");
+ meta.setRunConfigurationName("local");
+
+ WorkflowExecutor executor = newExecutor(meta, new WorkflowExecutorData());
+ assertTrue(executor.init());
+ }
+
+ @Test
+ void initFailsWhenFilenameFieldMissing() {
+ WorkflowExecutorMeta meta = new WorkflowExecutorMeta();
+ meta.setDefault();
+ meta.setFilenameInField(true);
+ meta.setFilenameField("");
+ meta.setRunConfigurationName("local");
+
+ WorkflowExecutor executor = newExecutor(meta, new WorkflowExecutorData());
+ assertFalse(executor.init());
+ }
+
+ @Test
+ void initLoadsChildWorkflowFromFilesystemPath() throws URISyntaxException {
+ String path =
+ Paths.get(
+ Objects.requireNonNull(
+ WorkflowExecutorTest.class.getResource(
+
"/org/apache/hop/pipeline/transforms/workflowexecutor/minimal-child.hwf"))
+ .toURI())
+ .toAbsolutePath()
+ .toString();
+
+ WorkflowExecutorMeta meta = new WorkflowExecutorMeta();
+ meta.setDefault();
+ meta.setFilenameInField(false);
+ meta.setFilename(path);
+ meta.setRunConfigurationName("local");
+
+ WorkflowExecutor executor = newExecutor(meta, new WorkflowExecutorData());
+ assertTrue(executor.init());
+ assertNotNull(executor.getData().executorWorkflowMeta);
+ }
+
+ @Test
+ void loadWorkflowMetaHonorsExplicitFilenameOverWrongMetaFilename()
+ throws HopException, URISyntaxException {
+ String goodPath =
+ Paths.get(
+ Objects.requireNonNull(
+ WorkflowExecutorTest.class.getResource(
+
"/org/apache/hop/pipeline/transforms/workflowexecutor/minimal-child.hwf"))
+ .toURI())
+ .toAbsolutePath()
+ .toString();
+
+ WorkflowExecutorMeta meta = new WorkflowExecutorMeta();
+ meta.setDefault();
+ meta.setFilename("/this/path/does/not/exist/bad-child.hwf");
+
+ WorkflowMeta loaded =
+ WorkflowExecutorMeta.loadWorkflowMeta(
+ meta, goodPath, new MemoryMetadataProvider(), new Variables());
+
+ assertNotNull(loaded);
+ assertFalse(loaded.getActions().isEmpty());
+ }
+}
diff --git a/integration-tests/transforms/main-0083-workflow-executor-test.hwf
b/plugins/transforms/workflowexecutor/src/test/resources/org/apache/hop/pipeline/transforms/workflowexecutor/minimal-child.hwf
similarity index 52%
copy from integration-tests/transforms/main-0083-workflow-executor-test.hwf
copy to
plugins/transforms/workflowexecutor/src/test/resources/org/apache/hop/pipeline/transforms/workflowexecutor/minimal-child.hwf
index 895b54c5da..ef669b313a 100644
--- a/integration-tests/transforms/main-0083-workflow-executor-test.hwf
+++
b/plugins/transforms/workflowexecutor/src/test/resources/org/apache/hop/pipeline/transforms/workflowexecutor/minimal-child.hwf
@@ -18,15 +18,15 @@ limitations under the License.
-->
<workflow>
- <name>main-0083-workflow-executor-test</name>
+ <name>minimal-child</name>
<name_sync_with_filename>Y</name_sync_with_filename>
<description/>
<extended_description/>
<workflow_version/>
<created_user>-</created_user>
- <created_date>2020/12/07 13:03:35.562</created_date>
+ <created_date>2025/01/01 00:00:00.000</created_date>
<modified_user>-</modified_user>
- <modified_date>2020/12/07 13:03:35.562</modified_date>
+ <modified_date>2025/01/01 00:00:00.000</modified_date>
<parameters>
</parameters>
<actions>
@@ -45,8 +45,8 @@ limitations under the License.
<schedulerType>0</schedulerType>
<weekDay>1</weekDay>
<parallel>N</parallel>
- <xloc>160</xloc>
- <yloc>96</yloc>
+ <xloc>112</xloc>
+ <yloc>112</yloc>
<attributes_hac/>
</action>
<action>
@@ -55,69 +55,18 @@ limitations under the License.
<type>SUCCESS</type>
<attributes/>
<parallel>N</parallel>
- <xloc>544</xloc>
- <yloc>96</yloc>
- <attributes_hac/>
- </action>
- <action>
- <name>Abort workflow</name>
- <description/>
- <type>ABORT</type>
- <attributes/>
- <always_log_rows>N</always_log_rows>
- <parallel>N</parallel>
- <xloc>320</xloc>
- <yloc>256</yloc>
- <attributes_hac/>
- </action>
- <action>
- <name>0083-workflow-executor-run-wf.hpl</name>
- <description/>
- <type>PIPELINE</type>
- <attributes/>
- <add_date>N</add_date>
- <add_time>N</add_time>
- <clear_files>N</clear_files>
- <clear_rows>N</clear_rows>
- <create_parent_folder>N</create_parent_folder>
- <exec_per_row>N</exec_per_row>
- <filename>${PROJECT_HOME}/0083-workflow-executor-run-wf.hpl</filename>
- <loglevel>Basic</loglevel>
- <parameters>
- <pass_all_parameters>Y</pass_all_parameters>
- </parameters>
- <params_from_previous>N</params_from_previous>
- <run_configuration>local</run_configuration>
- <set_append_logfile>N</set_append_logfile>
- <set_logfile>N</set_logfile>
- <wait_until_finished>Y</wait_until_finished>
- <parallel>N</parallel>
<xloc>320</xloc>
- <yloc>98</yloc>
+ <yloc>112</yloc>
<attributes_hac/>
</action>
</actions>
<hops>
<hop>
<from>START</from>
- <to>0083-workflow-executor-run-wf.hpl</to>
- <enabled>Y</enabled>
- <evaluation>Y</evaluation>
- <unconditional>Y</unconditional>
- </hop>
- <hop>
- <from>0083-workflow-executor-run-wf.hpl</from>
<to>Success</to>
<enabled>Y</enabled>
<evaluation>Y</evaluation>
- <unconditional>N</unconditional>
- </hop>
- <hop>
- <from>0083-workflow-executor-run-wf.hpl</from>
- <to>Abort workflow</to>
- <enabled>Y</enabled>
- <evaluation>N</evaluation>
- <unconditional>N</unconditional>
+ <unconditional>Y</unconditional>
</hop>
</hops>
<notepads>
diff --git
a/plugins/transforms/workflowexecutor/src/test/resources/workflow-executor-transform.xml
b/plugins/transforms/workflowexecutor/src/test/resources/workflow-executor-fromfield-transform.xml
similarity index 97%
copy from
plugins/transforms/workflowexecutor/src/test/resources/workflow-executor-transform.xml
copy to
plugins/transforms/workflowexecutor/src/test/resources/workflow-executor-fromfield-transform.xml
index 4064641e4a..0cc8e568ca 100644
---
a/plugins/transforms/workflowexecutor/src/test/resources/workflow-executor-transform.xml
+++
b/plugins/transforms/workflowexecutor/src/test/resources/workflow-executor-fromfield-transform.xml
@@ -29,6 +29,8 @@
</partitioning>
<run_configuration>local</run_configuration>
<filename>${PROJECT_HOME}/loops/child-workflow-executor.hwf</filename>
+ <filenameInField>Y</filenameInField>
+ <filenameField>wf_to_run</filenameField>
<group_size>1</group_size>
<group_field/>
<group_time/>
diff --git
a/plugins/transforms/workflowexecutor/src/test/resources/workflow-executor-transform.xml
b/plugins/transforms/workflowexecutor/src/test/resources/workflow-executor-transform.xml
index 4064641e4a..d2058dec6b 100644
---
a/plugins/transforms/workflowexecutor/src/test/resources/workflow-executor-transform.xml
+++
b/plugins/transforms/workflowexecutor/src/test/resources/workflow-executor-transform.xml
@@ -29,6 +29,8 @@
</partitioning>
<run_configuration>local</run_configuration>
<filename>${PROJECT_HOME}/loops/child-workflow-executor.hwf</filename>
+ <filenameInField>N</filenameInField>
+ <filenameField/>
<group_size>1</group_size>
<group_field/>
<group_time/>
