This is an automated email from the ASF dual-hosted git repository.
mattcasters 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 ee18184f78 Fix Sorted Merge when row comparison throws instead of
treating rows as equal (#8354)
ee18184f78 is described below
commit ee18184f789256a6686e9a202d0b3585c4ca9613
Author: Lance <[email protected]>
AuthorDate: Thu Sep 17 01:29:21 2026 +0800
Fix Sorted Merge when row comparison throws instead of treating rows as
equal (#8354)
Signed-off-by: lance <[email protected]>
---
.../pipeline/transforms/sortedmerge/RowSetRow.java | 59 +-----
.../transforms/sortedmerge/SortedMerge.java | 18 +-
.../transforms/sortedmerge/SortedMergeMeta.java | 2 +-
.../sortedmerge/messages/messages_en_US.properties | 1 +
.../sortedmerge/messages/messages_zh_CN.properties | 1 +
.../transforms/sortedmerge/RowSetRowTest.java | 59 ++++++
.../sortedmerge/SortedMergeMetaTest.java | 144 ++++++++++++++
.../transforms/sortedmerge/SortedMergeTest.java | 221 +++++++++++++++++++++
8 files changed, 444 insertions(+), 61 deletions(-)
diff --git
a/plugins/transforms/sortedmerge/src/main/java/org/apache/hop/pipeline/transforms/sortedmerge/RowSetRow.java
b/plugins/transforms/sortedmerge/src/main/java/org/apache/hop/pipeline/transforms/sortedmerge/RowSetRow.java
index b9b2413391..7fea2b4547 100644
---
a/plugins/transforms/sortedmerge/src/main/java/org/apache/hop/pipeline/transforms/sortedmerge/RowSetRow.java
+++
b/plugins/transforms/sortedmerge/src/main/java/org/apache/hop/pipeline/transforms/sortedmerge/RowSetRow.java
@@ -17,64 +17,17 @@
package org.apache.hop.pipeline.transforms.sortedmerge;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.Setter;
import org.apache.hop.core.IRowSet;
import org.apache.hop.core.row.IRowMeta;
+@Getter
+@Setter
+@AllArgsConstructor
public class RowSetRow {
private IRowSet rowSet;
private IRowMeta rowMeta;
private Object[] rowData;
-
- /**
- * @param rowSet
- * @param rowData
- */
- public RowSetRow(IRowSet rowSet, IRowMeta rowMeta, Object[] rowData) {
- super();
- this.rowSet = rowSet;
- this.rowMeta = rowMeta;
- this.rowData = rowData;
- }
-
- /**
- * @return the rowSet
- */
- public IRowSet getRowSet() {
- return rowSet;
- }
-
- /**
- * @param rowSet the rowSet to set
- */
- public void setRowSet(IRowSet rowSet) {
- this.rowSet = rowSet;
- }
-
- /**
- * @return the rowData
- */
- public Object[] getRowData() {
- return rowData;
- }
-
- /**
- * @param rowData the rowData to set
- */
- public void setRowData(Object[] rowData) {
- this.rowData = rowData;
- }
-
- /**
- * @return the rowMeta
- */
- public IRowMeta getRowMeta() {
- return rowMeta;
- }
-
- /**
- * @param rowMeta the rowMeta to set
- */
- public void setRowMeta(IRowMeta rowMeta) {
- this.rowMeta = rowMeta;
- }
}
diff --git
a/plugins/transforms/sortedmerge/src/main/java/org/apache/hop/pipeline/transforms/sortedmerge/SortedMerge.java
b/plugins/transforms/sortedmerge/src/main/java/org/apache/hop/pipeline/transforms/sortedmerge/SortedMerge.java
index 5606316fdc..e149aa26ee 100644
---
a/plugins/transforms/sortedmerge/src/main/java/org/apache/hop/pipeline/transforms/sortedmerge/SortedMerge.java
+++
b/plugins/transforms/sortedmerge/src/main/java/org/apache/hop/pipeline/transforms/sortedmerge/SortedMerge.java
@@ -22,6 +22,7 @@ import java.util.Collections;
import java.util.List;
import org.apache.hop.core.IRowSet;
import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.exception.HopRuntimeException;
import org.apache.hop.core.exception.HopTransformException;
import org.apache.hop.core.exception.HopValueException;
import org.apache.hop.i18n.BaseMessages;
@@ -69,8 +70,8 @@ public class SortedMerge extends
BaseTransform<SortedMergeMeta, SortedMergeData>
// now that we have all rows sorted, all we need to do is find out what
the smallest row is.
// The smallest row is the first in our case...
//
- RowSetRow smallestRow = data.sortedBuffer.get(0);
- data.sortedBuffer.remove(0);
+ RowSetRow smallestRow = data.sortedBuffer.getFirst();
+ data.sortedBuffer.removeFirst();
Object[] outputRowData = smallestRow.getRowData();
// We read another row from the row set where the smallest row came from.
@@ -164,8 +165,8 @@ public class SortedMerge extends
BaseTransform<SortedMergeMeta, SortedMergeData>
try {
return o1.getRowMeta().compare(o1.getRowData(), o2.getRowData(),
data.fieldIndices);
} catch (HopValueException e) {
- return 0; // TODO see if we should fire off alarms over here...
Perhaps throw a
- // HopRuntimeException.
+ throw new HopRuntimeException(
+ BaseMessages.getString(PKG,
"SortedMerge.Exception.ErrorComparingRows"), e);
}
};
@@ -177,13 +178,16 @@ public class SortedMerge extends
BaseTransform<SortedMergeMeta, SortedMergeData>
@Override
public boolean processRow() throws HopException {
- Object[] row = getRowSorted(); // get row, sorted
- if (row == null) { // no more input to be expected...
+ // get row, sorted
+ Object[] row = getRowSorted();
+ // no more input to be expected...
+ if (row == null) {
setOutputDone();
return false;
}
- putRow(data.rowMeta, row); // copy row to possible alternate rowset(s).
+ // copy row to possible alternate rowset(s).
+ putRow(data.rowMeta, row);
if (checkFeedback(getLinesRead()) && isBasic()) {
logBasic(BaseMessages.getString(PKG, "SortedMerge.Log.LineNumber") +
getLinesRead());
}
diff --git
a/plugins/transforms/sortedmerge/src/main/java/org/apache/hop/pipeline/transforms/sortedmerge/SortedMergeMeta.java
b/plugins/transforms/sortedmerge/src/main/java/org/apache/hop/pipeline/transforms/sortedmerge/SortedMergeMeta.java
index dbf7a35309..14db79b7c0 100644
---
a/plugins/transforms/sortedmerge/src/main/java/org/apache/hop/pipeline/transforms/sortedmerge/SortedMergeMeta.java
+++
b/plugins/transforms/sortedmerge/src/main/java/org/apache/hop/pipeline/transforms/sortedmerge/SortedMergeMeta.java
@@ -76,7 +76,7 @@ public class SortedMergeMeta extends
BaseTransformMeta<SortedMerge, SortedMergeD
int idx = inputRowMeta.indexOfValue(field.getFieldName());
if (idx >= 0) {
IValueMeta valueMeta = inputRowMeta.getValueMeta(idx);
- valueMeta.setSortedDescending(field.isAscending());
+ valueMeta.setSortedDescending(!field.isAscending());
}
}
}
diff --git
a/plugins/transforms/sortedmerge/src/main/resources/org/apache/hop/pipeline/transforms/sortedmerge/messages/messages_en_US.properties
b/plugins/transforms/sortedmerge/src/main/resources/org/apache/hop/pipeline/transforms/sortedmerge/messages/messages_en_US.properties
index d01b7c9ff2..ea9ea2057b 100644
---
a/plugins/transforms/sortedmerge/src/main/resources/org/apache/hop/pipeline/transforms/sortedmerge/messages/messages_en_US.properties
+++
b/plugins/transforms/sortedmerge/src/main/resources/org/apache/hop/pipeline/transforms/sortedmerge/messages/messages_en_US.properties
@@ -19,6 +19,7 @@ SortedMerge.Description=Sorted merge
SortedMerge.Injection.ASCENDING=Specify if the fields are ascending or not
(Y/N).
SortedMerge.Injection.FIELD_NAME=The name of the input field used for sorting.
SortedMerge.Injection.FIELDS=The names of the input fields that are being
merged.
+SortedMerge.Exception.ErrorComparingRows=Error comparing rows during sorted
merge
SortedMerge.Log.LineNumber=Linenr
SortedMerge.Name=Sorted merge
SortedMergeDialog.Ascending.Column=Ascending
diff --git
a/plugins/transforms/sortedmerge/src/main/resources/org/apache/hop/pipeline/transforms/sortedmerge/messages/messages_zh_CN.properties
b/plugins/transforms/sortedmerge/src/main/resources/org/apache/hop/pipeline/transforms/sortedmerge/messages/messages_zh_CN.properties
index 72ad3169a8..59384ac686 100644
---
a/plugins/transforms/sortedmerge/src/main/resources/org/apache/hop/pipeline/transforms/sortedmerge/messages/messages_zh_CN.properties
+++
b/plugins/transforms/sortedmerge/src/main/resources/org/apache/hop/pipeline/transforms/sortedmerge/messages/messages_zh_CN.properties
@@ -21,6 +21,7 @@
SortedMerge.Description=\u5408\u5E76\u6392\u5E8F\u540E\u7684\u6570\u636E\u96C6
SortedMerge.Injection.ASCENDING=\u6307\u5B9A\u5B57\u6BB5\u662F\u5426\u4E3A\u5347\u5E8F(Y/N).
SortedMerge.Injection.FIELD_NAME=\u8F93\u5165\u7528\u4E8E\u6392\u5E8F\u7684\u5B57\u6BB5\u540D\u79F0.
SortedMerge.Injection.FIELDS=\u6B63\u5728\u5408\u5E76\u7684\u8F93\u5165\u5B57\u6BB5\u540D\u79F0.
+SortedMerge.Exception.ErrorComparingRows=\u6392\u5E8F\u5408\u5E76\u65F6\u6BD4\u8F83\u884C\u5931\u8D25
SortedMerge.Log.LineNumber=Linenr
SortedMerge.Name=\u6392\u5E8F\u5408\u5E76
SortedMergeDialog.Ascending.Column=\u5347\u5E8F
diff --git
a/plugins/transforms/sortedmerge/src/test/java/org/apache/hop/pipeline/transforms/sortedmerge/RowSetRowTest.java
b/plugins/transforms/sortedmerge/src/test/java/org/apache/hop/pipeline/transforms/sortedmerge/RowSetRowTest.java
new file mode 100644
index 0000000000..43bd66abfb
--- /dev/null
+++
b/plugins/transforms/sortedmerge/src/test/java/org/apache/hop/pipeline/transforms/sortedmerge/RowSetRowTest.java
@@ -0,0 +1,59 @@
+/*
+ * 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.sortedmerge;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.mockito.Mockito.mock;
+
+import org.apache.hop.core.IRowSet;
+import org.apache.hop.core.row.IRowMeta;
+import org.junit.jupiter.api.Test;
+
+/** Unit test for {@link RowSetRow} */
+class RowSetRowTest {
+
+ @Test
+ void constructorStoresRowSetMetaAndData() {
+ IRowSet rowSet = mock(IRowSet.class);
+ IRowMeta rowMeta = mock(IRowMeta.class);
+ Object[] rowData = new Object[] {"a", 1};
+
+ RowSetRow row = new RowSetRow(rowSet, rowMeta, rowData);
+
+ assertSame(rowSet, row.getRowSet());
+ assertSame(rowMeta, row.getRowMeta());
+ assertArrayEquals(rowData, row.getRowData());
+ }
+
+ @Test
+ void settersReplaceStoredValues() {
+ RowSetRow row = new RowSetRow(mock(IRowSet.class), mock(IRowMeta.class),
new Object[] {"old"});
+
+ IRowSet rowSet = mock(IRowSet.class);
+ IRowMeta rowMeta = mock(IRowMeta.class);
+ Object[] rowData = new Object[] {"new"};
+ row.setRowSet(rowSet);
+ row.setRowMeta(rowMeta);
+ row.setRowData(rowData);
+
+ assertSame(rowSet, row.getRowSet());
+ assertSame(rowMeta, row.getRowMeta());
+ assertArrayEquals(rowData, row.getRowData());
+ }
+}
diff --git
a/plugins/transforms/sortedmerge/src/test/java/org/apache/hop/pipeline/transforms/sortedmerge/SortedMergeMetaTest.java
b/plugins/transforms/sortedmerge/src/test/java/org/apache/hop/pipeline/transforms/sortedmerge/SortedMergeMetaTest.java
index f5054a4884..125640025c 100644
---
a/plugins/transforms/sortedmerge/src/test/java/org/apache/hop/pipeline/transforms/sortedmerge/SortedMergeMetaTest.java
+++
b/plugins/transforms/sortedmerge/src/test/java/org/apache/hop/pipeline/transforms/sortedmerge/SortedMergeMetaTest.java
@@ -20,11 +20,36 @@ package org.apache.hop.pipeline.transforms.sortedmerge;
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 static org.mockito.Mockito.mock;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.hop.core.HopEnvironment;
+import org.apache.hop.core.ICheckResult;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.row.RowMeta;
+import org.apache.hop.core.row.value.ValueMetaString;
+import org.apache.hop.core.variables.Variables;
+import org.apache.hop.junit.rules.RestoreHopEngineEnvironmentExtension;
+import org.apache.hop.pipeline.PipelineMeta;
+import org.apache.hop.pipeline.PipelineMeta.PipelineType;
+import org.apache.hop.pipeline.transform.TransformMeta;
import org.apache.hop.pipeline.transform.TransformSerializationTestUtil;
+import
org.apache.hop.pipeline.transforms.sortedmerge.SortedMergeMeta.MergeField;
+import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
class SortedMergeMetaTest {
+
+ @RegisterExtension
+ static RestoreHopEngineEnvironmentExtension env = new
RestoreHopEngineEnvironmentExtension();
+
+ @BeforeAll
+ static void setUpClass() throws Exception {
+ HopEnvironment.init();
+ }
+
@Test
void testSerializationRoundTrip() throws Exception {
SortedMergeMeta meta =
@@ -37,4 +62,123 @@ class SortedMergeMetaTest {
assertEquals("field2", meta.getMergeFields().getLast().getFieldName());
assertFalse(meta.getMergeFields().getLast().isAscending());
}
+
+ @Test
+ void mergeFieldCopyConstructorCopiesValuesIndependently() {
+ MergeField original = new MergeField("key", true);
+ MergeField copy = new MergeField(original);
+
+ assertEquals("key", copy.getFieldName());
+ assertTrue(copy.isAscending());
+
+ copy.setFieldName("other");
+ copy.setAscending(false);
+ assertEquals("key", original.getFieldName());
+ assertTrue(original.isAscending());
+ }
+
+ @Test
+ void getSupportedPipelineTypesIsNormalOnly() {
+ PipelineType[] types = new SortedMergeMeta().getSupportedPipelineTypes();
+ assertEquals(1, types.length);
+ assertEquals(PipelineType.Normal, types[0]);
+ }
+
+ @Test
+ void getFieldsSetsSortedDescendingToMatchRuntime() throws Exception {
+ SortedMergeMeta meta = new SortedMergeMeta();
+ meta.getMergeFields().add(new MergeField("ascKey", true));
+ meta.getMergeFields().add(new MergeField("descKey", false));
+ meta.getMergeFields().add(new MergeField("missing", true));
+
+ RowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaString("ascKey"));
+ rowMeta.addValueMeta(new ValueMetaString("descKey"));
+ rowMeta.addValueMeta(new ValueMetaString("other"));
+
+ meta.getFields(rowMeta, "Sorted Merge", null, null, new Variables(), null);
+
+ assertFalse(rowMeta.searchValueMeta("ascKey").isSortedDescending());
+ assertTrue(rowMeta.searchValueMeta("descKey").isSortedDescending());
+ assertFalse(rowMeta.searchValueMeta("other").isSortedDescending());
+ }
+
+ @Test
+ void checkReportsErrorWhenNoPreviousFields() {
+ SortedMergeMeta meta = new SortedMergeMeta();
+ List<ICheckResult> remarks = check(meta, null, new String[] {"in"});
+
+ assertEquals(ICheckResult.TYPE_RESULT_ERROR, remarks.get(0).getType());
+ assertEquals(ICheckResult.TYPE_RESULT_OK, remarks.get(1).getType());
+ }
+
+ @Test
+ void checkReportsErrorWhenNoInputStreams() {
+ SortedMergeMeta meta = new SortedMergeMeta();
+ meta.getMergeFields().add(new MergeField("key", true));
+ List<ICheckResult> remarks = check(meta, stringMeta("key"), new String[0]);
+
+ assertTrue(hasType(remarks, ICheckResult.TYPE_RESULT_OK));
+ assertTrue(hasType(remarks, ICheckResult.TYPE_RESULT_ERROR));
+ }
+
+ @Test
+ void checkReportsOkWhenAllSortKeysArePresent() {
+ SortedMergeMeta meta = new SortedMergeMeta();
+ meta.getMergeFields().add(new MergeField("key", true));
+ List<ICheckResult> remarks = check(meta, stringMeta("key"), new String[]
{"in"});
+
+ assertEquals(3, remarks.size());
+ assertTrue(remarks.stream().allMatch(r -> r.getType() ==
ICheckResult.TYPE_RESULT_OK));
+ }
+
+ @Test
+ void checkReportsErrorWhenSortKeyIsMissing() {
+ SortedMergeMeta meta = new SortedMergeMeta();
+ meta.getMergeFields().add(new MergeField("missing", true));
+ List<ICheckResult> remarks = check(meta, stringMeta("key"), new String[]
{"in"});
+
+ assertTrue(
+ remarks.stream()
+ .anyMatch(
+ r ->
+ r.getType() == ICheckResult.TYPE_RESULT_ERROR
+ && r.getText().contains("missing")));
+ }
+
+ @Test
+ void checkReportsErrorWhenNoSortKeysEntered() {
+ SortedMergeMeta meta = new SortedMergeMeta();
+ List<ICheckResult> remarks = check(meta, stringMeta("key"), new String[]
{"in"});
+
+ assertTrue(hasType(remarks, ICheckResult.TYPE_RESULT_ERROR));
+ assertTrue(hasType(remarks, ICheckResult.TYPE_RESULT_OK));
+ }
+
+ private static List<ICheckResult> check(SortedMergeMeta meta, IRowMeta prev,
String[] input) {
+ List<ICheckResult> remarks = new ArrayList<>();
+ meta.check(
+ remarks,
+ mock(PipelineMeta.class),
+ mock(TransformMeta.class),
+ prev,
+ input,
+ new String[0],
+ null,
+ new Variables(),
+ null);
+ return remarks;
+ }
+
+ private static IRowMeta stringMeta(String... names) {
+ RowMeta rowMeta = new RowMeta();
+ for (String name : names) {
+ rowMeta.addValueMeta(new ValueMetaString(name));
+ }
+ return rowMeta;
+ }
+
+ private static boolean hasType(List<ICheckResult> remarks, int type) {
+ return remarks.stream().anyMatch(r -> r.getType() == type);
+ }
}
diff --git
a/plugins/transforms/sortedmerge/src/test/java/org/apache/hop/pipeline/transforms/sortedmerge/SortedMergeTest.java
b/plugins/transforms/sortedmerge/src/test/java/org/apache/hop/pipeline/transforms/sortedmerge/SortedMergeTest.java
new file mode 100644
index 0000000000..54f489da24
--- /dev/null
+++
b/plugins/transforms/sortedmerge/src/test/java/org/apache/hop/pipeline/transforms/sortedmerge/SortedMergeTest.java
@@ -0,0 +1,221 @@
+/*
+ * 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.sortedmerge;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.hop.core.HopEnvironment;
+import org.apache.hop.core.IRowSet;
+import org.apache.hop.core.QueueRowSet;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.exception.HopRuntimeException;
+import org.apache.hop.core.exception.HopTransformException;
+import org.apache.hop.core.exception.HopValueException;
+import org.apache.hop.core.logging.ILoggingObject;
+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.ValueMetaString;
+import org.apache.hop.junit.rules.RestoreHopEngineEnvironmentExtension;
+import org.apache.hop.pipeline.transforms.mock.TransformMockHelper;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+/** Unit test for {@link SortedMerge} */
+class SortedMergeTest {
+
+ @RegisterExtension
+ static RestoreHopEngineEnvironmentExtension env = new
RestoreHopEngineEnvironmentExtension();
+
+ private TransformMockHelper<SortedMergeMeta, SortedMergeData> mockHelper;
+
+ @BeforeAll
+ static void setUpClass() throws HopException {
+ HopEnvironment.init();
+ }
+
+ @BeforeEach
+ void setUp() {
+ mockHelper =
+ new TransformMockHelper<>("Sorted Merge", SortedMergeMeta.class,
SortedMergeData.class);
+ when(mockHelper.logChannelFactory.create(any(), any(ILoggingObject.class)))
+ .thenReturn(mockHelper.iLogChannel);
+ when(mockHelper.pipeline.isRunning()).thenReturn(true);
+ }
+
+ @AfterEach
+ void tearDown() {
+ mockHelper.cleanUp();
+ }
+
+ @Test
+ void processRowMergesTwoSortedStreamsByKey() throws Exception {
+ IRowMeta rowMeta = stringKeyMeta();
+ SortedMerge transform =
+ createTransform(
+ rowMeta,
+ new Object[][] {{"alpha"}, {"delta"}},
+ new Object[][] {{"bravo"}, {"charlie"}});
+
+ List<Object[]> rows = runToCompletion(transform);
+
+ assertEquals(List.of("alpha", "bravo", "charlie", "delta"), keys(rows));
+ }
+
+ @Test
+ void processRowKeepsEqualKeysWhenCompareSucceeds() throws Exception {
+ IRowMeta rowMeta = stringKeyMeta();
+ SortedMerge transform =
+ createTransform(rowMeta, new Object[][] {{"same"}}, new Object[][]
{{"same"}});
+
+ List<Object[]> rows = runToCompletion(transform);
+
+ assertEquals(List.of("same", "same"), keys(rows));
+ }
+
+ @Test
+ void processRowReturnsFalseWhenAllStreamsAreEmpty() throws Exception {
+ SortedMerge transform = createTransform(stringKeyMeta(), new Object[][]
{}, new Object[][] {});
+
+ assertFalse(transform.processRow());
+ assertTrue(drain(transform.getOutputRowSets().getFirst()).isEmpty());
+ }
+
+ @Test
+ void processRowFailsWhenSortKeyIsMissing() {
+ IRowMeta rowMeta = stringKeyMeta();
+ SortedMergeMeta meta = new SortedMergeMeta();
+ meta.getMergeFields().add(new SortedMergeMeta.MergeField("missing", true));
+
+ SortedMerge transform = createTransform(meta, rowMeta, new Object[][]
{{"left"}});
+
+ HopTransformException thrown = assertThrows(HopTransformException.class,
transform::processRow);
+ assertTrue(thrown.getMessage().contains("missing"));
+ }
+
+ @Test
+ void processRowFailsWhenCompareThrowsHopValueException() throws Exception {
+ RowMeta rowMeta = spy(new RowMeta());
+ rowMeta.addValueMeta(new ValueMetaString("key"));
+ doThrow(new HopValueException("compare failed"))
+ .when(rowMeta)
+ .compare(any(Object[].class), any(Object[].class), any(int[].class));
+
+ SortedMerge transform =
+ createTransform(rowMeta, new Object[][] {{"left"}}, new Object[][]
{{"right"}});
+
+ HopRuntimeException thrown = assertThrows(HopRuntimeException.class,
transform::processRow);
+ assertInstanceOf(HopValueException.class, thrown.getCause());
+ }
+
+ @Test
+ void processRowFailsWhenValueMetaCompareThrows() {
+ RowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(throwingKeyMeta());
+
+ SortedMerge transform =
+ createTransform(rowMeta, new Object[][] {{"left"}}, new Object[][]
{{"right"}});
+
+ HopRuntimeException thrown = assertThrows(HopRuntimeException.class,
transform::processRow);
+ assertInstanceOf(HopValueException.class, thrown.getCause());
+ }
+
+ private SortedMerge createTransform(IRowMeta rowMeta, Object[][]... streams)
{
+ SortedMergeMeta meta = new SortedMergeMeta();
+ meta.getMergeFields().add(new SortedMergeMeta.MergeField("key", true));
+ return createTransform(meta, rowMeta, streams);
+ }
+
+ private SortedMerge createTransform(
+ SortedMergeMeta meta, IRowMeta rowMeta, Object[][]... streams) {
+ SortedMerge transform =
+ new SortedMerge(
+ mockHelper.transformMeta,
+ meta,
+ new SortedMergeData(),
+ 0,
+ mockHelper.pipelineMeta,
+ mockHelper.pipeline);
+ transform.init();
+ transform.setInputRowMeta(rowMeta);
+
+ for (Object[][] rows : streams) {
+ QueueRowSet input = new QueueRowSet();
+ for (Object[] row : rows) {
+ input.putRow(rowMeta, row);
+ }
+ input.setDone();
+ transform.addRowSetToInputRowSets(input);
+ }
+ transform.addRowSetToOutputRowSets(new QueueRowSet());
+ return transform;
+ }
+
+ private static IRowMeta stringKeyMeta() {
+ RowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaString("key"));
+ return rowMeta;
+ }
+
+ private static IValueMeta throwingKeyMeta() {
+ return new ValueMetaString("key") {
+ @Override
+ public int compare(Object data1, Object data2) throws HopValueException {
+ throw new HopValueException("compare failed");
+ }
+ };
+ }
+
+ private static List<Object[]> runToCompletion(SortedMerge transform) throws
HopException {
+ while (transform.processRow()) {
+ // drain all merged rows
+ }
+ return drain(transform.getOutputRowSets().getFirst());
+ }
+
+ private static List<Object[]> drain(IRowSet output) {
+ List<Object[]> rows = new ArrayList<>();
+ Object[] row = output.getRow();
+ while (row != null) {
+ rows.add(row);
+ row = output.getRow();
+ }
+ return rows;
+ }
+
+ private static List<String> keys(List<Object[]> rows) {
+ List<String> keys = new ArrayList<>(rows.size());
+ for (Object[] row : rows) {
+ keys.add((String) row[0]);
+ }
+ return keys;
+ }
+}
