diff --git a/docker/integration-tests/integration-tests-base.yaml b/docker/integration-tests/integration-tests-base.yaml index 51d1a1028a2..2f1c9277c1c 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 fb62f84ecaa..28db04461a5 100644 --- a/docker/integration-tests/integration-tests-beam-base.yaml +++ b/docker/integration-tests/integration-tests-beam-base.yaml @@ -29,4 +29,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/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/mergerows.adoc b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/mergerows.adoc index a6e451b24ff..7dd38f33901 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 446a91e2850..9d0bbc8d8a7 100755 --- a/integration-tests/scripts/run-tests-docker.sh +++ b/integration-tests/scripts/run-tests-docker.sh @@ -29,6 +29,7 @@ for ARGUMENT in "$@"; do 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} ;; @@ -44,6 +45,14 @@ 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. Examples: +# TEST_FILTER=0077-merge-rows +# TEST_FILTER='*0077*' +if [ -z "${TEST_FILTER}" ]; then + TEST_FILTER="" +fi + if [ -z "${JENKINS_USER}" ]; then JENKINS_USER="jenkins" fi @@ -117,13 +126,17 @@ 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") - 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 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 4c465295fa1..872b4085aff 100755 --- a/integration-tests/scripts/run-tests.sh +++ b/integration-tests/scripts/run-tests.sh @@ -104,9 +104,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}" @@ -161,8 +212,16 @@ for d in "${CURRENT_DIR}"/../${PROJECT_NAME}/; do # # TODO: add hpl support when result is returned correctly # + if [ -n "${TEST_FILTER}" ]; then + echo "TEST_FILTER is set: ${TEST_FILTER}" + fi + find $d -name 'main*.hwf' | sort | while read 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 00000000000..a3b5a269bd0 --- /dev/null +++ b/integration-tests/transforms/0077-merge-rows-align-off.hpl @@ -0,0 +1,291 @@ + + + + + 0077-merge-rows-align-off + Y + Legacy strict layout mode (align off) with matching streams + + + Normal + + + N + 1000 + 100 + - + 2026/07/10 12:00:00.000 + - + 2026/07/10 12:00:00.000 + + + + + + source1 + Merge rows (diff) + Y + + + source2 + Merge rows (diff) + Y + + + Merge rows (diff) + validate + Y + + + + Merge rows (diff) + MergeRows + + Y + + 1 + + none + + + N + + id + + + name + score + + flagfield + source1 + source2 + + + 240 + 96 + + + + source1 + DataGrid + + Y + + 1 + + none + + + + + 1 + James + 3.6 + + + 2 + Marie + 8.3 + + + 3 + Michael + 9.1 + + + 4 + Patrcia + 4.2 + + + 5 + Robert + 5.0 + + + 6 + Linda + 1.6 + + + 7 + David + 7.2 + + + 8 + Elizabeth + 6.9 + + + 9 + Willian + 4.2 + + + 10 + Barbara + 0.5 + + + + + -1 + -1 + N + id + Integer + + + -1 + -1 + N + name + String + + + -1 + -1 + N + score + #.# + Number + + + + + 96 + 48 + + + + source2 + DataGrid + + Y + + 1 + + none + + + + + 1 + James + 3.6 + + + 3 + Michael + 9.1 + + + 4 + Patricia + 4.2 + + + 5 + Robert + 5.0 + + + 7 + David + 7.2 + + + 8 + Elizabeth + 6.9 + + + 9 + William + 4.2 + + + 10 + Barbara + 0.5 + + + 11 + Richard + 9.6 + + + 12 + Susan + 5.4 + + + + + -1 + -1 + N + id + Integer + + + -1 + -1 + N + name + String + + + -1 + -1 + N + score + #.# + Number + + + + + 96 + 144 + + + + validate + Dummy + + Y + + 1 + + none + + + + + 416 + 96 + + + + + + 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 00000000000..bc89f6ac586 --- /dev/null +++ b/integration-tests/transforms/0077-merge-rows-align.hpl @@ -0,0 +1,218 @@ + + + + + 0077-merge-rows-align + Y + Merge rows with mismatched layouts and automatic alignment enabled + + + Normal + + + N + 1000 + 100 + - + 2026/07/10 12:00:00.000 + - + 2026/07/10 12:00:00.000 + + + + + + source1 + Merge rows (diff) + Y + + + source2 + Merge rows (diff) + Y + + + Merge rows (diff) + validate + Y + + + + Merge rows (diff) + MergeRows + + Y + + 1 + + none + + + Y + + id + + + status + + flagfield + source1 + source2 + + + 240 + 96 + + + + source1 + DataGrid + Reference stream: id, name, status + Y + + 1 + + none + + + + + 1 + Alice + active + + + 2 + Bob + active + + + 3 + Carol + inactive + + + + + -1 + -1 + N + id + Integer + + + -1 + -1 + N + name + String + + + -1 + -1 + N + status + String + + + + + 96 + 48 + + + + source2 + DataGrid + Compare stream: id, status, region (different fields and order) + Y + + 1 + + none + + + + + 1 + active + US + + + 2 + retired + EU + + + 4 + pending + APAC + + + + + -1 + -1 + N + id + Integer + + + -1 + -1 + N + status + String + + + -1 + -1 + N + region + String + + + + + 96 + 144 + + + + validate + Dummy + + Y + + 1 + + none + + + + + 416 + 96 + + + + + + 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 00000000000..ae776bc7d0c --- /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 381b71ece96..f840fc84336 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. 0077-merge-rows-passthrough UNIT + + 0077-merge-rows-align UNIT + + + 0077-merge-rows-align-off UNIT + N 272 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 00000000000..7daa2be65cc --- /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 00000000000..8d2a1a62cf6 --- /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 00000000000..381e807e70a --- /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 8f0694b2f68..c1dab031a53 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 @@ /** * 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 { 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 boolean processRow() throws HopException { if (first) { first = false; - // Find the appropriate RowSet - // List 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 boolean processRow() throws HopException { } } - 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 boolean processRow() throws HopException { } } - // 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 boolean processRow() throws HopException { + 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 boolean processRow() throws HopException { 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 boolean processRow() throws HopException { 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 boolean processRow() throws HopException { } } } 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 boolean processRow() throws HopException { } 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 boolean processRow() throws HopException { tailIndex++; } - // send the row to the next transforms... putRow(data.outputRowMeta, outputRow); if (checkFeedback(getLinesRead()) && isBasic()) { @@ -301,20 +293,32 @@ public boolean processRow() throws HopException { 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 boolean init() { * * @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 00000000000..6a2b80c5a96 --- /dev/null +++ b/plugins/transforms/mergerows/src/main/java/org/apache/hop/pipeline/transforms/mergerows/MergeRowsAlignment.java @@ -0,0 +1,145 @@ +/* + * 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.HashSet; +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}. */ +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. */ + public static final class SchemaMapping { + private final IRowMeta outputRowMeta; + private final IRowMeta referenceRowMeta; + private final IRowMeta compareRowMeta; + private final int[] referenceMapping; + private final int[] compareMapping; + + private SchemaMapping( + IRowMeta outputRowMeta, + IRowMeta referenceRowMeta, + IRowMeta compareRowMeta, + int[] referenceMapping, + int[] compareMapping) { + this.outputRowMeta = outputRowMeta; + this.referenceRowMeta = referenceRowMeta; + this.compareRowMeta = compareRowMeta; + this.referenceMapping = referenceMapping; + this.compareMapping = compareMapping; + } + + public IRowMeta getOutputRowMeta() { + return outputRowMeta; + } + + 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 fieldNames = new HashSet<>(); + for (String fieldName : outputRowMeta.getFieldNames()) { + fieldNames.add(fieldName); + } + + 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, referenceRowMeta, 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 b65426267d0..253f13d64b3 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 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 6bcf8975d10..e020a7cc968 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 @@ private void addSourcesTab() { 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 void getData() { 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 @@ private void getInfo(MergeRowsMeta meta) { 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 5441f024b17..b561a7da71a 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 { injectionKeyDescription = "MergeRows.Injection.DIFF_FIELD") private String diffJsonField; + @HopMetadataProperty( + key = "align_input_layouts", + defaultBoolean = true, + 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 MergeRowsMeta(MergeRowsMeta m) { 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 MergeRowsMeta clone() { @Override public void setDefault() { flagField = "flagfield"; + alignInputLayouts = true; } @Override @@ -156,11 +166,16 @@ public void getFields( // 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 void check( 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 17ac21b5fc0..eedbed63d13 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 00000000000..09f93ebd35b --- /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 00000000000..c451f383e21 --- /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 4599efc94d3..3e1020ba798 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 @@ void setup() { } @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 @@ void testCheckInputRowsBothEmpty() throws HopTransformException { } @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 @@ void testCheckInputRowsBothNonEmpty() throws HopTransformException { } @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 @@ void testCheckInputRowsEmptyAndNonEmpty() throws HopTransformException { } @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 30831706351..503975fdda6 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 @@ -52,5 +52,29 @@ void testSerialization() throws Exception { 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=true + assertEquals(true, meta.isAlignInputLayouts()); + } + + @Test + void testDefaultAlignInputLayoutsEnabled() { + MergeRowsMeta meta = new MergeRowsMeta(); + meta.setDefault(); + assertEquals(true, 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()); + assertEquals(false, copy.isAlignInputLayouts()); + assertEquals(1, copy.getKeyFields().size()); } }