You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by sr...@apache.org on 2016/07/10 21:30:13 UTC
[1/3] storm git commit: [STORM-1833] Simple equi-join in storm-sql
standalone
Repository: storm
Updated Branches:
refs/heads/master 629ac13a3 -> 122bbefa2
[STORM-1833] Simple equi-join in storm-sql standalone
Added support for simple equi-joins with single condition in storm-sql
standalone mode.
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/2282fb0c
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/2282fb0c
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/2282fb0c
Branch: refs/heads/master
Commit: 2282fb0c68263cbb1396e9f949e35ce910331236
Parents: 3fe482e
Author: Arun Mahadevan <ai...@hortonworks.com>
Authored: Wed May 4 20:01:24 2016 +0530
Committer: Arun Mahadevan <ai...@hortonworks.com>
Committed: Wed May 18 15:24:58 2016 +0530
----------------------------------------------------------------------
.../sql/compiler/PostOrderRelNodeVisitor.java | 6 +
.../backends/standalone/PlanCompiler.java | 33 +++--
.../backends/standalone/RelNodeCompiler.java | 122 ++++++++++++++++++-
.../test/org/apache/storm/sql/TestStormSql.java | 80 ++++++++++++
.../sql/runtime/AbstractChannelHandler.java | 5 +
.../storm/sql/runtime/ChannelContext.java | 1 +
.../storm/sql/runtime/ChannelHandler.java | 2 +
.../org/apache/storm/sql/runtime/Channels.java | 16 +++
.../test/org/apache/storm/sql/TestUtils.java | 41 +++++++
9 files changed, 287 insertions(+), 19 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/2282fb0c/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/compiler/PostOrderRelNodeVisitor.java
----------------------------------------------------------------------
diff --git a/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/compiler/PostOrderRelNodeVisitor.java b/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/compiler/PostOrderRelNodeVisitor.java
index bb7c8d1..9dcfd29 100644
--- a/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/compiler/PostOrderRelNodeVisitor.java
+++ b/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/compiler/PostOrderRelNodeVisitor.java
@@ -55,6 +55,8 @@ public abstract class PostOrderRelNodeVisitor<T> {
return visitUncollect((Uncollect) n);
} else if (n instanceof Window) {
return visitWindow((Window) n);
+ } else if (n instanceof Join) {
+ return visitJoin((Join) n);
} else {
return defaultValue(n);
}
@@ -116,6 +118,10 @@ public abstract class PostOrderRelNodeVisitor<T> {
return defaultValue(window);
}
+ public T visitJoin(Join join) throws Exception {
+ return defaultValue(join);
+ }
+
public T defaultValue(RelNode n) {
return null;
}
http://git-wip-us.apache.org/repos/asf/storm/blob/2282fb0c/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/compiler/backends/standalone/PlanCompiler.java
----------------------------------------------------------------------
diff --git a/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/compiler/backends/standalone/PlanCompiler.java b/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/compiler/backends/standalone/PlanCompiler.java
index 3a92ae5..8131934 100644
--- a/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/compiler/backends/standalone/PlanCompiler.java
+++ b/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/compiler/backends/standalone/PlanCompiler.java
@@ -47,6 +47,8 @@ public class PlanCompiler {
"import org.apache.storm.sql.runtime.ChannelHandler;",
"import org.apache.storm.sql.runtime.DataSource;",
"import org.apache.storm.sql.runtime.AbstractValuesProcessor;",
+ "import com.google.common.collect.ArrayListMultimap;",
+ "import com.google.common.collect.Multimap;",
"public final class Processor extends AbstractValuesProcessor {", "");
private static final String INITIALIZER_PROLOGUE = NEW_LINE_JOINER.join(
" @Override",
@@ -91,25 +93,20 @@ public class PlanCompiler {
pw.print(" }\n");
}
- private void chainOperators(
- PrintWriter pw, RelNode root, Set<TableScan> tables) {
- String lastCtx = "r";
- Queue<RelNode> q = new ArrayDeque<>();
- q.add(root);
- RelNode n;
- while ((n = q.poll()) != null) {
- pw.print(
- String.format(" ChannelContext CTX_%d = Channels.chain(%2$s, %3$s);\n",
- n.getId(), lastCtx, RelNodeCompiler.getStageName(n)));
- lastCtx = String.format("CTX_%d", n.getId());
-
- if (n instanceof TableScan) {
- tables.add((TableScan)n);
- }
+ private void chainOperators(PrintWriter pw, RelNode root, Set<TableScan> tables) {
+ doChainOperators(pw, root, tables, "r");
+ }
- for (RelNode i : n.getInputs()) {
- q.add(i);
- }
+ private void doChainOperators(PrintWriter pw, RelNode node, Set<TableScan> tables, String parentCtx) {
+ pw.print(
+ String.format(" ChannelContext CTX_%d = Channels.chain(%2$s, %3$s);\n",
+ node.getId(), parentCtx, RelNodeCompiler.getStageName(node)));
+ String currentCtx = String.format("CTX_%d", node.getId());
+ if (node instanceof TableScan) {
+ tables.add((TableScan) node);
+ }
+ for (RelNode i : node.getInputs()) {
+ doChainOperators(pw, i, tables, currentCtx);
}
}
http://git-wip-us.apache.org/repos/asf/storm/blob/2282fb0c/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/compiler/backends/standalone/RelNodeCompiler.java
----------------------------------------------------------------------
diff --git a/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/compiler/backends/standalone/RelNodeCompiler.java b/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/compiler/backends/standalone/RelNodeCompiler.java
index 7782529..3ede7e9 100644
--- a/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/compiler/backends/standalone/RelNodeCompiler.java
+++ b/external/sql/storm-sql-core/src/jvm/org/apache/storm/sql/compiler/backends/standalone/RelNodeCompiler.java
@@ -20,12 +20,15 @@ package org.apache.storm.sql.compiler.backends.standalone;
import com.google.common.base.Joiner;
import com.google.common.primitives.Primitives;
import org.apache.calcite.adapter.java.JavaTypeFactory;
+import org.apache.calcite.plan.RelOptUtil;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.core.Aggregate;
import org.apache.calcite.rel.core.AggregateCall;
import org.apache.calcite.rel.core.Filter;
+import org.apache.calcite.rel.core.Join;
import org.apache.calcite.rel.core.Project;
import org.apache.calcite.rel.core.TableScan;
+import org.apache.calcite.rel.logical.LogicalJoin;
import org.apache.calcite.rel.stream.Delta;
import org.apache.calcite.schema.AggregateFunction;
import org.apache.calcite.schema.impl.AggregateFunctionImpl;
@@ -38,6 +41,7 @@ import java.io.PrintWriter;
import java.io.StringWriter;
import java.lang.reflect.Type;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -88,10 +92,101 @@ class RelNodeCompiler extends PostOrderRelNodeVisitor<Void> {
" public void dataReceived(ChannelContext ctx, Values _data) {",
""
);
+
+ private static final String JOIN_STAGE_PROLOGUE = NEW_LINE_JOINER.join(
+ " private static final ChannelHandler %1$s = ",
+ " new AbstractChannelHandler() {",
+ " Object left = %2$s;",
+ " Object right = %3$s;",
+ " Object source = null;",
+ " List<Values> leftRows = new ArrayList<>();",
+ " List<Values> rightRows = new ArrayList<>();",
+ " boolean leftDone = false;",
+ " boolean rightDone = false;",
+ " int[] ordinals = new int[] {%4$s, %5$s};",
+ "",
+ " Multimap<Object, Values> getJoinTable(List<Values> rows, int joinIndex) {",
+ " Multimap<Object, Values> m = ArrayListMultimap.create();",
+ " for(Values v: rows) {",
+ " m.put(v.get(joinIndex), v);",
+ " }",
+ " return m;",
+ " }",
+ "",
+ " List<Values> join(Multimap<Object, Values> tab, List<Values> rows, int rowIdx, boolean rev) {",
+ " List<Values> res = new ArrayList<>();",
+ " for (Values row: rows) {",
+ " for (Values mapValue: tab.get(row.get(rowIdx))) {",
+ " if (mapValue != null) {",
+ " Values joinedRow = new Values();",
+ " if(rev) {",
+ " joinedRow.addAll(row);",
+ " joinedRow.addAll(mapValue);",
+ " } else {",
+ " joinedRow.addAll(mapValue);",
+ " joinedRow.addAll(row);",
+ " }",
+ " res.add(joinedRow);",
+ " }",
+ " }",
+ " }",
+ " return res;",
+ " }",
+ "",
+ " @Override",
+ " public void setSource(ChannelContext ctx, Object source) {",
+ " this.source = source;",
+ " }",
+ "",
+ " @Override",
+ " public void flush(ChannelContext ctx) {",
+ " if (source == left) {",
+ " leftDone = true;",
+ " } else if (source == right) {",
+ " rightDone = true;",
+ " }",
+ " if (leftDone && rightDone) {",
+ " if (leftRows.size() <= rightRows.size()) {",
+ " for(Values res: join(getJoinTable(leftRows, ordinals[0]), rightRows, ordinals[1], false)) {",
+ " ctx.emit(res);",
+ " }",
+ " } else {",
+ " for(Values res: join(getJoinTable(rightRows, ordinals[1]), leftRows, ordinals[0], true)) {",
+ " ctx.emit(res);",
+ " }",
+ " }",
+ " leftDone = rightDone = false;",
+ " leftRows.clear();",
+ " rightRows.clear();",
+ " super.flush(ctx);",
+ " }",
+ " }",
+ "",
+ " @Override",
+ " public void dataReceived(ChannelContext ctx, Values _data) {",
+ ""
+ );
+
private static final String STAGE_PASSTHROUGH = NEW_LINE_JOINER.join(
" private static final ChannelHandler %1$s = AbstractChannelHandler.PASS_THROUGH;",
"");
+ private static final String STAGE_ENUMERABLE_TABLE_SCAN = NEW_LINE_JOINER.join(
+ " private static final ChannelHandler %1$s = new AbstractChannelHandler() {",
+ " @Override",
+ " public void flush(ChannelContext ctx) {",
+ " ctx.setSource(this);",
+ " super.flush(ctx);",
+ " }",
+ "",
+ " @Override",
+ " public void dataReceived(ChannelContext ctx, Values _data) {",
+ " ctx.setSource(this);",
+ " ctx.emit(_data);",
+ " }",
+ " };",
+ "");
+
private int nameCount;
private Map<AggregateCall, String> aggregateCallVarNames = new HashMap<>();
@@ -143,7 +238,7 @@ class RelNodeCompiler extends PostOrderRelNodeVisitor<Void> {
@Override
public Void visitTableScan(TableScan scan) throws Exception {
- pw.print(String.format(STAGE_PASSTHROUGH, getStageName(scan)));
+ pw.print(String.format(STAGE_ENUMERABLE_TABLE_SCAN, getStageName(scan)));
return null;
}
@@ -166,6 +261,18 @@ class RelNodeCompiler extends PostOrderRelNodeVisitor<Void> {
return null;
}
+ @Override
+ public Void visitJoin(Join join) {
+ beginJoinStage(join);
+ pw.println(" if (source == left) {");
+ pw.println(" leftRows.add(_data);");
+ pw.println(" } else if (source == right) {");
+ pw.println(" rightRows.add(_data);");
+ pw.println(" }");
+ endStage();
+ return null;
+ }
+
private String groupValueEmitStr(String var, int n) {
int count = 0;
StringBuilder sb = new StringBuilder();
@@ -318,6 +425,19 @@ class RelNodeCompiler extends PostOrderRelNodeVisitor<Void> {
pw.print(String.format(AGGREGATE_STAGE_PROLOGUE, getStageName(n), getGroupByIndices(n), emitAggregateStmts(n)));
}
+ private void beginJoinStage(Join join) {
+ int[] ordinals = new int[2];
+ if (!RelOptUtil.analyzeSimpleEquiJoin((LogicalJoin) join, ordinals)) {
+ throw new UnsupportedOperationException("Only simple equi joins are supported");
+ }
+
+ pw.print(String.format(JOIN_STAGE_PROLOGUE, getStageName(join),
+ getStageName(join.getLeft()),
+ getStageName(join.getRight()),
+ ordinals[0],
+ ordinals[1]));
+ }
+
private void endStage() {
pw.print(" }\n };\n");
}
http://git-wip-us.apache.org/repos/asf/storm/blob/2282fb0c/external/sql/storm-sql-core/src/test/org/apache/storm/sql/TestStormSql.java
----------------------------------------------------------------------
diff --git a/external/sql/storm-sql-core/src/test/org/apache/storm/sql/TestStormSql.java b/external/sql/storm-sql-core/src/test/org/apache/storm/sql/TestStormSql.java
index 2dd0ba0..7b8554e 100644
--- a/external/sql/storm-sql-core/src/test/org/apache/storm/sql/TestStormSql.java
+++ b/external/sql/storm-sql-core/src/test/org/apache/storm/sql/TestStormSql.java
@@ -100,12 +100,54 @@ public class TestStormSql {
}
}
+ private static class MockEmpDataSourceProvider implements DataSourcesProvider {
+ @Override
+ public String scheme() {
+ return "mockemp";
+ }
+
+ @Override
+ public DataSource construct(
+ URI uri, String inputFormatClass, String outputFormatClass,
+ List<FieldInfo> fields) {
+ return new TestUtils.MockEmpDataSource();
+ }
+
+ @Override
+ public ISqlTridentDataSource constructTrident(URI uri, String inputFormatClass, String outputFormatClass,
+ String properties, List<FieldInfo> fields) {
+ throw new UnsupportedOperationException("Not supported");
+ }
+ }
+
+ private static class MockDeptDataSourceProvider implements DataSourcesProvider {
+ @Override
+ public String scheme() {
+ return "mockdept";
+ }
+
+ @Override
+ public DataSource construct(
+ URI uri, String inputFormatClass, String outputFormatClass,
+ List<FieldInfo> fields) {
+ return new TestUtils.MockDeptDataSource();
+ }
+
+ @Override
+ public ISqlTridentDataSource constructTrident(URI uri, String inputFormatClass, String outputFormatClass,
+ String properties, List<FieldInfo> fields) {
+ throw new UnsupportedOperationException("Not supported");
+ }
+ }
+
@BeforeClass
public static void setUp() {
DataSourcesRegistry.providerMap().put("mock", new MockDataSourceProvider());
DataSourcesRegistry.providerMap().put("mocknested", new MockNestedDataSourceProvider());
DataSourcesRegistry.providerMap().put("mockgroup", new MockGroupDataSourceProvider());
+ DataSourcesRegistry.providerMap().put("mockemp", new MockEmpDataSourceProvider());
+ DataSourcesRegistry.providerMap().put("mockdept", new MockDeptDataSourceProvider());
}
@AfterClass
@@ -341,4 +383,42 @@ public class TestStormSql {
Assert.assertEquals(0, values.get(0).get(1));
Assert.assertEquals(2L, values.get(0).get(2));
}
+
+ @Test
+ public void testjoin() throws Exception {
+ List<String> stmt = new ArrayList<>();
+ stmt.add("CREATE EXTERNAL TABLE EMP (EMPID INT PRIMARY KEY, EMPNAME VARCHAR, DEPTID INT) LOCATION 'mockemp:///foo'");
+ stmt.add("CREATE EXTERNAL TABLE DEPT (DEPTID INT PRIMARY KEY, DEPTNAME VARCHAR) LOCATION 'mockdept:///foo'");
+ stmt.add("SELECT STREAM EMPID, EMPNAME, DEPTNAME FROM EMP AS e JOIN DEPT AS d ON e.DEPTID = d.DEPTID WHERE e.empid > 0");
+ StormSql sql = StormSql.construct();
+ List<Values> values = new ArrayList<>();
+ ChannelHandler h = new TestUtils.CollectDataChannelHandler(values);
+ sql.execute(stmt, h);
+ System.out.println(values);
+ Assert.assertEquals(3, values.size());
+ Assert.assertEquals("emp1", values.get(0).get(1));
+ Assert.assertEquals("dept1", values.get(0).get(2));
+ Assert.assertEquals("emp2", values.get(1).get(1));
+ Assert.assertEquals("dept1", values.get(1).get(2));
+ Assert.assertEquals("emp3", values.get(2).get(1));
+ Assert.assertEquals("dept2", values.get(2).get(2));
+ }
+
+ @Test
+ public void testjoinAndGroupby() throws Exception {
+ List<String> stmt = new ArrayList<>();
+ stmt.add("CREATE EXTERNAL TABLE EMP (EMPID INT PRIMARY KEY, EMPNAME VARCHAR, DEPTID INT) LOCATION 'mockemp:///foo'");
+ stmt.add("CREATE EXTERNAL TABLE DEPT (DEPTID INT PRIMARY KEY, DEPTNAME VARCHAR) LOCATION 'mockdept:///foo'");
+ stmt.add("SELECT STREAM d.DEPTID, count(EMPID) FROM EMP AS e JOIN DEPT AS d ON e.DEPTID = d.DEPTID WHERE e.empid > 0" +
+ "GROUP BY d.DEPTID");
+ StormSql sql = StormSql.construct();
+ List<Values> values = new ArrayList<>();
+ ChannelHandler h = new TestUtils.CollectDataChannelHandler(values);
+ sql.execute(stmt, h);
+ Assert.assertEquals(2, values.size());
+ Assert.assertEquals(1, values.get(0).get(0));
+ Assert.assertEquals(2L, values.get(0).get(1));
+ Assert.assertEquals(2, values.get(1).get(0));
+ Assert.assertEquals(1L, values.get(1).get(1));
+ }
}
http://git-wip-us.apache.org/repos/asf/storm/blob/2282fb0c/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/AbstractChannelHandler.java
----------------------------------------------------------------------
diff --git a/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/AbstractChannelHandler.java b/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/AbstractChannelHandler.java
index 019a495..a7ebbda 100644
--- a/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/AbstractChannelHandler.java
+++ b/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/AbstractChannelHandler.java
@@ -40,6 +40,11 @@ public abstract class AbstractChannelHandler implements ChannelHandler {
ctx.flush();
}
+ @Override
+ public void setSource(ChannelContext ctx, Object source) {
+
+ }
+
public static final AbstractChannelHandler PASS_THROUGH = new AbstractChannelHandler() {
@Override
public void dataReceived(ChannelContext ctx, Values data) {
http://git-wip-us.apache.org/repos/asf/storm/blob/2282fb0c/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/ChannelContext.java
----------------------------------------------------------------------
diff --git a/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/ChannelContext.java b/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/ChannelContext.java
index 6e5a120..9b21090 100644
--- a/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/ChannelContext.java
+++ b/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/ChannelContext.java
@@ -28,4 +28,5 @@ public interface ChannelContext {
void emit(Values data);
void fireChannelInactive();
void flush();
+ void setSource(Object source);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/2282fb0c/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/ChannelHandler.java
----------------------------------------------------------------------
diff --git a/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/ChannelHandler.java b/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/ChannelHandler.java
index 2369e76..5a6a60b 100644
--- a/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/ChannelHandler.java
+++ b/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/ChannelHandler.java
@@ -38,4 +38,6 @@ public interface ChannelHandler {
void exceptionCaught(Throwable cause);
void flush(ChannelContext ctx);
+
+ void setSource(ChannelContext ctx, Object source);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/2282fb0c/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/Channels.java
----------------------------------------------------------------------
diff --git a/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/Channels.java b/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/Channels.java
index 8f21919..4451d9e 100644
--- a/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/Channels.java
+++ b/external/sql/storm-sql-runtime/src/jvm/org/apache/storm/sql/runtime/Channels.java
@@ -33,6 +33,11 @@ public class Channels {
public void flush() {
}
+
+ @Override
+ public void setSource(java.lang.Object source) {
+
+ }
};
private static class ChannelContextAdapter implements ChannelContext {
@@ -59,6 +64,12 @@ public class Channels {
public void flush() {
handler.flush(next);
}
+
+ @Override
+ public void setSource(java.lang.Object source) {
+ handler.setSource(next, source);
+ next.setSource(source); // propagate through the chain
+ }
}
private static class ForwardingChannelContext implements ChannelContext {
@@ -82,6 +93,11 @@ public class Channels {
public void flush() {
next.flush();
}
+
+ @Override
+ public void setSource(Object source) {
+ next.setSource(source);
+ }
}
public static ChannelContext chain(
http://git-wip-us.apache.org/repos/asf/storm/blob/2282fb0c/external/sql/storm-sql-runtime/src/test/org/apache/storm/sql/TestUtils.java
----------------------------------------------------------------------
diff --git a/external/sql/storm-sql-runtime/src/test/org/apache/storm/sql/TestUtils.java b/external/sql/storm-sql-runtime/src/test/org/apache/storm/sql/TestUtils.java
index 60e1343..5e9d78c 100644
--- a/external/sql/storm-sql-runtime/src/test/org/apache/storm/sql/TestUtils.java
+++ b/external/sql/storm-sql-runtime/src/test/org/apache/storm/sql/TestUtils.java
@@ -98,6 +98,44 @@ public class TestUtils {
}
}
+ public static class MockEmpDataSource implements DataSource {
+ private final ArrayList<Values> RECORDS = new ArrayList<>();
+
+ public MockEmpDataSource() {
+ RECORDS.add(new Values(1, "emp1", 1));
+ RECORDS.add(new Values(2, "emp2", 1));
+ RECORDS.add(new Values(3, "emp3", 2));
+ }
+
+ @Override
+ public void open(ChannelContext ctx) {
+ for (Values v : RECORDS) {
+ ctx.emit(v);
+ }
+ ctx.flush();
+ ctx.fireChannelInactive();
+ }
+ }
+
+ public static class MockDeptDataSource implements DataSource {
+ private final ArrayList<Values> RECORDS = new ArrayList<>();
+
+ public MockDeptDataSource() {
+ RECORDS.add(new Values(1, "dept1"));
+ RECORDS.add(new Values(2, "dept2"));
+ RECORDS.add(new Values(3, "dept3"));
+ }
+
+ @Override
+ public void open(ChannelContext ctx) {
+ for (Values v : RECORDS) {
+ ctx.emit(v);
+ }
+ ctx.flush();
+ ctx.fireChannelInactive();
+ }
+ }
+
public static class MockNestedDataSource implements DataSource {
private final ArrayList<Values> RECORDS = new ArrayList<>();
@@ -218,6 +256,9 @@ public class TestUtils {
@Override
public void flush(ChannelContext ctx) {}
+
+ @Override
+ public void setSource(ChannelContext ctx, Object source) {}
}
public static long monotonicNow() {
[3/3] storm git commit: Added STORM-1833 to CHANGELOG.
Posted by sr...@apache.org.
Added STORM-1833 to CHANGELOG.
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/122bbefa
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/122bbefa
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/122bbefa
Branch: refs/heads/master
Commit: 122bbefa2264fa3a28156ad60b2e6e3f4dd0fae1
Parents: 05b4ab1
Author: Sriharsha Chintalapani <ha...@hortonworks.com>
Authored: Sun Jul 10 14:29:40 2016 -0700
Committer: Sriharsha Chintalapani <ha...@hortonworks.com>
Committed: Sun Jul 10 14:29:40 2016 -0700
----------------------------------------------------------------------
CHANGELOG.md | 1 +
1 file changed, 1 insertion(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/122bbefa/CHANGELOG.md
----------------------------------------------------------------------
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 1b12b69..9aa1f32 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -116,6 +116,7 @@
* STORM-1769: Added a test to check local nimbus with notifier plugin
## 1.1.0
+ * STORM-1833: Simple equi-join in storm-sql standalone mode
* STORM-1866: Update Resource Aware Scheduler Documentation
* STORM-1930: Kafka New Client API - Support for Topic Wildcards
* STORM-1924: Adding conf options for Persistent Word Count Topology
[2/3] storm git commit: Merge branch 'join' of
https://github.com/arunmahadevan/storm into STORM-1833
Posted by sr...@apache.org.
Merge branch 'join' of https://github.com/arunmahadevan/storm into STORM-1833
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/05b4ab1a
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/05b4ab1a
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/05b4ab1a
Branch: refs/heads/master
Commit: 05b4ab1ac7d5b32ce3de3cc415f7470354373460
Parents: 629ac13 2282fb0
Author: Sriharsha Chintalapani <ha...@hortonworks.com>
Authored: Sun Jul 10 14:11:30 2016 -0700
Committer: Sriharsha Chintalapani <ha...@hortonworks.com>
Committed: Sun Jul 10 14:11:30 2016 -0700
----------------------------------------------------------------------
.../sql/compiler/PostOrderRelNodeVisitor.java | 6 +
.../backends/standalone/PlanCompiler.java | 33 +++--
.../backends/standalone/RelNodeCompiler.java | 122 ++++++++++++++++++-
.../test/org/apache/storm/sql/TestStormSql.java | 80 ++++++++++++
.../sql/runtime/AbstractChannelHandler.java | 5 +
.../storm/sql/runtime/ChannelContext.java | 1 +
.../storm/sql/runtime/ChannelHandler.java | 2 +
.../org/apache/storm/sql/runtime/Channels.java | 16 +++
.../test/org/apache/storm/sql/TestUtils.java | 41 +++++++
9 files changed, 287 insertions(+), 19 deletions(-)
----------------------------------------------------------------------