com.pervasive.datarush.operators.io.paraccel

Class MatrixOperatorBase<ExecState>

java.lang.Object

com.pervasive.datarush.operators.AbstractLogicalOperator

com.pervasive.datarush.operators.CompositeOperator

com.pervasive.datarush.operators.io.paraccel.MatrixOperatorBase<ExecState>

All Implemented Interfaces:

LogicalOperator, RecordSinkOperator, SinkOperator<RecordPort>

Direct Known Subclasses:

LoadMatrix, MatrixSink

public abstract class MatrixOperatorBase<ExecState>
extends CompositeOperator
implements RecordSinkOperator

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
protected static class	MatrixOperatorBase.Pair<A,B>
protected static class	MatrixOperatorBase.Workers<E>

Field Summary

Fields

Modifier and Type	Field and Description
static int	END_PORT_RANGE
protected String	host
protected int	logFrequency
protected static org.apache.commons.logging.Log	logger
protected int	maxConnections
protected int	retryCount
static int	START_PORT_RANGE
protected int	timeoutInterval

Constructor Summary

Constructors

Constructor and Description
MatrixOperatorBase()

Method Summary

Modifier and Type	Method and Description
protected void	compose(CompositionContext ctx) Compose the body of this operator.
protected abstract MatrixOperatorBase.Workers<ExecState>	createWorkers(CompositionContext ctx, RecordTokenType inputType, com.pervasive.datarush.operators.io.paraccel.TableInfo tableInfo, List<String> columns, Map<String,String> columnMap, int startPortRange, int endPortRange, int dataWorkerParallelism)
String	getHost() Get the server host name property.
RecordPort	getInput() Gets the record port providing the input data to the sink.
int	getMaxConnections() Get the max connections property.
protected abstract MatrixOperatorBase.Pair<Integer,Integer>	portRange()
protected abstract Map<String,String>	renameMapping()
protected abstract MatrixOperatorBase.Pair<com.pervasive.datarush.operators.io.paraccel.TableInfo,List<com.pervasive.datarush.operators.io.paraccel.SliceIdentifier>>	schemaAndSliceInfo()
void	setHost(String hostName) Set the host name property.
void	setMaxConnections(int maxConnections) Set the max connections property.
protected abstract boolean	skipCompose()
protected abstract void	validateExtensions()
protected void	validatePortRange(String startPortName, int startPort, int endPort)

Methods inherited from class com.pervasive.datarush.operators.AbstractLogicalOperator

disableParallelism, getInputPorts, getOutputPorts, newInput, newInput, newOutput, newRecordInput, newRecordInput, newRecordOutput, notifyError

Methods inherited from class java.lang.Object

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

Methods inherited from interface com.pervasive.datarush.operators.LogicalOperator

disableParallelism, getInputPorts, getOutputPorts

Field Detail

START_PORT_RANGE

public static final int START_PORT_RANGE

See Also:

Constant Field Values

END_PORT_RANGE

public static final int END_PORT_RANGE

See Also:

Constant Field Values

host

protected String host

logFrequency

protected int logFrequency

retryCount

protected int retryCount

timeoutInterval

protected int timeoutInterval

maxConnections

protected int maxConnections

logger

protected static final org.apache.commons.logging.Log logger

Constructor Detail

MatrixOperatorBase

public MatrixOperatorBase()

Method Detail

getInput

public RecordPort getInput()

Description copied from interface: RecordSinkOperator

Gets the record port providing the input data to the sink.

Specified by:

getInput in interface RecordSinkOperator

Specified by:

getInput in interface SinkOperator<RecordPort>

Returns:

the input port for the sink

getHost

public String getHost()

Get the server host name property.

Returns:

server host name

setHost

public void setHost(String hostName)

Set the host name property. This is the host name of the server where Matrix is installed.

Parameters:

hostName - Matrix server host name

getMaxConnections

public int getMaxConnections()

Get the max connections property.

Returns:

max connections property value

setMaxConnections

public void setMaxConnections(int maxConnections)

Set the max connections property. The given value limits the number of connections to make to Matrix for data transfers. Defaults to zero which implies a connection per Matrix slice.

Note that the number of connections is limited by setting the parallelism of the data worker operator of the LoadMatrix instance. This may incur the cost of data redistribution to match the parallelism specified.

Parameters:

maxConnections - maximum number of connections to use for data transfer

compose

protected void compose(CompositionContext ctx)

Description copied from class: CompositeOperator

Compose the body of this operator. Implementations should do the following:

1. Perform any validation of configuration, input types, etc
2. Instantiate and configure sub-operators, adding them to the provided context via the method OperatorComposable.add(O)
3. Create necessary connections via the method OperatorComposable.connect(P, P). This includes connections from the composite's input ports to sub-operators, connections between sub-operators, and connections from sub-operators output ports to the composite's output ports

Specified by:

compose in class CompositeOperator

Parameters:

ctx - the context

renameMapping

protected abstract Map<String,String> renameMapping()

skipCompose

protected abstract boolean skipCompose()

validatePortRange

protected void validatePortRange(String startPortName,
                                 int startPort,
                                 int endPort)

createWorkers

protected abstract MatrixOperatorBase.Workers<ExecState> createWorkers(CompositionContext ctx,
                                                                       RecordTokenType inputType,
                                                                       com.pervasive.datarush.operators.io.paraccel.TableInfo tableInfo,
                                                                       List<String> columns,
                                                                       Map<String,String> columnMap,
                                                                       int startPortRange,
                                                                       int endPortRange,
                                                                       int dataWorkerParallelism)

validateExtensions

protected abstract void validateExtensions()

schemaAndSliceInfo

protected abstract MatrixOperatorBase.Pair<com.pervasive.datarush.operators.io.paraccel.TableInfo,List<com.pervasive.datarush.operators.io.paraccel.SliceIdentifier>> schemaAndSliceInfo()

portRange

protected abstract MatrixOperatorBase.Pair<Integer,Integer> portRange()