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/05/17 05:14:53 UTC
[3/5] storm git commit: [STORM-1709] Introduce flush to force evalute
group by instead of emitting null
[STORM-1709] Introduce flush to force evalute group by instead of emitting null
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/70aef859
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/70aef859
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/70aef859
Branch: refs/heads/master
Commit: 70aef859d7c4d2e9a595094b04b5d1c30e8f00a4
Parents: 56e6b61
Author: Arun Mahadevan <ai...@hortonworks.com>
Authored: Thu May 12 13:05:17 2016 +0530
Committer: Arun Mahadevan <ai...@hortonworks.com>
Committed: Thu May 12 14:44:49 2016 +0530
----------------------------------------------------------------------
.../backends/standalone/RelNodeCompiler.java | 63 ++++++++++++--------
.../sql/runtime/AbstractChannelHandler.java | 5 ++
.../storm/sql/runtime/ChannelContext.java | 1 +
.../storm/sql/runtime/ChannelHandler.java | 2 +
.../org/apache/storm/sql/runtime/Channels.java | 15 +++++
.../test/org/apache/storm/sql/TestUtils.java | 7 ++-
6 files changed, 65 insertions(+), 28 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/70aef859/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 22aabc9..7782529 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
@@ -35,6 +35,7 @@ import org.apache.storm.sql.compiler.ExprCompiler;
import org.apache.storm.sql.compiler.PostOrderRelNodeVisitor;
import java.io.PrintWriter;
+import java.io.StringWriter;
import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.HashMap;
@@ -54,7 +55,6 @@ class RelNodeCompiler extends PostOrderRelNodeVisitor<Void> {
" new AbstractChannelHandler() {",
" @Override",
" public void dataReceived(ChannelContext ctx, Values _data) {",
- " if (_data != null) {",
""
);
@@ -74,6 +74,17 @@ class RelNodeCompiler extends PostOrderRelNodeVisitor<Void> {
" }",
"",
" @Override",
+ " public void flush(ChannelContext ctx) {",
+ " emitAggregateResults(ctx);",
+ " super.flush(ctx);",
+ " prevGroupValues = null;",
+ " }",
+ "",
+ " private void emitAggregateResults(ChannelContext ctx) {",
+ " %3$s",
+ " }",
+ "",
+ " @Override",
" public void dataReceived(ChannelContext ctx, Values _data) {",
""
);
@@ -140,15 +151,8 @@ class RelNodeCompiler extends PostOrderRelNodeVisitor<Void> {
public Void visitAggregate(Aggregate aggregate) throws Exception {
beginAggregateStage(aggregate);
pw.println(" List<Object> curGroupValues = _data == null ? null : getGroupValues(_data);");
- pw.println(" if (curGroupValues == null || (prevGroupValues != null && ! prevGroupValues.equals(curGroupValues))) {");
- List<String> res = new ArrayList<>();
- for (AggregateCall call : aggregate.getAggCallList()) {
- res.add(aggregateResult(call));
- }
- pw.print(String.format(" ctx.emit(new Values(%s, %s));\n",
- groupValueEmitStr("prevGroupValues", aggregate.getGroupSet().cardinality()),
- Joiner.on(", ").join(res)));
- pw.println(" accumulators.clear();");
+ pw.println(" if (prevGroupValues != null && !prevGroupValues.equals(curGroupValues)) {");
+ pw.println(" emitAggregateResults(ctx);");
pw.println(" }");
pw.println(" if (curGroupValues != null) {");
for (AggregateCall call : aggregate.getAggCallList()) {
@@ -158,7 +162,7 @@ class RelNodeCompiler extends PostOrderRelNodeVisitor<Void> {
pw.println(" if (prevGroupValues != curGroupValues) {");
pw.println(" prevGroupValues = curGroupValues;");
pw.println(" }");
- endAggregateStage();
+ endStage();
return null;
}
@@ -174,26 +178,40 @@ class RelNodeCompiler extends PostOrderRelNodeVisitor<Void> {
return sb.toString();
}
- private String aggregateResult(AggregateCall call) {
+ private String emitAggregateStmts(Aggregate aggregate) {
+ List<String> res = new ArrayList<>();
+ StringWriter sw = new StringWriter();
+ for (AggregateCall call : aggregate.getAggCallList()) {
+ res.add(aggregateResult(call, new PrintWriter(sw)));
+ }
+ return NEW_LINE_JOINER.join(sw.toString(),
+ String.format(" ctx.emit(new Values(%s, %s));",
+ groupValueEmitStr("prevGroupValues", aggregate.getGroupSet().cardinality()),
+ Joiner.on(", ").join(res)),
+ " accumulators.clear();"
+ );
+ }
+
+ private String aggregateResult(AggregateCall call, PrintWriter pw) {
SqlAggFunction aggFunction = call.getAggregation();
String aggregationName = call.getAggregation().getName();
Type ty = typeFactory.getJavaClass(call.getType());
String result;
if (aggFunction instanceof SqlUserDefinedAggFunction) {
AggregateFunction aggregateFunction = ((SqlUserDefinedAggFunction) aggFunction).function;
- result = doAggregateResult((AggregateFunctionImpl) aggregateFunction, reserveAggVarName(call), ty);
+ result = doAggregateResult((AggregateFunctionImpl) aggregateFunction, reserveAggVarName(call), ty, pw);
} else {
List<BuiltinAggregateFunctions.TypeClass> typeClasses = BuiltinAggregateFunctions.TABLE.get(aggregationName);
if (typeClasses == null) {
throw new UnsupportedOperationException(aggregationName + " Not implemented");
}
result = doAggregateResult(AggregateFunctionImpl.create(findMatchingClass(aggregationName, typeClasses, ty)),
- reserveAggVarName(call), ty);
+ reserveAggVarName(call), ty, pw);
}
return result;
}
- private String doAggregateResult(AggregateFunctionImpl aggFn, String varName, Type ty) {
+ private String doAggregateResult(AggregateFunctionImpl aggFn, String varName, Type ty, PrintWriter pw) {
String resultName = varName + "_result";
List<String> args = new ArrayList<>();
if (!aggFn.isStatic) {
@@ -202,15 +220,15 @@ class RelNodeCompiler extends PostOrderRelNodeVisitor<Void> {
boolean genericType = aggFn.initMethod.getDeclaringClass().getTypeParameters().length > 0;
if (genericType) {
pw.println(" @SuppressWarnings(\"unchecked\")");
- pw.println(String.format(" final %1$s<%3$s> %2$s = (%1$s<%3$s>) accumulators.get(\"%2$s\");", aggObjClassName,
+ pw.print(String.format(" final %1$s<%3$s> %2$s = (%1$s<%3$s>) accumulators.get(\"%2$s\");", aggObjClassName,
aggObjName, Primitives.wrap((Class<?>) ty).getCanonicalName()));
} else {
- pw.println(String.format(" final %1$s %2$s = (%1$s) accumulators.get(\"%2$s\");", aggObjClassName, aggObjName));
+ pw.print(String.format(" final %1$s %2$s = (%1$s) accumulators.get(\"%2$s\");", aggObjClassName, aggObjName));
}
args.add(aggObjName);
}
args.add(String.format("(%s)accumulators.get(\"%s\")", ((Class<?>) ty).getCanonicalName(), varName));
- pw.println(String.format(" final %s %s = %s;", ((Class<?>) ty).getCanonicalName(),
+ pw.print(String.format(" final %s %s = %s;", ((Class<?>) ty).getCanonicalName(),
resultName, ExprCompiler.printMethodCall(aggFn.resultMethod, args)));
return resultName;
@@ -297,17 +315,10 @@ class RelNodeCompiler extends PostOrderRelNodeVisitor<Void> {
}
private void beginAggregateStage(Aggregate n) {
- pw.print(String.format(AGGREGATE_STAGE_PROLOGUE, getStageName(n), getGroupByIndices(n)));
+ pw.print(String.format(AGGREGATE_STAGE_PROLOGUE, getStageName(n), getGroupByIndices(n), emitAggregateStmts(n)));
}
private void endStage() {
- pw.println(" } else {");
- pw.println(" ctx.emit(_data);");
- pw.println(" }");
- pw.print(" }\n };\n");
- }
-
- private void endAggregateStage() {
pw.print(" }\n };\n");
}
http://git-wip-us.apache.org/repos/asf/storm/blob/70aef859/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 6a8bbe5..019a495 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
@@ -35,6 +35,11 @@ public abstract class AbstractChannelHandler implements ChannelHandler {
}
+ @Override
+ public void flush(ChannelContext ctx) {
+ ctx.flush();
+ }
+
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/70aef859/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 c29c6b1..6e5a120 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
@@ -27,4 +27,5 @@ public interface ChannelContext {
*/
void emit(Values data);
void fireChannelInactive();
+ void flush();
}
http://git-wip-us.apache.org/repos/asf/storm/blob/70aef859/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 f30d0a3..2369e76 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
@@ -36,4 +36,6 @@ public interface ChannelHandler {
void channelInactive(ChannelContext ctx);
void exceptionCaught(Throwable cause);
+
+ void flush(ChannelContext ctx);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/70aef859/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 47dab11..8f21919 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
@@ -28,6 +28,11 @@ public class Channels {
@Override
public void fireChannelInactive() {}
+
+ @Override
+ public void flush() {
+
+ }
};
private static class ChannelContextAdapter implements ChannelContext {
@@ -49,6 +54,11 @@ public class Channels {
public void fireChannelInactive() {
handler.channelInactive(next);
}
+
+ @Override
+ public void flush() {
+ handler.flush(next);
+ }
}
private static class ForwardingChannelContext implements ChannelContext {
@@ -67,6 +77,11 @@ public class Channels {
public void fireChannelInactive() {
next.fireChannelInactive();
}
+
+ @Override
+ public void flush() {
+ next.flush();
+ }
}
public static ChannelContext chain(
http://git-wip-us.apache.org/repos/asf/storm/blob/70aef859/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 28d3506..60e1343 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
@@ -85,8 +85,6 @@ public class TestUtils {
for (int i = 0; i < 10; ++i) {
RECORDS.add(new Values(i/3, i, (i+1)* 0.5, "x", i/2));
}
- // force evaluation of the aggregate function on the last group by emitting null row
- RECORDS.add(null);
}
@Override
@@ -94,6 +92,8 @@ public class TestUtils {
for (Values v : RECORDS) {
ctx.emit(v);
}
+ // force evaluation of the aggregate function on the last group
+ ctx.flush();
ctx.fireChannelInactive();
}
}
@@ -215,6 +215,9 @@ public class TestUtils {
public void exceptionCaught(Throwable cause) {
throw new RuntimeException(cause);
}
+
+ @Override
+ public void flush(ChannelContext ctx) {}
}
public static long monotonicNow() {