ElementaryOperator extends Operator<Self> { @SuppressWarnings({ "unchecked", "rawtypes" }) protected eu.stratosphere.api.common.operators.Operator getOperator(final SopremoRecordLayout layout) { final Class<? extends Function> stubClass = this.getFunctionClass(); if (stubClass == null) throw new IllegalStateException("no implementing stub found"); final Class<? extends eu.stratosphere.api.common.operators.Operator> contractClass = OperatorUtil.getContractClass(stubClass); if (contractClass == null) throw new IllegalStateException("no associated contract found"); final String name = this.toString(); try { if (contractClass == ReduceOperatorBase.class) { final int[] keyIndices = this.getKeyIndices(layout, this.getKeyExpressions(0)); final SopremoReduceOperator contract = new SopremoReduceOperator(this, stubClass, keyIndices, name); if (!this.getInnerGroupOrder(0).isEmpty()) contract.setInnerGroupOrder(this.createOrdering(layout, this.getInnerGroupOrder(0))); return contract; } else if (contractClass == CoGroupOperatorBase.class) { final int[] keyIndices1 = this.getKeyIndices(layout, this.getKeyExpressions(0)); final int[] keyIndices2 = this.getKeyIndices(layout, this.getKeyExpressions(1)); final SopremoCoGroupOperator contract = new SopremoCoGroupOperator(this, stubClass, keyIndices1, keyIndices2, name); if (!this.getInnerGroupOrder(0).isEmpty()) contract.setFirstInnerGroupOrdering(this.createOrdering(layout, this.getInnerGroupOrder(0))); if (!this.getInnerGroupOrder(1).isEmpty()) contract.setSecondInnerGroupOrdering(this.createOrdering(layout, this.getInnerGroupOrder(1))); return contract; } else if (contractClass == JoinOperatorBase.class) { final int[] keyIndices1 = this.getKeyIndices(layout, this.getKeyExpressions(0)); final int[] keyIndices2 = this.getKeyIndices(layout, this.getKeyExpressions(1)); return new JoinOperatorBase(stubClass, keyIndices1, keyIndices2, name); } else if (contractClass == MapOperatorBase.class) return new MapOperatorBase(stubClass, name); else if (contractClass == CrossOperatorBase.class) return new CrossOperatorBase(stubClass, name); else throw new UnsupportedOperationException("Unknown contract type"); } catch (final Exception e) { throw new IllegalStateException("Cannot create contract from stub " + stubClass, e); } } ElementaryOperator();  ElementaryOperator(final int inputs);  ElementaryOperator(final int minInputs, final int maxInputs); @Override void appendAsString(final Appendable appendable); @Override ElementarySopremoModule asElementaryOperators(); PactModule asPactModule(); @Override boolean equals(final Object obj); Set<EvaluationExpression> getAllKeyExpressions(); @SuppressWarnings("unchecked") List<OrderingExpression> getInnerGroupOrder(final int inputIndex); @SuppressWarnings("unchecked") List<? extends EvaluationExpression> getKeyExpressions(final int inputIndex); EvaluationExpression getResultProjection(); @Override int hashCode(); boolean isCombinable(); boolean isCombinableFirst(); boolean isCombinableSecond(); void setCombinable(final boolean combinable); void setCombinableFirst(final boolean combinableFirst); void setCombinableSecond(final boolean combinableSecond); void setInnerGroupOrder(final int inputIndex, final List<OrderingExpression> innerGroupOrder); void setInnerGroupOrder(final int inputIndex, final OrderingExpression... innerGroupOrder); void setKeyExpressions(final int index, final EvaluationExpression... keyExpressions); void setKeyExpressions(final int inputIndex,
			final List<? extends EvaluationExpression> keyExpressions); @Property @Name(preposition = "into") void setResultProjection(final EvaluationExpression resultProjection); Self withCombinable(final boolean combinable); Self withCombinableFirst(final boolean combinableFirst); void withCombinableSecond(final boolean combinableSecond); Self withInnerGroupOrdering(final int index, final List<OrderingExpression> innerGroupOrder); Self withInnerGroupOrdering(final int index, final OrderingExpression... innerGroupOrder); Self withKeyExpression(final int index, final EvaluationExpression... keyExpressions); Self withKeyExpressions(final int index, final List<? extends EvaluationExpression> keyExpressions); Self withResultProjection(final EvaluationExpression resultProjection);  }
@Test public void getOperatorShouldReturnTheJoiningOperatorToTheFirstFunction() { final SopremoRecordLayout layout = SopremoRecordLayout.create(new ObjectAccess("someField")); final eu.stratosphere.api.common.operators.Operator contract = new OperatorWithTwoFunctions().getOperator(layout); assertEquals(SopremoReduceOperator.class, contract.getClass()); assertTrue(Arrays.asList(OperatorWithTwoFunctions.Implementation1.class, OperatorWithTwoFunctions.Implementation2.class).contains(contract.getUserCodeWrapper().getUserCodeClass())); }
@Test public void getOperatorShouldReturnTheJoiningOperatorToTheOnlyFunction() { final eu.stratosphere.api.common.operators.Operator contract = new OperatorWithOneFunction().getOperator(LAYOUT); assertEquals(MapOperatorBase.class, contract.getClass()); assertEquals(OperatorWithOneFunction.Implementation.class, contract.getUserCodeWrapper().getUserCodeClass()); }
@Test(expected = IllegalStateException.class) public void getOperatorShouldFailIfNoFunction() { new OperatorWithNoFunctions().getOperator(LAYOUT); }
@Test(expected = IllegalStateException.class) public void getOperatorShouldFailIfOnlyInstanceFunction() { new OperatorWithInstanceFunction().getOperator(LAYOUT); }
@Test(expected = IllegalStateException.class) public void getOperatorShouldFailIfOnlyUnknownFunction() { new OperatorWithUnknownFunction().getOperator(LAYOUT); }
@Test(expected = IllegalStateException.class) public void getOperatorShouldFailIfOperatorNotInstancable() { new OperatorWithUninstantiableFunction().getOperator(LAYOUT); }