org.knime.core.node.streamable.simple

Class SimpleStreamableFunctionWithInternalsNodeModel<T extends StreamableOperatorInternals>

java.lang.Object

org.knime.core.node.NodeModel

org.knime.core.node.streamable.simple.SimpleStreamableFunctionNodeModel

org.knime.core.node.streamable.simple.SimpleStreamableFunctionWithInternalsNodeModel<T>

Type Parameters:

T - The sub type of the internals used by the implementation

All Implemented Interfaces:

StreamableFunctionProducer

Direct Known Subclasses:

DoubleToIntNodeModel, DoubleToIntNodeModel, RegexSplitNodeModel, StringToNumberNodeModel

public abstract class SimpleStreamableFunctionWithInternalsNodeModel<T extends StreamableOperatorInternals>
extends SimpleStreamableFunctionNodeModel

Abstract class of node that computes a simple (1:1) function but needs to have some final processing afterwards (mostly only setting an error message).

Since:

2.6

Author:

Bernd Wiswedel, KNIME.com, Zurich, Switzerland

Constructor Summary

Constructors

Constructor and Description
SimpleStreamableFunctionWithInternalsNodeModel(Class<T> cl) New node model (one in, one out).
SimpleStreamableFunctionWithInternalsNodeModel(Class<T> cl, PortType[] inPortTypes, PortType[] outPortTypes, int streamableInPortIdx, int streamableOutPortIdx) Constructor for a node with multiple in or out ports.

Method Summary

Modifier and Type	Method and Description
protected ColumnRearranger	createColumnRearranger(DataTableSpec spec) Creates a column rearranger that describes the changes to the input table.
protected abstract ColumnRearranger	createColumnRearranger(DataTableSpec spec, T emptyInternals) Extends the behavior of SimpleStreamableFunctionNodeModel.createColumnRearranger(DataTableSpec) by an empty internals object that is filled while processing the data.
MergeOperator	createMergeOperator() Streaming API (pending): Factory method to create a merge operator that combines results created in different StreamableOperator objects.
StreamableFunction	createStreamableOperator(PartitionInfo partitionInfo, PortObjectSpec[] inSpecs) Streaming API (pending): Factory method for a streamable operator that is used to execute this node.
protected T	createStreamingOperatorInternals() Creates new empty instance of the internals.
void	finishStreamableExecution(StreamableOperatorInternals internals, ExecutionContext exec, PortOutput[] output) Streaming API (pending): Called by the executor if the data is processed in a distributed fashion to create the final output result or update node internals (for instance a warning message or view content).
protected abstract void	finishStreamableExecution(T operatorInternals) Finalizes execution with a merged internals object.
protected abstract T	mergeStreamingOperatorInternals(T[] operatorInternals) Called the merge operator to merge internals created by different streamble operators (possibly on remote machines).

Methods inherited from class org.knime.core.node.streamable.simple.SimpleStreamableFunctionNodeModel

configure, execute, getInputPortRoles, getOutputPortRoles, getStreamableInPortIdx, getStreamableOutPortIdx, isDistributable, loadInternals, reset, saveInternals

Methods inherited from class org.knime.core.node.NodeModel

addWarningListener, computeFinalOutputSpecs, configure, continueLoop, createInitialStreamableOperatorInternals, execute, getAvailableFlowVariables, getAvailableInputFlowVariables, getCredentialsProvider, getInHiLiteHandler, getInPortType, getInteractiveNodeView, getLogger, getLoopEndNode, getLoopStartNode, getNrInPorts, getNrOutPorts, getOutHiLiteHandler, getOutPortType, getWarningMessage, iterate, loadValidatedSettingsFrom, notifyViews, notifyWarningListeners, onDispose, peekFlowVariableDouble, peekFlowVariableInt, peekFlowVariableString, pushFlowVariableDouble, pushFlowVariableInt, pushFlowVariableString, removeWarningListener, resetAndConfigureLoopBody, saveSettingsTo, setInHiLiteHandler, setWarningMessage, stateChanged, validateSettings

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

SimpleStreamableFunctionWithInternalsNodeModel

public SimpleStreamableFunctionWithInternalsNodeModel(Class<T> cl)

New node model (one in, one out).

Parameters:

cl - The class of the StreamableOperatorInternals (used for instantiation of arrays and new instances).

SimpleStreamableFunctionWithInternalsNodeModel

public SimpleStreamableFunctionWithInternalsNodeModel(Class<T> cl,
                                                      PortType[] inPortTypes,
                                                      PortType[] outPortTypes,
                                                      int streamableInPortIdx,
                                                      int streamableOutPortIdx)

Constructor for a node with multiple in or out ports.

Parameters:

cl - The class of the StreamableOperatorInternals (used for instantiation of arrays and new instances).

inPortTypes - in-port types. The ports at the index streamableInPortIdx MUST be a non-optional BufferedDataTable!

