This is an automated email from the ASF dual-hosted git repository.
bamaer 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 18df7242ee Issue #5193 : Accept sequence start, end, and increment
from one info row (#8696)
18df7242ee is described below
commit 18df7242ee4d5211c81a38c3bdbdced2e46071b6
Author: Matt Casters <[email protected]>
AuthorDate: Thu Oct 1 15:38:10 2026 +0200
Issue #5193 : Accept sequence start, end, and increment from one info row
(#8696)
* Issue #5193 : Accept sequence start, end, and increment from one info row
* Issue #5193 : Report a distributed configuration row when Add Sequence
runs in multiple copies
---------
Co-authored-by: Bart Maertens <[email protected]>
---
.../pages/pipeline/transforms/addsequence.adoc | 12 +-
.../transforms/addsequence/AddSequence.java | 152 ++++++++++++--
.../transforms/addsequence/AddSequenceDialog.java | 167 ++++++++++++++-
.../transforms/addsequence/AddSequenceMeta.java | 214 +++++++++++++++++++
.../addsequence/messages/messages_en_US.properties | 29 +++
.../addsequence/AddSequenceMetaTest.java | 230 +++++++++++++++++++-
.../transforms/addsequence/AddSequenceTest.java | 231 +++++++++++++++++++++
7 files changed, 1005 insertions(+), 30 deletions(-)
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/addsequence.adoc
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/addsequence.adoc
index 5a1e071543..fbfb1167ec 100644
---
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/addsequence.adoc
+++
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/addsequence.adoc
@@ -31,6 +31,8 @@ You can either use a database sequence (if supported, e.g.
Oracle, PostgreSQL) t
Hop-generated sequence values are unique when used in the same pipeline, and
return to the same starting value each time you run the pipeline. Hop sequences
are unique only when used in the same pipeline.
Also, they are not stored, so the values start back at the same value every
time the pipeline is launched.
+The start, increment, and maximum can be typed in, or read from another
transform. Connect that transform to Add Sequence and select it as the
configuration transform. It must return exactly one row. More than one row
stops the pipeline with an error. The hop is an info hop, so that row is not
part of the main stream. The end field is the maximum value: after it, the
sequence returns to the start value. The configuration transform cannot run in
multiple copies. If Add Sequence itself run [...]
+
== Options
[options="header"]
@@ -46,7 +48,11 @@ This name must be unique within a single pipeline.
|Use a pipeline counter to generate the sequence|Enable if you want the
sequence to be generated by Hop, then set these parameters: Counter name
(optional), Start at, Increment by, Maximum value.
|Counter name (optional)|If multiple transforms in a pipeline generate the
same value name, this option enables you to specify the name of the counter to
associate with.
Avoids forcing unique sequencing across multiple transforms.
-|Start at|The value to begin the sequence with.
-|Increment by|The amount by which the sequence increases or decreases.
-|Maximum value|The value after which the sequence returns to the Start At
value.
+|Start at|The value to begin the sequence with. Not used when a configuration
transform is selected.
+|Increment by|The amount by which the sequence increases or decreases. Not
used when a configuration transform is selected.
+|Maximum value|The value after which the sequence returns to the Start At
value. Not used when a configuration transform is selected.
+|Configuration transform|Optional transform that provides exactly one row with
the start, end, and increment. Leave this empty to use Start at, Increment by,
and Maximum value.
+|Start field|Field in the configuration row that contains the value to begin
the sequence with.
+|End field|Field in the configuration row that contains the maximum value.
After this value the sequence returns to the start value.
+|Increment field|Field in the configuration row that contains the amount by
which the sequence increases or decreases.
|===
\ No newline at end of file
diff --git
a/plugins/transforms/addsequence/src/main/java/org/apache/hop/pipeline/transforms/addsequence/AddSequence.java
b/plugins/transforms/addsequence/src/main/java/org/apache/hop/pipeline/transforms/addsequence/AddSequence.java
index 55a6c3fd49..4b98db9dc5 100644
---
a/plugins/transforms/addsequence/src/main/java/org/apache/hop/pipeline/transforms/addsequence/AddSequence.java
+++
b/plugins/transforms/addsequence/src/main/java/org/apache/hop/pipeline/transforms/addsequence/AddSequence.java
@@ -17,13 +17,16 @@
package org.apache.hop.pipeline.transforms.addsequence;
+import org.apache.hop.core.Const;
import org.apache.hop.core.Counter;
import org.apache.hop.core.Counters;
+import org.apache.hop.core.IRowSet;
import org.apache.hop.core.database.Database;
import org.apache.hop.core.database.DatabaseMeta;
import org.apache.hop.core.exception.HopDatabaseException;
import org.apache.hop.core.exception.HopException;
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.RowDataUtil;
import org.apache.hop.core.util.Utils;
@@ -93,6 +96,19 @@ public class AddSequence extends
BaseTransform<AddSequenceMeta, AddSequenceData>
@Override
public boolean processRow() throws HopException {
+ // Read the info row before the main input so its rowset is removed and
cannot be mixed in.
+ if (first && meta.isConfigurationFromTransform()) {
+ try {
+ readConfigurationRow();
+ } catch (HopException e) {
+ logError(BaseMessages.getString(PKG,
"AddSequence.Log.ErrorInTransform") + e.getMessage());
+ setErrors(1);
+ stopAll();
+ setOutputDone();
+ return false;
+ }
+ }
+
// Get row from input rowset & set row busy!
Object[] r = getRow();
if (r == null) {
@@ -173,6 +189,17 @@ public class AddSequence extends
BaseTransform<AddSequenceMeta, AddSequenceData>
+ dbe.getMessage());
}
} else if (meta.isCounterUsed()) {
+ if (meta.isConfigurationFromTransform()) {
+ // The counter is created from the info row on the first processRow
call.
+ if (Utils.isEmpty(Const.trim(resolve(meta.getStartField())))
+ || Utils.isEmpty(Const.trim(resolve(meta.getEndField())))
+ || Utils.isEmpty(Const.trim(resolve(meta.getIncrementField()))))
{
+ logError(BaseMessages.getString(PKG,
"AddSequence.Log.ConfigurationFieldsMissing"));
+ return false;
+ }
+ return true;
+ }
+
// Do the environment translations of the counter values.
boolean doAbort = false;
try {
@@ -221,26 +248,7 @@ public class AddSequence extends
BaseTransform<AddSequenceMeta, AddSequenceData>
return false;
}
- String realCounterName = resolve(meta.getCounterName());
- if (!Utils.isEmpty(realCounterName)) {
- data.setLookup(lookupCounterName(realCounterName));
- } else {
- data.setLookup(lookupCounterName(meta.getValueName()));
- }
-
- // We need to synchronize over the whole pipeline to make sure that we
always get the same
- // counter
- // regardless of the number of transform copies asking for it.
- //
- synchronized (getPipeline()) {
- if (isDetailed()) {
- logDetailed("init counter name: {0}", data.getLookup());
- }
- data.counter =
- Counters.getInstance()
- .getOrUpdateCounter(
- data.getLookup(), new Counter(data.start,
data.increment, data.maximum));
- }
+ createCounter(data.start, data.increment, data.maximum);
return true;
} else {
logError(
@@ -275,6 +283,110 @@ public class AddSequence extends
BaseTransform<AddSequenceMeta, AddSequenceData>
super.cleanup();
}
+ /**
+ * Read the single configuration row and create the counter from its start,
end, and increment
+ * fields. Zero rows or more than one row is an error.
+ */
+ private void readConfigurationRow() throws HopException {
+ String sourceName = meta.getConfigurationTransform();
+ IRowSet rowSet = findInputRowSet(sourceName);
+ if (rowSet == null) {
+ throw new HopTransformException(
+ BaseMessages.getString(
+ PKG, "AddSequence.Exception.ConfigurationTransformNotFound",
sourceName));
+ }
+
+ Object[] row = getRowFrom(rowSet);
+ if (row == null) {
+ // Distributed over several copies of this transform, the single row
only reaches one of them.
+ String key =
+ getTransformMeta().getCopies(this) > 1
+ ? "AddSequence.Exception.ConfigurationRowMissingInCopy"
+ : "AddSequence.Exception.ConfigurationRowMissing";
+ throw new HopTransformException(
+ BaseMessages.getString(PKG, key, sourceName,
Integer.toString(getCopy())));
+ }
+ Object[] extra = getRowFrom(rowSet);
+ if (extra != null) {
+ throw new HopTransformException(
+ BaseMessages.getString(
+ PKG, "AddSequence.Exception.ConfigurationRowNotSingle",
sourceName));
+ }
+
+ IRowMeta rowMeta = rowSet.getRowMeta();
+ if (rowMeta == null) {
+ throw new HopTransformException(
+ BaseMessages.getString(
+ PKG, "AddSequence.Exception.ConfigurationRowHasNoFields",
sourceName));
+ }
+
+ long start = readConfigurationValue(rowMeta, row,
resolve(meta.getStartField()));
+ long end = readConfigurationValue(rowMeta, row,
resolve(meta.getEndField()));
+ long increment = readConfigurationValue(rowMeta, row,
resolve(meta.getIncrementField()));
+ if (isDetailed()) {
+ logDetailed(
+ BaseMessages.getString(
+ PKG, "AddSequence.Log.ReadConfiguration", sourceName, start,
end, increment));
+ }
+ createCounter(start, increment, end);
+ }
+
+ private long readConfigurationValue(IRowMeta rowMeta, Object[] row, String
fieldName)
+ throws HopException {
+ String name = Const.trim(fieldName);
+ int index = Utils.isEmpty(name) ? -1 : rowMeta.indexOfValue(name);
+ if (index < 0) {
+ throw new HopTransformException(
+ BaseMessages.getString(PKG,
"AddSequence.Exception.ConfigurationFieldNotFound", name));
+ }
+ try {
+ Long value = rowMeta.getValueMeta(index).getInteger(row[index]);
+ if (value == null) {
+ throw new HopTransformException(
+ BaseMessages.getString(PKG,
"AddSequence.Exception.ConfigurationFieldIsNull", name));
+ }
+ return value;
+ } catch (HopTransformException e) {
+ throw e;
+ } catch (HopValueException e) {
+ throw new HopTransformException(
+ BaseMessages.getString(
+ PKG,
+ "AddSequence.Exception.ConfigurationFieldNotNumeric",
+ name,
+ String.valueOf(row[index]),
+ e.getMessage()),
+ e);
+ }
+ }
+
+ /**
+ * One counter is shared by every copy of this transform. Synchronize on the
pipeline so the first
+ * copy creates it and the others reuse it.
+ */
+ private void createCounter(long start, long increment, long maximum) {
+ data.start = start;
+ data.increment = increment;
+ data.maximum = maximum;
+
+ String realCounterName = resolve(meta.getCounterName());
+ if (!Utils.isEmpty(realCounterName)) {
+ data.setLookup(lookupCounterName(realCounterName));
+ } else {
+ data.setLookup(lookupCounterName(meta.getValueName()));
+ }
+
+ synchronized (getPipeline()) {
+ if (isDetailed()) {
+ logDetailed("init counter name: {0}", data.getLookup());
+ }
+ data.counter =
+ Counters.getInstance()
+ .getOrUpdateCounter(
+ data.getLookup(), new Counter(data.start, data.increment,
data.maximum));
+ }
+ }
+
/**
* Build a unique identifier for this pipeline run.
*
diff --git
a/plugins/transforms/addsequence/src/main/java/org/apache/hop/pipeline/transforms/addsequence/AddSequenceDialog.java
b/plugins/transforms/addsequence/src/main/java/org/apache/hop/pipeline/transforms/addsequence/AddSequenceDialog.java
index 22f6a971f5..2b3a57ccfd 100644
---
a/plugins/transforms/addsequence/src/main/java/org/apache/hop/pipeline/transforms/addsequence/AddSequenceDialog.java
+++
b/plugins/transforms/addsequence/src/main/java/org/apache/hop/pipeline/transforms/addsequence/AddSequenceDialog.java
@@ -20,6 +20,8 @@ package org.apache.hop.pipeline.transforms.addsequence;
import org.apache.hop.core.Const;
import org.apache.hop.core.database.Database;
import org.apache.hop.core.database.DatabaseMeta;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.row.IRowMeta;
import org.apache.hop.core.util.Utils;
import org.apache.hop.core.variables.IVariables;
import org.apache.hop.i18n.BaseMessages;
@@ -44,6 +46,7 @@ import org.eclipse.swt.layout.FormAttachment;
import org.eclipse.swt.layout.FormData;
import org.eclipse.swt.layout.FormLayout;
import org.eclipse.swt.widgets.Button;
+import org.eclipse.swt.widgets.Combo;
import org.eclipse.swt.widgets.Composite;
import org.eclipse.swt.widgets.Group;
import org.eclipse.swt.widgets.Label;
@@ -83,6 +86,18 @@ public class AddSequenceDialog extends BaseTransformDialog {
private Label wlMaxVal;
private TextVar wMaxVal;
+ private Label wlConfigTransform;
+ private Combo wConfigTransform;
+
+ private Label wlStartField;
+ private Combo wStartField;
+
+ private Label wlEndField;
+ private Combo wEndField;
+
+ private Label wlIncrementField;
+ private Combo wIncrementField;
+
private final AddSequenceMeta input;
public AddSequenceDialog(
@@ -354,6 +369,95 @@ public class AddSequenceDialog extends BaseTransformDialog
{
fdMaxVal.right = new FormAttachment(100, 0);
wMaxVal.setLayoutData(fdMaxVal);
+ wlConfigTransform = new Label(gCounter, SWT.RIGHT);
+ wlConfigTransform.setText(
+ BaseMessages.getString(PKG,
"AddSequenceDialog.ConfigurationTransform.Label"));
+ PropsUi.setLook(wlConfigTransform);
+ wlConfigTransform.setToolTipText(
+ BaseMessages.getString(PKG,
"AddSequenceDialog.ConfigurationTransform.Tooltip"));
+ FormData fdlConfigTransform = new FormData();
+ fdlConfigTransform.left = new FormAttachment(0, 0);
+ fdlConfigTransform.right = new FormAttachment(middle, -margin);
+ fdlConfigTransform.top = new FormAttachment(wMaxVal, margin);
+ wlConfigTransform.setLayoutData(fdlConfigTransform);
+ wConfigTransform = new Combo(gCounter, SWT.SINGLE | SWT.LEFT | SWT.BORDER);
+ PropsUi.setLook(wConfigTransform);
+ wConfigTransform.setToolTipText(
+ BaseMessages.getString(PKG,
"AddSequenceDialog.ConfigurationTransform.Tooltip"));
+
wConfigTransform.setItems(pipelineMeta.getPrevTransformNames(transformName));
+ FormData fdConfigTransform = new FormData();
+ fdConfigTransform.left = new FormAttachment(middle, 0);
+ fdConfigTransform.top = new FormAttachment(wMaxVal, margin);
+ fdConfigTransform.right = new FormAttachment(100, 0);
+ wConfigTransform.setLayoutData(fdConfigTransform);
+
+ wlStartField = new Label(gCounter, SWT.RIGHT);
+ wlStartField.setText(BaseMessages.getString(PKG,
"AddSequenceDialog.StartField.Label"));
+ PropsUi.setLook(wlStartField);
+ wlStartField.setToolTipText(
+ BaseMessages.getString(PKG, "AddSequenceDialog.StartField.Tooltip"));
+ FormData fdlStartField = new FormData();
+ fdlStartField.left = new FormAttachment(0, 0);
+ fdlStartField.right = new FormAttachment(middle, -margin);
+ fdlStartField.top = new FormAttachment(wConfigTransform, margin);
+ wlStartField.setLayoutData(fdlStartField);
+ wStartField = new Combo(gCounter, SWT.SINGLE | SWT.LEFT | SWT.BORDER);
+ PropsUi.setLook(wStartField);
+ wStartField.setToolTipText(BaseMessages.getString(PKG,
"AddSequenceDialog.StartField.Tooltip"));
+ wStartField.addModifyListener(lsMod);
+ FormData fdStartField = new FormData();
+ fdStartField.left = new FormAttachment(middle, 0);
+ fdStartField.top = new FormAttachment(wConfigTransform, margin);
+ fdStartField.right = new FormAttachment(100, 0);
+ wStartField.setLayoutData(fdStartField);
+
+ wlEndField = new Label(gCounter, SWT.RIGHT);
+ wlEndField.setText(BaseMessages.getString(PKG,
"AddSequenceDialog.EndField.Label"));
+ PropsUi.setLook(wlEndField);
+ wlEndField.setToolTipText(BaseMessages.getString(PKG,
"AddSequenceDialog.EndField.Tooltip"));
+ FormData fdlEndField = new FormData();
+ fdlEndField.left = new FormAttachment(0, 0);
+ fdlEndField.right = new FormAttachment(middle, -margin);
+ fdlEndField.top = new FormAttachment(wStartField, margin);
+ wlEndField.setLayoutData(fdlEndField);
+ wEndField = new Combo(gCounter, SWT.SINGLE | SWT.LEFT | SWT.BORDER);
+ PropsUi.setLook(wEndField);
+ wEndField.setToolTipText(BaseMessages.getString(PKG,
"AddSequenceDialog.EndField.Tooltip"));
+ wEndField.addModifyListener(lsMod);
+ FormData fdEndField = new FormData();
+ fdEndField.left = new FormAttachment(middle, 0);
+ fdEndField.top = new FormAttachment(wStartField, margin);
+ fdEndField.right = new FormAttachment(100, 0);
+ wEndField.setLayoutData(fdEndField);
+
+ wlIncrementField = new Label(gCounter, SWT.RIGHT);
+ wlIncrementField.setText(BaseMessages.getString(PKG,
"AddSequenceDialog.IncrementField.Label"));
+ PropsUi.setLook(wlIncrementField);
+ wlIncrementField.setToolTipText(
+ BaseMessages.getString(PKG,
"AddSequenceDialog.IncrementField.Tooltip"));
+ FormData fdlIncrementField = new FormData();
+ fdlIncrementField.left = new FormAttachment(0, 0);
+ fdlIncrementField.right = new FormAttachment(middle, -margin);
+ fdlIncrementField.top = new FormAttachment(wEndField, margin);
+ wlIncrementField.setLayoutData(fdlIncrementField);
+ wIncrementField = new Combo(gCounter, SWT.SINGLE | SWT.LEFT | SWT.BORDER);
+ PropsUi.setLook(wIncrementField);
+ wIncrementField.setToolTipText(
+ BaseMessages.getString(PKG,
"AddSequenceDialog.IncrementField.Tooltip"));
+ wIncrementField.addModifyListener(lsMod);
+ FormData fdIncrementField = new FormData();
+ fdIncrementField.left = new FormAttachment(middle, 0);
+ fdIncrementField.top = new FormAttachment(wEndField, margin);
+ fdIncrementField.right = new FormAttachment(100, 0);
+ wIncrementField.setLayoutData(fdIncrementField);
+
+ wConfigTransform.addModifyListener(
+ e -> {
+ input.setChanged();
+ enableFields();
+ });
+ wConfigTransform.addListener(SWT.Selection, e ->
refreshConfigurationFields());
+
wContent.pack();
Rectangle bounds = wContent.getBounds();
sc.setContent(wContent);
@@ -380,17 +484,58 @@ public class AddSequenceDialog extends
BaseTransformDialog {
wlSeqname.setEnabled(useDatabase);
wSeqname.setEnabled(useDatabase);
+ boolean fromTransform = useCounter &&
!Utils.isEmpty(wConfigTransform.getText());
+
wlCounterName.setEnabled(useCounter);
wCounterName.setEnabled(useCounter);
- wlStartAt.setEnabled(useCounter);
- wStartAt.setEnabled(useCounter);
- wlIncrBy.setEnabled(useCounter);
- wIncrBy.setEnabled(useCounter);
- wlMaxVal.setEnabled(useCounter);
- wMaxVal.setEnabled(useCounter);
+ wlStartAt.setEnabled(useCounter && !fromTransform);
+ wStartAt.setEnabled(useCounter && !fromTransform);
+ wlIncrBy.setEnabled(useCounter && !fromTransform);
+ wIncrBy.setEnabled(useCounter && !fromTransform);
+ wlMaxVal.setEnabled(useCounter && !fromTransform);
+ wMaxVal.setEnabled(useCounter && !fromTransform);
+ wlConfigTransform.setEnabled(useCounter);
+ wConfigTransform.setEnabled(useCounter);
+ wlStartField.setEnabled(fromTransform);
+ wStartField.setEnabled(fromTransform);
+ wlEndField.setEnabled(fromTransform);
+ wEndField.setEnabled(fromTransform);
+ wlIncrementField.setEnabled(fromTransform);
+ wIncrementField.setEnabled(fromTransform);
activeSequence();
}
+ /**
+ * Fill the start, end, and increment combos from the configuration
transform. A failure to
+ * calculate those fields leaves the typed names in place.
+ */
+ private void refreshConfigurationFields() {
+ if (wStartField == null || wStartField.isDisposed()) {
+ return;
+ }
+ String start = wStartField.getText();
+ String end = wEndField.getText();
+ String increment = wIncrementField.getText();
+ String[] names = new String[0];
+ try {
+ String source = wConfigTransform.getText();
+ if (!Utils.isEmpty(source)) {
+ IRowMeta rowMeta = pipelineMeta.getTransformFields(variables, source);
+ if (rowMeta != null) {
+ names = rowMeta.getFieldNames();
+ }
+ }
+ } catch (HopException e) {
+ log.logDebug("Unable to read fields from the configuration transform: "
+ e.getMessage());
+ }
+ wStartField.setItems(names);
+ wEndField.setItems(names);
+ wIncrementField.setItems(names);
+ wStartField.setText(Const.NVL(start, ""));
+ wEndField.setText(Const.NVL(end, ""));
+ wIncrementField.setText(Const.NVL(increment, ""));
+ }
+
/** Copy information from the meta-data input to the dialog fields. */
public void getData() {
logDebug(BaseMessages.getString(PKG,
"AddSequenceDialog.Log.GettingKeyInfo"));
@@ -417,6 +562,11 @@ public class AddSequenceDialog extends BaseTransformDialog
{
wStartAt.setText(input.getStartAt());
wIncrBy.setText(input.getIncrementBy());
wMaxVal.setText(input.getMaxValue());
+ wConfigTransform.setText(Const.NVL(input.getConfigurationTransform(), ""));
+ wStartField.setText(Const.NVL(input.getStartField(), ""));
+ wEndField.setText(Const.NVL(input.getEndField(), ""));
+ wIncrementField.setText(Const.NVL(input.getIncrementField(), ""));
+ refreshConfigurationFields();
enableFields();
}
@@ -446,6 +596,11 @@ public class AddSequenceDialog extends BaseTransformDialog
{
input.setStartAt(wStartAt.getText());
input.setIncrementBy(wIncrBy.getText());
input.setMaxValue(wMaxVal.getText());
+ input.setConfigurationTransform(wConfigTransform.getText());
+ input.setStartField(wStartField.getText());
+ input.setEndField(wEndField.getText());
+ input.setIncrementField(wIncrementField.getText());
+ input.searchInfoAndTargetTransforms(pipelineMeta.getTransforms());
if (input.isDatabaseUsed()
&& pipelineMeta.findDatabase(wConnection.getText(), variables) ==
null) {
diff --git
a/plugins/transforms/addsequence/src/main/java/org/apache/hop/pipeline/transforms/addsequence/AddSequenceMeta.java
b/plugins/transforms/addsequence/src/main/java/org/apache/hop/pipeline/transforms/addsequence/AddSequenceMeta.java
index ade6e663d5..e9221e163a 100644
---
a/plugins/transforms/addsequence/src/main/java/org/apache/hop/pipeline/transforms/addsequence/AddSequenceMeta.java
+++
b/plugins/transforms/addsequence/src/main/java/org/apache/hop/pipeline/transforms/addsequence/AddSequenceMeta.java
@@ -40,7 +40,13 @@ import org.apache.hop.metadata.api.IHopMetadataProvider;
import org.apache.hop.metadata.api.IOptionalDatabaseConnection;
import org.apache.hop.pipeline.PipelineMeta;
import org.apache.hop.pipeline.transform.BaseTransformMeta;
+import org.apache.hop.pipeline.transform.ITransformIOMeta;
+import org.apache.hop.pipeline.transform.TransformIOMeta;
import org.apache.hop.pipeline.transform.TransformMeta;
+import org.apache.hop.pipeline.transform.stream.IStream;
+import org.apache.hop.pipeline.transform.stream.IStream.StreamType;
+import org.apache.hop.pipeline.transform.stream.Stream;
+import org.apache.hop.pipeline.transform.stream.StreamIcon;
/** Meta data for the Add Sequence transform. */
@Transform(
@@ -109,6 +115,35 @@ public class AddSequenceMeta extends
BaseTransformMeta<AddSequence, AddSequenceD
injectionKeyDescription = "AddSequenceMeta.Injection.MaxValue")
private String maxValue;
+ /** Info transform that provides one row with the counter start, end, and
increment. */
+ @HopMetadataProperty(
+ key = "configuration_transform",
+ injectionKeyDescription =
"AddSequenceMeta.Injection.ConfigurationTransform")
+ private String configurationTransform;
+
+ @HopMetadataProperty(
+ key = "start_field",
+ injectionKeyDescription = "AddSequenceMeta.Injection.StartField")
+ private String startField;
+
+ @HopMetadataProperty(
+ key = "end_field",
+ injectionKeyDescription = "AddSequenceMeta.Injection.EndField")
+ private String endField;
+
+ @HopMetadataProperty(
+ key = "increment_field",
+ injectionKeyDescription = "AddSequenceMeta.Injection.IncrementField")
+ private String incrementField;
+
+ /**
+ * Counter values come from one row of {@link #configurationTransform}
instead of the typed start,
+ * increment, and maximum.
+ */
+ public boolean isConfigurationFromTransform() {
+ return counterUsed && !databaseUsed &&
!Utils.isEmpty(configurationTransform);
+ }
+
/**
* @param maxValue The maxValue to set.
*/
@@ -200,6 +235,105 @@ public class AddSequenceMeta extends
BaseTransformMeta<AddSequence, AddSequenceD
transformMeta);
remarks.add(cr);
}
+
+ checkConfigurationTransform(remarks, pipelineMeta, transformMeta, info,
variables);
+ }
+
+ /**
+ * The configuration transform is optional. When it is set, the start, end,
and increment field
+ * names have to be set as well, and the transform has to exist. Running in
several copies needs
+ * the configuration row copied to every copy: distributed, it only reaches
the first one.
+ */
+ private void checkConfigurationTransform(
+ List<ICheckResult> remarks,
+ PipelineMeta pipelineMeta,
+ TransformMeta transformMeta,
+ IRowMeta info,
+ IVariables variables) {
+ if (!isConfigurationFromTransform()) {
+ return;
+ }
+
+ boolean fieldsMissing =
+ Utils.isEmpty(startField) || Utils.isEmpty(endField) ||
Utils.isEmpty(incrementField);
+ if (fieldsMissing) {
+ remarks.add(
+ new CheckResult(
+ ICheckResult.TYPE_RESULT_ERROR,
+ BaseMessages.getString(PKG,
"AddSequenceMeta.CheckResult.ConfigurationFieldsMissing"),
+ transformMeta));
+ }
+
+ if (pipelineMeta != null) {
+ TransformMeta source =
pipelineMeta.findTransform(configurationTransform);
+ if (source != null
+ && source.isDistributes()
+ && transformMeta != null
+ && transformMeta.getCopies(variables) > 1) {
+ remarks.add(
+ new CheckResult(
+ ICheckResult.TYPE_RESULT_ERROR,
+ BaseMessages.getString(
+ PKG,
+ "AddSequenceMeta.CheckResult.ConfigurationRowDistributed",
+ configurationTransform),
+ transformMeta));
+ }
+ if (source == null) {
+ remarks.add(
+ new CheckResult(
+ ICheckResult.TYPE_RESULT_ERROR,
+ BaseMessages.getString(
+ PKG,
+
"AddSequenceMeta.CheckResult.ConfigurationTransformNotFound",
+ configurationTransform),
+ transformMeta));
+ } else {
+ remarks.add(
+ new CheckResult(
+ ICheckResult.TYPE_RESULT_OK,
+ BaseMessages.getString(
+ PKG,
+
"AddSequenceMeta.CheckResult.ConfigurationTransformSelected",
+ configurationTransform),
+ transformMeta));
+ }
+ }
+
+ if (fieldsMissing || info == null || info.isEmpty()) {
+ return;
+ }
+
+ StringBuilder missing = new StringBuilder();
+ appendMissingConfigurationField(missing, info, startField);
+ appendMissingConfigurationField(missing, info, endField);
+ appendMissingConfigurationField(missing, info, incrementField);
+ if (!missing.isEmpty()) {
+ remarks.add(
+ new CheckResult(
+ ICheckResult.TYPE_RESULT_ERROR,
+ BaseMessages.getString(
+ PKG,
"AddSequenceMeta.CheckResult.ConfigurationFieldsNotFound", missing),
+ transformMeta));
+ } else {
+ remarks.add(
+ new CheckResult(
+ ICheckResult.TYPE_RESULT_OK,
+ BaseMessages.getString(PKG,
"AddSequenceMeta.CheckResult.ConfigurationFieldsFound"),
+ transformMeta));
+ }
+ }
+
+ private static void appendMissingConfigurationField(
+ StringBuilder missing, IRowMeta info, String fieldName) {
+ String name = Const.trim(fieldName);
+ if (Utils.isEmpty(name) || info.indexOfValue(name) >= 0) {
+ return;
+ }
+ if (!missing.isEmpty()) {
+ missing.append(", ");
+ }
+ missing.append(name);
}
/**
@@ -300,4 +434,84 @@ public class AddSequenceMeta extends
BaseTransformMeta<AddSequence, AddSequenceD
return retval;
}
+
+ /**
+ * Keeps {@link #configurationTransform} in sync when the info hop is drawn,
split, or detached.
+ * {@link #searchInfoAndTargetTransforms} resolves the stream from that name.
+ */
+ @Override
+ public void handleStreamSelection(IStream stream) {
+ List<IStream> infoStreams = getTransformIOMeta().getInfoStreams();
+ if (infoStreams.isEmpty() || stream == null ||
!infoStreams.contains(stream)) {
+ return;
+ }
+ TransformMeta source = stream.getTransformMeta();
+ if (source == null) {
+ return;
+ }
+ setConfigurationTransform(source.getName());
+ stream.setSubject(source.getName());
+ }
+
+ @Override
+ public void searchInfoAndTargetTransforms(List<TransformMeta> transforms) {
+ List<IStream> infoStreams = getTransformIOMeta().getInfoStreams();
+ if (infoStreams.isEmpty()) {
+ return;
+ }
+ IStream stream = infoStreams.get(0);
+ if (!isConfigurationFromTransform()) {
+ stream.setTransformMeta(null);
+ return;
+ }
+ String lookupName = stream.getSubject();
+ if (!Utils.isEmpty(configurationTransform)) {
+ lookupName = configurationTransform;
+ stream.setSubject(configurationTransform);
+ }
+ stream.setTransformMeta(TransformMeta.findTransform(transforms,
Const.trim(lookupName)));
+ }
+
+ @Override
+ public void convertIOMetaToTransformNames() {
+ List<IStream> infoStreams = getTransformIOMeta().getInfoStreams();
+ if (infoStreams.isEmpty()) {
+ return;
+ }
+ String name = infoStreams.get(0).getTransformName();
+ if (!Utils.isEmpty(name)) {
+ configurationTransform = name;
+ }
+ }
+
+ @Override
+ public ITransformIOMeta getTransformIOMeta() {
+ ITransformIOMeta ioMeta = super.getTransformIOMeta(false);
+ if (ioMeta == null) {
+ ioMeta = new TransformIOMeta(true, true, false, false, false, false);
+ ioMeta.addStream(
+ new Stream(
+ StreamType.INFO,
+ null,
+ BaseMessages.getString(PKG,
"AddSequenceMeta.InfoStream.Description"),
+ StreamIcon.INFO,
+ configurationTransform));
+ setTransformIOMeta(ioMeta);
+ }
+ return ioMeta;
+ }
+
+ @Override
+ public void resetTransformIoMeta() {
+ // Keep the configuration info stream. Recreating it here drops the
transform it points at.
+ }
+
+ /**
+ * The configuration row does not have the same layout as the main input.
Skip the safe-mode row
+ * mixing check while that info hop is in use.
+ */
+ @Override
+ public boolean excludeFromRowLayoutVerification() {
+ return isConfigurationFromTransform();
+ }
}
diff --git
a/plugins/transforms/addsequence/src/main/resources/org/apache/hop/pipeline/transforms/addsequence/messages/messages_en_US.properties
b/plugins/transforms/addsequence/src/main/resources/org/apache/hop/pipeline/transforms/addsequence/messages/messages_en_US.properties
index 3d84741d40..1efb891dde 100644
---
a/plugins/transforms/addsequence/src/main/resources/org/apache/hop/pipeline/transforms/addsequence/messages/messages_en_US.properties
+++
b/plugins/transforms/addsequence/src/main/resources/org/apache/hop/pipeline/transforms/addsequence/messages/messages_en_US.properties
@@ -15,9 +15,18 @@
# limitations under the License.
#
+AddSequence.Exception.ConfigurationFieldIsNull=Sequence configuration field
[{0}] is null.
+AddSequence.Exception.ConfigurationFieldNotFound=Sequence configuration field
[{0}] could not be found.
+AddSequence.Exception.ConfigurationFieldNotNumeric=Sequence configuration
field [{0}] could not be read as an integer. Value\: [{1}]. {2}
+AddSequence.Exception.ConfigurationRowHasNoFields=Sequence configuration
transform [{0}] returned a row without fields.
+AddSequence.Exception.ConfigurationRowMissing=Sequence configuration transform
[{0}] did not return a row. It must return exactly one row.
+AddSequence.Exception.ConfigurationRowMissingInCopy=Sequence configuration
transform [{0}] did not send a row to copy {1} of this transform. When Add
Sequence runs in multiple copies, set the data movement of [{0}] to copy, so
every copy receives the configuration row.
+AddSequence.Exception.ConfigurationRowNotSingle=Sequence configuration
transform [{0}] returned more than one row. It must return exactly one row.
+AddSequence.Exception.ConfigurationTransformNotFound=Unable to read sequence
configuration from transform [{0}]. Connect that transform to this one.
AddSequence.Exception.CouldNotFindNextValueForSequence=Couldn''t find next
value for sequence \:
AddSequence.Exception.ErrorReadingSequence=Error reading next value of
sequence [{0}] from database
AddSequence.Exception.NoSpecifiedMethod=No method is specified in this
transform\!
+AddSequence.Log.ConfigurationFieldsMissing=Specify the start, end, and
increment fields to read from the configuration transform.
AddSequence.Log.ConnectedDB=Connected to database...
AddSequence.Log.CouldNotConnectToDB=transform couldn''t connect to the
database\:
AddSequence.Log.CouldNotParseCounterValue=Counter value for {0} could not be
parsed, original value is [{1}] which becomes [{2}]. Error\: {3}.
@@ -25,14 +34,21 @@ AddSequence.Log.ErrorInTransform=Because of an error, this
transform can''t cont
AddSequence.Log.LineNumber=linenr
AddSequence.Log.NeedToSelectSequence=You need to select to create a sequence
either using a database or an internal counter.
AddSequence.Log.PipelineCountersHashtableNotAllocated=Sorry, Pipeline counters
hash table not allocated\! (internal error)
+AddSequence.Log.ReadConfiguration=Sequence configuration from [{0}]\:
start={1}, end={2}, increment={3}
AddSequence.Log.ReadRow=Read row \#
AddSequence.Log.WriteRow=Wrote row \#
+AddSequenceDialog.ConfigurationTransform.Label=Configuration transform
+AddSequenceDialog.ConfigurationTransform.Tooltip=Optional transform that
provides exactly one row with the start, end, and increment. More than one row
stops the pipeline. When this is set, Start at, Increment by, and Maximum value
are not used.
AddSequenceDialog.CounterName.Label=Counter name (optional)
+AddSequenceDialog.EndField.Label=End field
+AddSequenceDialog.EndField.Tooltip=Field in the configuration row that
contains the maximum value. After this value the sequence returns to the start
value.
AddSequenceDialog.ErrorGettingSchemas=Error getting schemas!
AddSequenceDialog.ErrorGettingSequences=Error getting sequences!
AddSequenceDialog.GetSchemas.Label=Schemas...
AddSequenceDialog.GetSequences.Label=Sequences...
AddSequenceDialog.IncrBy.Label=Increment by
+AddSequenceDialog.IncrementField.Label=Increment field
+AddSequenceDialog.IncrementField.Tooltip=Field in the configuration row that
contains the amount by which the sequence increases or decreases.
AddSequenceDialog.Log.GettingKeyInfo=getting key info...
AddSequenceDialog.MaxVal.Label=Maximum value
AddSequenceDialog.NoSchema.Message=No schema is available!
@@ -46,6 +62,8 @@ AddSequenceDialog.SelectSequence.Title=Available sequences on
[{0}]
AddSequenceDialog.Seqname.Label=Sequence name
AddSequenceDialog.Shell.Title=Add sequence
AddSequenceDialog.StartAt.Label=Start at value
+AddSequenceDialog.StartField.Label=Start field
+AddSequenceDialog.StartField.Tooltip=Field in the configuration row that
contains the value to begin the sequence with.
AddSequenceDialog.TargetSchema.Label=Schema name
AddSequenceDialog.TransformName.Label=Transform name
AddSequenceDialog.UseCounter.Label=Use counter to calculate sequence
@@ -55,6 +73,12 @@ AddSequenceDialog.UseDatabase.Label=Use DB to get sequence
AddSequenceDialog.UseDatabase.Tooltip=Check this to use a database (probably
Oracle) to get the value of a sequence.
AddSequenceDialog.UseDatabaseGroup.Label=Use a database to generate the
sequence
AddSequenceDialog.Valuename.Label=Name of value
+AddSequenceMeta.CheckResult.ConfigurationFieldsFound=Start, end, and increment
fields were found in the configuration transform.
+AddSequenceMeta.CheckResult.ConfigurationFieldsMissing=The configuration
transform needs a start field, an end field, and an increment field.
+AddSequenceMeta.CheckResult.ConfigurationFieldsNotFound=These fields could not
be found in the configuration transform\: {0}
+AddSequenceMeta.CheckResult.ConfigurationTransformNotFound=Configuration
transform [{0}] could not be found.
+AddSequenceMeta.CheckResult.ConfigurationTransformSelected=Configuration
transform [{0}] is selected.
+AddSequenceMeta.CheckResult.ConfigurationRowDistributed=Add Sequence runs in
multiple copies, but configuration transform [{0}] distributes its rows. Only
the first copy receives the configuration row. Set the data movement of [{0}]
to copy.
AddSequenceMeta.CheckResult.NoInputReceived.Title=No input received from other
transforms\!
AddSequenceMeta.CheckResult.SequenceCouldNotBeFound.Title=The sequence [{0}]
couldn''t be found
AddSequenceMeta.CheckResult.SequenceExists.Title=Sequence exits.
@@ -62,13 +86,18 @@
AddSequenceMeta.CheckResult.TransformIsReceving.Title=Transform is receiving inf
AddSequenceMeta.CheckResult.UnableToConnectDB.Title=Unable to connect to
database to verify sequence because of an error\:
AddSequenceMeta.ErrorMessage.NoConnectionDefined=There is no connection
defined in this transform.
AddSequenceMeta.ErrorMessage.UnableToConnectDB=I was unable to connect to the
database to verify the status of the table.
+AddSequenceMeta.InfoStream.Description=One row with the sequence start, end,
and increment
+AddSequenceMeta.Injection.ConfigurationTransform=Configuration transform
AddSequenceMeta.Injection.Connection=Database connection
AddSequenceMeta.Injection.CounterName=Counter name
+AddSequenceMeta.Injection.EndField=End field
AddSequenceMeta.Injection.IncrementBy=Increment by
+AddSequenceMeta.Injection.IncrementField=Increment field
AddSequenceMeta.Injection.MaxValue=Maximum value
AddSequenceMeta.Injection.SchemaName=Schema name
AddSequenceMeta.Injection.SequenceName=Sequence name
AddSequenceMeta.Injection.StartAt=Start at
+AddSequenceMeta.Injection.StartField=Start field
AddSequenceMeta.Injection.UseCounter=Use counter?
AddSequenceMeta.keyword=sequence,counter,increment,autonumber,nextval
BaseTransform.TypeLongDesc.AddSequence=Add sequence
diff --git
a/plugins/transforms/addsequence/src/test/java/org/apache/hop/pipeline/transforms/addsequence/AddSequenceMetaTest.java
b/plugins/transforms/addsequence/src/test/java/org/apache/hop/pipeline/transforms/addsequence/AddSequenceMetaTest.java
index ff3e87cf19..0a7b81775a 100644
---
a/plugins/transforms/addsequence/src/test/java/org/apache/hop/pipeline/transforms/addsequence/AddSequenceMetaTest.java
+++
b/plugins/transforms/addsequence/src/test/java/org/apache/hop/pipeline/transforms/addsequence/AddSequenceMetaTest.java
@@ -35,13 +35,17 @@ import org.apache.hop.core.ICheckResult;
import org.apache.hop.core.SqlStatement;
import org.apache.hop.core.database.DatabaseMeta;
import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.row.RowMeta;
+import org.apache.hop.core.row.value.ValueMetaInteger;
import org.apache.hop.core.variables.Variables;
import org.apache.hop.i18n.BaseMessages;
import org.apache.hop.junit.rules.RestoreHopEngineEnvironmentExtension;
import org.apache.hop.metadata.api.IHopMetadataProvider;
import org.apache.hop.metadata.api.IHopMetadataSerializer;
import org.apache.hop.metadata.validation.ReferencedDatabaseConnectionChecker;
+import org.apache.hop.pipeline.PipelineMeta;
import org.apache.hop.pipeline.transform.TransformMeta;
+import org.apache.hop.pipeline.transform.stream.IStream;
import org.apache.hop.pipeline.transforms.loadsave.LoadSaveTester;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
@@ -70,7 +74,11 @@ class AddSequenceMetaTest {
"counterName",
"startAt",
"incrementBy",
- "maxValue");
+ "maxValue",
+ "configurationTransform",
+ "startField",
+ "endField",
+ "incrementField");
LoadSaveTester<AddSequenceMeta> loadSaveTester =
new LoadSaveTester<>(
@@ -143,6 +151,15 @@ class AddSequenceMetaTest {
meta.setMaxValueByValue(10000L);
assertEquals("10000", meta.getMaxValue());
+
+ meta.setConfigurationTransform("Max query");
+ assertEquals("Max query", meta.getConfigurationTransform());
+ meta.setStartField("start_value");
+ assertEquals("start_value", meta.getStartField());
+ meta.setEndField("end_value");
+ assertEquals("end_value", meta.getEndField());
+ meta.setIncrementField("increment_value");
+ assertEquals("increment_value", meta.getIncrementField());
}
@Test
@@ -160,6 +177,217 @@ class AddSequenceMetaTest {
assertEquals(meta.isDatabaseUsed(), cloned.isDatabaseUsed());
assertEquals(meta.getConnection(), cloned.getConnection());
assertEquals(meta.getStartAt(), cloned.getStartAt());
+
+ meta.setConfigurationTransform("Max query");
+ meta.setStartField("start_value");
+ meta.setEndField("end_value");
+ meta.setIncrementField("increment_value");
+ cloned = (AddSequenceMeta) meta.clone();
+ assertEquals("Max query", cloned.getConfigurationTransform());
+ assertEquals("start_value", cloned.getStartField());
+ assertEquals("end_value", cloned.getEndField());
+ assertEquals("increment_value", cloned.getIncrementField());
+ }
+
+ @Test
+ void configurationTransformIsAnInfoStream() {
+ AddSequenceMeta meta = new AddSequenceMeta();
+ meta.setDefault();
+ meta.setConfigurationTransform("Max query");
+ meta.setStartField("start_value");
+ meta.setEndField("end_value");
+ meta.setIncrementField("increment_value");
+
+ TransformMeta source = new TransformMeta();
+ source.setName("Max query");
+ meta.searchInfoAndTargetTransforms(List.of(source));
+
+ assertEquals(1, meta.getTransformIOMeta().getInfoStreams().size());
+ IStream stream = meta.getTransformIOMeta().getInfoStreams().get(0);
+ assertEquals(source, stream.getTransformMeta());
+ assertEquals("Max query", stream.getSubject());
+ assertTrue(meta.isConfigurationFromTransform());
+ assertTrue(meta.excludeFromRowLayoutVerification());
+
+ meta.handleStreamSelection(stream);
+ assertEquals("Max query", meta.getConfigurationTransform());
+
+ meta.resetTransformIoMeta();
+ assertEquals(1, meta.getTransformIOMeta().getInfoStreams().size());
+ assertEquals(source,
meta.getTransformIOMeta().getInfoStreams().get(0).getTransformMeta());
+
+ meta.setDatabaseUsed(true);
+ meta.setCounterUsed(false);
+ meta.searchInfoAndTargetTransforms(List.of(source));
+
assertNull(meta.getTransformIOMeta().getInfoStreams().get(0).getTransformMeta());
+ assertFalse(meta.isConfigurationFromTransform());
+ assertFalse(meta.excludeFromRowLayoutVerification());
+ }
+
+ @Test
+ void checkReportsConfigurationTransformProblems() throws Exception {
+ AddSequenceMeta meta = new AddSequenceMeta();
+ meta.setDefault();
+ meta.setConfigurationTransform("Max query");
+ TransformMeta transformMeta = new TransformMeta("Add sequence", meta);
+
+ List<ICheckResult> missingFields = new ArrayList<>();
+ meta.check(
+ missingFields,
+ null,
+ transformMeta,
+ null,
+ new String[] {"Generate rows"},
+ new String[0],
+ null,
+ new Variables(),
+ metadataProviderThatCannotLoad());
+ assertTrue(
+ missingFields.stream()
+ .anyMatch(
+ remark ->
+ remark.getType() == ICheckResult.TYPE_RESULT_ERROR
+ && remark
+ .getText()
+ .equals(
+ BaseMessages.getString(
+ AddSequenceMeta.class,
+
"AddSequenceMeta.CheckResult.ConfigurationFieldsMissing"))));
+
+ meta.setStartField("start_value");
+ meta.setEndField("end_value");
+ meta.setIncrementField("increment_value");
+ PipelineMeta pipelineMeta = mock(PipelineMeta.class);
+ when(pipelineMeta.findTransform("Max query")).thenReturn(null);
+ List<ICheckResult> missingTransform = new ArrayList<>();
+ meta.check(
+ missingTransform,
+ pipelineMeta,
+ transformMeta,
+ null,
+ new String[] {"Generate rows"},
+ new String[0],
+ null,
+ new Variables(),
+ metadataProviderThatCannotLoad());
+ assertTrue(
+ missingTransform.stream()
+ .anyMatch(
+ remark ->
+ remark.getType() == ICheckResult.TYPE_RESULT_ERROR
+ && remark.getText().contains("Max query")));
+
+ TransformMeta source = new TransformMeta();
+ source.setName("Max query");
+ when(pipelineMeta.findTransform("Max query")).thenReturn(source);
+ RowMeta info = new RowMeta();
+ info.addValueMeta(new ValueMetaInteger("start_value"));
+ info.addValueMeta(new ValueMetaInteger("other"));
+ List<ICheckResult> missingInfoFields = new ArrayList<>();
+ meta.check(
+ missingInfoFields,
+ pipelineMeta,
+ transformMeta,
+ null,
+ new String[] {"Generate rows"},
+ new String[0],
+ info,
+ new Variables(),
+ metadataProviderThatCannotLoad());
+ assertTrue(
+ missingInfoFields.stream()
+ .anyMatch(
+ remark ->
+ remark.getType() == ICheckResult.TYPE_RESULT_ERROR
+ && remark.getText().contains("end_value")
+ && remark.getText().contains("increment_value")));
+
+ info.addValueMeta(new ValueMetaInteger("end_value"));
+ info.addValueMeta(new ValueMetaInteger("increment_value"));
+ List<ICheckResult> ok = new ArrayList<>();
+ meta.check(
+ ok,
+ pipelineMeta,
+ transformMeta,
+ null,
+ new String[] {"Generate rows"},
+ new String[0],
+ info,
+ new Variables(),
+ metadataProviderThatCannotLoad());
+ assertTrue(
+ ok.stream()
+ .anyMatch(
+ remark ->
+ remark.getType() == ICheckResult.TYPE_RESULT_OK
+ && remark
+ .getText()
+ .equals(
+ BaseMessages.getString(
+ AddSequenceMeta.class,
+
"AddSequenceMeta.CheckResult.ConfigurationFieldsFound"))));
+ assertTrue(ok.stream().noneMatch(remark -> remark.getType() ==
ICheckResult.TYPE_RESULT_ERROR));
+ }
+
+ /**
+ * A distributing configuration transform only sends its single row to the
first copy of Add
+ * Sequence. Verify has to say so before the other copies fail at runtime.
+ */
+ @Test
+ void checkReportsADistributedConfigurationRowWithMultipleCopies() throws
Exception {
+ AddSequenceMeta meta = new AddSequenceMeta();
+ meta.setDefault();
+ meta.setConfigurationTransform("Max query");
+ meta.setStartField("start_value");
+ meta.setEndField("end_value");
+ meta.setIncrementField("increment_value");
+ TransformMeta transformMeta = new TransformMeta("Add sequence", meta);
+ TransformMeta source = new TransformMeta();
+ source.setName("Max query");
+ source.setDistributes(true);
+ PipelineMeta pipelineMeta = mock(PipelineMeta.class);
+ when(pipelineMeta.findTransform("Max query")).thenReturn(source);
+ // Copies are only resolved against variables for a transform in a
pipeline.
+ transformMeta.setParentPipelineMeta(pipelineMeta);
+ String distributed =
+ BaseMessages.getString(
+ AddSequenceMeta.class,
+ "AddSequenceMeta.CheckResult.ConfigurationRowDistributed",
+ "Max query");
+
+ // One copy: distributing is fine.
+ assertFalse(hasError(meta, pipelineMeta, transformMeta, distributed));
+
+ // Several copies, also from a variable, and the source distributes: an
error.
+ transformMeta.setCopiesString("${COPIES}");
+ assertTrue(hasError(meta, pipelineMeta, transformMeta, distributed));
+
+ // Several copies, and the source copies its rows to every copy: fine.
+ source.setDistributes(false);
+ assertFalse(hasError(meta, pipelineMeta, transformMeta, distributed));
+ }
+
+ private boolean hasError(
+ AddSequenceMeta meta, PipelineMeta pipelineMeta, TransformMeta
transformMeta, String text)
+ throws Exception {
+ Variables variables = new Variables();
+ variables.setVariable("COPIES", "3");
+ List<ICheckResult> remarks = new ArrayList<>();
+ meta.check(
+ remarks,
+ pipelineMeta,
+ transformMeta,
+ null,
+ new String[] {"Generate rows"},
+ new String[0],
+ null,
+ variables,
+ metadataProviderThatCannotLoad());
+ return remarks.stream()
+ .anyMatch(
+ remark ->
+ remark.getType() == ICheckResult.TYPE_RESULT_ERROR
+ && remark.getText().equals(text));
}
/**
diff --git
a/plugins/transforms/addsequence/src/test/java/org/apache/hop/pipeline/transforms/addsequence/AddSequenceTest.java
b/plugins/transforms/addsequence/src/test/java/org/apache/hop/pipeline/transforms/addsequence/AddSequenceTest.java
index c0b5fe806a..d74875c30c 100644
---
a/plugins/transforms/addsequence/src/test/java/org/apache/hop/pipeline/transforms/addsequence/AddSequenceTest.java
+++
b/plugins/transforms/addsequence/src/test/java/org/apache/hop/pipeline/transforms/addsequence/AddSequenceTest.java
@@ -23,22 +23,28 @@ import static
org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.argThat;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+import java.math.BigDecimal;
import java.util.List;
import java.util.stream.Stream;
import org.apache.hop.core.Counters;
import org.apache.hop.core.HopEnvironment;
+import org.apache.hop.core.IRowSet;
import org.apache.hop.core.database.Database;
import org.apache.hop.core.exception.HopException;
import org.apache.hop.core.logging.ILoggingObject;
import org.apache.hop.core.row.IRowMeta;
import org.apache.hop.core.row.RowMeta;
+import org.apache.hop.core.row.value.ValueMetaBigNumber;
import org.apache.hop.core.row.value.ValueMetaInteger;
import org.apache.hop.core.row.value.ValueMetaString;
+import org.apache.hop.i18n.BaseMessages;
import org.apache.hop.junit.rules.RestoreHopEngineEnvironmentExtension;
import org.apache.hop.pipeline.PipelineTestingUtil;
import org.apache.hop.pipeline.transforms.mock.TransformMockHelper;
@@ -430,4 +436,229 @@ class AddSequenceTest {
addSequence.dispose();
}
+
+ /** Start, end, and increment come from one info row. Typed counter values
are ignored. */
+ @Test
+ void testProcessRowReadsSingleConfigurationRow() throws Exception {
+ AddSequenceMeta meta = configurationMeta();
+ RowMeta configMeta = new RowMeta();
+ configMeta.addValueMeta(new ValueMetaBigNumber("start_value"));
+ configMeta.addValueMeta(new ValueMetaBigNumber("end_value"));
+ configMeta.addValueMeta(new ValueMetaInteger("increment_value"));
+ IRowSet rowSet = mock(IRowSet.class);
+ when(rowSet.getRowMeta()).thenReturn(configMeta);
+
+ AddSequence addSequence = spyConfiguration(meta, rowSet);
+ doReturn(new Object[] {new BigDecimal("100"), new BigDecimal("102"), 2L})
+ .doReturn(null)
+ .when(addSequence)
+ .getRowFrom(rowSet);
+ doReturn(new Object[] {"A"})
+ .doReturn(new Object[] {"B"})
+ .doReturn(new Object[] {"C"})
+ .doReturn(null)
+ .when(addSequence)
+ .getRow();
+
+ List<Object[]> result = PipelineTestingUtil.execute(addSequence, 3, false);
+
+ assertEquals(100L, result.get(0)[1]);
+ assertEquals(102L, result.get(1)[1]);
+ // End is the maximum: the next value wraps back to the start.
+ assertEquals(100L, result.get(2)[1]);
+ assertNotNull(addSequence.getData().counter);
+ addSequence.dispose();
+ }
+
+ @Test
+ void testProcessRowReadsStringConfigurationValues() throws Exception {
+ AddSequenceMeta meta = configurationMeta();
+ RowMeta configMeta = new RowMeta();
+ configMeta.addValueMeta(new ValueMetaString("start_value"));
+ configMeta.addValueMeta(new ValueMetaString("end_value"));
+ configMeta.addValueMeta(new ValueMetaString("increment_value"));
+ IRowSet rowSet = mock(IRowSet.class);
+ when(rowSet.getRowMeta()).thenReturn(configMeta);
+
+ AddSequence addSequence = spyConfiguration(meta, rowSet);
+ doReturn(new Object[] {"50", "1000",
"5"}).doReturn(null).when(addSequence).getRowFrom(rowSet);
+ doReturn(new Object[] {"A"})
+ .doReturn(new Object[] {"B"})
+ .doReturn(null)
+ .when(addSequence)
+ .getRow();
+
+ List<Object[]> result = PipelineTestingUtil.execute(addSequence, 2, false);
+
+ assertEquals(50L, result.get(0)[1]);
+ assertEquals(55L, result.get(1)[1]);
+ addSequence.dispose();
+ }
+
+ @Test
+ void testConfigurationTransformWithNoRowFails() throws Exception {
+ AddSequence addSequence = spyConfiguration(configurationMeta(),
rowSetWith(configRowMeta()));
+ doReturn(null).when(addSequence).getRowFrom(any(IRowSet.class));
+
+ assertFalse(addSequence.processRow());
+ assertEquals(1L, addSequence.getErrors());
+ assertNull(addSequence.getData().counter);
+ addSequence.dispose();
+ }
+
+ /**
+ * With several copies of Add Sequence, a distributed configuration row only
reaches one of them.
+ * The others have to say how to fix that, not just that the row is missing.
+ */
+ @Test
+ void testConfigurationRowMissingInACopyExplainsHowToFixIt() throws Exception
{
+ when(transformMockHelper.transformMeta.getCopies(any())).thenReturn(2);
+ AddSequenceMeta meta = configurationMeta();
+ AddSequence addSequence = spyConfiguration(meta,
rowSetWith(configRowMeta()));
+ doReturn(null).when(addSequence).getRowFrom(any(IRowSet.class));
+
+ assertFalse(addSequence.processRow());
+ assertEquals(1L, addSequence.getErrors());
+ String expected =
+ BaseMessages.getString(
+ AddSequence.class,
+ "AddSequence.Exception.ConfigurationRowMissingInCopy",
+ meta.getConfigurationTransform(),
+ "0");
+ verify(transformMockHelper.iLogChannel)
+ .logError(argThat((String message) -> message != null &&
message.contains(expected)));
+ addSequence.dispose();
+ }
+
+ @Test
+ void testConfigurationTransformWithTwoRowsFails() throws Exception {
+ AddSequence addSequence = spyConfiguration(configurationMeta(),
rowSetWith(configRowMeta()));
+ doReturn(new Object[] {1L, 10L, 1L})
+ .doReturn(new Object[] {2L, 10L, 1L})
+ .when(addSequence)
+ .getRowFrom(any(IRowSet.class));
+
+ assertFalse(addSequence.processRow());
+ assertEquals(1L, addSequence.getErrors());
+ assertNull(addSequence.getData().counter);
+ addSequence.dispose();
+ }
+
+ @Test
+ void testConfigurationTransformNotConnectedFails() throws Exception {
+ AddSequenceMeta meta = configurationMeta();
+ AddSequence addSequence =
+ new AddSequence(
+ transformMockHelper.transformMeta,
+ meta,
+ new AddSequenceData(),
+ 0,
+ transformMockHelper.pipelineMeta,
+ transformMockHelper.pipeline);
+ when(transformMockHelper.pipeline.getContainerId()).thenReturn("cfg-" +
System.nanoTime());
+ assertTrue(addSequence.init());
+ addSequence = spy(addSequence);
+
doReturn(null).when(addSequence).findInputRowSet(meta.getConfigurationTransform());
+
+ assertFalse(addSequence.processRow());
+ assertEquals(1L, addSequence.getErrors());
+ addSequence.dispose();
+ }
+
+ @Test
+ void testConfigurationFieldMissingFails() throws Exception {
+ RowMeta configMeta = new RowMeta();
+ configMeta.addValueMeta(new ValueMetaInteger("other"));
+ AddSequence addSequence = spyConfiguration(configurationMeta(),
rowSetWith(configMeta));
+ doReturn(new Object[]
{1L}).doReturn(null).when(addSequence).getRowFrom(any(IRowSet.class));
+
+ assertFalse(addSequence.processRow());
+ assertEquals(1L, addSequence.getErrors());
+ assertNull(addSequence.getData().counter);
+ addSequence.dispose();
+ }
+
+ @Test
+ void testConfigurationFieldNullFails() throws Exception {
+ AddSequence addSequence = spyConfiguration(configurationMeta(),
rowSetWith(configRowMeta()));
+ doReturn(new Object[] {null, 10L, 1L})
+ .doReturn(null)
+ .when(addSequence)
+ .getRowFrom(any(IRowSet.class));
+
+ assertFalse(addSequence.processRow());
+ assertEquals(1L, addSequence.getErrors());
+ addSequence.dispose();
+ }
+
+ @Test
+ void testInitSkipsTypedCounterValuesWhenConfigurationTransformIsSet() {
+ AddSequenceMeta meta = configurationMeta();
+ when(transformMockHelper.pipeline.getContainerId()).thenReturn("cfg-" +
System.nanoTime());
+ AddSequence addSequence =
+ new AddSequence(
+ transformMockHelper.transformMeta,
+ meta,
+ new AddSequenceData(),
+ 0,
+ transformMockHelper.pipelineMeta,
+ transformMockHelper.pipeline);
+
+ assertTrue(addSequence.init());
+ assertNull(addSequence.getData().counter);
+
+ meta.setStartField("");
+ assertFalse(addSequence.init());
+ addSequence.dispose();
+ }
+
+ private static AddSequenceMeta configurationMeta() {
+ AddSequenceMeta meta = new AddSequenceMeta();
+ meta.setDefault();
+ meta.setValueName("id");
+ meta.setCounterUsed(true);
+ meta.setDatabaseUsed(false);
+ meta.setConfigurationTransform("Max query");
+ meta.setStartField("start_value");
+ meta.setEndField("end_value");
+ meta.setIncrementField("increment_value");
+ // Proves the typed values are not parsed when the configuration transform
is set.
+ meta.setStartAt("not-a-number");
+ meta.setIncrementBy("not-a-number");
+ meta.setMaxValue("not-a-number");
+ return meta;
+ }
+
+ private static RowMeta configRowMeta() {
+ RowMeta configMeta = new RowMeta();
+ configMeta.addValueMeta(new ValueMetaInteger("start_value"));
+ configMeta.addValueMeta(new ValueMetaInteger("end_value"));
+ configMeta.addValueMeta(new ValueMetaInteger("increment_value"));
+ return configMeta;
+ }
+
+ private static IRowSet rowSetWith(RowMeta configMeta) {
+ IRowSet rowSet = mock(IRowSet.class);
+ when(rowSet.getRowMeta()).thenReturn(configMeta);
+ return rowSet;
+ }
+
+ private AddSequence spyConfiguration(AddSequenceMeta meta, IRowSet rowSet)
throws Exception {
+ when(transformMockHelper.pipeline.getContainerId()).thenReturn("cfg-" +
System.nanoTime());
+ AddSequence addSequence =
+ new AddSequence(
+ transformMockHelper.transformMeta,
+ meta,
+ new AddSequenceData(),
+ 0,
+ transformMockHelper.pipelineMeta,
+ transformMockHelper.pipeline);
+ assertTrue(addSequence.init());
+ RowMeta inputRowMeta = new RowMeta();
+ inputRowMeta.addValueMeta(new ValueMetaString("name"));
+ addSequence.setInputRowMeta(inputRowMeta);
+ addSequence = spy(addSequence);
+
doReturn(rowSet).when(addSequence).findInputRowSet(meta.getConfigurationTransform());
+ return addSequence;
+ }
}
