Patch for DRILL-705

Currently only supports partitioning/ordering, not yet preceding or
after offsets

8 files changed, 128 insertions, 107 deletions

diff --git a/common/src/main/java/org/apache/drill/common/expression/CastExpression.java b/common/src/main/java/org/apache/drill/common/expression/CastExpression.java
index b73a447b5..5d09a2427 100644
--- a/common/src/main/java/org/apache/drill/common/expression/CastExpression.java
+++ b/common/src/main/java/org/apache/drill/common/expression/CastExpression.java
@@ -23,6 +23,8 @@ import java.util.Iterator;
 import org.apache.drill.common.expression.visitors.ExprVisitor;
 import org.apache.drill.common.types.TypeProtos.MajorType;
 
+import static com.google.common.base.Preconditions.checkNotNull;
+
 public class CastExpression extends LogicalExpressionBase implements Iterable<LogicalExpression>{
 
   static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(CastExpression.class);
@@ -33,7 +35,7 @@ public class CastExpression extends LogicalExpressionBase implements Iterable<Lo
   public CastExpression(LogicalExpression input, MajorType type, ExpressionPosition pos) {
     super(pos);
     this.input = input;
-    this.type = type;
+    this.type = checkNotNull(type, "Major type cannot be null");
   }
 
   @Override
diff --git a/common/src/main/java/org/apache/drill/common/logical/data/AbstractBuilder.java b/common/src/main/java/org/apache/drill/common/logical/data/AbstractBuilder.java
index 427b04038..28424a583 100644
--- a/common/src/main/java/org/apache/drill/common/logical/data/AbstractBuilder.java
+++ b/common/src/main/java/org/apache/drill/common/logical/data/AbstractBuilder.java
@@ -19,6 +19,7 @@ package org.apache.drill.common.logical.data;
 
 import java.util.List;
 
+import org.apache.drill.common.expression.FieldReference;
 import org.apache.drill.common.expression.LogicalExpression;
 
 public abstract class AbstractBuilder<T extends LogicalOperator> {
@@ -30,6 +31,10 @@ public abstract class AbstractBuilder<T extends LogicalOperator> {
     return exprs.toArray(new LogicalExpression[exprs.size()]);
   }
 
+  protected FieldReference[] aF(List<FieldReference> exprs){
+    return exprs.toArray(new FieldReference[exprs.size()]);
+  }
+
   protected NamedExpression[] aN(List<NamedExpression> exprs){
     return exprs.toArray(new NamedExpression[exprs.size()]);
   }
diff --git a/common/src/main/java/org/apache/drill/common/logical/data/LogicalOperatorBase.java b/common/src/main/java/org/apache/drill/common/logical/data/LogicalOperatorBase.java
index 58a00fef3..6c534ebf3 100644
--- a/common/src/main/java/org/apache/drill/common/logical/data/LogicalOperatorBase.java
+++ b/common/src/main/java/org/apache/drill/common/logical/data/LogicalOperatorBase.java
@@ -89,5 +89,4 @@ public abstract class LogicalOperatorBase implements LogicalOperator{
     logger.debug("Adding Logical Operator sub types: {}", ((Object) ops) );
     return ops;
   }
-
-}
+}
\ No newline at end of file
diff --git a/common/src/main/java/org/apache/drill/common/logical/data/NamedExpression.java b/common/src/main/java/org/apache/drill/common/logical/data/NamedExpression.java
index 1550ddfd6..a166c251c 100644
--- a/common/src/main/java/org/apache/drill/common/logical/data/NamedExpression.java
+++ b/common/src/main/java/org/apache/drill/common/logical/data/NamedExpression.java
@@ -30,7 +30,7 @@ public class NamedExpression {
   private final FieldReference ref;
 
   @JsonCreator
-  public NamedExpression( @JsonProperty("expr") LogicalExpression expr, @JsonProperty("ref") FieldReference ref) {
+  public NamedExpression(@JsonProperty("expr") LogicalExpression expr, @JsonProperty("ref") FieldReference ref) {
     super();
     this.expr = expr;
     this.ref = ref;
diff --git a/common/src/main/java/org/apache/drill/common/logical/data/Window.java b/common/src/main/java/org/apache/drill/common/logical/data/Window.java
new file mode 100644
index 000000000..6dba77cc7
--- /dev/null
+++ b/common/src/main/java/org/apache/drill/common/logical/data/Window.java
@@ -0,0 +1,112 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.drill.common.logical.data;
+
+import com.google.common.collect.Iterators;
+import com.google.common.collect.Lists;
+import org.apache.drill.common.expression.FieldReference;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import org.apache.drill.common.expression.LogicalExpression;
+import org.apache.drill.common.logical.data.visitors.LogicalVisitor;
+
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkState;
+
+@JsonTypeName("window")
+public class Window extends SingleInputOperator {
+  private final NamedExpression[] withins;
+  private final NamedExpression[] aggregations;
+  private final long start;
+  private final long end;
+
+
+  @JsonCreator
+  public Window(@JsonProperty("withins") NamedExpression[] withins,
+                @JsonProperty("aggregations") NamedExpression[] aggregations,
+                @JsonProperty("start") Long start,
+                @JsonProperty("end") Long end) {
+    super();
+    this.withins = withins;
+    this.start = start == null ? Long.MIN_VALUE : start;
+    this.end = end == null ? Long.MIN_VALUE : end;
+    this.aggregations = aggregations;
+  }
+
+  public NamedExpression[] getWithins() {
+    return withins;
+  }
+
+  public long getStart() {
+    return start;
+  }
+
+  public long getEnd() {
+    return end;
+  }
+
+  public NamedExpression[] getAggregations() {
+    return aggregations;
+  }
+
+  @Override
+  public <T, X, E extends Throwable> T accept(LogicalVisitor<T, X, E> logicalVisitor, X value) throws E {
+    return logicalVisitor.visitWindow(this, value);
+  }
+
+  @Override
+  public Iterator<LogicalOperator> iterator() {
+    return Iterators.singletonIterator(getInput());
+  }
+
+  public static class Builder extends AbstractSingleBuilder<Window, Builder>{
+    private List<NamedExpression> aggregations = Lists.newArrayList();
+    private List<NamedExpression> withins = Lists.newArrayList();
+    private List<Order.Ordering> orderings = Lists.newArrayList();
+    private long start = Long.MIN_VALUE;
+    private long end = Long.MIN_VALUE;
+
+
+    public Builder addAggregation(FieldReference ref, LogicalExpression expr){
+      aggregations.add(new NamedExpression(expr, ref));
+      return this;
+    }
+
+    public Builder addWithin(FieldReference within, LogicalExpression expr) {
+      withins.add(new NamedExpression(expr, within));
+      return this;
+    }
+
+    public Window internalBuild() {
+      checkState(!withins.isEmpty(), "Withins in window must not be empty.");
+      checkState(!aggregations.isEmpty(), "Aggregations in window must not be empty.");
+      return new Window(aN(withins), aN(aggregations), start, end);
+    }
+
+    public Builder addOrdering(Order.Ordering ordering) {
+      orderings.add(ordering);
+      return this;
+    }
+  }
+}
diff --git a/common/src/main/java/org/apache/drill/common/logical/data/WindowFrame.java b/common/src/main/java/org/apache/drill/common/logical/data/WindowFrame.java
deleted file mode 100644
index 6fb64bc81..000000000
--- a/common/src/main/java/org/apache/drill/common/logical/data/WindowFrame.java
+++ /dev/null
@@ -1,98 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.drill.common.logical.data;
-
-import java.util.Iterator;
-
-import org.apache.drill.common.expression.FieldReference;
-import org.apache.drill.common.logical.data.visitors.LogicalVisitor;
-
-import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonProperty;
-import com.fasterxml.jackson.annotation.JsonTypeName;
-import com.google.common.collect.Iterators;
-
-@JsonTypeName("windowframe")
-public class WindowFrame extends SingleInputOperator{
-
-
-  private final FieldReference within;
-  private final FrameRef frame;
-  private final long start;
-  private final long end;
-
-
-  @JsonCreator
-  public WindowFrame(@JsonProperty("within") FieldReference within, @JsonProperty("ref") FrameRef frame, @JsonProperty("start") Long start, @JsonProperty("end") Long end) {
-    super();
-    this.within = within;
-    this.frame = frame;
-    this.start = start == null ? Long.MIN_VALUE : start;
-    this.end = end == null ? Long.MIN_VALUE : end;
-  }
-
-
-  @JsonProperty("ref")
-  public FrameRef getFrame() {
-    return frame;
-  }
-
-  public FieldReference getWithin() {
-    return within;
-  }
-
-  public long getStart() {
-    return start;
-  }
-
-  public long getEnd() {
-    return end;
-  }
-
-    @Override
-    public <T, X, E extends Throwable> T accept(LogicalVisitor<T, X, E> logicalVisitor, X value) throws E {
-        return logicalVisitor.visitWindowFrame(this, value);
-    }
-
-    @Override
-    public Iterator<LogicalOperator> iterator() {
-        return Iterators.singletonIterator(getInput());
-    }
-
-
-    public static class FrameRef{
-    private final FieldReference segment;
-    private final FieldReference position;
-
-    @JsonCreator
-    public FrameRef(@JsonProperty("segment") FieldReference segment, @JsonProperty("position") FieldReference position) {
-      super();
-      this.segment = segment;
-      this.position = position;
-    }
-
-    public FieldReference getSegment() {
-      return segment;
-    }
-    public FieldReference getPosition() {
-      return position;
-    }
-
-
-  }
-}
diff --git a/common/src/main/java/org/apache/drill/common/logical/data/visitors/AbstractLogicalVisitor.java b/common/src/main/java/org/apache/drill/common/logical/data/visitors/AbstractLogicalVisitor.java
index d128b1030..92e370f09 100644
--- a/common/src/main/java/org/apache/drill/common/logical/data/visitors/AbstractLogicalVisitor.java
+++ b/common/src/main/java/org/apache/drill/common/logical/data/visitors/AbstractLogicalVisitor.java
@@ -32,7 +32,7 @@ import org.apache.drill.common.logical.data.Sequence;
 import org.apache.drill.common.logical.data.Store;
 import org.apache.drill.common.logical.data.Transform;
 import org.apache.drill.common.logical.data.Union;
-import org.apache.drill.common.logical.data.WindowFrame;
+import org.apache.drill.common.logical.data.Window;
 import org.apache.drill.common.logical.data.Writer;
 
 
@@ -110,8 +110,8 @@ public abstract class AbstractLogicalVisitor<T, X, E extends Throwable> implemen
     }
 
     @Override
-    public T visitWindowFrame(WindowFrame windowFrame, X value) throws E {
-        return visitOp(windowFrame, value);
+    public T visitWindow(Window window, X value) throws E {
+        return visitOp(window, value);
     }
 
     @Override
diff --git a/common/src/main/java/org/apache/drill/common/logical/data/visitors/LogicalVisitor.java b/common/src/main/java/org/apache/drill/common/logical/data/visitors/LogicalVisitor.java
index 4bf9fbfca..3a426bf13 100644
--- a/common/src/main/java/org/apache/drill/common/logical/data/visitors/LogicalVisitor.java
+++ b/common/src/main/java/org/apache/drill/common/logical/data/visitors/LogicalVisitor.java
@@ -32,7 +32,8 @@ import org.apache.drill.common.logical.data.Sequence;
 import org.apache.drill.common.logical.data.Store;
 import org.apache.drill.common.logical.data.Transform;
 import org.apache.drill.common.logical.data.Union;
-import org.apache.drill.common.logical.data.WindowFrame;
+import org.apache.drill.common.logical.data.Window;
+import org.apache.drill.common.logical.data.Window;
 import org.apache.drill.common.logical.data.Writer;
 
 /**
@@ -60,6 +61,6 @@ public interface LogicalVisitor<RETURN, EXTRA, EXCEP extends Throwable> {
     public RETURN visitSequence(Sequence sequence, EXTRA value) throws EXCEP;
     public RETURN visitTransform(Transform transform, EXTRA value) throws EXCEP;
     public RETURN visitUnion(Union union, EXTRA value) throws EXCEP;
-    public RETURN visitWindowFrame(WindowFrame windowFrame, EXTRA value) throws EXCEP;
+    public RETURN visitWindow(Window window, EXTRA value) throws EXCEP;
     public RETURN visitWriter(Writer writer, EXTRA value) throws EXCEP;
 }