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 7023f0ee7e issue #7393 : Merge Rows option to automatically line up
input fields (#7540)
7023f0ee7e is described below
commit 7023f0ee7e1b0db37dc1f7fae32066fdf5466bae
Author: Matt Casters <[email protected]>
AuthorDate: Fri Jul 17 13:48:59 2026 +0200
issue #7393 : Merge Rows option to automatically line up input fields
(#7540)
* issue #7393 : Merge Rows option to automatically line up input fields
* issue #7393 : restore integration tests, docs, and TEST_FILTER support
Bring over the pieces missed when re-creating the PR after the main merge:
align/align-off integration tests and golden data, mergerows docs, and
TEST_FILTER support for the IT runner scripts (applied on top of current
main without clobbering Spark client / CLIENT_UNZIP changes).
* issue #7393 : Merge Rows option to automatically line up input fields
* issue #7393 : honor TEST_FILTER under default IT runner
Force the classic per-workflow path when TEST_FILTER is set so filtered
Docker integration tests actually run only matching main*.hwf files, and
document the option for run-tests-docker.sh.
---
.../integration-tests/integration-tests-base.yaml | 2 +
.../integration-tests-beam-base.yaml | 2 +
.../modules/ROOT/pages/integration-testing.adoc | 6 +
.../ROOT/pages/pipeline/transforms/mergerows.adoc | 6 +
integration-tests/scripts/run-tests-docker.sh | 28 +-
integration-tests/scripts/run-tests.sh | 64 ++++-
.../transforms/0077-merge-rows-align-off.hpl | 291 +++++++++++++++++++++
.../transforms/0077-merge-rows-align.hpl | 218 +++++++++++++++
.../datasets/golden-merge-rows-align.csv | 5 +
.../transforms/main-0077-merge-rows.hwf | 6 +
.../metadata/dataset/golden-merge-rows-align.json | 48 ++++
.../unit-test/0077-merge-rows-align UNIT.json | 44 ++++
.../unit-test/0077-merge-rows-align-off UNIT.json | 40 +++
.../pipeline/transforms/mergerows/MergeRows.java | 159 +++++------
.../transforms/mergerows/MergeRowsAlignment.java | 139 ++++++++++
.../transforms/mergerows/MergeRowsData.java | 7 +-
.../transforms/mergerows/MergeRowsDialog.java | 23 ++
.../transforms/mergerows/MergeRowsMeta.java | 101 ++++---
.../mergerows/messages/messages_en_US.properties | 25 +-
.../mergerows/MergeRowsAlignmentTest.java | 137 ++++++++++
.../transforms/mergerows/MergeRowsLogicTest.java | 135 ++++++++++
.../mergerows/MergeRowsMetaCheckTest.java | 33 ++-
.../transforms/mergerows/MergeRowsMetaTest.java | 27 ++
23 files changed, 1410 insertions(+), 136 deletions(-)
diff --git a/docker/integration-tests/integration-tests-base.yaml
b/docker/integration-tests/integration-tests-base.yaml
index 51d1a1028a..2f1c9277c1 100644
--- a/docker/integration-tests/integration-tests-base.yaml
+++ b/docker/integration-tests/integration-tests-base.yaml
@@ -31,4 +31,6 @@ services:
- ../../integration-tests/:/files
environment:
- FLASK_ENV=docker
+ # Optional workflow basename filter (substring or glob). Set by
run-tests-docker.sh.
+ - TEST_FILTER=${TEST_FILTER:-}
command: [ "bash", "-c", "/files/scripts/run-tests.sh ${PROJECT_NAME}" ]
diff --git a/docker/integration-tests/integration-tests-beam-base.yaml
b/docker/integration-tests/integration-tests-beam-base.yaml
index eb2537f4a0..8521fd7ee8 100644
--- a/docker/integration-tests/integration-tests-beam-base.yaml
+++ b/docker/integration-tests/integration-tests-beam-base.yaml
@@ -33,4 +33,6 @@ services:
- FLASK_ENV=docker
# Keep driver Spark client pack in sync with the fat jar (empty =
default pack)
- HOP_SPARK_CLIENT_VERSION=${HOP_SPARK_CLIENT_VERSION:-}
+ # Optional workflow basename filter (substring or glob). Set by
run-tests-docker.sh.
+ - TEST_FILTER=${TEST_FILTER:-}
command: [ "bash", "-c", "/files/scripts/run-tests.sh ${PROJECT_NAME}" ]
diff --git a/docs/hop-dev-manual/modules/ROOT/pages/integration-testing.adoc
b/docs/hop-dev-manual/modules/ROOT/pages/integration-testing.adoc
index 96546e9cd1..16c9553ed9 100644
--- a/docs/hop-dev-manual/modules/ROOT/pages/integration-testing.adoc
+++ b/docs/hop-dev-manual/modules/ROOT/pages/integration-testing.adoc
@@ -96,6 +96,12 @@ cd integration-tests/scripts
# Optionally you can run a single project by adding the PROJECT_NAME variable
to the script
./run-tests-docker.sh PROJECT_NAME=database
+
+# Optionally filter which main*.hwf workflows run (basename substring or shell
glob;
+# comma-separated lists are supported). Quote globs so the shell does not
expand them.
+
+./run-tests-docker.sh PROJECT_NAME=transforms TEST_FILTER=0077-merge-rows
+./run-tests-docker.sh PROJECT_NAME=transforms TEST_FILTER='*0077*'
----
+
. Surefire Reports
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/mergerows.adoc
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/mergerows.adoc
index a6e451b24f..7dd38f3390 100644
--- a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/mergerows.adoc
+++ b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/mergerows.adoc
@@ -78,11 +78,17 @@ In the subsequent transform, you can use the flag field
generated by **Merge row
|Flag field name|Specify the name of the flag field on the output stream.
|Difference field name|Specify the name of the field to contain the difference
in case the status is **changed**.
The difference will be in the form of a "changes" array in JSON format.
+|Align input layouts|When enabled (the default), the transform builds a
**reference-authoritative** union of the reference and compare field layouts
before comparing.
+Missing fields on either side are filled with null.
+Shared fields keep the reference data type; compatible compare values are
converted to that type.
+Compare-only fields are appended to the output layout.
+Disable this option to require identical layouts (legacy strict mode).
|Keys to match|Specify fields containing the keys on which to match. Click
"Get key fields" to insert all the fields from the reference rows
|Values to compare|Specify fields containing the values to compare. Click "Get
value fields" to insert all the fields from the compare rows.
Key fields do not need to be repeated here.
|===
+
== Passing through fields
It's possible to pass through fields from the reference or compare data
streams.
diff --git a/integration-tests/scripts/run-tests-docker.sh
b/integration-tests/scripts/run-tests-docker.sh
index ffac427059..2d095aad19 100755
--- a/integration-tests/scripts/run-tests-docker.sh
+++ b/integration-tests/scripts/run-tests-docker.sh
@@ -24,11 +24,14 @@
EXECUTED_COMPOSE_FILES=("${DOCKER_FILES_DIR}/integration-tests-base.yaml")
for ARGUMENT in "$@"; do
- KEY=$(echo $ARGUMENT | cut -f1 -d=)
- VALUE=$(echo $ARGUMENT | cut -f2 -d=)
+ # Quote so glob characters in values (e.g. TEST_FILTER='*0077*') are
preserved.
+ # cut -f2- keeps values that themselves contain '='.
+ KEY=$(echo "${ARGUMENT}" | cut -f1 -d=)
+ VALUE=$(echo "${ARGUMENT}" | cut -f2- -d=)
case "$KEY" in
PROJECT_NAME) PROJECT_NAME=${VALUE} ;;
+ TEST_FILTER) TEST_FILTER=${VALUE} ;;
JENKINS_USER) JENKINS_USER=${VALUE} ;;
JENKINS_UID) JENKINS_UID=${VALUE} ;;
JENKINS_GROUP) JENKINS_GROUP=${VALUE} ;;
@@ -65,6 +68,17 @@ if [ -z "${PROJECT_NAME}" ]; then
PROJECT_NAME="*"
fi
+# Optional filter for main*.hwf basenames (substring or glob, comma-separated).
+# Passed into the test container as TEST_FILTER; when set, run-tests.sh uses
the
+# classic per-workflow runner so only matching main*.hwf files execute.
Examples:
+# ./run-tests-docker.sh PROJECT_NAME=transforms TEST_FILTER=0077-merge-rows
+# ./run-tests-docker.sh PROJECT_NAME=transforms TEST_FILTER='*0077*'
+# ./run-tests-docker.sh PROJECT_NAME=transforms
TEST_FILTER='0077-merge-rows,0076-other'
+if [ -z "${TEST_FILTER}" ]; then
+ TEST_FILTER=""
+fi
+export TEST_FILTER
+
if [ -z "${JENKINS_USER}" ]; then
JENKINS_USER="jenkins"
fi
@@ -204,20 +218,24 @@ for d in "${CURRENT_DIR}"/../${PROJECT_NAME}/; do
# Check if specific compose exists
+ if [ -n "${TEST_FILTER}" ]; then
+ echo "TEST_FILTER: ${TEST_FILTER}"
+ fi
+
if [ -f "${DOCKER_FILES_DIR}/integration-tests-${PROJECT_NAME}.yaml" ];
then
echo "Project compose exists."
EXECUTED_COMPOSE_FILES=("${EXECUTED_COMPOSE_FILES[@]}"
"${DOCKER_FILES_DIR}/integration-tests-${PROJECT_NAME}.yaml")
# Rebuild project images so SPARK_VERSION (and similar) build args
take effect
if [ "${PROJECT_NAME}" = "spark" ]; then
echo "Spark IT cluster version: ${SPARK_VERSION} (hadoop
${HADOOP_VERSION})"
- PROJECT_NAME=${PROJECT_NAME} SPARK_VERSION=${SPARK_VERSION}
HADOOP_VERSION=${HADOOP_VERSION} SPARK_BASE_URL=${SPARK_BASE_URL} \
+ PROJECT_NAME=${PROJECT_NAME} TEST_FILTER=${TEST_FILTER}
SPARK_VERSION=${SPARK_VERSION} HADOOP_VERSION=${HADOOP_VERSION}
SPARK_BASE_URL=${SPARK_BASE_URL} \
docker compose -f
${DOCKER_FILES_DIR}/integration-tests-${PROJECT_NAME}.yaml up --build
--abort-on-container-exit
else
- PROJECT_NAME=${PROJECT_NAME} docker compose -f
${DOCKER_FILES_DIR}/integration-tests-${PROJECT_NAME}.yaml up
--abort-on-container-exit
+ PROJECT_NAME=${PROJECT_NAME} TEST_FILTER=${TEST_FILTER} docker
compose -f ${DOCKER_FILES_DIR}/integration-tests-${PROJECT_NAME}.yaml up
--abort-on-container-exit
fi
else
echo "Project compose does not exists."
- PROJECT_NAME=${PROJECT_NAME} docker compose -f
${DOCKER_FILES_DIR}/integration-tests-base.yaml up --abort-on-container-exit
+ PROJECT_NAME=${PROJECT_NAME} TEST_FILTER=${TEST_FILTER} docker compose
-f ${DOCKER_FILES_DIR}/integration-tests-base.yaml up --abort-on-container-exit
fi
fi
fi
diff --git a/integration-tests/scripts/run-tests.sh
b/integration-tests/scripts/run-tests.sh
index 3d5fa333f6..6db92c699c 100755
--- a/integration-tests/scripts/run-tests.sh
+++ b/integration-tests/scripts/run-tests.sh
@@ -121,9 +121,60 @@ if [ -z "${PROJECT_NAME}" ]; then
PROJECT_NAME="*"
fi
+# Optional filter for main*.hwf workflows (substring or glob against basename).
+# Comma-separated list is supported. Examples:
+# TEST_FILTER=0077-merge-rows
+# TEST_FILTER='*0077*','*0078*'
+# TEST_FILTER=main-0077-merge-rows.hwf
+if [ -z "${TEST_FILTER}" ]; then
+ TEST_FILTER=""
+fi
+
#set global variables
SPACER="==========================================="
+# Return 0 if the workflow file should run under the current TEST_FILTER.
+should_run_workflow() {
+ local file="$1"
+ local base
+ base=$(basename "$file")
+
+ if [ -z "${TEST_FILTER}" ]; then
+ return 0
+ fi
+
+ local old_ifs=$IFS
+ IFS=','
+ local pattern
+ # shellcheck disable=SC2086
+ for pattern in ${TEST_FILTER}; do
+ # trim whitespace
+ pattern="${pattern#"${pattern%%[![:space:]]*}"}"
+ pattern="${pattern%"${pattern##*[![:space:]]}"}"
+ [ -z "${pattern}" ] && continue
+
+ # No glob meta-characters: treat as basename substring
+ if [[ "${pattern}" != *[\*\?[]* ]]; then
+ case "${base}" in
+ *"${pattern}"*)
+ IFS=$old_ifs
+ return 0
+ ;;
+ esac
+ else
+ # Glob match against basename
+ case "${base}" in
+ ${pattern})
+ IFS=$old_ifs
+ return 0
+ ;;
+ esac
+ fi
+ done
+ IFS=$old_ifs
+ return 1
+}
+
# Set up a temporary folder
export TMP_FOLDER=/tmp/hop-it-$$
rm -rf "${TMP_FOLDER}"
@@ -198,8 +249,9 @@ for d in "${CURRENT_DIR}"/../${PROJECT_NAME}/; do
SUITE_RUN_CONFIG="hop-local"
fi
- # Prefer single-JVM suite runner when available (unless isolation mode
is requested)
- if [ "${HOP_IT_PER_TEST_JVM}" != "true" ] && [ -f "${RUNNER_PIPELINE}"
]; then
+ # Prefer single-JVM suite runner when available (unless isolation mode
is requested).
+ # TEST_FILTER is only applied on the classic per-workflow path, so force
that mode when set.
+ if [ "${HOP_IT_PER_TEST_JVM}" != "true" ] && [ -z "${TEST_FILTER}" ] &&
[ -f "${RUNNER_PIPELINE}" ]; then
echo ${SPACER}
echo "Running project tests in single JVM via run-project-tests.hpl
(run config: ${SUITE_RUN_CONFIG})"
@@ -250,8 +302,16 @@ for d in "${CURRENT_DIR}"/../${PROJECT_NAME}/; do
# Classic path: one hop-run JVM per main-*.hwf
#
+ if [ -n "${TEST_FILTER}" ]; then
+ echo "TEST_FILTER is set: ${TEST_FILTER}"
+ fi
+
find "$d" -name 'main*.hwf' | sort | while read -r f; do
+ if ! should_run_workflow "$f"; then
+ continue
+ fi
+
#cleanup temp files
rm -f /tmp/test_output
rm -f /tmp/test_output_err
diff --git a/integration-tests/transforms/0077-merge-rows-align-off.hpl
b/integration-tests/transforms/0077-merge-rows-align-off.hpl
new file mode 100644
index 0000000000..a3b5a269bd
--- /dev/null
+++ b/integration-tests/transforms/0077-merge-rows-align-off.hpl
@@ -0,0 +1,291 @@
+<?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>0077-merge-rows-align-off</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description>Legacy strict layout mode (align off) with matching
streams</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/07/10 12:00:00.000</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/07/10 12:00:00.000</modified_date>
+ </info>
+ <notepads>
+ </notepads>
+ <order>
+ <hop>
+ <from>source1</from>
+ <to>Merge rows (diff)</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>source2</from>
+ <to>Merge rows (diff)</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>Merge rows (diff)</from>
+ <to>validate</to>
+ <enabled>Y</enabled>
+ </hop>
+ </order>
+ <transform>
+ <name>Merge rows (diff)</name>
+ <type>MergeRows</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <align_input_layouts>N</align_input_layouts>
+ <keys>
+ <key>id</key>
+ </keys>
+ <values>
+ <value>name</value>
+ <value>score</value>
+ </values>
+ <flag_field>flagfield</flag_field>
+ <reference>source1</reference>
+ <compare>source2</compare>
+ <attributes/>
+ <GUI>
+ <xloc>240</xloc>
+ <yloc>96</yloc>
+ </GUI>
+ </transform>
+ <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>1</item>
+ <item>James</item>
+ <item>3.6</item>
+ </line>
+ <line>
+ <item>2</item>
+ <item>Marie</item>
+ <item>8.3</item>
+ </line>
+ <line>
+ <item>3</item>
+ <item>Michael</item>
+ <item>9.1</item>
+ </line>
+ <line>
+ <item>4</item>
+ <item>Patrcia</item>
+ <item>4.2</item>
+ </line>
+ <line>
+ <item>5</item>
+ <item>Robert</item>
+ <item>5.0</item>
+ </line>
+ <line>
+ <item>6</item>
+ <item>Linda</item>
+ <item>1.6</item>
+ </line>
+ <line>
+ <item>7</item>
+ <item>David</item>
+ <item>7.2</item>
+ </line>
+ <line>
+ <item>8</item>
+ <item>Elizabeth</item>
+ <item>6.9</item>
+ </line>
+ <line>
+ <item>9</item>
+ <item>Willian</item>
+ <item>4.2</item>
+ </line>
+ <line>
+ <item>10</item>
+ <item>Barbara</item>
+ <item>0.5</item>
+ </line>
+ </data>
+ <fields>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>id</name>
+ <type>Integer</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>name</name>
+ <type>String</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>score</name>
+ <format>#.#</format>
+ <type>Number</type>
+ </field>
+ </fields>
+ <attributes/>
+ <GUI>
+ <xloc>96</xloc>
+ <yloc>48</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>1</item>
+ <item>James</item>
+ <item>3.6</item>
+ </line>
+ <line>
+ <item>3</item>
+ <item>Michael</item>
+ <item>9.1</item>
+ </line>
+ <line>
+ <item>4</item>
+ <item>Patricia</item>
+ <item>4.2</item>
+ </line>
+ <line>
+ <item>5</item>
+ <item>Robert</item>
+ <item>5.0</item>
+ </line>
+ <line>
+ <item>7</item>
+ <item>David</item>
+ <item>7.2</item>
+ </line>
+ <line>
+ <item>8</item>
+ <item>Elizabeth</item>
+ <item>6.9</item>
+ </line>
+ <line>
+ <item>9</item>
+ <item>William</item>
+ <item>4.2</item>
+ </line>
+ <line>
+ <item>10</item>
+ <item>Barbara</item>
+ <item>0.5</item>
+ </line>
+ <line>
+ <item>11</item>
+ <item>Richard</item>
+ <item>9.6</item>
+ </line>
+ <line>
+ <item>12</item>
+ <item>Susan</item>
+ <item>5.4</item>
+ </line>
+ </data>
+ <fields>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>id</name>
+ <type>Integer</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>name</name>
+ <type>String</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>score</name>
+ <format>#.#</format>
+ <type>Number</type>
+ </field>
+ </fields>
+ <attributes/>
+ <GUI>
+ <xloc>96</xloc>
+ <yloc>144</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>416</xloc>
+ <yloc>96</yloc>
+ </GUI>
+ </transform>
+ <transform_error_handling>
+ </transform_error_handling>
+ <attributes/>
+</pipeline>
diff --git a/integration-tests/transforms/0077-merge-rows-align.hpl
b/integration-tests/transforms/0077-merge-rows-align.hpl
new file mode 100644
index 0000000000..bc89f6ac58
--- /dev/null
+++ b/integration-tests/transforms/0077-merge-rows-align.hpl
@@ -0,0 +1,218 @@
+<?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>0077-merge-rows-align</name>
+ <name_sync_with_filename>Y</name_sync_with_filename>
+ <description>Merge rows with mismatched layouts and automatic alignment
enabled</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/07/10 12:00:00.000</created_date>
+ <modified_user>-</modified_user>
+ <modified_date>2026/07/10 12:00:00.000</modified_date>
+ </info>
+ <notepads>
+ </notepads>
+ <order>
+ <hop>
+ <from>source1</from>
+ <to>Merge rows (diff)</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>source2</from>
+ <to>Merge rows (diff)</to>
+ <enabled>Y</enabled>
+ </hop>
+ <hop>
+ <from>Merge rows (diff)</from>
+ <to>validate</to>
+ <enabled>Y</enabled>
+ </hop>
+ </order>
+ <transform>
+ <name>Merge rows (diff)</name>
+ <type>MergeRows</type>
+ <description/>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <align_input_layouts>Y</align_input_layouts>
+ <keys>
+ <key>id</key>
+ </keys>
+ <values>
+ <value>status</value>
+ </values>
+ <flag_field>flagfield</flag_field>
+ <reference>source1</reference>
+ <compare>source2</compare>
+ <attributes/>
+ <GUI>
+ <xloc>240</xloc>
+ <yloc>96</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>source1</name>
+ <type>DataGrid</type>
+ <description>Reference stream: id, name, status</description>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <data>
+ <line>
+ <item>1</item>
+ <item>Alice</item>
+ <item>active</item>
+ </line>
+ <line>
+ <item>2</item>
+ <item>Bob</item>
+ <item>active</item>
+ </line>
+ <line>
+ <item>3</item>
+ <item>Carol</item>
+ <item>inactive</item>
+ </line>
+ </data>
+ <fields>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>id</name>
+ <type>Integer</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>name</name>
+ <type>String</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>status</name>
+ <type>String</type>
+ </field>
+ </fields>
+ <attributes/>
+ <GUI>
+ <xloc>96</xloc>
+ <yloc>48</yloc>
+ </GUI>
+ </transform>
+ <transform>
+ <name>source2</name>
+ <type>DataGrid</type>
+ <description>Compare stream: id, status, region (different fields and
order)</description>
+ <distribute>Y</distribute>
+ <custom_distribution/>
+ <copies>1</copies>
+ <partitioning>
+ <method>none</method>
+ <schema_name/>
+ </partitioning>
+ <data>
+ <line>
+ <item>1</item>
+ <item>active</item>
+ <item>US</item>
+ </line>
+ <line>
+ <item>2</item>
+ <item>retired</item>
+ <item>EU</item>
+ </line>
+ <line>
+ <item>4</item>
+ <item>pending</item>
+ <item>APAC</item>
+ </line>
+ </data>
+ <fields>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>id</name>
+ <type>Integer</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>status</name>
+ <type>String</type>
+ </field>
+ <field>
+ <length>-1</length>
+ <precision>-1</precision>
+ <set_empty_string>N</set_empty_string>
+ <name>region</name>
+ <type>String</type>
+ </field>
+ </fields>
+ <attributes/>
+ <GUI>
+ <xloc>96</xloc>
+ <yloc>144</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>416</xloc>
+ <yloc>96</yloc>
+ </GUI>
+ </transform>
+ <transform_error_handling>
+ </transform_error_handling>
+ <attributes/>
+</pipeline>
diff --git a/integration-tests/transforms/datasets/golden-merge-rows-align.csv
b/integration-tests/transforms/datasets/golden-merge-rows-align.csv
new file mode 100644
index 0000000000..ae776bc7d0
--- /dev/null
+++ b/integration-tests/transforms/datasets/golden-merge-rows-align.csv
@@ -0,0 +1,5 @@
+id,name,status,region,flagfield
+1,,active,US,identical
+2,,retired,EU,changed
+3,Carol,inactive,,deleted
+4,,pending,APAC,new
diff --git a/integration-tests/transforms/main-0077-merge-rows.hwf
b/integration-tests/transforms/main-0077-merge-rows.hwf
index 381b71ece9..f840fc8433 100644
--- a/integration-tests/transforms/main-0077-merge-rows.hwf
+++ b/integration-tests/transforms/main-0077-merge-rows.hwf
@@ -64,6 +64,12 @@ limitations under the License.
<test_name>
<name>0077-merge-rows-passthrough UNIT</name>
</test_name>
+ <test_name>
+ <name>0077-merge-rows-align UNIT</name>
+ </test_name>
+ <test_name>
+ <name>0077-merge-rows-align-off UNIT</name>
+ </test_name>
</test_names>
<parallel>N</parallel>
<xloc>272</xloc>
diff --git
a/integration-tests/transforms/metadata/dataset/golden-merge-rows-align.json
b/integration-tests/transforms/metadata/dataset/golden-merge-rows-align.json
new file mode 100644
index 0000000000..7daa2be65c
--- /dev/null
+++ b/integration-tests/transforms/metadata/dataset/golden-merge-rows-align.json
@@ -0,0 +1,48 @@
+{
+ "base_filename": "golden-merge-rows-align.csv",
+ "name": "golden-merge-rows-align",
+ "description": "Golden data for merge rows with layout alignment",
+ "dataset_fields": [
+ {
+ "field_comment": "",
+ "field_length": -1,
+ "field_type": 5,
+ "field_precision": 0,
+ "field_name": "id",
+ "field_format": "####0;-####0"
+ },
+ {
+ "field_comment": "",
+ "field_length": -1,
+ "field_type": 2,
+ "field_precision": -1,
+ "field_name": "name",
+ "field_format": ""
+ },
+ {
+ "field_comment": "",
+ "field_length": -1,
+ "field_type": 2,
+ "field_precision": -1,
+ "field_name": "status",
+ "field_format": ""
+ },
+ {
+ "field_comment": "",
+ "field_length": -1,
+ "field_type": 2,
+ "field_precision": -1,
+ "field_name": "region",
+ "field_format": ""
+ },
+ {
+ "field_comment": "",
+ "field_length": -1,
+ "field_type": 2,
+ "field_precision": -1,
+ "field_name": "flagfield",
+ "field_format": ""
+ }
+ ],
+ "folder_name": ""
+}
diff --git
a/integration-tests/transforms/metadata/unit-test/0077-merge-rows-align
UNIT.json
b/integration-tests/transforms/metadata/unit-test/0077-merge-rows-align
UNIT.json
new file mode 100644
index 0000000000..8d2a1a62cf
--- /dev/null
+++ b/integration-tests/transforms/metadata/unit-test/0077-merge-rows-align
UNIT.json
@@ -0,0 +1,44 @@
+{
+ "database_replacements": [],
+ "autoOpening": true,
+ "description": "Merge rows with automatic layout alignment on mismatched
streams",
+ "persist_filename": "",
+ "test_type": "UNIT_TEST",
+ "variableValues": [],
+ "basePath": "",
+ "golden_data_sets": [
+ {
+ "field_mappings": [
+ {
+ "transform_field": "id",
+ "data_set_field": "id"
+ },
+ {
+ "transform_field": "name",
+ "data_set_field": "name"
+ },
+ {
+ "transform_field": "status",
+ "data_set_field": "status"
+ },
+ {
+ "transform_field": "region",
+ "data_set_field": "region"
+ },
+ {
+ "transform_field": "flagfield",
+ "data_set_field": "flagfield"
+ }
+ ],
+ "field_order": [
+ "id"
+ ],
+ "data_set_name": "golden-merge-rows-align",
+ "transform_name": "validate"
+ }
+ ],
+ "input_data_sets": [],
+ "name": "0077-merge-rows-align UNIT",
+ "trans_test_tweaks": [],
+ "pipeline_filename": "./0077-merge-rows-align.hpl"
+}
diff --git
a/integration-tests/transforms/metadata/unit-test/0077-merge-rows-align-off
UNIT.json
b/integration-tests/transforms/metadata/unit-test/0077-merge-rows-align-off
UNIT.json
new file mode 100644
index 0000000000..381e807e70
--- /dev/null
+++ b/integration-tests/transforms/metadata/unit-test/0077-merge-rows-align-off
UNIT.json
@@ -0,0 +1,40 @@
+{
+ "database_replacements": [],
+ "autoOpening": true,
+ "description": "Merge rows with layout alignment disabled (legacy strict
mode)",
+ "persist_filename": "",
+ "test_type": "UNIT_TEST",
+ "variableValues": [],
+ "basePath": "",
+ "golden_data_sets": [
+ {
+ "field_mappings": [
+ {
+ "transform_field": "id",
+ "data_set_field": "id"
+ },
+ {
+ "transform_field": "name",
+ "data_set_field": "name"
+ },
+ {
+ "transform_field": "score",
+ "data_set_field": "score"
+ },
+ {
+ "transform_field": "flagfield",
+ "data_set_field": "flagfield"
+ }
+ ],
+ "field_order": [
+ "id"
+ ],
+ "data_set_name": "golden-merge-rows",
+ "transform_name": "validate"
+ }
+ ],
+ "input_data_sets": [],
+ "name": "0077-merge-rows-align-off UNIT",
+ "trans_test_tweaks": [],
+ "pipeline_filename": "./0077-merge-rows-align-off.hpl"
+}
diff --git
a/plugins/transforms/mergerows/src/main/java/org/apache/hop/pipeline/transforms/mergerows/MergeRows.java
b/plugins/transforms/mergerows/src/main/java/org/apache/hop/pipeline/transforms/mergerows/MergeRows.java
index 8f0694b2f6..c1dab031a5 100644
---
a/plugins/transforms/mergerows/src/main/java/org/apache/hop/pipeline/transforms/mergerows/MergeRows.java
+++
b/plugins/transforms/mergerows/src/main/java/org/apache/hop/pipeline/transforms/mergerows/MergeRows.java
@@ -41,15 +41,16 @@ import org.json.simple.JSONObject;
/**
* Merge rows from 2 sorted streams to detect changes. Use this as feed for a
dimension in case you
- * have no time stamps in your source system.
+ * have no time stamps in your source system. Optionally aligns mismatched
input layouts before
+ * comparing.
*/
public class MergeRows extends BaseTransform<MergeRowsMeta, MergeRowsData> {
private static final Class<?> PKG = MergeRowsMeta.class;
- private static final String VALUE_IDENTICAL = "identical";
- private static final String VALUE_CHANGED = "changed";
- private static final String VALUE_NEW = "new";
- private static final String VALUE_DELETED = "deleted";
+ static final String VALUE_IDENTICAL = "identical";
+ static final String VALUE_CHANGED = "changed";
+ static final String VALUE_NEW = "new";
+ static final String VALUE_DELETED = "deleted";
public MergeRows(
TransformMeta transformMeta,
@@ -66,35 +67,53 @@ public class MergeRows extends BaseTransform<MergeRowsMeta,
MergeRowsData> {
if (first) {
first = false;
- // Find the appropriate RowSet
- //
List<IStream> infoStreams = meta.getTransformIOMeta().getInfoStreams();
- // oneRowSet is the "Reference" stream
data.oneRowSet = findInputRowSet(infoStreams.get(0).getTransformName());
- // twoRowSet is the "Comparison" stream
data.twoRowSet = findInputRowSet(infoStreams.get(1).getTransformName());
- data.one = getRowFrom(data.oneRowSet);
- data.two = getRowFrom(data.twoRowSet);
+ data.referenceRowMeta = data.oneRowSet.getRowMeta();
+ if (data.referenceRowMeta == null) {
+ data.referenceRowMeta =
+ getPipelineMeta().getTransformFields(this,
meta.getReferenceTransform());
+ }
+ data.compareRowMeta = data.twoRowSet.getRowMeta();
+ if (data.compareRowMeta == null) {
+ data.compareRowMeta =
+ getPipelineMeta().getTransformFields(this,
meta.getCompareTransform());
+ }
- try {
- checkInputLayoutValid(data.oneRowSet.getRowMeta(),
data.twoRowSet.getRowMeta());
- } catch (HopRowException e) {
- throw new HopException(
- BaseMessages.getString(PKG,
"MergeRows.Exception.InvalidLayoutDetected"), e);
+ if (meta.isAlignInputLayouts()) {
+ data.schemaMapping =
+ MergeRowsAlignment.buildSchemaMapping(data.referenceRowMeta,
data.compareRowMeta);
+ data.oneRowMeta = data.schemaMapping.getOutputRowMeta();
+ data.twoRowMeta = data.oneRowMeta;
+ } else {
+ try {
+ checkInputLayoutValid(data.referenceRowMeta, data.compareRowMeta);
+ } catch (HopRowException e) {
+ throw new HopException(
+ BaseMessages.getString(PKG,
"MergeRows.Exception.InvalidLayoutDetected"), e);
+ }
+ data.oneRowMeta = data.referenceRowMeta;
+ data.twoRowMeta = data.compareRowMeta;
}
- if (data.one != null) {
- // Find the key indexes:
+ data.one = getAlignedRow(0, data.oneRowSet);
+ data.two = getAlignedRow(1, data.twoRowSet);
+
+ IRowMeta keyLookupMeta = data.oneRowMeta;
+ if (data.one != null || data.two != null) {
data.keyNrs = new int[meta.getKeyFields().size()];
for (int i = 0; i < data.keyNrs.length; i++) {
- data.keyNrs[i] =
data.oneRowSet.getRowMeta().indexOfValue(meta.getKeyFields().get(i));
+ data.keyNrs[i] =
keyLookupMeta.indexOfValue(meta.getKeyFields().get(i));
if (data.keyNrs[i] < 0) {
String message =
BaseMessages.getString(
PKG,
- "MergeRows.Exception.UnableToFindFieldInReferenceStream",
+ meta.isAlignInputLayouts()
+ ?
"MergeRows.Exception.UnableToFindFieldInAlignedStream"
+ :
"MergeRows.Exception.UnableToFindFieldInReferenceStream",
meta.getKeyFields().get(i));
logError(message);
throw new HopTransformException(message);
@@ -102,15 +121,17 @@ public class MergeRows extends
BaseTransform<MergeRowsMeta, MergeRowsData> {
}
}
- if (data.two != null) {
+ if (data.one != null || data.two != null) {
data.valueNrs = new int[meta.getValueFields().size()];
for (int i = 0; i < data.valueNrs.length; i++) {
- data.valueNrs[i] =
data.twoRowSet.getRowMeta().indexOfValue(meta.getValueFields().get(i));
+ data.valueNrs[i] =
keyLookupMeta.indexOfValue(meta.getValueFields().get(i));
if (data.valueNrs[i] < 0) {
String message =
BaseMessages.getString(
PKG,
- "MergeRows.Exception.UnableToFindFieldInReferenceStream",
+ meta.isAlignInputLayouts()
+ ?
"MergeRows.Exception.UnableToFindFieldInAlignedStream"
+ :
"MergeRows.Exception.UnableToFindFieldInReferenceStream",
meta.getValueFields().get(i));
logError(message);
throw new HopTransformException(message);
@@ -118,21 +139,12 @@ public class MergeRows extends
BaseTransform<MergeRowsMeta, MergeRowsData> {
}
}
- // Calculate the indices for the passthrough fields.
- //
+ // Passthrough indexes use the original (unaligned) row metas.
data.passThroughIndexes = new ArrayList<>();
- data.oneRowMeta = data.oneRowSet.getRowMeta();
- if (data.oneRowMeta == null) {
- data.oneRowMeta = getPipelineMeta().getTransformFields(this,
meta.getReferenceTransform());
- }
- data.twoRowMeta = data.twoRowSet.getRowMeta();
- if (data.twoRowMeta == null) {
- data.twoRowMeta = getPipelineMeta().getTransformFields(this,
meta.getCompareTransform());
- }
for (PassThroughField field : meta.getPassThroughFields()) {
int index;
if (field.isReferenceField()) {
- index = data.oneRowMeta.indexOfValue(field.getSourceField());
+ index = data.referenceRowMeta.indexOfValue(field.getSourceField());
if (index < 0) {
throw new HopTransformException(
"Unable to find passthrough field '"
@@ -141,7 +153,7 @@ public class MergeRows extends BaseTransform<MergeRowsMeta,
MergeRowsData> {
+ meta.getReferenceTransform());
}
} else {
- index = data.twoRowMeta.indexOfValue(field.getSourceField());
+ index = data.compareRowMeta.indexOfValue(field.getSourceField());
if (index < 0) {
throw new HopTransformException(
"Unable to find passthrough field '"
@@ -170,7 +182,7 @@ public class MergeRows extends BaseTransform<MergeRowsMeta,
MergeRowsData> {
meta.getFields(
data.outputRowMeta,
getTransformName(),
- new IRowMeta[] {data.oneRowMeta, data.twoRowMeta},
+ new IRowMeta[] {data.referenceRowMeta, data.compareRowMeta},
null,
this,
metadataProvider);
@@ -181,34 +193,28 @@ public class MergeRows extends
BaseTransform<MergeRowsMeta, MergeRowsData> {
String differenceJson = "{\"changes\":[]}";
boolean getDifference = StringUtils.isNotEmpty(meta.getDiffJsonField());
- // Remember the rows used to compare: copy rows 'one' and 'two'.
copyOneTwo();
- if (data.one == null && data.two != null) { // Record 2 is flagged as new!
+ if (data.one == null && data.two != null) {
outputRow = data.two;
flagField = VALUE_NEW;
+ data.two = getAlignedRow(1, data.twoRowSet);
- // Also get a next row from compare rowset...
- data.two = getRowFrom(data.twoRowSet);
-
- } else if (data.one != null && data.two == null) { // Record 1 is flagged
as deleted!
+ } else if (data.one != null && data.two == null) {
outputRow = data.one;
flagField = VALUE_DELETED;
+ data.one = getAlignedRow(0, data.oneRowSet);
- // Also get a next row from reference rowset...
- data.one = getRowFrom(data.oneRowSet);
-
- } else { // OK, Here is the real start of the compare code!
-
- int compare = data.oneRowSet.getRowMeta().compare(data.one, data.two,
data.keyNrs);
- if (compare == 0) { // The Key matches, we CAN compare the two rows...
+ } else {
+ int compare = data.oneRowMeta.compare(data.one, data.two, data.keyNrs);
+ if (compare == 0) {
int compareValues = 0;
if (getDifference) {
JSONObject j = new JSONObject();
JSONArray jChanges = new JSONArray();
j.put("changes", jChanges);
for (int valueNr : data.valueNrs) {
- IValueMeta valueMeta =
data.oneRowSet.getRowMeta().getValueMeta(valueNr);
+ IValueMeta valueMeta = data.oneRowMeta.getValueMeta(valueNr);
Object refData = data.one[valueNr];
Object cmpData = data.two[valueNr];
int compareValue = valueMeta.compare(refData, cmpData);
@@ -226,39 +232,34 @@ public class MergeRows extends
BaseTransform<MergeRowsMeta, MergeRowsData> {
}
}
} else {
- compareValues = data.oneRowSet.getRowMeta().compare(data.one,
data.two, data.valueNrs);
+ compareValues = data.oneRowMeta.compare(data.one, data.two,
data.valueNrs);
}
if (compareValues == 0) {
- outputRow = data.two; // documented behavior: use the comparison
stream
+ outputRow = data.two;
flagField = VALUE_IDENTICAL;
} else {
- // Return the compare (most recent) row
- //
outputRow = data.two;
flagField = VALUE_CHANGED;
}
- // Get a new row from both streams...
- data.one = getRowFrom(data.oneRowSet);
- data.two = getRowFrom(data.twoRowSet);
+ data.one = getAlignedRow(0, data.oneRowSet);
+ data.two = getAlignedRow(1, data.twoRowSet);
} else {
- if (compare < 0) { // one < two
+ if (compare < 0) {
outputRow = data.one;
flagField = VALUE_DELETED;
- data.one = getRowFrom(data.oneRowSet);
+ data.one = getAlignedRow(0, data.oneRowSet);
} else {
outputRow = data.two;
flagField = VALUE_NEW;
- data.two = getRowFrom(data.twoRowSet);
+ data.two = getAlignedRow(1, data.twoRowSet);
}
}
}
int extraPassthroughFields = meta.getPassThroughFields().size();
- // Optionally add the difference JSON field
- //
outputRow = RowDataUtil.resizeArray(outputRow, data.outputRowMeta.size());
int tailIndex = data.outputRowMeta.size() - 2 - extraPassthroughFields;
if (getDifference && differenceJson != null) {
@@ -268,22 +269,14 @@ public class MergeRows extends
BaseTransform<MergeRowsMeta, MergeRowsData> {
}
outputRow[tailIndex++] = flagField;
- // Copy the passthrough fields
- //
for (int i = 0; i < extraPassthroughFields; i++) {
int sourceIndex = data.passThroughIndexes.get(i);
PassThroughField field = meta.getPassThroughFields().get(i);
if (field.isReferenceField()) {
- // If the record is deleted, identical or changed we copy data
- // from the reference row.
- //
if (data.oneCopy != null && !VALUE_NEW.equals(flagField)) {
outputRow[tailIndex] = data.oneCopy[sourceIndex];
}
} else {
- // If the record is new, identical or changed we copy data
- // from the compared-to row.
- //
if (data.twoCopy != null && !VALUE_DELETED.equals(flagField)) {
outputRow[tailIndex] = data.twoCopy[sourceIndex];
}
@@ -291,7 +284,6 @@ public class MergeRows extends BaseTransform<MergeRowsMeta,
MergeRowsData> {
tailIndex++;
}
- // send the row to the next transforms...
putRow(data.outputRowMeta, outputRow);
if (checkFeedback(getLinesRead()) && isBasic()) {
@@ -301,20 +293,32 @@ public class MergeRows extends
BaseTransform<MergeRowsMeta, MergeRowsData> {
return true;
}
+ private Object[] getAlignedRow(int streamIndex, org.apache.hop.core.IRowSet
rowSet)
+ throws HopTransformException {
+ Object[] row = getRowFrom(rowSet);
+ if (streamIndex == 0) {
+ data.oneOriginal = row;
+ } else {
+ data.twoOriginal = row;
+ }
+ if (row == null || data.schemaMapping == null) {
+ return row;
+ }
+ return data.schemaMapping.mapRow(streamIndex, row);
+ }
+
private void copyOneTwo() throws HopValueException {
data.oneCopy = null;
data.twoCopy = null;
- // We only need the exact current record values in case we want to pass
through fields
- //
if (meta.getPassThroughFields().isEmpty()) {
return;
}
- if (data.one != null) {
- data.oneCopy = data.oneRowMeta.cloneRow(data.one);
+ if (data.oneOriginal != null) {
+ data.oneCopy = data.referenceRowMeta.cloneRow(data.oneOriginal);
}
- if (data.two != null) {
- data.twoCopy = data.twoRowMeta.cloneRow(data.two);
+ if (data.twoOriginal != null) {
+ data.twoCopy = data.compareRowMeta.cloneRow(data.twoOriginal);
}
}
@@ -342,7 +346,6 @@ public class MergeRows extends BaseTransform<MergeRowsMeta,
MergeRowsData> {
*
* @param referenceRowMeta Reference row
* @param compareRowMeta Row to compare to
- * @return true when templates are compatible.
* @throws HopRowException in case there is a compatibility error.
*/
static void checkInputLayoutValid(IRowMeta referenceRowMeta, IRowMeta
compareRowMeta)
diff --git
a/plugins/transforms/mergerows/src/main/java/org/apache/hop/pipeline/transforms/mergerows/MergeRowsAlignment.java
b/plugins/transforms/mergerows/src/main/java/org/apache/hop/pipeline/transforms/mergerows/MergeRowsAlignment.java
new file mode 100644
index 0000000000..8395f8162a
--- /dev/null
+++
b/plugins/transforms/mergerows/src/main/java/org/apache/hop/pipeline/transforms/mergerows/MergeRowsAlignment.java
@@ -0,0 +1,139 @@
+/*
+ * 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.mergerows;
+
+import java.util.Arrays;
+import java.util.HashSet;
+import lombok.Getter;
+import lombok.Setter;
+import org.apache.hop.core.exception.HopTransformException;
+import org.apache.hop.core.exception.HopValueException;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.row.IValueMeta;
+import org.apache.hop.core.row.value.ValueMetaFactory;
+import org.apache.hop.i18n.BaseMessages;
+
+/** Reference-authoritative schema union and row mapping for {@link
MergeRows}. */
+@Getter
+@Setter
+public final class MergeRowsAlignment {
+ private static final Class<?> PKG = MergeRowsMeta.class;
+
+ private MergeRowsAlignment() {}
+
+ /** Union of reference and compare layouts plus per-stream field index
mappings. */
+ @Getter
+ public static final class SchemaMapping {
+ private final IRowMeta outputRowMeta;
+ private final IRowMeta compareRowMeta;
+ private final int[] referenceMapping;
+ private final int[] compareMapping;
+
+ private SchemaMapping(
+ IRowMeta outputRowMeta,
+ IRowMeta compareRowMeta,
+ int[] referenceMapping,
+ int[] compareMapping) {
+ this.outputRowMeta = outputRowMeta;
+ this.compareRowMeta = compareRowMeta;
+ this.referenceMapping = referenceMapping;
+ this.compareMapping = compareMapping;
+ }
+
+ public Object[] mapRow(int streamIndex, Object[] sourceRow) throws
HopTransformException {
+ Object[] outputRow = new Object[outputRowMeta.size()];
+ if (sourceRow == null) {
+ return outputRow;
+ }
+ if (streamIndex == 0) {
+ mapReferenceRow(sourceRow, outputRow);
+ } else if (streamIndex == 1) {
+ mapCompareRow(sourceRow, outputRow);
+ }
+ return outputRow;
+ }
+
+ private void mapReferenceRow(Object[] sourceRow, Object[] outputRow) {
+ for (int sourceIndex = 0; sourceIndex < referenceMapping.length;
sourceIndex++) {
+ int outputIndex = referenceMapping[sourceIndex];
+ if (outputIndex >= 0 && outputIndex < outputRow.length) {
+ outputRow[outputIndex] = sourceRow[sourceIndex];
+ }
+ }
+ }
+
+ private void mapCompareRow(Object[] sourceRow, Object[] outputRow)
+ throws HopTransformException {
+ for (int sourceIndex = 0; sourceIndex < compareMapping.length;
sourceIndex++) {
+ int outputIndex = compareMapping[sourceIndex];
+ if (outputIndex < 0 || outputIndex >= outputRow.length) {
+ continue;
+ }
+ IValueMeta compareField = compareRowMeta.getValueMeta(sourceIndex);
+ IValueMeta outputField = outputRowMeta.getValueMeta(outputIndex);
+ Object value = sourceRow[sourceIndex];
+ if (compareField.getType() != outputField.getType()) {
+ value = convertCompareValue(compareField, outputField, value);
+ }
+ outputRow[outputIndex] = value;
+ }
+ }
+
+ private Object convertCompareValue(
+ IValueMeta compareField, IValueMeta referenceField, Object value)
+ throws HopTransformException {
+ try {
+ return referenceField.convertData(compareField, value);
+ } catch (HopValueException e) {
+ throw new HopTransformException(
+ BaseMessages.getString(
+ PKG,
+ "MergeRowsAlignment.Exception.UnableToConvertCompareField",
+ compareField.getName(),
+ ValueMetaFactory.getValueMetaName(compareField.getType()),
+ ValueMetaFactory.getValueMetaName(referenceField.getType())),
+ e);
+ }
+ }
+ }
+
+ public static SchemaMapping buildSchemaMapping(
+ IRowMeta referenceRowMeta, IRowMeta compareRowMeta) {
+ IRowMeta outputRowMeta = referenceRowMeta.clone();
+ HashSet<String> fieldNames = new
HashSet<>(Arrays.asList(outputRowMeta.getFieldNames()));
+
+ int[] referenceMapping = new int[referenceRowMeta.size()];
+ for (int fieldIndex = 0; fieldIndex < referenceRowMeta.size();
fieldIndex++) {
+ String name = referenceRowMeta.getValueMeta(fieldIndex).getName();
+ referenceMapping[fieldIndex] = outputRowMeta.indexOfValue(name);
+ }
+
+ int[] compareMapping = new int[compareRowMeta.size()];
+ for (int fieldIndex = 0; fieldIndex < compareRowMeta.size(); fieldIndex++)
{
+ IValueMeta compareField = compareRowMeta.getValueMeta(fieldIndex);
+ String name = compareField.getName();
+ if (!fieldNames.contains(name)) {
+ outputRowMeta.addValueMeta(compareField.clone());
+ fieldNames.add(name);
+ }
+ compareMapping[fieldIndex] = outputRowMeta.indexOfValue(name);
+ }
+
+ return new SchemaMapping(outputRowMeta, compareRowMeta, referenceMapping,
compareMapping);
+ }
+}
diff --git
a/plugins/transforms/mergerows/src/main/java/org/apache/hop/pipeline/transforms/mergerows/MergeRowsData.java
b/plugins/transforms/mergerows/src/main/java/org/apache/hop/pipeline/transforms/mergerows/MergeRowsData.java
index b65426267d..253f13d64b 100644
---
a/plugins/transforms/mergerows/src/main/java/org/apache/hop/pipeline/transforms/mergerows/MergeRowsData.java
+++
b/plugins/transforms/mergerows/src/main/java/org/apache/hop/pipeline/transforms/mergerows/MergeRowsData.java
@@ -29,13 +29,18 @@ public class MergeRowsData extends BaseTransformData
implements ITransformData {
public Object[] one;
public Object[] two;
+ public Object[] oneOriginal;
+ public Object[] twoOriginal;
public int[] keyNrs;
public int[] valueNrs;
public IRowSet oneRowSet;
- public IRowMeta oneRowMeta;
public IRowSet twoRowSet;
+ public IRowMeta referenceRowMeta;
+ public IRowMeta compareRowMeta;
+ public IRowMeta oneRowMeta;
public IRowMeta twoRowMeta;
+ public MergeRowsAlignment.SchemaMapping schemaMapping;
public List<Integer> passThroughIndexes;
public Object[] oneCopy;
diff --git
a/plugins/transforms/mergerows/src/main/java/org/apache/hop/pipeline/transforms/mergerows/MergeRowsDialog.java
b/plugins/transforms/mergerows/src/main/java/org/apache/hop/pipeline/transforms/mergerows/MergeRowsDialog.java
index 6bcf8975d1..e020a7cc96 100644
---
a/plugins/transforms/mergerows/src/main/java/org/apache/hop/pipeline/transforms/mergerows/MergeRowsDialog.java
+++
b/plugins/transforms/mergerows/src/main/java/org/apache/hop/pipeline/transforms/mergerows/MergeRowsDialog.java
@@ -64,6 +64,7 @@ public class MergeRowsDialog extends BaseTransformDialog {
private CCombo wCompare;
private Text wFlagField;
private Text wDiffField;
+ private Button wAlignInputLayouts;
private TableView wKeys;
private TableView wValues;
@@ -202,6 +203,26 @@ public class MergeRowsDialog extends BaseTransformDialog {
fdDiffField.right = new FormAttachment(100, 0);
wDiffField.setLayoutData(fdDiffField);
+ Label wlAlign = new Label(wSourcesComp, SWT.RIGHT);
+ wlAlign.setText(BaseMessages.getString(PKG,
"MergeRowsDialog.AlignInputLayouts.Label"));
+ PropsUi.setLook(wlAlign);
+ FormData fdlAlign = new FormData();
+ fdlAlign.left = new FormAttachment(0, 0);
+ fdlAlign.right = new FormAttachment(middle, -margin);
+ fdlAlign.top = new FormAttachment(wDiffField, margin);
+ wlAlign.setLayoutData(fdlAlign);
+ wAlignInputLayouts = new Button(wSourcesComp, SWT.CHECK);
+ PropsUi.setLook(wAlignInputLayouts);
+ wAlignInputLayouts.setText(
+ BaseMessages.getString(PKG, "MergeRowsDialog.AlignInputLayouts.Text"));
+ wAlignInputLayouts.setSelection(true);
+ wAlignInputLayouts.addListener(SWT.Selection, e -> input.setChanged());
+ FormData fdAlign = new FormData();
+ fdAlign.top = new FormAttachment(wlAlign, 0, SWT.CENTER);
+ fdAlign.left = new FormAttachment(middle, 0);
+ fdAlign.right = new FormAttachment(100, 0);
+ wAlignInputLayouts.setLayoutData(fdAlign);
+
FormData fdSourcesComp = new FormData();
fdSourcesComp.left = new FormAttachment(0, 0);
fdSourcesComp.top = new FormAttachment(0, 0);
@@ -447,6 +468,7 @@ public class MergeRowsDialog extends BaseTransformDialog {
wCompare.setText(Const.NVL(infoStreams.get(1).getTransformName(), ""));
wFlagField.setText(Const.NVL(input.getFlagField(), ""));
wDiffField.setText(Const.NVL(input.getDiffJsonField(), ""));
+ wAlignInputLayouts.setSelection(input.isAlignInputLayouts());
for (int i = 0; i < input.getKeyFields().size(); i++) {
TableItem item = new TableItem(wKeys.table, SWT.NONE);
@@ -512,6 +534,7 @@ public class MergeRowsDialog extends BaseTransformDialog {
infoStreams.get(1).setTransformMeta(pipelineMeta.findTransform(wCompare.getText()));
meta.setFlagField(wFlagField.getText());
meta.setDiffJsonField(wDiffField.getText());
+ meta.setAlignInputLayouts(wAlignInputLayouts.getSelection());
meta.getKeyFields().clear();
for (TableItem item : wKeys.getNonEmptyItems()) {
diff --git
a/plugins/transforms/mergerows/src/main/java/org/apache/hop/pipeline/transforms/mergerows/MergeRowsMeta.java
b/plugins/transforms/mergerows/src/main/java/org/apache/hop/pipeline/transforms/mergerows/MergeRowsMeta.java
index 5441f024b1..b8db0961d7 100644
---
a/plugins/transforms/mergerows/src/main/java/org/apache/hop/pipeline/transforms/mergerows/MergeRowsMeta.java
+++
b/plugins/transforms/mergerows/src/main/java/org/apache/hop/pipeline/transforms/mergerows/MergeRowsMeta.java
@@ -99,6 +99,13 @@ public class MergeRowsMeta extends
BaseTransformMeta<MergeRows, MergeRowsData> {
injectionKeyDescription = "MergeRows.Injection.DIFF_FIELD")
private String diffJsonField;
+ @HopMetadataProperty(
+ key = "align_input_layouts",
+ defaultBoolean = false,
+ injectionKey = "ALIGN_INPUT_LAYOUTS",
+ injectionKeyDescription = "MergeRows.Injection.ALIGN_INPUT_LAYOUTS")
+ private boolean alignInputLayouts = true;
+
@HopMetadataProperty(
groupKey = "passthrough-fields",
key = "passthrough-field",
@@ -122,6 +129,8 @@ public class MergeRowsMeta extends
BaseTransformMeta<MergeRows, MergeRowsData> {
this.valueFields = new ArrayList<>(m.valueFields);
this.referenceTransform = m.referenceTransform;
this.compareTransform = m.compareTransform;
+ this.diffJsonField = m.diffJsonField;
+ this.alignInputLayouts = m.alignInputLayouts;
m.getPassThroughFields().forEach(f -> this.passThroughFields.add(new
PassThroughField(f)));
}
@@ -133,6 +142,7 @@ public class MergeRowsMeta extends
BaseTransformMeta<MergeRows, MergeRowsData> {
@Override
public void setDefault() {
flagField = "flagfield";
+ alignInputLayouts = true;
}
@Override
@@ -156,11 +166,16 @@ public class MergeRowsMeta extends
BaseTransformMeta<MergeRows, MergeRowsData> {
// So we just merge in the info fields.
//
if (info != null) {
- boolean found = false;
- for (int i = 0; i < info.length && !found; i++) {
- if (info[i] != null) {
- r.mergeRowMeta(info[i], name);
- found = true;
+ if (alignInputLayouts && info.length == 2 && info[0] != null && info[1]
!= null) {
+ r.mergeRowMeta(
+ MergeRowsAlignment.buildSchemaMapping(info[0],
info[1]).getOutputRowMeta(), name);
+ } else {
+ boolean found = false;
+ for (int i = 0; i < info.length && !found; i++) {
+ if (info[i] != null) {
+ r.mergeRowMeta(info[i], name);
+ found = true;
+ }
}
}
}
@@ -255,42 +270,54 @@ public class MergeRowsMeta extends
BaseTransformMeta<MergeRows, MergeRowsData> {
remarks.add(cr);
}
- IRowMeta referenceRowMeta = null;
- IRowMeta compareRowMeta = null;
- try {
- referenceRowMeta =
- pipelineMeta.getPrevTransformFields(variables,
referenceStream.getTransformName());
- compareRowMeta =
- pipelineMeta.getPrevTransformFields(variables,
compareStream.getTransformName());
- } catch (HopTransformException kse) {
- new CheckResult(
- ICheckResult.TYPE_RESULT_ERROR,
- BaseMessages.getString(PKG,
"MergeRowsMeta.CheckResult.ErrorGettingPrevTransformFields"),
- transformMeta);
- }
- if (referenceRowMeta != null && compareRowMeta != null) {
- boolean rowsMatch = false;
+ if (!alignInputLayouts) {
+ IRowMeta referenceRowMeta = null;
+ IRowMeta compareRowMeta = null;
try {
- MergeRows.checkInputLayoutValid(referenceRowMeta, compareRowMeta);
- rowsMatch = true;
- } catch (HopRowException kre) {
- // Ignore
- }
- if (rowsMatch) {
- cr =
- new CheckResult(
- ICheckResult.TYPE_RESULT_OK,
- BaseMessages.getString(PKG,
"MergeRowsMeta.CheckResult.RowDefinitionMatch"),
- transformMeta);
- remarks.add(cr);
- } else {
- cr =
+ referenceRowMeta =
+ pipelineMeta.getPrevTransformFields(variables,
referenceStream.getTransformName());
+ compareRowMeta =
+ pipelineMeta.getPrevTransformFields(variables,
compareStream.getTransformName());
+ } catch (HopTransformException kse) {
+ remarks.add(
new CheckResult(
ICheckResult.TYPE_RESULT_ERROR,
- BaseMessages.getString(PKG,
"MergeRowsMeta.CheckResult.RowDefinitionNotMatch"),
- transformMeta);
- remarks.add(cr);
+ BaseMessages.getString(
+ PKG,
"MergeRowsMeta.CheckResult.ErrorGettingPrevTransformFields"),
+ transformMeta));
}
+ if (referenceRowMeta != null && compareRowMeta != null) {
+ boolean rowsMatch = false;
+ try {
+ MergeRows.checkInputLayoutValid(referenceRowMeta, compareRowMeta);
+ rowsMatch = true;
+ } catch (HopRowException kre) {
+ // Ignore
+ }
+ if (rowsMatch) {
+ cr =
+ new CheckResult(
+ ICheckResult.TYPE_RESULT_OK,
+ BaseMessages.getString(PKG,
"MergeRowsMeta.CheckResult.RowDefinitionMatch"),
+ transformMeta);
+ remarks.add(cr);
+ } else {
+ cr =
+ new CheckResult(
+ ICheckResult.TYPE_RESULT_ERROR,
+ BaseMessages.getString(PKG,
"MergeRowsMeta.CheckResult.RowDefinitionNotMatch"),
+ transformMeta);
+ remarks.add(cr);
+ }
+ }
+ } else if (referenceStream.getTransformName() != null
+ && compareStream.getTransformName() != null) {
+ cr =
+ new CheckResult(
+ ICheckResult.TYPE_RESULT_OK,
+ BaseMessages.getString(PKG,
"MergeRowsMeta.CheckResult.AlignmentEnabled"),
+ transformMeta);
+ remarks.add(cr);
}
}
diff --git
a/plugins/transforms/mergerows/src/main/resources/org/apache/hop/pipeline/transforms/mergerows/messages/messages_en_US.properties
b/plugins/transforms/mergerows/src/main/resources/org/apache/hop/pipeline/transforms/mergerows/messages/messages_en_US.properties
index 17ac21b5fc..eedbed63d1 100644
---
a/plugins/transforms/mergerows/src/main/resources/org/apache/hop/pipeline/transforms/mergerows/messages/messages_en_US.properties
+++
b/plugins/transforms/mergerows/src/main/resources/org/apache/hop/pipeline/transforms/mergerows/messages/messages_en_US.properties
@@ -15,17 +15,19 @@
# limitations under the License.
#
-MergeRows.Description=Merge two streams of rows, sorted on a certain key. The
two streams are compared and the equals, changed, deleted and new rows are
flagged.
+MergeRows.Description=Merge two streams of rows, sorted on a certain key. The
two streams are compared and the equals, changed, deleted and new rows are
flagged. Optionally aligns mismatched input layouts before comparing.
MergeRows.Exception.InvalidLayoutDetected=Invalid layout detected in input
streams, keys and values to merge have to be of identical structure and be in
the same place in the rows
MergeRows.Exception.UnableToFindFieldInReferenceStream=Unable to find field
[{0}] in reference stream.
+MergeRows.Exception.UnableToFindFieldInAlignedStream=Unable to find field
[{0}] in the aligned input layout.
MergeRows.Injection.FLAG_FIELD=The name of the flag field.
MergeRows.Injection.DIFF_FIELD=The name of the field for the difference JSON.
MergeRows.Injection.KEY_FIELDS=Specify key fields to compare.
MergeRows.Injection.KEY_FIELD=The key field to match with
MergeRows.Injection.VALUE_FIELDS=Specify value fields to compare.
MergeRows.Injection.VALUE_FIELD=The field to compare
-MergeRows.Injection.PASSTHROUGH_FIELDS = The fields to pass through to the
result
-MergeRows.Injection.PASSTHROUGH_FIELD = One fields to pass through to the
result
+MergeRows.Injection.ALIGN_INPUT_LAYOUTS=Automatically align reference and
compare row layouts before comparing
+MergeRows.Injection.PASSTHROUGH_FIELDS=The fields to pass through to the result
+MergeRows.Injection.PASSTHROUGH_FIELD=One field to pass through to the result
MergeRows.LineNumber=linenr
MergeRows.Log.BothTrueAndFalseNeeded=Both the ''true'' and the ''false''
transforms need to be supplied, or neither
MergeRows.Log.DataInfo=ONE\: {0} / TWO\:
@@ -37,6 +39,8 @@ MergeRowsDialog.ErrorGettingFields.DialogMessage=Unable to
get the fields becaus
MergeRowsDialog.ErrorGettingFields.DialogTitle=Error getting fields
MergeRowsDialog.FlagField.Label=Flag field name
MergeRowsDialog.DiffField.Label=Difference JSON field name
+MergeRowsDialog.AlignInputLayouts.Label=Align input layouts\:
+MergeRowsDialog.AlignInputLayouts.Text=Automatically align reference and
compare row layouts before comparing
MergeRowsDialog.KeyFields.Button=Get &key fields
MergeRowsDialog.Keys.Label=Keys to match \:
MergeRowsDialog.MergeRowsWarningDialog.DialogMessage=If the incoming data is
not sorted on the specified keys, the output results may not be correct. We
recommend sorting the incoming data within the pipeline.
@@ -48,6 +52,7 @@ MergeRowsDialog.Shell.Label=Merge rows (diff)
MergeRowsDialog.TransformName.Label=Transform name
MergeRowsDialog.ValueFields.Button=Get &value fields
MergeRowsDialog.Values.Label=Values to compare \:
+MergeRowsMeta.CheckResult.AlignmentEnabled=Input layout alignment is enabled;
reference and compare streams may use different field layouts.
MergeRowsMeta.CheckResult.ErrorGettingPrevTransformFields=Error when
retrieving fields from previous transforms
MergeRowsMeta.CheckResult.OneSourceTransformMissing=Please specify both the
'reference' AND 'compare' source transforms.
MergeRowsMeta.CheckResult.RowDefinitionMatch=The field layout from both input
transforms match
@@ -58,13 +63,15 @@ MergeRowsMeta.Exception.FlagFieldNotSpecified=The flag
field is not specified.
MergeRowsMeta.Exception.UnableToLoadTransformMeta=Unable to load transform
info from XML
MergeRowsMeta.InfoStream.FirstStream.Description=Reference stream to merge
MergeRowsMeta.InfoStream.SecondStream.Description=Compare (changed data)
stream to merge
-MergeRowsMeta.keyword=merge,diff,compare,changed,deleted
-MergeRowsDialog.ExtraFields.Button = Get fields to pass through
+MergeRowsMeta.keyword=merge,diff,compare,changed,deleted,align,layout
+MergeRowsDialog.ExtraFields.Button=Get fields to pass through
MergeRowsDialog.Extra.Label=Extra fields to pass through from sources to the
output\:
MergeRowsDialog.ExtraColumn.Reference.Label=Reference?
MergeRowsDialog.ExtraColumn.Field.Label=Source field
MergeRowsDialog.ExtraColumn.RenameTo.Label=Rename to
-MergeRowsDialog.SourcesTab.CTabItem = Sources
-MergeRowsDialog.KeysTab.CTabItem = Keys
-MergeRowsDialog.ValuesTab.CTabItem = Values
-MergeRowsDialog.ExtraTab.CTabItem = Extra
+MergeRowsDialog.SourcesTab.CTabItem=Sources
+MergeRowsDialog.KeysTab.CTabItem=Keys
+MergeRowsDialog.ValuesTab.CTabItem=Values
+MergeRowsDialog.ExtraTab.CTabItem=Extra
+MergeRowsAlignment.Exception.UnableToConvertCompareField=Unable to convert
compare field ''{0}'' from {1} to reference type {2}
+
diff --git
a/plugins/transforms/mergerows/src/test/java/org/apache/hop/pipeline/transforms/mergerows/MergeRowsAlignmentTest.java
b/plugins/transforms/mergerows/src/test/java/org/apache/hop/pipeline/transforms/mergerows/MergeRowsAlignmentTest.java
new file mode 100644
index 0000000000..09f93ebd35
--- /dev/null
+++
b/plugins/transforms/mergerows/src/test/java/org/apache/hop/pipeline/transforms/mergerows/MergeRowsAlignmentTest.java
@@ -0,0 +1,137 @@
+/*
+ * 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.mergerows;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.apache.hop.core.HopEnvironment;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.exception.HopTransformException;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.row.IValueMeta;
+import org.apache.hop.core.row.RowMeta;
+import org.apache.hop.core.row.value.ValueMetaInteger;
+import org.apache.hop.core.row.value.ValueMetaString;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+class MergeRowsAlignmentTest {
+
+ @BeforeAll
+ static void initHop() throws HopException {
+ HopEnvironment.init();
+ }
+
+ @Test
+ void buildSchemaMappingUnionsFieldsAndNullFillsMissingColumns() throws
HopTransformException {
+ IRowMeta reference = row("hub_hk", "attr_a");
+ IRowMeta compare = row("hub_hk", "attr_b");
+
+ MergeRowsAlignment.SchemaMapping mapping =
+ MergeRowsAlignment.buildSchemaMapping(reference, compare);
+
+ assertEquals(3, mapping.getOutputRowMeta().size());
+ Object[] mappedReference = mapping.mapRow(0, new Object[] {"hk-1",
"value-a"});
+ Object[] mappedCompare = mapping.mapRow(1, new Object[] {"hk-1",
"value-b"});
+
+ assertEquals("hk-1", mappedReference[index(mapping, "hub_hk")]);
+ assertEquals("value-a", mappedReference[index(mapping, "attr_a")]);
+ assertNull(mappedReference[index(mapping, "attr_b")]);
+
+ assertEquals("hk-1", mappedCompare[index(mapping, "hub_hk")]);
+ assertNull(mappedCompare[index(mapping, "attr_a")]);
+ assertEquals("value-b", mappedCompare[index(mapping, "attr_b")]);
+ }
+
+ @Test
+ void buildSchemaMappingKeepsReferenceTypeForSharedFields() {
+ IRowMeta reference = new RowMeta();
+ reference.addValueMeta(new ValueMetaString("shared"));
+
+ IRowMeta compare = new RowMeta();
+ compare.addValueMeta(new ValueMetaInteger("shared"));
+
+ MergeRowsAlignment.SchemaMapping mapping =
+ MergeRowsAlignment.buildSchemaMapping(reference, compare);
+
+ IValueMeta sharedField =
mapping.getOutputRowMeta().getValueMeta(index(mapping, "shared"));
+ assertEquals(IValueMeta.TYPE_STRING, sharedField.getType());
+ }
+
+ @Test
+ void mapCompareRowConvertsCompatibleValuesToReferenceType() throws
HopTransformException {
+ IRowMeta reference = new RowMeta();
+ reference.addValueMeta(new ValueMetaInteger("amount"));
+
+ IRowMeta compare = new RowMeta();
+ compare.addValueMeta(new ValueMetaString("amount"));
+
+ MergeRowsAlignment.SchemaMapping mapping =
+ MergeRowsAlignment.buildSchemaMapping(reference, compare);
+
+ Object[] mappedCompare = mapping.mapRow(1, new Object[] {"42"});
+ assertEquals(42L, mappedCompare[index(mapping, "amount")]);
+ }
+
+ @Test
+ void mapCompareRowFailsWithClearErrorForIncompatibleConversion() {
+ IRowMeta reference = new RowMeta();
+ reference.addValueMeta(new ValueMetaInteger("amount"));
+
+ IRowMeta compare = new RowMeta();
+ compare.addValueMeta(new ValueMetaString("amount"));
+
+ MergeRowsAlignment.SchemaMapping mapping =
+ MergeRowsAlignment.buildSchemaMapping(reference, compare);
+
+ HopTransformException error =
+ assertThrows(
+ HopTransformException.class, () -> mapping.mapRow(1, new Object[]
{"not-a-number"}));
+ assertTrue(error.getMessage().contains("amount"));
+ assertTrue(error.getMessage().contains("String"));
+ assertTrue(error.getMessage().contains("Integer"));
+ }
+
+ @Test
+ void compareOnlyFieldKeepsCompareType() {
+ IRowMeta reference = row("hub_hk");
+ IRowMeta compare = row("hub_hk", "attr_b");
+
+ MergeRowsAlignment.SchemaMapping mapping =
+ MergeRowsAlignment.buildSchemaMapping(reference, compare);
+
+ assertEquals(
+ IValueMeta.TYPE_STRING,
+ mapping.getOutputRowMeta().getValueMeta(index(mapping,
"attr_b")).getType());
+ }
+
+ private static int index(MergeRowsAlignment.SchemaMapping mapping, String
fieldName) {
+ return mapping.getOutputRowMeta().indexOfValue(fieldName);
+ }
+
+ private static IRowMeta row(String... fieldNames) {
+ RowMeta rowMeta = new RowMeta();
+ for (String fieldName : fieldNames) {
+ rowMeta.addValueMeta(new ValueMetaString(fieldName));
+ }
+ return rowMeta;
+ }
+}
diff --git
a/plugins/transforms/mergerows/src/test/java/org/apache/hop/pipeline/transforms/mergerows/MergeRowsLogicTest.java
b/plugins/transforms/mergerows/src/test/java/org/apache/hop/pipeline/transforms/mergerows/MergeRowsLogicTest.java
new file mode 100644
index 0000000000..c451f383e2
--- /dev/null
+++
b/plugins/transforms/mergerows/src/test/java/org/apache/hop/pipeline/transforms/mergerows/MergeRowsLogicTest.java
@@ -0,0 +1,135 @@
+/*
+ * 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.mergerows;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import org.apache.hop.core.HopEnvironment;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.exception.HopRowException;
+import org.apache.hop.core.exception.HopValueException;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.row.RowMeta;
+import org.apache.hop.core.row.value.ValueMetaString;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+class MergeRowsLogicTest {
+
+ @BeforeAll
+ static void initHop() throws HopException {
+ HopEnvironment.init();
+ }
+
+ @Test
+ void alignedRowsProduceChangedFlagWhenValuesDiffer() throws Exception {
+ IRowMeta reference = row("id", "name");
+ IRowMeta compare = row("id", "extra");
+
+ MergeRowsAlignment.SchemaMapping mapping =
+ MergeRowsAlignment.buildSchemaMapping(reference, compare);
+ IRowMeta aligned = mapping.getOutputRowMeta();
+
+ Object[] refRow = mapping.mapRow(0, new Object[] {"1", "Alice"});
+ Object[] cmpRow = mapping.mapRow(1, new Object[] {"1", "extra-value"});
+
+ int[] keyNrs = new int[] {aligned.indexOfValue("id")};
+ int[] valueNrs = new int[] {aligned.indexOfValue("name")};
+
+ assertEquals(0, aligned.compare(refRow, cmpRow, keyNrs));
+ assertEquals(MergeRows.VALUE_CHANGED, flagForRows(aligned, refRow, cmpRow,
keyNrs, valueNrs));
+ }
+
+ @Test
+ void alignedRowsProduceIdenticalFlagWhenValuesMatch() throws Exception {
+ IRowMeta reference = row("id", "name", "status");
+ IRowMeta compare = row("id", "name");
+
+ MergeRowsAlignment.SchemaMapping mapping =
+ MergeRowsAlignment.buildSchemaMapping(reference, compare);
+ IRowMeta aligned = mapping.getOutputRowMeta();
+
+ Object[] refRow = mapping.mapRow(0, new Object[] {"1", "Alice", "active"});
+ Object[] cmpRow = mapping.mapRow(1, new Object[] {"1", "Alice"});
+
+ int[] keyNrs = new int[] {aligned.indexOfValue("id")};
+ int[] valueNrs = new int[] {aligned.indexOfValue("name")};
+
+ assertEquals(MergeRows.VALUE_IDENTICAL, flagForRows(aligned, refRow,
cmpRow, keyNrs, valueNrs));
+ }
+
+ @Test
+ void alignedRowsProduceNewAndDeletedFlagsForKeyOrdering() throws Exception {
+ IRowMeta reference = row("id", "name");
+ IRowMeta compare = row("id", "name");
+
+ MergeRowsAlignment.SchemaMapping mapping =
+ MergeRowsAlignment.buildSchemaMapping(reference, compare);
+ IRowMeta aligned = mapping.getOutputRowMeta();
+ int[] keyNrs = new int[] {aligned.indexOfValue("id")};
+ int[] valueNrs = new int[] {aligned.indexOfValue("name")};
+
+ Object[] refOnly = mapping.mapRow(0, new Object[] {"1", "Alice"});
+ Object[] cmpLater = mapping.mapRow(1, new Object[] {"2", "Bob"});
+
+ assertEquals(
+ MergeRows.VALUE_DELETED, flagForRows(aligned, refOnly, cmpLater,
keyNrs, valueNrs));
+
+ Object[] refLater = mapping.mapRow(0, new Object[] {"2", "Bob"});
+ Object[] cmpOnly = mapping.mapRow(1, new Object[] {"1", "Alice"});
+
+ assertEquals(MergeRows.VALUE_NEW, flagForRows(aligned, refLater, cmpOnly,
keyNrs, valueNrs));
+ }
+
+ @Test
+ void checkInputLayoutValidRejectsMismatchedLayouts() {
+ IRowMeta reference = row("id", "name");
+ IRowMeta compare = row("id", "extra");
+
+ assertThrows(HopRowException.class, () ->
MergeRows.checkInputLayoutValid(reference, compare));
+ }
+
+ private static String flagForRows(
+ IRowMeta aligned, Object[] referenceRow, Object[] compareRow, int[]
keyNrs, int[] valueNrs)
+ throws HopValueException {
+ if (referenceRow == null && compareRow != null) {
+ return MergeRows.VALUE_NEW;
+ }
+ if (referenceRow != null && compareRow == null) {
+ return MergeRows.VALUE_DELETED;
+ }
+
+ int compare = aligned.compare(referenceRow, compareRow, keyNrs);
+ if (compare == 0) {
+ if (aligned.compare(referenceRow, compareRow, valueNrs) == 0) {
+ return MergeRows.VALUE_IDENTICAL;
+ }
+ return MergeRows.VALUE_CHANGED;
+ }
+ return compare < 0 ? MergeRows.VALUE_DELETED : MergeRows.VALUE_NEW;
+ }
+
+ private static IRowMeta row(String... fieldNames) {
+ RowMeta rowMeta = new RowMeta();
+ for (String fieldName : fieldNames) {
+ rowMeta.addValueMeta(new ValueMetaString(fieldName));
+ }
+ return rowMeta;
+ }
+}
diff --git
a/plugins/transforms/mergerows/src/test/java/org/apache/hop/pipeline/transforms/mergerows/MergeRowsMetaCheckTest.java
b/plugins/transforms/mergerows/src/test/java/org/apache/hop/pipeline/transforms/mergerows/MergeRowsMetaCheckTest.java
index 4599efc94d..3e1020ba79 100644
---
a/plugins/transforms/mergerows/src/test/java/org/apache/hop/pipeline/transforms/mergerows/MergeRowsMetaCheckTest.java
+++
b/plugins/transforms/mergerows/src/test/java/org/apache/hop/pipeline/transforms/mergerows/MergeRowsMetaCheckTest.java
@@ -92,7 +92,29 @@ class MergeRowsMetaCheckTest {
}
@Test
- void testCheckInputRowsBothEmpty() throws HopTransformException {
+ void testCheckWithAlignmentEnabledReportsOkWithoutLayoutValidation() {
+ meta.setAlignInputLayouts(true);
+
+ meta.check(
+ remarks,
+ pipelineMeta,
+ transformMeta,
+ null,
+ new String[0],
+ new String[0],
+ (RowMeta) null,
+ new Variables(),
+ null);
+
+ assertNotNull(remarks);
+ assertTrue(remarks.size() >= 2);
+ assertEquals(ICheckResult.TYPE_RESULT_OK, remarks.get(0).getType());
+ assertEquals(ICheckResult.TYPE_RESULT_OK, remarks.get(1).getType());
+ }
+
+ @Test
+ void testCheckInputRowsBothEmptyStrict() throws HopTransformException {
+ meta.setAlignInputLayouts(false);
when(pipelineMeta.getPrevTransformFields(any(IVariables.class),
eq(REFERENCE_TRANSFORM_NAME)))
.thenReturn(generateRowMetaEmpty());
when(pipelineMeta.getPrevTransformFields(any(IVariables.class),
eq(COMPARISON_TRANSFORM_NAME)))
@@ -115,7 +137,8 @@ class MergeRowsMetaCheckTest {
}
@Test
- void testCheckInputRowsBothNonEmpty() throws HopTransformException {
+ void testCheckInputRowsBothNonEmptyStrict() throws HopTransformException {
+ meta.setAlignInputLayouts(false);
when(pipelineMeta.getPrevTransformFields(any(IVariables.class),
eq(REFERENCE_TRANSFORM_NAME)))
.thenReturn(generateRowMeta10Strings());
when(pipelineMeta.getPrevTransformFields(any(IVariables.class),
eq(COMPARISON_TRANSFORM_NAME)))
@@ -138,7 +161,8 @@ class MergeRowsMetaCheckTest {
}
@Test
- void testCheckInputRowsEmptyAndNonEmpty() throws HopTransformException {
+ void testCheckInputRowsEmptyAndNonEmptyStrict() throws HopTransformException
{
+ meta.setAlignInputLayouts(false);
when(pipelineMeta.getPrevTransformFields(any(),
eq(REFERENCE_TRANSFORM_NAME)))
.thenReturn(generateRowMetaEmpty());
when(pipelineMeta.getPrevTransformFields(any(),
eq(COMPARISON_TRANSFORM_NAME)))
@@ -161,7 +185,8 @@ class MergeRowsMetaCheckTest {
}
@Test
- void testCheckInputRowsDifferentRowMetaTypes() throws HopTransformException {
+ void testCheckInputRowsDifferentRowMetaTypesStrict() throws
HopTransformException {
+ meta.setAlignInputLayouts(false);
when(pipelineMeta.getPrevTransformFields(any(),
eq(REFERENCE_TRANSFORM_NAME)))
.thenReturn(generateRowMeta10MixedTypes());
when(pipelineMeta.getPrevTransformFields(any(),
eq(COMPARISON_TRANSFORM_NAME)))
diff --git
a/plugins/transforms/mergerows/src/test/java/org/apache/hop/pipeline/transforms/mergerows/MergeRowsMetaTest.java
b/plugins/transforms/mergerows/src/test/java/org/apache/hop/pipeline/transforms/mergerows/MergeRowsMetaTest.java
index 3083170635..52e5c11766 100644
---
a/plugins/transforms/mergerows/src/test/java/org/apache/hop/pipeline/transforms/mergerows/MergeRowsMetaTest.java
+++
b/plugins/transforms/mergerows/src/test/java/org/apache/hop/pipeline/transforms/mergerows/MergeRowsMetaTest.java
@@ -19,6 +19,8 @@
package org.apache.hop.pipeline.transforms.mergerows;
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.core.HopEnvironment;
import org.apache.hop.core.plugins.PluginRegistry;
@@ -52,5 +54,30 @@ class MergeRowsMetaTest {
assertEquals(2, meta.getValueFields().size());
assertEquals("name", meta.getValueFields().get(0));
assertEquals("score", meta.getValueFields().get(1));
+ // Missing align_input_layouts in XML uses defaultBoolean=false
+ // This keeps previously built pipelines compatible.
+ assertFalse(meta.isAlignInputLayouts());
+ }
+
+ @Test
+ void testDefaultAlignInputLayoutsEnabled() {
+ MergeRowsMeta meta = new MergeRowsMeta();
+ meta.setDefault();
+ assertTrue(meta.isAlignInputLayouts());
+ }
+
+ @Test
+ void testCloneCopiesDiffAndAlign() {
+ MergeRowsMeta meta = new MergeRowsMeta();
+ meta.setFlagField("flag");
+ meta.setDiffJsonField("difference");
+ meta.setAlignInputLayouts(false);
+ meta.getKeyFields().add("id");
+
+ MergeRowsMeta copy = meta.clone();
+ assertEquals("flag", copy.getFlagField());
+ assertEquals("difference", copy.getDiffJsonField());
+ assertFalse(copy.isAlignInputLayouts());
+ assertEquals(1, copy.getKeyFields().size());
}
}
