This is an automated email from the ASF dual-hosted git repository.
hansva pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/main by this push:
new f10a477714 issue #3137 : sample target hops and show data preview
icons on hop arrows (#7595)
f10a477714 is described below
commit f10a47771495187401088c164a59db098f58259a
Author: Matt Casters <[email protected]>
AuthorDate: Wed Jul 22 15:02:24 2026 +0200
issue #3137 : sample target hops and show data preview icons on hop arrows
(#7595)
Add IRowToListener so putRowTo destinations (Filter, Switch/Case, etc.) can
be sampled separately by hop. Draw clickable data-grid icons on hops near
the
source transform for full pipeline transparency.
---
.../java/org/apache/hop/core/gui/AreaOwner.java | 5 +
.../org/apache/hop/pipeline/PipelinePainter.java | 65 ++++++++
.../pipeline/canvas/PipelineCanvasSvgRenderer.java | 2 +
.../hop/pipeline/engine/EngineComponent.java | 20 +++
.../hop/pipeline/engine/IEngineComponent.java | 21 +++
.../hop/pipeline/transform/BaseTransform.java | 28 +++-
.../hop/pipeline/transform/IRowToListener.java | 46 ++++++
.../apache/hop/pipeline/transform/ITransform.java | 22 +++
.../hop/pipeline/transform/RowToAdapter.java | 42 +++++
.../hop/pipeline/transform/BaseTransformTest.java | 40 +++++
.../hop/beam/core/transform/BeamRowHandler.java | 8 +-
.../org/apache/hop/spark/core/SparkRowHandler.java | 8 +-
.../transforms/javascript/ScriptValuesDummy.java | 16 ++
.../hopgui/file/pipeline/HopGuiPipelineGraph.java | 68 +++++++-
.../ui/hopgui/file/shared/DrillDownGuiPlugin.java | 23 ++-
.../file/shared/PipelineRowSamplerHelper.java | 183 ++++++++++++++++-----
16 files changed, 534 insertions(+), 63 deletions(-)
diff --git a/engine/src/main/java/org/apache/hop/core/gui/AreaOwner.java
b/engine/src/main/java/org/apache/hop/core/gui/AreaOwner.java
index 156703c6fd..9c5f1e0434 100644
--- a/engine/src/main/java/org/apache/hop/core/gui/AreaOwner.java
+++ b/engine/src/main/java/org/apache/hop/core/gui/AreaOwner.java
@@ -56,6 +56,11 @@ public class AreaOwner {
* there are available output rows.
*/
TRANSFORM_OUTPUT_DATA(true),
+ /**
+ * Data preview icon on a pipeline hop (near the source) for rows sampled
on that hop, including
+ * target hops such as Filter true/false branches.
+ */
+ HOP_OUTPUT_DATA(true),
/** The pipeline hop decoration */
TRANSFORM_TARGET_HOP_ICON(true),
diff --git a/engine/src/main/java/org/apache/hop/pipeline/PipelinePainter.java
b/engine/src/main/java/org/apache/hop/pipeline/PipelinePainter.java
index f1fd885fd3..d1da36872b 100644
--- a/engine/src/main/java/org/apache/hop/pipeline/PipelinePainter.java
+++ b/engine/src/main/java/org/apache/hop/pipeline/PipelinePainter.java
@@ -80,6 +80,10 @@ public class PipelinePainter extends
BasePainter<PipelineHopMeta, TransformMeta>
private IPipelineEngine<PipelineMeta> pipeline;
private boolean slowTransformIndicatorEnabled;
private Map<String, RowBuffer> outputRowsMap;
+
+ /** Hop key (origin\\tdestination) → sampled rows for target hops /
putRowTo. */
+ private Map<String, RowBuffer> outputHopRowsMap;
+
private Map<String, Object> stateMap;
private boolean showingSelectedTransformMetrics = true;
@@ -131,6 +135,7 @@ public class PipelinePainter extends
BasePainter<PipelineHopMeta, TransformMeta>
this.slowTransformIndicatorEnabled = slowTransformIndicatorEnabled;
this.outputRowsMap = outputRowsMap;
+ this.outputHopRowsMap = null;
transformLogMap = null;
@@ -737,6 +742,46 @@ public class PipelinePainter extends
BasePainter<PipelineHopMeta, TransformMeta>
}
}
+ /**
+ * Draw a data-preview icon on the hop near the source transform when
hop-level samples exist. Key
+ * format matches UI hop sampling: {@code origin + "\t" + destination}.
+ */
+ private void drawHopOutputDataIndicator(
+ PipelineHopMeta pipelineHop,
+ TransformMeta fromTransform,
+ TransformMeta toTransform,
+ int x1,
+ int y1,
+ int x2,
+ int y2)
+ throws HopException {
+ if (Utils.isEmpty(outputHopRowsMap) || pipelineHop == null ||
fromTransform == null) {
+ return;
+ }
+ String hopKey = fromTransform.getName() + "\t" + toTransform.getName();
+ RowBuffer rowBuffer = outputHopRowsMap.get(hopKey);
+ if (rowBuffer == null || rowBuffer.isEmpty()) {
+ return;
+ }
+
+ // Place at ~30% of the hop length from the source transform
+ double hopDataPosition = 0.30;
+ int iconX = (int) (x1 + hopDataPosition * (x2 - x1)) - miniIconSize / 2;
+ int iconY = (int) (y1 + hopDataPosition * (y2 - y1)) - miniIconSize / 2;
+
+ gc.drawImage(EImage.DATA, iconX, iconY, magnification);
+ areaOwners.add(
+ new AreaOwner(
+ AreaType.HOP_OUTPUT_DATA,
+ iconX,
+ iconY,
+ miniIconSize,
+ miniIconSize,
+ offset,
+ pipelineHop,
+ rowBuffer));
+ }
+
private void drawTextRightAligned(String txt, int x, int y) {
int off = gc.textExtent(txt).x;
x -= off;
@@ -1288,6 +1333,10 @@ public class PipelinePainter extends
BasePainter<PipelineHopMeta, TransformMeta>
}
}
}
+
+ // Data preview icon near the source transform for hop-level samples
(target hops, etc.)
+ //
+ drawHopOutputDataIndicator(pipelineHop, fs, ts, x1, y1, x2, y2);
}
PipelinePainterExtension extension =
@@ -1429,6 +1478,22 @@ public class PipelinePainter extends
BasePainter<PipelineHopMeta, TransformMeta>
this.outputRowsMap = outputRowsMap;
}
+ /**
+ * Gets outputHopRowsMap
+ *
+ * @return hop-keyed sample buffers
+ */
+ public Map<String, RowBuffer> getOutputHopRowsMap() {
+ return outputHopRowsMap;
+ }
+
+ /**
+ * @param outputHopRowsMap hop key → RowBuffer samples for target hops
+ */
+ public void setOutputHopRowsMap(Map<String, RowBuffer> outputHopRowsMap) {
+ this.outputHopRowsMap = outputHopRowsMap;
+ }
+
/**
* Gets stateMap
*
diff --git
a/engine/src/main/java/org/apache/hop/pipeline/canvas/PipelineCanvasSvgRenderer.java
b/engine/src/main/java/org/apache/hop/pipeline/canvas/PipelineCanvasSvgRenderer.java
index 8f2f986101..fe519c4344 100644
---
a/engine/src/main/java/org/apache/hop/pipeline/canvas/PipelineCanvasSvgRenderer.java
+++
b/engine/src/main/java/org/apache/hop/pipeline/canvas/PipelineCanvasSvgRenderer.java
@@ -63,6 +63,7 @@ public final class PipelineCanvasSvgRenderer {
public boolean slowTransformIndicatorEnabled;
public double zoomFactor;
public Map<String, RowBuffer> outputRowsMap;
+ public Map<String, RowBuffer> outputHopRowsMap;
public boolean drawingBorderAroundName;
public String mouseOverName;
public Map<String, Object> stateMap;
@@ -124,6 +125,7 @@ public final class PipelineCanvasSvgRenderer {
ctx.stateMap);
pipelinePainter.setMagnification(ctx.magnification);
+ pipelinePainter.setOutputHopRowsMap(ctx.outputHopRowsMap);
pipelinePainter.setTransformLogMap(ctx.transformLogMap);
pipelinePainter.setStartHopTransform(ctx.startHopTransform);
pipelinePainter.setEndHopLocation(ctx.endHopLocation);
diff --git
a/engine/src/main/java/org/apache/hop/pipeline/engine/EngineComponent.java
b/engine/src/main/java/org/apache/hop/pipeline/engine/EngineComponent.java
index cc6b6d3867..7cb2aac259 100644
--- a/engine/src/main/java/org/apache/hop/pipeline/engine/EngineComponent.java
+++ b/engine/src/main/java/org/apache/hop/pipeline/engine/EngineComponent.java
@@ -17,7 +17,9 @@
package org.apache.hop.pipeline.engine;
+import java.util.Collections;
import java.util.Date;
+import java.util.List;
import java.util.Objects;
import lombok.Setter;
import org.apache.hop.core.exception.HopRuntimeException;
@@ -27,6 +29,7 @@ import org.apache.hop.core.logging.LogLevel;
import org.apache.hop.i18n.BaseMessages;
import org.apache.hop.pipeline.transform.BaseTransform;
import org.apache.hop.pipeline.transform.IRowListener;
+import org.apache.hop.pipeline.transform.IRowToListener;
public class EngineComponent implements IEngineComponent {
@@ -195,6 +198,23 @@ public class EngineComponent implements IEngineComponent {
"Removing a row listener to this transform is not possible as it's not
part of a running engine");
}
+ @Override
+ public void addRowToListener(IRowToListener rowToListener) {
+ throw new HopRuntimeException(
+ "Adding a row-to listener to this transform is not possible as it's
not part of a running engine");
+ }
+
+ @Override
+ public void removeRowToListener(IRowToListener rowToListener) {
+ throw new HopRuntimeException(
+ "Removing a row-to listener to this transform is not possible as it's
not part of a running engine");
+ }
+
+ @Override
+ public List<IRowToListener> getRowToListeners() {
+ return Collections.emptyList();
+ }
+
/**
* Gets linesRead
*
diff --git
a/engine/src/main/java/org/apache/hop/pipeline/engine/IEngineComponent.java
b/engine/src/main/java/org/apache/hop/pipeline/engine/IEngineComponent.java
index 9b3a74929a..39c77c9713 100644
--- a/engine/src/main/java/org/apache/hop/pipeline/engine/IEngineComponent.java
+++ b/engine/src/main/java/org/apache/hop/pipeline/engine/IEngineComponent.java
@@ -18,9 +18,11 @@
package org.apache.hop.pipeline.engine;
import java.util.Date;
+import java.util.List;
import org.apache.hop.core.logging.ILogChannel;
import org.apache.hop.core.logging.LogLevel;
import org.apache.hop.pipeline.transform.IRowListener;
+import org.apache.hop.pipeline.transform.IRowToListener;
/**
* An identifiable component of an execution engine {@link IPipelineEngine} In
a pipeline engine
@@ -135,6 +137,25 @@ public interface IEngineComponent {
void removeRowListener(IRowListener rowListener);
+ /**
+ * Add a destination-aware row listener for rows written via target hops
({@code putRowTo}).
+ *
+ * @param rowToListener the listener to add
+ */
+ void addRowToListener(IRowToListener rowToListener);
+
+ /**
+ * Remove a destination-aware row listener.
+ *
+ * @param rowToListener the listener to remove
+ */
+ void removeRowToListener(IRowToListener rowToListener);
+
+ /**
+ * @return the installed destination-aware row listeners
+ */
+ List<IRowToListener> getRowToListeners();
+
/**
* Get the execution status of the component
*
diff --git
a/engine/src/main/java/org/apache/hop/pipeline/transform/BaseTransform.java
b/engine/src/main/java/org/apache/hop/pipeline/transform/BaseTransform.java
index 5600a65210..3cac9a4d45 100644
--- a/engine/src/main/java/org/apache/hop/pipeline/transform/BaseTransform.java
+++ b/engine/src/main/java/org/apache/hop/pipeline/transform/BaseTransform.java
@@ -252,6 +252,9 @@ public class BaseTransform<Meta extends ITransformMeta,
Data extends ITransformD
/** The list of IRowListener interfaces */
protected List<IRowListener> rowListeners;
+ /** The list of destination-aware IRowToListener interfaces (target hops /
putRowTo) */
+ protected List<IRowToListener> rowToListeners;
+
/**
* Map of files that are generated or used by this transform. After
execution, these can be added
* to result. The entry to the map is the filename
@@ -428,6 +431,7 @@ public class BaseTransform<Meta extends ITransformMeta,
Data extends ITransformD
rowDistribution = transformMeta.getRowDistribution();
rowListeners = new CopyOnWriteArrayList<>();
+ rowToListeners = new CopyOnWriteArrayList<>();
resultFiles = new HashMap<>();
resultFilesLock = new ReentrantReadWriteLock();
@@ -1422,9 +1426,12 @@ public class BaseTransform<Meta extends ITransformMeta,
Data extends ITransformD
}
}
- // Do not call the row listeners for targeted rows.
- // It can cause rows with varying layouts to arrive at the same listener
without a way to keep
- // them apart.
+ // Do not call IRowListener for targeted rows: mixed layouts can arrive
without a way to keep
+ // them apart. Use IRowToListener instead — it includes the destination
IRowSet.
+ //
+ for (IRowToListener listener : rowToListeners) {
+ listener.rowWrittenTo(rowMeta, row, rowSet);
+ }
// Keep adding to terminator_rows buffer...
if (terminator && terminatorRows != null) {
@@ -3280,6 +3287,21 @@ public class BaseTransform<Meta extends ITransformMeta,
Data extends ITransformD
return Collections.unmodifiableList(rowListeners);
}
+ @Override
+ public void addRowToListener(IRowToListener rowToListener) {
+ rowToListeners.add(rowToListener);
+ }
+
+ @Override
+ public void removeRowToListener(IRowToListener rowToListener) {
+ rowToListeners.remove(rowToListener);
+ }
+
+ @Override
+ public List<IRowToListener> getRowToListeners() {
+ return Collections.unmodifiableList(rowToListeners);
+ }
+
/**
* Adds the result file.
*
diff --git
a/engine/src/main/java/org/apache/hop/pipeline/transform/IRowToListener.java
b/engine/src/main/java/org/apache/hop/pipeline/transform/IRowToListener.java
new file mode 100644
index 0000000000..1e9b60c437
--- /dev/null
+++ b/engine/src/main/java/org/apache/hop/pipeline/transform/IRowToListener.java
@@ -0,0 +1,46 @@
+/*
+ * 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.transform;
+
+import org.apache.hop.core.IRowSet;
+import org.apache.hop.core.exception.HopTransformException;
+import org.apache.hop.core.row.IRowMeta;
+
+/**
+ * Listener for rows written to a specific destination rowset (target hops).
+ *
+ * <p>Unlike {@link IRowListener}, this interface includes the destination
{@link IRowSet} so
+ * listeners can keep rows with different layouts apart (e.g. Filter
true/false branches,
+ * Switch/Case targets). Used by GUI hop sampling and other destination-aware
consumers.
+ *
+ * @see IRowListener
+ * @see BaseTransform#putRowTo(IRowMeta, Object[], IRowSet)
+ */
+public interface IRowToListener {
+
+ /**
+ * Called when a row is written to a specific destination rowset (typically
via {@code putRowTo} /
+ * {@code handlePutRowTo}).
+ *
+ * @param rowMeta the metadata of the row
+ * @param row the data of the row
+ * @param rowSet the destination rowset the row was written to
+ * @throws HopTransformException an exception that can be thrown to hard
stop the transform
+ */
+ void rowWrittenTo(IRowMeta rowMeta, Object[] row, IRowSet rowSet) throws
HopTransformException;
+}
diff --git
a/engine/src/main/java/org/apache/hop/pipeline/transform/ITransform.java
b/engine/src/main/java/org/apache/hop/pipeline/transform/ITransform.java
index 30e8f8ffed..a31892dfdc 100644
--- a/engine/src/main/java/org/apache/hop/pipeline/transform/ITransform.java
+++ b/engine/src/main/java/org/apache/hop/pipeline/transform/ITransform.java
@@ -244,6 +244,28 @@ public interface ITransform extends IVariables,
IHasLogChannel, IEngineComponent
*/
List<IRowListener> getRowListeners();
+ /**
+ * Add a destination-aware row listener for rows written via target hops
({@code putRowTo}).
+ *
+ * @param rowToListener the listener to add
+ */
+ @Override
+ void addRowToListener(IRowToListener rowToListener);
+
+ /**
+ * Remove a destination-aware row listener from this transform.
+ *
+ * @param rowToListener the listener to remove
+ */
+ @Override
+ void removeRowToListener(IRowToListener rowToListener);
+
+ /**
+ * @return a list of the installed destination-aware row listeners
+ */
+ @Override
+ List<IRowToListener> getRowToListeners();
+
/**
* @return The list of active input rowsets for the transform
*/
diff --git
a/engine/src/main/java/org/apache/hop/pipeline/transform/RowToAdapter.java
b/engine/src/main/java/org/apache/hop/pipeline/transform/RowToAdapter.java
new file mode 100644
index 0000000000..2b5af37297
--- /dev/null
+++ b/engine/src/main/java/org/apache/hop/pipeline/transform/RowToAdapter.java
@@ -0,0 +1,42 @@
+/*
+ * 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.transform;
+
+import org.apache.hop.core.IRowSet;
+import org.apache.hop.core.exception.HopTransformException;
+import org.apache.hop.core.row.IRowMeta;
+
+/**
+ * Empty adapter for {@link IRowToListener}. Exists as a convenience for
creating destination-aware
+ * row listeners.
+ *
+ * @see IRowToListener
+ * @see RowAdapter
+ */
+public class RowToAdapter implements IRowToListener {
+
+ public RowToAdapter() {
+ // Do nothing
+ }
+
+ @Override
+ public void rowWrittenTo(IRowMeta rowMeta, Object[] row, IRowSet rowSet)
+ throws HopTransformException {
+ // Do nothing
+ }
+}
diff --git
a/engine/src/test/java/org/apache/hop/pipeline/transform/BaseTransformTest.java
b/engine/src/test/java/org/apache/hop/pipeline/transform/BaseTransformTest.java
index 9410f39796..e1d096436f 100644
---
a/engine/src/test/java/org/apache/hop/pipeline/transform/BaseTransformTest.java
+++
b/engine/src/test/java/org/apache/hop/pipeline/transform/BaseTransformTest.java
@@ -17,9 +17,11 @@
package org.apache.hop.pipeline.transform;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNotSame;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;
@@ -29,6 +31,7 @@ import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.lenient;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -362,6 +365,43 @@ class BaseTransformTest {
}
}
+ @Test
+ void putRowToInvokesRowToListenersNotRowListeners() throws HopException {
+ BaseTransform<ITransformMeta, ITransformData> baseTransform =
+ new BaseTransform<>(
+ mockHelper.transformMeta,
+ mockHelper.iTransformMeta,
+ mockHelper.iTransformData,
+ 0,
+ mockHelper.pipelineMeta,
+ mockHelper.pipeline);
+
+ IRowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaString("col"));
+ Object[] row = new Object[] {"value"};
+ QueueRowSet rowSet = new QueueRowSet();
+
+ IRowListener rowListener = mock(IRowListener.class);
+ baseTransform.addRowListener(rowListener);
+
+ final IRowSet[] receivedRowSet = new IRowSet[1];
+ final Object[][] receivedRow = new Object[1][];
+ baseTransform.addRowToListener(
+ (meta, data, dest) -> {
+ receivedRowSet[0] = dest;
+ receivedRow[0] = data;
+ });
+
+ baseTransform.putRowTo(rowMeta, row, rowSet);
+
+ assertSame(rowSet, receivedRowSet[0]);
+ assertSame(row, receivedRow[0]);
+ assertEquals(1L, baseTransform.getLinesWritten());
+ verify(rowListener, never()).rowWrittenEvent(any(), any());
+ verify(rowListener, never()).rowReadEvent(any(), any());
+ verify(rowListener, never()).errorRowWrittenEvent(any(), any());
+ }
+
private IRowHandler rowHandlerWithDefaultMethods() {
return new IRowHandler() {
@Override
diff --git
a/plugins/engines/beam/src/main/java/org/apache/hop/beam/core/transform/BeamRowHandler.java
b/plugins/engines/beam/src/main/java/org/apache/hop/beam/core/transform/BeamRowHandler.java
index 87dac2a9ec..cf77d28cb6 100644
---
a/plugins/engines/beam/src/main/java/org/apache/hop/beam/core/transform/BeamRowHandler.java
+++
b/plugins/engines/beam/src/main/java/org/apache/hop/beam/core/transform/BeamRowHandler.java
@@ -26,6 +26,7 @@ import org.apache.hop.core.row.IRowMeta;
import org.apache.hop.pipeline.transform.BaseTransform;
import org.apache.hop.pipeline.transform.IRowHandler;
import org.apache.hop.pipeline.transform.IRowListener;
+import org.apache.hop.pipeline.transform.IRowToListener;
/**
* Reading and writing rows from/to row sets is simpler in a Beam context
since we're using a single
@@ -104,9 +105,10 @@ public class BeamRowHandler implements IRowHandler {
@Override
public void putRowTo(IRowMeta rowMeta, Object[] row, IRowSet rowSet)
throws HopTransformException {
- List<IRowListener> rowListeners = transform.getRowListeners();
- for (IRowListener listener : rowListeners) {
- listener.rowWrittenEvent(rowMeta, row);
+ // Destination-aware listeners (not IRowListener — mixed layouts need the
target rowset)
+ List<IRowToListener> rowToListeners = transform.getRowToListeners();
+ for (IRowToListener listener : rowToListeners) {
+ listener.rowWrittenTo(rowMeta, row, rowSet);
}
rowSet.putRow(rowMeta, row);
diff --git
a/plugins/engines/spark/src/main/java/org/apache/hop/spark/core/SparkRowHandler.java
b/plugins/engines/spark/src/main/java/org/apache/hop/spark/core/SparkRowHandler.java
index dff5a7a393..b72ae64360 100644
---
a/plugins/engines/spark/src/main/java/org/apache/hop/spark/core/SparkRowHandler.java
+++
b/plugins/engines/spark/src/main/java/org/apache/hop/spark/core/SparkRowHandler.java
@@ -25,6 +25,7 @@ import org.apache.hop.core.row.IRowMeta;
import org.apache.hop.pipeline.transform.BaseTransform;
import org.apache.hop.pipeline.transform.IRowHandler;
import org.apache.hop.pipeline.transform.IRowListener;
+import org.apache.hop.pipeline.transform.IRowToListener;
/**
* Non-blocking row I/O for single-threaded mini-pipelines on Spark executors
(same idea as Beam's
@@ -105,9 +106,10 @@ public class SparkRowHandler implements IRowHandler {
@Override
public void putRowTo(IRowMeta rowMeta, Object[] row, IRowSet rowSet)
throws HopTransformException {
- List<IRowListener> rowListeners = transform.getRowListeners();
- for (IRowListener listener : rowListeners) {
- listener.rowWrittenEvent(rowMeta, row);
+ // Destination-aware listeners (not IRowListener — mixed layouts need the
target rowset)
+ List<IRowToListener> rowToListeners = transform.getRowToListeners();
+ for (IRowToListener listener : rowToListeners) {
+ listener.rowWrittenTo(rowMeta, row, rowSet);
}
rowSet.putRow(rowMeta, row);
transform.incrementLinesWritten();
diff --git
a/plugins/transforms/javascript/src/main/java/org/apache/hop/pipeline/transforms/javascript/ScriptValuesDummy.java
b/plugins/transforms/javascript/src/main/java/org/apache/hop/pipeline/transforms/javascript/ScriptValuesDummy.java
index 2606272718..41e303e52c 100644
---
a/plugins/transforms/javascript/src/main/java/org/apache/hop/pipeline/transforms/javascript/ScriptValuesDummy.java
+++
b/plugins/transforms/javascript/src/main/java/org/apache/hop/pipeline/transforms/javascript/ScriptValuesDummy.java
@@ -36,6 +36,7 @@ import org.apache.hop.pipeline.Pipeline;
import org.apache.hop.pipeline.PipelineMeta;
import org.apache.hop.pipeline.engine.EngineComponent;
import org.apache.hop.pipeline.transform.IRowListener;
+import org.apache.hop.pipeline.transform.IRowToListener;
import org.apache.hop.pipeline.transform.ITransform;
import org.apache.hop.pipeline.transform.ITransformData;
import org.apache.hop.pipeline.transform.ITransformFinishedListener;
@@ -63,6 +64,11 @@ public class ScriptValuesDummy implements ITransform {
// Do nothing
}
+ @Override
+ public void addRowToListener(IRowToListener rowToListener) {
+ // Do nothing
+ }
+
@Override
public void dispose() {
// Do nothing
@@ -147,6 +153,11 @@ public class ScriptValuesDummy implements ITransform {
return null;
}
+ @Override
+ public List<IRowToListener> getRowToListeners() {
+ return null;
+ }
+
@Override
public String getTransformPluginId() {
return null;
@@ -210,6 +221,11 @@ public class ScriptValuesDummy implements ITransform {
// Do nothing
}
+ @Override
+ public void removeRowToListener(IRowToListener rowToListener) {
+ // Do nothing
+ }
+
public void run() {
// Do nothing
}
diff --git
a/ui/src/main/java/org/apache/hop/ui/hopgui/file/pipeline/HopGuiPipelineGraph.java
b/ui/src/main/java/org/apache/hop/ui/hopgui/file/pipeline/HopGuiPipelineGraph.java
index 3d6edc3ad8..6d839e21fb 100644
---
a/ui/src/main/java/org/apache/hop/ui/hopgui/file/pipeline/HopGuiPipelineGraph.java
+++
b/ui/src/main/java/org/apache/hop/ui/hopgui/file/pipeline/HopGuiPipelineGraph.java
@@ -439,6 +439,10 @@ public class HopGuiPipelineGraph extends
HopGuiAbstractGraph
private PipelineHopMeta clickedPipelineHop;
@Getter @Setter protected Map<String, RowBuffer> outputRowsMap;
+
+ /** Hop key (origin\\tdestination) → sampled rows for target hops (Filter,
Switch/Case, …). */
+ @Getter @Setter protected Map<String, RowBuffer> outputHopRowsMap;
+
private boolean avoidContextDialog;
public void setCurrentNote(NotePadMeta ni) {
@@ -812,6 +816,10 @@ public class HopGuiPipelineGraph extends
HopGuiAbstractGraph
done = true;
break;
+ case HOP_OUTPUT_DATA:
+ done = true;
+ break;
+
case HOP_COPY_ICON:
done = true;
break;
@@ -1141,6 +1149,11 @@ public class HopGuiPipelineGraph extends
HopGuiAbstractGraph
return;
}
break;
+ case HOP_OUTPUT_DATA:
+ if (showHopOutputData(areaOwner)) {
+ return;
+ }
+ break;
case TRANSFORM_ICON:
if (startHopTransform != null) {
// Mouse up while drawing a hop candidate
@@ -1429,21 +1442,37 @@ public class HopGuiPipelineGraph extends
HopGuiAbstractGraph
return showTransformOutputData(dataTransform, rowBuffer);
}
+ public boolean showHopOutputData(AreaOwner areaOwner) {
+ PipelineHopMeta hop = (PipelineHopMeta) areaOwner.getParent();
+ RowBuffer rowBuffer = (RowBuffer) areaOwner.getOwner();
+ if (hop == null || rowBuffer == null) {
+ return false;
+ }
+ String fromName = hop.getFromTransform() != null ?
hop.getFromTransform().getName() : "?";
+ String toName = hop.getToTransform() != null ?
hop.getToTransform().getName() : "?";
+ String hopLabel = fromName + " → " + toName;
+ return showOutputDataDialog(hopLabel, hopLabel, rowBuffer);
+ }
+
private boolean showTransformOutputData(TransformMeta dataTransformMeta,
RowBuffer rowBuffer) {
+ if (dataTransformMeta == null) {
+ return false;
+ }
+ return showOutputDataDialog(
+ dataTransformMeta.getName(), dataTransformMeta.getName(), rowBuffer);
+ }
+
+ private boolean showOutputDataDialog(String titleName, String messageName,
RowBuffer rowBuffer) {
if (rowBuffer != null) {
synchronized (rowBuffer.getBuffer()) {
if (!rowBuffer.isEmpty()) {
try {
String title =
BaseMessages.getString(
- PKG,
- "PipelineGraph.ViewOutput.OutputDialog.Header",
- dataTransformMeta.getName());
+ PKG, "PipelineGraph.ViewOutput.OutputDialog.Header",
titleName);
String message =
BaseMessages.getString(
- PKG,
- "PipelineGraph.ViewOutput.OutputDialog.OutputRows.Text",
- dataTransformMeta.getName());
+ PKG,
"PipelineGraph.ViewOutput.OutputDialog.OutputRows.Text", messageName);
String prefix = "";
if (pipeline != null && pipeline.getPipelineRunConfiguration() !=
null) {
@@ -3740,6 +3769,22 @@ public class HopGuiPipelineGraph extends
HopGuiAbstractGraph
tipImage = GuiResource.getInstance().getImageData();
}
break;
+ case HOP_OUTPUT_DATA:
+ RowBuffer hopRowBuffer = (RowBuffer) areaOwner.getOwner();
+ if (hopRowBuffer != null && !hopRowBuffer.isEmpty()) {
+ PipelineHopMeta hopMeta = (PipelineHopMeta) areaOwner.getParent();
+ if (hopMeta != null
+ && hopMeta.getFromTransform() != null
+ && hopMeta.getToTransform() != null) {
+ tip.append(hopMeta.getFromTransform().getName())
+ .append(" → ")
+ .append(hopMeta.getToTransform().getName())
+ .append(Const.CR);
+ }
+ tip.append("Available output rows: " + hopRowBuffer.size());
+ tipImage = GuiResource.getInstance().getImageData();
+ }
+ break;
default:
// For plugins...
//
@@ -4030,6 +4075,7 @@ public class HopGuiPipelineGraph extends
HopGuiAbstractGraph
context.slowTransformIndicatorEnabled =
propsUi.isIndicateSlowPipelineTransformsEnabled();
context.zoomFactor = propsUi.getZoomFactor();
context.outputRowsMap = outputRowsMap;
+ context.outputHopRowsMap = outputHopRowsMap;
context.drawingBorderAroundName = propsUi.isBorderDrawnAroundCanvasNames();
context.mouseOverName = mouseOverName;
context.stateMap = stateMap;
@@ -4090,6 +4136,7 @@ public class HopGuiPipelineGraph extends
HopGuiAbstractGraph
stateMap);
pipelinePainter.setMagnification((float) (magnification *
PropsUi.getNativeZoomFactor()));
+ pipelinePainter.setOutputHopRowsMap(outputHopRowsMap);
pipelinePainter.setTransformLogMap(transformLogMap);
pipelinePainter.setStartHopTransform(startHopTransform);
pipelinePainter.setEndHopLocation(endHopLocation);
@@ -5088,7 +5135,8 @@ public class HopGuiPipelineGraph extends
HopGuiAbstractGraph
private void addRowsSamplerToPipeline(IPipelineEngine<PipelineMeta>
pipeline) {
try {
outputRowsMap = new HashMap<>();
- PipelineRowSamplerHelper.addRowSamplersToPipeline(pipeline,
outputRowsMap);
+ outputHopRowsMap = new HashMap<>();
+ PipelineRowSamplerHelper.addRowSamplersToPipeline(pipeline,
outputRowsMap, outputHopRowsMap);
} catch (Exception e) {
// Ignore: run config not local or sample type not set
}
@@ -5104,7 +5152,8 @@ public class HopGuiPipelineGraph extends
HopGuiAbstractGraph
return;
}
outputRowsMap = new HashMap<>();
- PipelineRowSamplerHelper.addRowSamplersToPipeline(pipeline, outputRowsMap);
+ outputHopRowsMap = new HashMap<>();
+ PipelineRowSamplerHelper.addRowSamplersToPipeline(pipeline, outputRowsMap,
outputHopRowsMap);
}
public void showSaveFileMessage() {
@@ -5429,8 +5478,11 @@ public class HopGuiPipelineGraph extends
HopGuiAbstractGraph
// wasn't open)
Map<String, RowBuffer> existingBuffers =
DrillDownGuiPlugin.getDataSnifferBuffersForPipeline(runningPipeline.getLogChannelId());
+ Map<String, RowBuffer> existingHopBuffers =
+
DrillDownGuiPlugin.getDataSnifferHopBuffersForPipeline(runningPipeline.getLogChannelId());
if (existingBuffers != null) {
outputRowsMap = existingBuffers;
+ outputHopRowsMap = existingHopBuffers != null ? existingHopBuffers : new
HashMap<>();
} else {
addRowsSamplerToAttachedPipeline();
}
diff --git
a/ui/src/main/java/org/apache/hop/ui/hopgui/file/shared/DrillDownGuiPlugin.java
b/ui/src/main/java/org/apache/hop/ui/hopgui/file/shared/DrillDownGuiPlugin.java
index 5ef8525bb3..a33b656b19 100644
---
a/ui/src/main/java/org/apache/hop/ui/hopgui/file/shared/DrillDownGuiPlugin.java
+++
b/ui/src/main/java/org/apache/hop/ui/hopgui/file/shared/DrillDownGuiPlugin.java
@@ -86,6 +86,7 @@ public class DrillDownGuiPlugin {
runningPipelines.clear();
runningWorkflows.clear();
dataSnifferBuffersByLogChannelId.clear();
+ dataSnifferHopBuffersByLogChannelId.clear();
}
/**
@@ -96,6 +97,13 @@ public class DrillDownGuiPlugin {
private static final Map<String, Map<String, RowBuffer>>
dataSnifferBuffersByLogChannelId =
new ConcurrentHashMap<>();
+ /**
+ * Per-run hop-level sniffer buffers (logChannelId -> hop key -> RowBuffer)
for target hops
+ * ({@code putRowTo}).
+ */
+ private static final Map<String, Map<String, RowBuffer>>
dataSnifferHopBuffersByLogChannelId =
+ new ConcurrentHashMap<>();
+
/**
* Attach row listeners to a pipeline at start so we capture output rows for
debugging. Uses the
* pipeline's run configuration (sample type and size in GUI) when
available; otherwise no
@@ -110,7 +118,10 @@ public class DrillDownGuiPlugin {
Map<String, RowBuffer> buffers =
dataSnifferBuffersByLogChannelId.computeIfAbsent(
logChannelId, k -> new ConcurrentHashMap<>());
- PipelineRowSamplerHelper.addRowSamplersToPipeline(pipeline, buffers);
+ Map<String, RowBuffer> hopBuffers =
+ dataSnifferHopBuffersByLogChannelId.computeIfAbsent(
+ logChannelId, k -> new ConcurrentHashMap<>());
+ PipelineRowSamplerHelper.addRowSamplersToPipeline(pipeline, buffers,
hopBuffers);
}
/**
@@ -121,6 +132,16 @@ public class DrillDownGuiPlugin {
return dataSnifferBuffersByLogChannelId.get(logChannelId);
}
+ /**
+ * Return hop-level data sniffer buffers for a pipeline run (if any).
+ *
+ * @param logChannelId pipeline log channel id
+ * @return hop key → RowBuffer, or null
+ */
+ public static Map<String, RowBuffer>
getDataSnifferHopBuffersForPipeline(String logChannelId) {
+ return dataSnifferHopBuffersByLogChannelId.get(logChannelId);
+ }
+
// ==================== TRANSFORM CONTEXT ====================
@GuiContextAction(
diff --git
a/ui/src/main/java/org/apache/hop/ui/hopgui/file/shared/PipelineRowSamplerHelper.java
b/ui/src/main/java/org/apache/hop/ui/hopgui/file/shared/PipelineRowSamplerHelper.java
index f8bd920cfa..632858f014 100644
---
a/ui/src/main/java/org/apache/hop/ui/hopgui/file/shared/PipelineRowSamplerHelper.java
+++
b/ui/src/main/java/org/apache/hop/ui/hopgui/file/shared/PipelineRowSamplerHelper.java
@@ -22,8 +22,11 @@ import java.util.Collections;
import java.util.LinkedList;
import java.util.Map;
import java.util.Random;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.lang3.StringUtils;
import org.apache.hop.core.Const;
+import org.apache.hop.core.IRowSet;
import org.apache.hop.core.exception.HopTransformException;
import org.apache.hop.core.exception.HopValueException;
import org.apache.hop.core.row.IRowMeta;
@@ -35,16 +38,49 @@ import
org.apache.hop.pipeline.engines.IMeasuringLocalPipelineRunConfiguration;
import org.apache.hop.pipeline.engines.local.LocalPipelineEngine;
import
org.apache.hop.pipeline.engines.local.LocalPipelineRunConfiguration.SampleType;
import org.apache.hop.pipeline.transform.RowAdapter;
+import org.apache.hop.pipeline.transform.RowToAdapter;
/**
* Helper to attach row listeners to a pipeline so output rows are sampled
into a target map (e.g.
* for execution preview or drill-down). Uses the pipeline's run configuration
(sample type and size
* in GUI).
+ *
+ * <p>Supports both transform-level samples ({@code IRowListener} / {@code
putRow}) and hop-level
+ * samples ({@code IRowToListener} / {@code putRowTo}) for target hops such as
Filter and
+ * Switch/Case.
*/
public final class PipelineRowSamplerHelper {
+ private static final String HOP_KEY_SEPARATOR = "\t";
+
private PipelineRowSamplerHelper() {}
+ /**
+ * Build a stable map key for a hop from origin transform name to
destination transform name.
+ *
+ * @param originTransformName source transform name
+ * @param destinationTransformName target transform name
+ * @return hop key used in hop sample maps
+ */
+ public static String hopKey(String originTransformName, String
destinationTransformName) {
+ return Const.NVL(originTransformName, "")
+ + HOP_KEY_SEPARATOR
+ + Const.NVL(destinationTransformName, "");
+ }
+
+ /**
+ * Build a hop key from a rowset's origin/destination names.
+ *
+ * @param rowSet the destination rowset
+ * @return hop key, or null if rowSet is null
+ */
+ public static String hopKey(IRowSet rowSet) {
+ if (rowSet == null) {
+ return null;
+ }
+ return hopKey(rowSet.getOriginTransformName(),
rowSet.getDestinationTransformName());
+ }
+
/**
* Attach row listeners to a pipeline so output rows are sampled into {@code
targetMap} (transform
* name → RowBuffer). Uses the pipeline's run configuration (sample type and
size in GUI). Does
@@ -55,6 +91,21 @@ public final class PipelineRowSamplerHelper {
*/
public static void addRowSamplersToPipeline(
IPipelineEngine<PipelineMeta> pipeline, Map<String, RowBuffer>
targetMap) {
+ addRowSamplersToPipeline(pipeline, targetMap, null);
+ }
+
+ /**
+ * Attach transform-level and optional hop-level row samplers.
+ *
+ * @param pipeline the pipeline engine
+ * @param targetMap transform name → RowBuffer (for {@code putRow} / {@code
IRowListener})
+ * @param hopTargetMap hop key → RowBuffer (for {@code putRowTo} / {@code
IRowToListener}); may be
+ * null to skip hop sampling
+ */
+ public static void addRowSamplersToPipeline(
+ IPipelineEngine<PipelineMeta> pipeline,
+ Map<String, RowBuffer> targetMap,
+ Map<String, RowBuffer> hopTargetMap) {
if (pipeline == null || targetMap == null) {
return;
}
@@ -85,60 +136,102 @@ public final class PipelineRowSamplerHelper {
return;
}
final Random random = new Random();
+ // Per-key counters for Random sampling (transform name or hop key)
+ final Map<String, AtomicInteger> rowCounters = new ConcurrentHashMap<>();
for (final String transformName : meta.getTransformNames()) {
IEngineComponent component = pipeline.findComponent(transformName, 0);
- if (component != null) {
- component.addRowListener(
- new RowAdapter() {
- int nrRows = 0;
+ if (component == null) {
+ continue;
+ }
+
+ // Transform-level samples (main putRow path)
+ component.addRowListener(
+ new RowAdapter() {
+ @Override
+ public void rowWrittenEvent(IRowMeta rowMeta, Object[] row)
+ throws HopTransformException {
+ applySample(
+ targetMap,
+ transformName,
+ rowMeta,
+ row,
+ sampleType,
+ sampleSize,
+ random,
+ rowCounters);
+ }
+ });
+ // Hop-level samples (putRowTo / target hops)
+ if (hopTargetMap != null) {
+ component.addRowToListener(
+ new RowToAdapter() {
@Override
- public void rowWrittenEvent(IRowMeta rowMeta, Object[] row)
+ public void rowWrittenTo(IRowMeta rowMeta, Object[] row, IRowSet
rowSet)
throws HopTransformException {
- RowBuffer rowBuffer = targetMap.get(transformName);
- if (rowBuffer == null) {
- rowBuffer = new RowBuffer(rowMeta);
- if (sampleType == SampleType.Last) {
- rowBuffer.setBuffer(Collections.synchronizedList(new
LinkedList<>()));
- } else {
- rowBuffer.setBuffer(Collections.synchronizedList(new
ArrayList<>()));
- }
- targetMap.put(transformName, rowBuffer);
- }
- try {
- row = rowMeta.cloneRow(row);
- } catch (HopValueException e) {
- throw new HopTransformException("Error copying row for
preview", e);
- }
- switch (sampleType) {
- case First:
- if (rowBuffer.size() < sampleSize) {
- rowBuffer.addRow(row);
- }
- break;
- case Last:
- rowBuffer.addRow(0, row);
- if (rowBuffer.size() > sampleSize) {
- rowBuffer.removeRow(rowBuffer.size() - 1);
- }
- break;
- case Random:
- nrRows++;
- if (rowBuffer.size() < sampleSize) {
- rowBuffer.addRow(row);
- } else {
- int randomIndex = random.nextInt(nrRows);
- if (randomIndex < sampleSize) {
- rowBuffer.setRow(randomIndex, row);
- }
- }
- break;
- default:
- break;
+ String key = hopKey(rowSet);
+ if (StringUtils.isEmpty(key) || key.equals(HOP_KEY_SEPARATOR))
{
+ return;
}
+ applySample(
+ hopTargetMap, key, rowMeta, row, sampleType, sampleSize,
random, rowCounters);
}
});
}
}
}
+
+ /** Apply First/Last/Random sampling into the map. */
+ private static void applySample(
+ Map<String, RowBuffer> targetMap,
+ String key,
+ IRowMeta rowMeta,
+ Object[] row,
+ SampleType sampleType,
+ int sampleSize,
+ Random random,
+ Map<String, AtomicInteger> rowCounters)
+ throws HopTransformException {
+ RowBuffer rowBuffer = targetMap.get(key);
+ if (rowBuffer == null) {
+ rowBuffer = new RowBuffer(rowMeta);
+ if (sampleType == SampleType.Last) {
+ rowBuffer.setBuffer(Collections.synchronizedList(new LinkedList<>()));
+ } else {
+ rowBuffer.setBuffer(Collections.synchronizedList(new ArrayList<>()));
+ }
+ targetMap.put(key, rowBuffer);
+ }
+ try {
+ row = rowMeta.cloneRow(row);
+ } catch (HopValueException e) {
+ throw new HopTransformException("Error copying row for preview", e);
+ }
+ switch (sampleType) {
+ case First:
+ if (rowBuffer.size() < sampleSize) {
+ rowBuffer.addRow(row);
+ }
+ break;
+ case Last:
+ rowBuffer.addRow(0, row);
+ if (rowBuffer.size() > sampleSize) {
+ rowBuffer.removeRow(rowBuffer.size() - 1);
+ }
+ break;
+ case Random:
+ int nrRows = rowCounters.computeIfAbsent(key, k -> new
AtomicInteger(0)).incrementAndGet();
+ if (rowBuffer.size() < sampleSize) {
+ rowBuffer.addRow(row);
+ } else {
+ int randomIndex = random.nextInt(nrRows);
+ if (randomIndex < sampleSize) {
+ rowBuffer.setRow(randomIndex, row);
+ }
+ }
+ break;
+ default:
+ break;
+ }
+ }
}
