com.pervasive.datarush.operators.join

Class CrossJoin

java.lang.Object

com.pervasive.datarush.operators.AbstractLogicalOperator

com.pervasive.datarush.operators.StreamingOperator

com.pervasive.datarush.operators.ExecutableOperator

com.pervasive.datarush.operators.join.CrossJoin

All Implemented Interfaces:

LogicalOperator

public final class CrossJoin
extends ExecutableOperator

Produce the cartesian product of two sets of records. The output is typed using the merge of the two input types; if the right type contains a field already named in the left type, it will be renamed to avoid collision. See TypeUtil.merge(RecordTokenType...) for details on the renaming process.

To generate pairs, the right-hand data is temporarily stored to disk for multiple passes. The left-hand data is read into memory in chunks, which are then used to generate a full set of pairs with the iterable right-hand data. This process is repeated until the left-hand data is exhausted.

No guarantee is made in respect to the ordering of the output data.

Constructor Summary

Constructors

Constructor and Description
CrossJoin() Performs a cross join between two inputs

Method Summary

Modifier and Type	Method and Description
protected void	computeMetadata(StreamingMetadataContext ctx) Implementations must adhere to the following contracts
protected void	execute(ExecutionContext ctx) The controller handles chunking the LHS data into reasonable chunks that can be loaded into memory.
int	getBufferRows() Gets the size (in rows) of the memory buffer used to generate output records.
RecordPort	getLeftInput() Returns the left input port
RecordPort	getOutput() Returns the input port
RecordPort	getRightInput() Returns the right input port
void	setBufferRows(int rows) Sets the size (in rows) of the memory buffer used to generate output record.

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

cloneForExecution, getNumInputCopies, getPortSettings, handleInactiveOutput

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

Constructor Detail

CrossJoin

public CrossJoin()

Performs a cross join between two inputs

Method Detail

getLeftInput

public RecordPort getLeftInput()

Returns the left input port

Returns:

the left input port

getRightInput

public RecordPort getRightInput()

Returns the right input port

Returns:

the right input port

getOutput

public RecordPort getOutput()

Returns the input port

Returns:

the output port

setBufferRows

public void setBufferRows(int rows)

Sets the size (in rows) of the memory buffer used to generate output record. Larger values can increase performance due to decreased intermediate file buffering.

Parameters:

rows - number of rows to store in memory buffer

getBufferRows

public int getBufferRows()

Gets the size (in rows) of the memory buffer used to generate output records.

Returns:

the number of rows allowed in the memory buffer

computeMetadata

protected void computeMetadata(StreamingMetadataContext ctx)

Description copied from class: StreamingOperator

Implementations must adhere to the following contracts

General

Regardless of input ports/output port types, all implementations must do the following:

1. Validation. Validation of configuration should always be performed first.
2. Declare parallelizability.. Implementations must declare parallelizability by calling StreamingMetadataContext.parallelize(ParallelismStrategy).

Input record ports

Implementations with input record ports must declare the following:

1. Required data ordering:
Implementations that have data ordering requirements must declare them by calling RecordPort#setRequiredDataOrdering, otherwise data may arrive in any order.
2. Required data distribution (only applies to parallelizable operators):
Implementations that have data distribution requirements must declare them by calling RecordPort#setRequiredDataDistribution, otherwise data will arrive in an unspecified partial distribution.

Note that if the upstream operator's output distribution/ordering is compatible with those required, we avoid a re-sort/re-distribution which is generally a very large savings from a performance standpoint. In addition, some operators may chose to query the upstream output distribution/ordering by calling RecordPort#getSourceDataDistribution and RecordPort#getSourceDataOrdering. These should be viewed as a hints to help chose a more efficient algorithm. In such cases, though, operators must still declare data ordering and data distribution requirements; otherwise there is no guarantee that data will arrive sorted/distributed as required.

Output record ports

Implementations with output record ports must declare the following:

1. Type: Implementations must declare their output type by calling RecordPort#setType.

Implementations with output record ports may declare the following:

1. Output data ordering: Implementations that can make guarantees as to their output ordering may do so by calling RecordPort#setOutputDataOrdering
2. Output data distribution (only applies to parallelizable operators): Implementations that can make guarantees as to their output distribution may do so by calling RecordPort#setOutputDataDistribution

Note that both of these properties are optional; if unspecified, performance may suffer since the framework may unnecessarily re-sort/re-distributed the data.

Input model ports

In general, there is nothing special to declare for input model ports. Models are implicitly duplicated to all partitions when going from non-parallel to parallel operators. The case of a model going from a parallel to a non-parallel node is a special case of a "model reducer" operator. In the case of a model reducer, the downstream operator, must declare the following:

1. Merge handler: Model reducers must declare a merge handler by calling AbstractModelPort#setMergeHandler.

Note that MergeModel is a convenient, re-usable model reducer, parameterized with a merge-handler.

Output model ports

SimpleModelPort's have no associated metadata and therefore there is never any output metadata to declare. PMMLPort's, on the other hand, do have associated metadata. For all PMMLPorts, implementations must declare the following:

1. pmmlModelSpec: Implementations must declare the PMML model spec by calling PMMLPort.setPMMLModelSpec.

Specified by:

computeMetadata in class StreamingOperator

Parameters:

ctx - the context

execute

protected void execute(ExecutionContext ctx)

The controller handles chunking the LHS data into reasonable chunks that can be loaded into memory. The LHS chunks are then joined (cartesian product) against an iteration of the RHS data. Having to use an application graph and export many ports from that graph, the partitioning has to be handled directly. Subgraphs are run until the LHS data is exhausted.

Specified by:

execute in class ExecutableOperator

Parameters:

ctx - context in which to lookup physical ports bound to logical ports