@@ -0,0 +1,3080 @@
+package edu.washington.escience.myria.operator.apply;%0A%0Aimport static org.junit.Assert.assertEquals;%0A%0Aimport java.util.Iterator;%0Aimport java.util.List;%0Aimport java.util.stream.Stream;%0A%0Aimport org.junit.Test;%0A%0Aimport com.google.common.base.Joiner;%0Aimport com.google.common.collect.ImmutableList;%0A%0Aimport edu.washington.escience.myria.DbException;%0Aimport edu.washington.escience.myria.Schema;%0Aimport edu.washington.escience.myria.Type;%0Aimport edu.washington.escience.myria.expression.ConstantExpression;%0Aimport edu.washington.escience.myria.expression.Expression;%0Aimport edu.washington.escience.myria.expression.ExpressionOperator;%0Aimport edu.washington.escience.myria.expression.SplitExpression;%0Aimport edu.washington.escience.myria.expression.VariableExpression;%0Aimport edu.washington.escience.myria.operator.Apply;%0Aimport edu.washington.escience.myria.operator.TupleSource;%0Aimport edu.washington.escience.myria.storage.TupleBatch;%0Aimport edu.washington.escience.myria.storage.TupleBatchBuffer;%0Aimport edu.washington.escience.myria.util.TestEnvVars;%0A%0Apublic class CartesianProductApplyTest %7B%0A%0A  private final String SEPARATOR = %22,%22;%0A%0A  @Test%0A  public void testApply() throws DbException %7B%0A    final Schema schema = Schema.ofFields(%22col_1%22, Type.STRING_TYPE, %22col_2%22, Type.STRING_TYPE);%0A    final Schema expectedResultSchema =%0A        Schema.ofFields(%22col_1_split%22, Type.STRING_TYPE, %22col_2_split%22, Type.STRING_TYPE);%0A    final TupleBatchBuffer input = new TupleBatchBuffer(schema);%0A%0A    final ImmutableList%3CString%3E input1 = ImmutableList.of(%22a%22, %22b%22, %22c%22);%0A    final ImmutableList%3CString%3E input2 = ImmutableList.of(%22d%22, %22e%22, %22f%22);%0A%0A    Stream%3CList%3CString%3E%3E inputPairsStream =%0A        input1.stream().flatMap(i1 -%3E input2.stream().map(i2 -%3E ImmutableList.of(i1, i2)));%0A    Iterator%3CList%3CString%3E%3E inputPairs = inputPairsStream.iterator();%0A%0A    input.putString(0, Joiner.on(SEPARATOR).join(input1));%0A    input.putString(1, Joiner.on(SEPARATOR).join(input2));%0A    ImmutableList.Builder%3CExpression%3E Expressions = ImmutableList.builder();%0A%0A    ExpressionOperator regex = new ConstantExpression(SEPARATOR);%0A    ExpressionOperator col1splits = new SplitExpression(new VariableExpression(0), regex);%0A    ExpressionOperator col2splits = new SplitExpression(new VariableExpression(1), regex);%0A    Expressions.add(new Expression(%22col_1_split%22, col1splits));%0A    Expressions.add(new Expression(%22col_2_split%22, col2splits));%0A%0A    Apply apply = new Apply(new TupleSource(input), Expressions.build());%0A    apply.open(TestEnvVars.get());%0A    int rowIdx = 0;%0A    while (!apply.eos()) %7B%0A      TupleBatch result = apply.nextReady();%0A      if (result != null) %7B%0A        assertEquals(expectedResultSchema, result.getSchema());%0A        for (int batchIdx = 0; batchIdx %3C result.numTuples(); ++batchIdx, ++rowIdx) %7B%0A          List%3CString%3E pair = inputPairs.next();%0A          assertEquals(pair.get(0), result.getString(0, batchIdx));%0A          assertEquals(pair.get(1), result.getString(1, batchIdx));%0A        %7D%0A      %7D%0A    %7D%0A    assertEquals(input1.size() * input2.size(), rowIdx);%0A    apply.close();%0A  %7D%0A%7D%0A

add test for cartesian product semantics of flatmap