You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hop.apache.org by mc...@apache.org on 2026/03/27 12:00:39 UTC
(hop) branch main updated: issue #1937 (Stream Schema merge XML cleanup with integration test) (#6880)
This is an automated email from the ASF dual-hosted git repository.
mcasters 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 069171ff6d issue #1937 (Stream Schema merge XML cleanup with integration test) (#6880)
069171ff6d is described below
commit 069171ff6da49283d0acb1c152296e50625ca658
Author: Matt Casters <ma...@gmail.com>
AuthorDate: Fri Mar 27 13:00:34 2026 +0100
issue #1937 (Stream Schema merge XML cleanup with integration test) (#6880)
---
.../transforms/0091-stream-schema-merge.hpl | 279 +++++++++++++++++++++
.../datasets/golden-stream-schema-merge.csv | 10 +
.../transforms/main-0091-stream-schema-merge.hwf | 58 +++++
.../dataset/golden-stream-schema-merge.json | 48 ++++
.../unit-test/0091-stream-schema-merge UNIT.json | 48 ++++
.../transforms/streamschemamerge/StreamSchema.java | 82 +++---
.../streamschemamerge/StreamSchemaDialog.java | 116 +++------
.../streamschemamerge/StreamSchemaMeta.java | 195 +++++---------
.../messages/messages_en_US.properties | 1 +
.../streamschemamerge/StreamSchemaMetaTest.java | 36 +++
.../src/test/resources/stream-schema-merge.xml | 32 +++
11 files changed, 635 insertions(+), 270 deletions(-)
diff --git a/integration-tests/transforms/0091-stream-schema-merge.hpl b/integration-tests/transforms/0091-stream-schema-merge.hpl
new file mode 100644
index 0000000000..489ce3936d
--- /dev/null
+++ b/integration-tests/transforms/0091-stream-schema-merge.hpl
@@ -0,0 +1,279 @@
+<?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>0091-stream-schema-merge</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/26 11:04:28.367</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/03/26 11:04:28.367</modified_date>
+ </info>
+ <notepads>
+ </notepads>
+ <order>
+ <hop>
+ <from>source1</from>
+ <to>Stream Schema Merge</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>source2</from>
+ <to>Stream Schema Merge</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>source3</from>
+ <to>Stream Schema Merge</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>Stream Schema Merge</from>
+ <to>validate</to>
+ <enabled>Y</enabled>
+ </hop>
+ </order>
+ <transform>
+ <name>source1</name>
+ <type>DataGrid</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <data>
+ <line>
+ <item>value-1-1</item>
+ <item>1</item>
+ </line>
+ <line>
+ <item>value-1-2</item>
+ <item>1</item>
+ </line>
+ <line>
+ <item>value-1-3</item>
+ <item>1</item>
+ </line>
+ </data>
+ <fields>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <currency/>
+ <set_empty_string>N</set_empty_string>
+ <name>f1</name>
+ <format/>
+ <group/>
+ <decimal/>
+ <type>String</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <currency/>
+ <set_empty_string>N</set_empty_string>
+ <name>source</name>
+ <format/>
+ <group/>
+ <decimal/>
+ <type>String</type>
+ </field>
+ </fields>
+ <attributes/>
+ <GUI>
+ <xloc>80</xloc>
+ <yloc>32</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>source2</name>
+ <type>DataGrid</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <data>
+ <line>
+ <item>value-2-1</item>
+ <item>value-3-1</item>
+ <item>2</item>
+ </line>
+ <line>
+ <item>value-2-2</item>
+ <item>value-3-2</item>
+ <item>2</item>
+ </line>
+ <line>
+ <item>value-2-3</item>
+ <item>value-3-3</item>
+ <item>2</item>
+ </line>
+ </data>
+ <fields>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>f2</name>
+ <type>String</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>f3</name>
+ <type>String</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>source</name>
+ <type>String</type>
+ </field>
+ </fields>
+ <attributes/>
+ <GUI>
+ <xloc>80</xloc>
+ <yloc>112</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>source3</name>
+ <type>DataGrid</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <data>
+ <line>
+ <item>value-1-4</item>
+ <item>value-4-1</item>
+ <item>3</item>
+ </line>
+ <line>
+ <item>value-1-5</item>
+ <item>value-4-2</item>
+ <item>3</item>
+ </line>
+ <line>
+ <item>value-1-6</item>
+ <item>value-4-3</item>
+ <item>3</item>
+ </line>
+ </data>
+ <fields>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>f1</name>
+ <type>String</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>f4</name>
+ <type>String</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>source</name>
+ <type>String</type>
+ </field>
+ </fields>
+ <attributes/>
+ <GUI>
+ <xloc>80</xloc>
+ <yloc>192</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>validate</name>
+ <type>Dummy</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <attributes/>
+ <GUI>
+ <xloc>336</xloc>
+ <yloc>192</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>Stream Schema Merge</name>
+ <type>StreamSchema</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <transforms>
+ <transform>
+ <name>source1</name>
+ </transform>
+ <transform>
+ <name>source2</name>
+ </transform>
+ <transform>
+ <name>source3</name>
+ </transform>
+ </transforms>
+ <attributes/>
+ <GUI>
+ <xloc>192</xloc>
+ <yloc>192</yloc>
+ </GUI>
+ </transform>
+ <transform_error_handling>
+ </transform_error_handling>
+ <attributes/>
+</pipeline>
diff --git a/integration-tests/transforms/datasets/golden-stream-schema-merge.csv b/integration-tests/transforms/datasets/golden-stream-schema-merge.csv
new file mode 100644
index 0000000000..5c892cce58
--- /dev/null
+++ b/integration-tests/transforms/datasets/golden-stream-schema-merge.csv
@@ -0,0 +1,10 @@
+f1,source,f2,f3,f4
+value-1-1,1,,,
+value-1-2,1,,,
+value-1-3,1,,,
+value-1-4,3,,,value-4-1
+value-1-5,3,,,value-4-2
+value-1-6,3,,,value-4-3
+,2,value-2-1,value-3-1,
+,2,value-2-2,value-3-2,
+,2,value-2-3,value-3-3,
diff --git a/integration-tests/transforms/main-0091-stream-schema-merge.hwf b/integration-tests/transforms/main-0091-stream-schema-merge.hwf
new file mode 100644
index 0000000000..e84efdeeba
--- /dev/null
+++ b/integration-tests/transforms/main-0091-stream-schema-merge.hwf
@@ -0,0 +1,58 @@
+<?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.
+
+-->
+<workflow>
+ <name>main-0091-stream-schema-merge</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description/>
+ <extended_description/>
+ <workflow_version/>
+ <created_user>-</created_user>
+ <created_date>2026/03/26 11:03:53.373</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/03/26 11:03:53.373</modified_date>
+ <parameters>
+ </parameters>
+ <actions>
+ <action>
+ <name>Start</name>
+ <description/>
+ <type>SPECIAL</type>
+ <attributes/>
+ <DayOfMonth>1</DayOfMonth>
+ <doNotWaitOnFirstExecution>N</doNotWaitOnFirstExecution>
+ <hour>12</hour>
+ <intervalMinutes>60</intervalMinutes>
+ <intervalSeconds>0</intervalSeconds>
+ <minutes>0</minutes>
+ <repeat>N</repeat>
+ <schedulerType>0</schedulerType>
+ <weekDay>1</weekDay>
+ <parallel>N</parallel>
+ <xloc>80</xloc>
+ <yloc>80</yloc>
+ <attributes_hac/>
+ </action>
+ </actions>
+ <hops>
+ </hops>
+ <notepads>
+ </notepads>
+ <attributes/>
+</workflow>
diff --git a/integration-tests/transforms/metadata/dataset/golden-stream-schema-merge.json b/integration-tests/transforms/metadata/dataset/golden-stream-schema-merge.json
new file mode 100644
index 0000000000..73d8f97a9f
--- /dev/null
+++ b/integration-tests/transforms/metadata/dataset/golden-stream-schema-merge.json
@@ -0,0 +1,48 @@
+{
+ "base_filename": "golden-stream-schema-merge.csv",
+ "name": "golden-stream-schema-merge",
+ "description": "",
+ "dataset_fields": [
+ {
+ "field_comment": "",
+ "field_length": -1,
+ "field_type": 2,
+ "field_precision": -1,
+ "field_name": "f1",
+ "field_format": ""
+ },
+ {
+ "field_comment": "",
+ "field_length": -1,
+ "field_type": 2,
+ "field_precision": -1,
+ "field_name": "source",
+ "field_format": ""
+ },
+ {
+ "field_comment": "",
+ "field_length": -1,
+ "field_type": 2,
+ "field_precision": -1,
+ "field_name": "f2",
+ "field_format": ""
+ },
+ {
+ "field_comment": "",
+ "field_length": -1,
+ "field_type": 2,
+ "field_precision": -1,
+ "field_name": "f3",
+ "field_format": ""
+ },
+ {
+ "field_comment": "",
+ "field_length": -1,
+ "field_type": 2,
+ "field_precision": -1,
+ "field_name": "f4",
+ "field_format": ""
+ }
+ ],
+ "folder_name": ""
+}
\ No newline at end of file
diff --git a/integration-tests/transforms/metadata/unit-test/0091-stream-schema-merge UNIT.json b/integration-tests/transforms/metadata/unit-test/0091-stream-schema-merge UNIT.json
new file mode 100644
index 0000000000..854c202356
--- /dev/null
+++ b/integration-tests/transforms/metadata/unit-test/0091-stream-schema-merge UNIT.json
@@ -0,0 +1,48 @@
+{
+ "database_replacements": [],
+ "autoOpening": true,
+ "description": "",
+ "persist_filename": "",
+ "test_type": "UNIT_TEST",
+ "variableValues": [],
+ "basePath": "${HOP_UNIT_TESTS_FOLDER}",
+ "golden_data_sets": [
+ {
+ "field_mappings": [
+ {
+ "transform_field": "f1",
+ "data_set_field": "f1"
+ },
+ {
+ "transform_field": "source",
+ "data_set_field": "source"
+ },
+ {
+ "transform_field": "f2",
+ "data_set_field": "f2"
+ },
+ {
+ "transform_field": "f3",
+ "data_set_field": "f3"
+ },
+ {
+ "transform_field": "f4",
+ "data_set_field": "f4"
+ }
+ ],
+ "field_order": [
+ "f1",
+ "source",
+ "f2",
+ "f3",
+ "f4"
+ ],
+ "data_set_name": "golden-stream-schema-merge",
+ "transform_name": "validate"
+ }
+ ],
+ "input_data_sets": [],
+ "name": "0091-stream-schema-merge UNIT",
+ "trans_test_tweaks": [],
+ "pipeline_filename": "./0091-stream-schema-merge.hpl"
+}
\ No newline at end of file
diff --git a/plugins/transforms/streamschemamerge/src/main/java/org/apache/hop/pipeline/transforms/streamschemamerge/StreamSchema.java b/plugins/transforms/streamschemamerge/src/main/java/org/apache/hop/pipeline/transforms/streamschemamerge/StreamSchema.java
index ce1eabbd49..453b191d8b 100644
--- a/plugins/transforms/streamschemamerge/src/main/java/org/apache/hop/pipeline/transforms/streamschemamerge/StreamSchema.java
+++ b/plugins/transforms/streamschemamerge/src/main/java/org/apache/hop/pipeline/transforms/streamschemamerge/StreamSchema.java
@@ -21,6 +21,8 @@ import java.util.ArrayList;
import java.util.List;
import org.apache.hop.core.IRowSet;
import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.exception.HopPluginException;
+import org.apache.hop.core.exception.HopTransformException;
import org.apache.hop.core.row.IRowMeta;
import org.apache.hop.core.row.RowDataUtil;
import org.apache.hop.pipeline.Pipeline;
@@ -38,10 +40,9 @@ import org.apache.hop.pipeline.transform.stream.IStream;
*
* <p>Because this transform combines multiple streams with different RowMetas together, it is
* deemed "not safe" and will fail if you try to run the pipeline with the "Enable Safe Mode
- * checked". Therefore it disables safe mode
+ * checked". Therefor it disables safe mode
*/
public class StreamSchema extends BaseTransform<StreamSchemaMeta, StreamSchemaData> {
-
/**
* The constructor should simply pass on its arguments to the parent class.
*
@@ -62,17 +63,6 @@ public class StreamSchema extends BaseTransform<StreamSchemaMeta, StreamSchemaDa
pipeline.setSafeModeEnabled(false); // safe mode is incompatible with this transform
}
- /**
- * Initialize data structures that we have information for at init time
- *
- * @return true if initialization completed successfully, false if there was an error
- */
- @Override
- public boolean init() {
-
- return super.init();
- }
-
/**
* For each row, create a new output row in the model of the master output row and copy the data
* values in to the appropriate indexes
@@ -82,47 +72,17 @@ public class StreamSchema extends BaseTransform<StreamSchemaMeta, StreamSchemaDa
*/
@Override
public boolean processRow() throws HopException {
-
/*
* Code in first method is responsible for finishing the initialization that we couldn't do earlier
*/
if (first) {
first = false;
-
- List<String> tNames = new ArrayList<>();
- List<IStream> streams = new ArrayList<>();
- List<IRowMeta> rowMetas = new ArrayList<>();
- data.rowSets = new ArrayList<>();
-
- int inputStreamsNum = meta.getTransformIOMeta().getInfoStreams().size();
-
- for (int i = 0; i < inputStreamsNum; i++) {
- IStream s = meta.getTransformIOMeta().getInfoStreams().get(i);
- IRowSet r = findInputRowSet(s.getTransformName());
- streams.add(s);
- tNames.add(r.getName());
- data.rowSets.add(r);
- rowMetas.add(getPipelineMeta().getTransformFields(this, s.getTransformName()));
- }
-
- data.infoStreams = streams;
- data.numTransforms = inputStreamsNum;
- data.transformNames = tNames.toArray(new String[0]);
- data.rowMetas = rowMetas.toArray(new IRowMeta[0]);
- // creates mapping and master output row
- data.schemaMapping = new SchemaMapper(data.rowMetas);
- data.mapping = data.schemaMapping.getMapping();
- data.outputRowMeta = data.schemaMapping.getRowMeta();
- // set the order of the inputrowsets to match the order we've defined
- setInputRowSets(data.rowSets);
- if (isDetailed()) {
- logDetailed("Finished generating mapping");
- }
+ processRowFirstCall();
}
Object[] incomingRow = getRow(); // get the next available row
- // if no more rows are expected, indicate transform is finished and processRow() should not be
+ // If no more rows are expected, indicate transform is finished and processRow() should not be
// called again
if (incomingRow == null) {
setOutputDone();
@@ -169,6 +129,38 @@ public class StreamSchema extends BaseTransform<StreamSchemaMeta, StreamSchemaDa
return true;
}
+ private void processRowFirstCall() throws HopTransformException, HopPluginException {
+ List<String> tNames = new ArrayList<>();
+ List<IStream> streams = new ArrayList<>();
+ List<IRowMeta> rowMetas = new ArrayList<>();
+ data.rowSets = new ArrayList<>();
+
+ int inputStreamsNum = meta.getTransformIOMeta().getInfoStreams().size();
+
+ for (int i = 0; i < inputStreamsNum; i++) {
+ IStream s = meta.getTransformIOMeta().getInfoStreams().get(i);
+ IRowSet r = findInputRowSet(s.getTransformName());
+ streams.add(s);
+ tNames.add(r.getName());
+ data.rowSets.add(r);
+ rowMetas.add(getPipelineMeta().getTransformFields(this, s.getTransformName()));
+ }
+
+ data.infoStreams = streams;
+ data.numTransforms = inputStreamsNum;
+ data.transformNames = tNames.toArray(new String[0]);
+ data.rowMetas = rowMetas.toArray(new IRowMeta[0]);
+ // creates mapping and master output row
+ data.schemaMapping = new SchemaMapper(data.rowMetas);
+ data.mapping = data.schemaMapping.getMapping();
+ data.outputRowMeta = data.schemaMapping.getRowMeta();
+ // set the order of the inputrowsets to match the order we've defined
+ setInputRowSets(data.rowSets);
+ if (isDetailed()) {
+ logDetailed("Finished generating mapping");
+ }
+ }
+
/** Clear transforms from transform data */
@Override
public void dispose() {
diff --git a/plugins/transforms/streamschemamerge/src/main/java/org/apache/hop/pipeline/transforms/streamschemamerge/StreamSchemaDialog.java b/plugins/transforms/streamschemamerge/src/main/java/org/apache/hop/pipeline/transforms/streamschemamerge/StreamSchemaDialog.java
index ab1b592d13..66cbae30c5 100644
--- a/plugins/transforms/streamschemamerge/src/main/java/org/apache/hop/pipeline/transforms/streamschemamerge/StreamSchemaDialog.java
+++ b/plugins/transforms/streamschemamerge/src/main/java/org/apache/hop/pipeline/transforms/streamschemamerge/StreamSchemaDialog.java
@@ -17,14 +17,10 @@
package org.apache.hop.pipeline.transforms.streamschemamerge;
-import java.util.List;
+import org.apache.hop.core.Const;
import org.apache.hop.core.variables.IVariables;
import org.apache.hop.i18n.BaseMessages;
import org.apache.hop.pipeline.PipelineMeta;
-import org.apache.hop.pipeline.transform.TransformMeta;
-import org.apache.hop.pipeline.transform.stream.IStream;
-import org.apache.hop.pipeline.transform.stream.Stream;
-import org.apache.hop.pipeline.transform.stream.StreamIcon;
import org.apache.hop.ui.core.PropsUi;
import org.apache.hop.ui.core.dialog.BaseDialog;
import org.apache.hop.ui.core.widget.ColumnInfo;
@@ -72,7 +68,11 @@ public class StreamSchemaDialog extends BaseTransformDialog {
public String open() {
createShell(BaseMessages.getString(PKG, "StreamSchemaTransform.Shell.Title"));
- buildButtonBar().ok(e -> ok()).get(e -> get()).cancel(e -> cancel()).build();
+ buildButtonBar()
+ .ok(e -> ok())
+ .get(e -> getSourceTransformNames())
+ .cancel(e -> cancel())
+ .build();
// Save the value of the changed flag on the meta object. If the user cancels
// the dialog, it will be restored to this saved value.
@@ -93,26 +93,24 @@ public class StreamSchemaDialog extends BaseTransformDialog {
fdlTransforms.top = new FormAttachment(wSpacer, margin);
wlTransforms.setLayoutData(fdlTransforms);
- final int FieldsCols = 1;
- final int FieldsRows = meta.getNumberOfTransforms();
-
previousTransforms = pipelineMeta.getPrevTransformNames(transformName);
- ColumnInfo[] colinf = new ColumnInfo[FieldsCols];
- colinf[0] =
- new ColumnInfo(
- BaseMessages.getString(PKG, "StreamSchemaTransformDialog.TransformName.Column"),
- ColumnInfo.COLUMN_TYPE_CCOMBO,
- previousTransforms,
- false);
+ ColumnInfo[] columnInfos =
+ new ColumnInfo[] {
+ new ColumnInfo(
+ BaseMessages.getString(PKG, "StreamSchemaTransformDialog.TransformName.Column"),
+ ColumnInfo.COLUMN_TYPE_CCOMBO,
+ previousTransforms,
+ false)
+ };
wTransforms =
new TableView(
variables,
shell,
SWT.BORDER | SWT.FULL_SELECTION | SWT.MULTI,
- colinf,
- FieldsRows,
+ columnInfos,
+ 1,
lsMod,
props);
@@ -144,37 +142,23 @@ public class StreamSchemaDialog extends BaseTransformDialog {
* the dialog controls.
*/
private void populateDialog() {
- Table table = wTransforms.table;
- if (meta.getNumberOfTransforms() > 0) {
- table.removeAll();
- }
- String[] transformNames = meta.getTransformsToMerge();
- for (int i = 0; i < transformNames.length; i++) {
- TableItem ti = new TableItem(table, SWT.NONE);
- ti.setText(0, "" + (i + 1));
- if (transformNames[i] != null) {
- ti.setText(1, transformNames[i]);
- }
+ for (StreamSchemaMeta.TransformToMerge transformToMerge : meta.getTransformsToMerge()) {
+ TableItem ti = new TableItem(wTransforms.table, SWT.NONE);
+ ti.setText(1, Const.NVL(transformToMerge.getName(), ""));
}
-
- wTransforms.removeEmptyRows();
- wTransforms.setRowNums();
- wTransforms.optWidth(true);
+ wTransforms.optimizeTableView();
}
/** Populates the table with a list of fields that have incoming hops */
- private void get() {
+ private void getSourceTransformNames() {
wTransforms.removeAll();
Table table = wTransforms.table;
- for (int i = 0; i < previousTransforms.length; i++) {
+ for (String previousTransform : previousTransforms) {
TableItem ti = new TableItem(table, SWT.NONE);
- ti.setText(0, "" + (i + 1));
- ti.setText(1, previousTransforms[i]);
+ ti.setText(1, previousTransform);
}
- wTransforms.removeEmptyRows();
- wTransforms.setRowNums();
- wTransforms.optWidth(true);
+ wTransforms.optimizeTableView();
}
/** Called when the user cancels the dialog. */
@@ -188,42 +172,6 @@ public class StreamSchemaDialog extends BaseTransformDialog {
dispose();
}
- /**
- * Helping method to update meta information when ok is selected
- *
- * @param inputTransforms Names of the transforms that are being merged together
- */
- private void getMeta(String[] inputTransforms) {
- List<IStream> infoStreams = meta.getTransformIOMeta().getInfoStreams();
-
- if (infoStreams.isEmpty() || inputTransforms.length < infoStreams.size()) {
- if (inputTransforms.length != 0) {
- for (String inputTransform : inputTransforms) {
- meta.getTransformIOMeta()
- .addStream(new Stream(IStream.StreamType.INFO, null, "", StreamIcon.INFO, null));
- }
- infoStreams = meta.getTransformIOMeta().getInfoStreams();
- }
- } else if (infoStreams.size() < inputTransforms.length) {
- int requiredStreams = inputTransforms.length - infoStreams.size();
-
- for (int i = 0; i < requiredStreams; i++) {
- meta.getTransformIOMeta()
- .addStream(new Stream(IStream.StreamType.INFO, null, "", StreamIcon.INFO, null));
- }
- infoStreams = meta.getTransformIOMeta().getInfoStreams();
- }
- int streamCount = infoStreams.size();
-
- String[] transformsToMerge = meta.getTransformsToMerge();
- for (int i = 0; i < streamCount; i++) {
- String transform = transformsToMerge[i];
- IStream infoStream = infoStreams.get(i);
- infoStream.setTransformMeta(pipelineMeta.findTransform(transform));
- infoStream.setSubject(transform);
- }
- }
-
/** Called when the user confirms the dialog */
private void ok() {
// The "TransformName" variable will be the return value for the open() method.
@@ -231,17 +179,13 @@ public class StreamSchemaDialog extends BaseTransformDialog {
transformName = wTransformName.getText();
// set output field name
- int nrtransforms = wTransforms.nrNonEmpty();
- String[] transformNames = new String[nrtransforms];
- for (int i = 0; i < nrtransforms; i++) {
- TableItem ti = wTransforms.getNonEmpty(i);
- TransformMeta tm = pipelineMeta.findTransform(ti.getText(1));
- if (tm != null) {
- transformNames[i] = tm.getName();
- }
+ meta.getTransformsToMerge().clear();
+ for (TableItem item : wTransforms.getNonEmptyItems()) {
+ StreamSchemaMeta.TransformToMerge transformToMerge = new StreamSchemaMeta.TransformToMerge();
+ transformToMerge.setName(item.getText(1));
+ meta.getTransformsToMerge().add(transformToMerge);
}
- meta.setTransformsToMerge(transformNames);
- getMeta(transformNames);
+ meta.resetTransformIoMeta();
// close the SWT dialog window
dispose();
diff --git a/plugins/transforms/streamschemamerge/src/main/java/org/apache/hop/pipeline/transforms/streamschemamerge/StreamSchemaMeta.java b/plugins/transforms/streamschemamerge/src/main/java/org/apache/hop/pipeline/transforms/streamschemamerge/StreamSchemaMeta.java
index 9931c11842..63d8252ec8 100644
--- a/plugins/transforms/streamschemamerge/src/main/java/org/apache/hop/pipeline/transforms/streamschemamerge/StreamSchemaMeta.java
+++ b/plugins/transforms/streamschemamerge/src/main/java/org/apache/hop/pipeline/transforms/streamschemamerge/StreamSchemaMeta.java
@@ -18,28 +18,27 @@
package org.apache.hop.pipeline.transforms.streamschemamerge;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.List;
+import lombok.Getter;
+import lombok.Setter;
import org.apache.hop.core.CheckResult;
-import org.apache.hop.core.Const;
import org.apache.hop.core.ICheckResult;
import org.apache.hop.core.annotations.Transform;
import org.apache.hop.core.exception.HopPluginException;
import org.apache.hop.core.exception.HopTransformException;
-import org.apache.hop.core.exception.HopValueException;
-import org.apache.hop.core.exception.HopXmlException;
import org.apache.hop.core.row.IRowMeta;
import org.apache.hop.core.variables.IVariables;
-import org.apache.hop.core.xml.XmlHandler;
import org.apache.hop.i18n.BaseMessages;
+import org.apache.hop.metadata.api.HopMetadataProperty;
import org.apache.hop.metadata.api.IHopMetadataProvider;
import org.apache.hop.pipeline.PipelineMeta;
import org.apache.hop.pipeline.transform.BaseTransformMeta;
+import org.apache.hop.pipeline.transform.ITransformIOMeta;
+import org.apache.hop.pipeline.transform.TransformIOMeta;
import org.apache.hop.pipeline.transform.TransformMeta;
import org.apache.hop.pipeline.transform.stream.IStream;
import org.apache.hop.pipeline.transform.stream.Stream;
import org.apache.hop.pipeline.transform.stream.StreamIcon;
-import org.w3c.dom.Node;
@Transform(
id = "StreamSchema",
@@ -49,12 +48,27 @@ import org.w3c.dom.Node;
categoryDescription = "i18n:org.apache.hop.pipeline.transform:BaseTransform.Category.Flow",
keywords = "i18n::StreamSchemaMeta.keyword",
documentationUrl = "/pipeline/transforms/streamschemamerge.html")
+@Getter
+@Setter
public class StreamSchemaMeta extends BaseTransformMeta<StreamSchema, StreamSchemaData> {
-
private static final Class<?> PKG = StreamSchemaMeta.class;
+ @Getter
+ @Setter
+ public static class TransformToMerge {
+ @HopMetadataProperty(key = "name")
+ private String name;
+
+ public TransformToMerge() {}
+
+ public TransformToMerge(TransformToMerge m) {
+ this.name = m.name;
+ }
+ }
+
/** Stores the names of the transforms to merge into the output */
- private ArrayList<String> transformsToMerge = new ArrayList<>();
+ @HopMetadataProperty(key = "transform", groupKey = "transforms")
+ private List<TransformToMerge> transformsToMerge;
/**
* Constructor should call super() to make sure the base class has a chance to initialize
@@ -62,63 +76,12 @@ public class StreamSchemaMeta extends BaseTransformMeta<StreamSchema, StreamSche
*/
public StreamSchemaMeta() {
super();
+ transformsToMerge = new ArrayList<>();
}
- /**
- * Prevents error box from popping up when sending in different row formats. Note you will still
- * get an error if you try to run the pipeline in safe mode.
- *
- * @return true
- */
- @Override
- public boolean excludeFromRowLayoutVerification() {
- return true;
- }
-
- /**
- * This method is called every time a new transform is created and should allocate/set the
- * transform configuration to sensible defaults. The values set here will be used by Spoon when a
- * new transform is created.
- */
- @Override
- public void setDefault() {
- // intentionally empty
- }
-
- /**
- * Getter for the fields that should be merged
- *
- * @return array of field names
- */
- public String[] getTransformsToMerge() {
- if (transformsToMerge == null) {
- return new String[0];
- } else {
- return transformsToMerge.toArray(new String[transformsToMerge.size()]);
- }
- }
-
- /**
- * Determine the number of transforms we're planning to merge
- *
- * @return number of items to merge, 0 if none
- */
- public int getNumberOfTransforms() {
- if (transformsToMerge == null) {
- return 0;
- } else {
- return transformsToMerge.size();
- }
- }
-
- /**
- * Set transforms to merge
- *
- * @param arrayOfTransforms Names of transforms to merge
- */
- public void setTransformsToMerge(String[] arrayOfTransforms) {
- transformsToMerge = new ArrayList<>();
- Collections.addAll(transformsToMerge, arrayOfTransforms);
+ public StreamSchemaMeta(StreamSchemaMeta m) {
+ this();
+ m.transformsToMerge.forEach(t -> this.transformsToMerge.add(new TransformToMerge(t)));
}
/**
@@ -131,80 +94,19 @@ public class StreamSchemaMeta extends BaseTransformMeta<StreamSchema, StreamSche
* @return a deep copy of this
*/
@Override
- public Object clone() {
- Object retval = super.clone();
- return retval;
- }
-
- /**
- * This method is called by Hop Gui when a transform needs to serialize its configuration to XML.
- * The expected return value is an XML fragment consisting of one or more XML tags.
- *
- * <p>Please use XmlHandler to conveniently generate the XML.
- *
- * @return a string containing the XML serialization of this transform
- */
- @Override
- public String getXml() throws HopValueException {
- StringBuilder xml = new StringBuilder();
- xml.append(" <transforms>" + Const.CR);
- for (String transformName : transformsToMerge) {
- xml.append(" <transform>" + Const.CR);
- xml.append(" " + XmlHandler.addTagValue("name", transformName));
- xml.append(" </transform>" + Const.CR);
- }
- xml.append(" </transforms>" + Const.CR);
- return xml.toString();
+ public StreamSchemaMeta clone() {
+ return new StreamSchemaMeta(this);
}
/**
- * This method is called by Hop when a transform needs to load its configuration from XML.
- *
- * <p>Please use XmlHandler to conveniently read from the XML node passed in.
+ * Prevents error box from popping up when sending in different row formats. Note you will still
+ * get an error if you try to run the pipeline in safe mode.
*
- * @param transformNode the XML node containing the configuration
- * @param metadataProvider the metadataProvider to optionally read from
+ * @return true
*/
@Override
- public void loadXml(Node transformNode, IHopMetadataProvider metadataProvider)
- throws HopXmlException {
-
- readData(transformNode);
- }
-
- /**
- * Helper methods to read in the XML
- *
- * @param transformNode XML node for the transform
- * @throws HopXmlException If there is an error reading the configuration
- */
- private void readData(Node transformNode) throws HopXmlException {
- try {
- Node transforms = XmlHandler.getSubNode(transformNode, "transforms");
- int nrtransforms = XmlHandler.countNodes(transforms, "transform");
-
- transformsToMerge.clear();
-
- // we need to add a stream for each transform we want to merge to ensure it gets treated as an
- // info
- // stream
- for (int i = 0; i < nrtransforms; i++) {
- getTransformIOMeta()
- .addStream(
- new Stream(
- IStream.StreamType.INFO, null, "Streams to Merge", StreamIcon.INFO, null));
- }
-
- List<IStream> infoStreams = getTransformIOMeta().getInfoStreams();
- for (int i = 0; i < nrtransforms; i++) {
- Node fnode = XmlHandler.getSubNodeByNr(transforms, "transform", i);
- String name = XmlHandler.getTagValue(fnode, "name");
- transformsToMerge.add(name);
- infoStreams.get(i).setSubject(name);
- }
- } catch (Exception e) {
- throw new HopXmlException("Unable to load transform info from XML", e);
- }
+ public boolean excludeFromRowLayoutVerification() {
+ return true;
}
/**
@@ -231,7 +133,7 @@ public class StreamSchemaMeta extends BaseTransformMeta<StreamSchema, StreamSche
throws HopTransformException {
/*
- * We don't have any input fields so we ingore inputRowMeta
+ * We don't have any input fields so we ignore inputRowMeta
*/
try {
SchemaMapper schemaMapping =
@@ -304,13 +206,28 @@ public class StreamSchemaMeta extends BaseTransformMeta<StreamSchema, StreamSche
@Override
public void resetTransformIoMeta() {
- // Do nothing, don't reset as there is no need to do this.
+ setTransformIOMeta(null);
}
- /** Has original function of resetTransformIoMeta, but we only want to call it when appropriate */
- /*
- public void wipeTransformIoMeta() {
- ioMeta = null;
- }
- */
+ @Override
+ public ITransformIOMeta getTransformIOMeta() {
+ ITransformIOMeta ioMeta = super.getTransformIOMeta(false);
+ if (ioMeta == null) {
+ ioMeta = new TransformIOMeta(true, true, false, false, false, false);
+
+ for (TransformToMerge transformToMerge : transformsToMerge) {
+ ioMeta.addStream(
+ new Stream(
+ IStream.StreamType.INFO,
+ null,
+ BaseMessages.getString(
+ PKG, "StreamSchemaMeta.InfoStream.Description", transformToMerge.name),
+ StreamIcon.INFO,
+ transformToMerge.name));
+ }
+ setTransformIOMeta(ioMeta);
+ }
+
+ return ioMeta;
+ }
}
diff --git a/plugins/transforms/streamschemamerge/src/main/resources/org/apache/hop/pipeline/transforms/streamschemamerge/messages/messages_en_US.properties b/plugins/transforms/streamschemamerge/src/main/resources/org/apache/hop/pipeline/transforms/streamschemamerge/messages/messages_en_US.properties
index ee5d2ce7f8..9e3c84e8cf 100644
--- a/plugins/transforms/streamschemamerge/src/main/resources/org/apache/hop/pipeline/transforms/streamschemamerge/messages/messages_en_US.properties
+++ b/plugins/transforms/streamschemamerge/src/main/resources/org/apache/hop/pipeline/transforms/streamschemamerge/messages/messages_en_US.properties
@@ -26,3 +26,4 @@ StreamSchemaTransform.Shell.Title=Stream Schema Merge
StreamSchemaTransform.TooltipDesc=Merge multiple streams with different schemas into one unified schema
StreamSchemaTransformDialog.TransformName.Column=Transform
StreamSchemaTransformDialog.Transforms.Label=Transforms to Merge
+StreamSchemaMeta.InfoStream.Description = This transform merges rows from source\: {0}
\ No newline at end of file
diff --git a/plugins/transforms/streamschemamerge/src/test/java/org/apache/hop/pipeline/transforms/streamschemamerge/StreamSchemaMetaTest.java b/plugins/transforms/streamschemamerge/src/test/java/org/apache/hop/pipeline/transforms/streamschemamerge/StreamSchemaMetaTest.java
new file mode 100644
index 0000000000..f77665ebfe
--- /dev/null
+++ b/plugins/transforms/streamschemamerge/src/test/java/org/apache/hop/pipeline/transforms/streamschemamerge/StreamSchemaMetaTest.java
@@ -0,0 +1,36 @@
+/*
+ * 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.streamschemamerge;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import org.apache.hop.pipeline.transform.TransformSerializationTestUtil;
+import org.junit.jupiter.api.Test;
+
+class StreamSchemaMetaTest {
+ @Test
+ void testSerializationRoundTrip() throws Exception {
+ StreamSchemaMeta meta =
+ TransformSerializationTestUtil.testSerialization(
+ "/stream-schema-merge.xml", StreamSchemaMeta.class);
+ assertEquals(3, meta.getTransformsToMerge().size());
+ assertEquals("transform1", meta.getTransformsToMerge().get(0).getName());
+ assertEquals("transform2", meta.getTransformsToMerge().get(1).getName());
+ assertEquals("transform3", meta.getTransformsToMerge().get(2).getName());
+ }
+}
diff --git a/plugins/transforms/streamschemamerge/src/test/resources/stream-schema-merge.xml b/plugins/transforms/streamschemamerge/src/test/resources/stream-schema-merge.xml
new file mode 100644
index 0000000000..3987f781c3
--- /dev/null
+++ b/plugins/transforms/streamschemamerge/src/test/resources/stream-schema-merge.xml
@@ -0,0 +1,32 @@
+<?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.
+ -->
+<transform>
+ <transforms>
+ <transform>
+ <name>transform1</name>
+ </transform>
+ <transform>
+ <name>transform2</name>
+ </transform>
+ <transform>
+ <name>transform3</name>
+ </transform>
+ </transforms>
+</transform>
+
+