From 8ff61e773733ce30e05dc548f575ae83e38e192c Mon Sep 17 00:00:00 2001
From: Keuntae Park <sirpkt@apache.org>
Date: Thu, 8 Jan 2015 16:30:36 +0900
Subject: [PATCH] WindowAggExec bug is fixed to support occurence of the same
 order by key in multiple window clauses

---
 .../planner/physical/WindowAggExec.java       | 11 ++++++-
 .../tajo/engine/query/TestWindowQuery.java    | 29 +++++++++++++++++++
 2 files changed, 39 insertions(+), 1 deletion(-)

diff --git a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/WindowAggExec.java b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/WindowAggExec.java
index e36dcd8483..a36bd4f00c 100644
--- a/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/WindowAggExec.java
+++ b/tajo-core/src/main/java/org/apache/tajo/engine/planner/physical/WindowAggExec.java
@@ -143,7 +143,16 @@ public WindowAggExec(TaskAttemptContext context, WindowAggNode plan, PhysicalExe
 
           for (SortSpec sortSpec : functions[i].getSortSpecs()) {
             if (!rewrittenSchema.contains(sortSpec.getSortKey())) {
-              additionalSortKeyColumns.add(sortSpec.getSortKey());
+              // check if additionalSortKeyColumns already has that sort key
+              boolean newKey = true;
+              for (Column c : additionalSortKeyColumns) {
+                if (c.equals(sortSpec.getSortKey())) {
+                  newKey = false;
+                }
+              }
+              if (newKey) {
+                additionalSortKeyColumns.add(sortSpec.getSortKey());
+              }
             }
           }
         }
diff --git a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestWindowQuery.java b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestWindowQuery.java
index 668ba70093..14ab58f7f9 100644
--- a/tajo-core/src/test/java/org/apache/tajo/engine/query/TestWindowQuery.java
+++ b/tajo-core/src/test/java/org/apache/tajo/engine/query/TestWindowQuery.java
@@ -321,4 +321,33 @@ public final void lastValueTime() throws Exception {
       executeString("DROP TABLE lastvaluetime PURGE");
     }
   }
+
+  @Test
+  public final void testMultipleWindow() throws Exception {
+    KeyValueSet tableOptions = new KeyValueSet();
+    tableOptions.set(StorageConstants.TEXT_DELIMITER, StorageConstants.DEFAULT_FIELD_DELIMITER);
+    tableOptions.set(StorageConstants.TEXT_NULL, "\\\\N");
+
+    Schema schema = new Schema();
+    schema.addColumn("id", TajoDataTypes.Type.INT4);
+    schema.addColumn("time", TajoDataTypes.Type.TIME);
+    schema.addColumn("name", TajoDataTypes.Type.TEXT);
+    String[] data = new String[]{ "1|12:11:12|abc", "2|10:11:13|def", "2|05:42:41|ghi" };
+    TajoTestingCluster.createTable("multiwindow", schema, tableOptions, data, 1);
+
+    try {
+      ResultSet res = executeString(
+          "select id, last_value(time) over ( partition by id order by time ) as time_last, last_value(name) over ( partition by id order by time ) as name_last from multiwindow");
+      String ascExpected = "id,time_last,name_last\n" +
+          "-------------------------------\n" +
+          "1,12:11:12,abc\n" +
+          "2,10:11:13,def\n" +
+          "2,10:11:13,def\n";
+
+      assertEquals(ascExpected, resultSetToString(res));
+      res.close();
+    } finally {
+      executeString("DROP TABLE multiwindow PURGE");
+    }
+  }
 }
\ No newline at end of file