outPortTypes - out-port types.The ports at the index streamableOutPortIdx MUST be a non-optional BufferedDataTable!

streamableInPortIdx - the index of the port that is streamable

streamableOutPortIdx - the index of the port that is streamable

Since:

3.1

Method Detail

createStreamingOperatorInternals

protected T createStreamingOperatorInternals()

Creates new empty instance of the internals. Default implementation uses reflection and calls no-arg constructor.

Returns:

A new instance of the internals (not null!)

mergeStreamingOperatorInternals

protected abstract T mergeStreamingOperatorInternals(T[] operatorInternals)

Called the merge operator to merge internals created by different streamble operators (possibly on remote machines).

Parameters:

operatorInternals - The internals to merge.

Returns:

A new merged internals object.

finishStreamableExecution

protected abstract void finishStreamableExecution(T operatorInternals)

Finalizes execution with a merged internals object. Clients can access its fields and update view content or set warning messages.

Parameters:

operatorInternals - The merged internals object.

createColumnRearranger

protected final ColumnRearranger createColumnRearranger(DataTableSpec spec)
                                                 throws InvalidSettingsException

Creates a column rearranger that describes the changes to the input table. Sub classes will check the consistency of the input table with their settings (fail with InvalidSettingsException if necessary) and then return a customized ColumnRearranger.

Specified by:

createColumnRearranger in class SimpleStreamableFunctionNodeModel

Parameters:

spec - The spec of the input table.

Returns:

A column rearranger describing the changes, never null.

Throws:

InvalidSettingsException - If the settings or the input are invalid.

createStreamableOperator

public StreamableFunction createStreamableOperator(PartitionInfo partitionInfo,
                                                   PortObjectSpec[] inSpecs)
                                            throws InvalidSettingsException

Streaming API (pending): Factory method for a streamable operator that is used to execute this node. The default implementation returns a standard operator that wraps the NodeModel.execute(PortObject[], ExecutionContext) method. Subclasses may override it to return a new operator that follows the data handling described by the NodeModel.getInputPortRoles() and NodeModel.getOutputPortRoles() methods.

This method is called by the node executor once or multiple times depending on the input roles. If any input is distributable, the method is called multiple times (for each partition once), possibly on different (remote) clones of this NodeModel.

Specified by:

createStreamableOperator in interface StreamableFunctionProducer

Overrides:

createStreamableOperator in class SimpleStreamableFunctionNodeModel

Parameters:

partitionInfo - The partition info describing the chunk (if distributable).

inSpecs - The port object specs of the input ports. These are identical to the specs that NodeModel.configure(PortObjectSpec[]) was last called with (also on the remote side).

Returns:

A new operator for the (chunk) execution.

Throws:

InvalidSettingsException - Usually not thrown in the client but still part of the method signature as implementations often run the same methods as during configure. (This method is not being called when configure fails.)

finishStreamableExecution

public void finishStreamableExecution(StreamableOperatorInternals internals,
                                      ExecutionContext exec,
                                      PortOutput[] output)
                               throws Exception

Streaming API (pending): Called by the executor if the data is processed in a distributed fashion to create the final output result or update node internals (for instance a warning message or view content). This method is called on the local side. See also the API description for NodeModel.getOutputPortRoles().

Overrides:

finishStreamableExecution in class NodeModel

Parameters:

internals - The merged internals of the streamable operators that processed the data. The internals object is created by one or multiple MergeOperator (as created by NodeModel.createMergeOperator()).

exec - For progress reporting, cancelation, output creation.

output - The array of the output representations. This method must only write to the slots that it is responsible for (non-distributed output).

Throws:

Exception - Any exception to indicate an error, including cancelation.

createColumnRearranger

protected abstract ColumnRearranger createColumnRearranger(DataTableSpec spec,
                                                           T emptyInternals)
                                                    throws InvalidSettingsException

Extends the behavior of SimpleStreamableFunctionNodeModel.createColumnRearranger(DataTableSpec) by an empty internals object that is filled while processing the data.

Parameters:

spec - ...

emptyInternals - The empty internals. Should be passed on to the cell factory (and filled in the AbstractCellFactory.afterProcessing() method).

Returns:

...

Throws:

InvalidSettingsException - ...

createMergeOperator

public MergeOperator createMergeOperator()

Streaming API (pending): Factory method to create a merge operator that combines results created in different StreamableOperator objects. This method must be overwritten if the input is distributable but the output is not (as it needs to prepare the final output that is then published by the NodeModel.finishStreamableExecution(StreamableOperatorInternals, ExecutionContext, PortOutput[]) method). It may be overwritten if the output is distributable but some work needs to be done after all operators have finished (e.g. setting some internals or warning message).

The default implementation returns null because input and output are non distributable.

Overrides:

createMergeOperator in class NodeModel

Returns:

A new merge operator or null.