You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by ja...@apache.org on 2017/06/07 17:04:27 UTC
[4/4] phoenix git commit: PHOENIX-3773 Implement FIRST_VALUES
aggregate function (Loknath Priyatham Teja Singamsetty)
PHOENIX-3773 Implement FIRST_VALUES aggregate function (Loknath Priyatham Teja Singamsetty)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/3733a4ef
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/3733a4ef
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/3733a4ef
Branch: refs/heads/4.x-HBase-0.98
Commit: 3733a4ef05cb43c8fd0b52846148fd21e36ef8b3
Parents: cdda8b1
Author: James Taylor <ja...@apache.org>
Authored: Tue Jun 6 18:44:49 2017 -0700
Committer: James Taylor <ja...@apache.org>
Committed: Wed Jun 7 10:03:59 2017 -0700
----------------------------------------------------------------------
.../phoenix/end2end/FirstValuesFunctionIT.java | 638 ++++++++++++++++++
.../phoenix/end2end/LastValuesFunctionIT.java | 640 +++++++++++++++++++
.../phoenix/expression/ExpressionType.java | 117 +++-
.../FirstLastValueBaseClientAggregator.java | 65 +-
.../FirstLastValueServerAggregator.java | 13 +-
.../expression/function/FirstValueFunction.java | 2 +-
.../function/FirstValuesFunction.java | 97 +++
.../expression/function/LastValueFunction.java | 2 +-
.../expression/function/LastValuesFunction.java | 97 +++
.../expression/function/NthValueFunction.java | 4 +-
.../parse/FirstValuesAggregateParseNode.java | 38 ++
.../parse/LastValuesAggregateParseNode.java | 38 ++
.../util/FirstLastNthValueDataContainer.java | 7 +-
13 files changed, 1729 insertions(+), 29 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/3733a4ef/phoenix-core/src/it/java/org/apache/phoenix/end2end/FirstValuesFunctionIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/FirstValuesFunctionIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/FirstValuesFunctionIT.java
new file mode 100644
index 0000000..1c4b419
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/FirstValuesFunctionIT.java
@@ -0,0 +1,638 @@
+/*
+ * 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.phoenix.end2end;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.util.ArrayList;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+
+public class FirstValuesFunctionIT extends ParallelStatsDisabledIT {
+
+ @Test
+ public void simpleTest() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL PRIMARY KEY, page_id UNSIGNED_LONG,"
+ + " dates INTEGER, val INTEGER)";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (2, 8, 1, 7)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (3, 8, 2, 9)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (4, 8, 3, 4)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (5, 8, 4, 2)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (6, 9, 5, 10)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (7, 9, 6, 13)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (8, 9, 7, 8)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (9, 9, 8, 6)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT FIRST_VALUES(val, 3) WITHIN GROUP (ORDER BY dates DESC) FROM " + tableName
+ + " GROUP BY page_id");
+
+ assertTrue(rs.next());
+ Assert.assertArrayEquals(new int[]{2, 4, 9}, (int[]) rs.getArray(1).getArray());
+ assertTrue(rs.next());
+ Assert.assertArrayEquals(new int[]{6, 8, 13}, (int[]) rs.getArray(1).getArray());
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void varcharDatatypeSimpleTest() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String table_name = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + table_name + " "
+ + "(id INTEGER NOT NULL PRIMARY KEY, page_id UNSIGNED_LONG, "
+ + "\"DATE\" VARCHAR(3), \"value\" VARCHAR(3))";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + table_name + " (id, page_id, \"DATE\", \"value\") VALUES (1, 8, '1', '3')");
+ conn.createStatement().execute(
+ "UPSERT INTO " + table_name + " (id, page_id, \"DATE\", \"value\") VALUES (2, 8, '2', '7')");
+ conn.createStatement().execute(
+ "UPSERT INTO " + table_name + " (id, page_id, \"DATE\", \"value\") VALUES (3, 8, '3', '9')");
+ conn.createStatement().execute(
+ "UPSERT INTO " + table_name + " (id, page_id, \"DATE\", \"value\") VALUES (5, 8, '4', '2')");
+ conn.createStatement().execute(
+ "UPSERT INTO " + table_name + " (id, page_id, \"DATE\", \"value\") VALUES (4, 8, '5', '4')");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT FIRST_VALUES(\"value\", 2) WITHIN GROUP (ORDER BY \"DATE\" ASC) FROM " + table_name
+ + " GROUP BY page_id");
+
+ assertTrue(rs.next());
+ Assert.assertArrayEquals(new String[]{"3", "7"}, (String[]) rs.getArray(1).getArray());
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void floatDatatypeSimpleTest() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL PRIMARY KEY, page_id UNSIGNED_LONG,"
+ + " \"DATE\" FLOAT, \"value\" FLOAT)";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (1, 8, 1, 300)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (2, 8, 2, 7)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (3, 8, 3, 9)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (5, 8, 4, 2)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (4, 8, 5, 400)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT FIRST_VALUES(\"value\", 2) WITHIN GROUP (ORDER BY \"DATE\" ASC) FROM " + tableName
+ + " GROUP BY page_id");
+
+ assertTrue(rs.next());
+ assertEquals(((float[])rs.getArray(1).getArray())[0], 300, 0.000001);
+ assertEquals(((float[])rs.getArray(1).getArray())[1], 7, 0.000001);
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void doubleDatatypeSimpleTest() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL PRIMARY KEY, page_id UNSIGNED_LONG, "
+ + "\"DATE\" DOUBLE, \"value\" DOUBLE)";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (1, 8, 1, 300)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (2, 8, 2, 7)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (3, 8, 3, 9)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (5, 8, 4, 2)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (4, 8, 5, 400)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT FIRST_VALUES(\"value\", 2) WITHIN GROUP (ORDER BY \"DATE\" ASC) FROM " + tableName
+ + " GROUP BY page_id");
+
+ assertTrue(rs.next());
+ assertEquals(((double[])rs.getArray(1).getArray())[0], 300, 0.000001);
+ assertEquals(((double[])rs.getArray(1).getArray())[1], 7, 0.000001);
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void offsetValueAscOrder() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL PRIMARY KEY, page_id UNSIGNED_LONG,"
+ + " \"DATE\" INTEGER, \"value\" UNSIGNED_LONG)";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (1, 8, 0, 300)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (2, 8, 1, 7)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (3, 8, 2, 9)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (4, 8, 3, 4)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (5, 8, 4, 2)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (6, 8, 5, 150)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT FIRST_VALUES(\"value\", 2) WITHIN GROUP (ORDER BY \"DATE\" ASC) FROM "
+ + tableName + " GROUP BY page_id");
+
+ assertTrue(rs.next());
+ Assert.assertArrayEquals(new long[]{300, 7}, (long[]) rs.getArray(1).getArray());
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void simpleTestNoGroupBy() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL PRIMARY KEY, page_id UNSIGNED_LONG,"
+ + " dates INTEGER, val INTEGER)";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (2, 8, 1, 7)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (3, 8, 2, 9)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT FIRST_VALUES(val, 2) WITHIN GROUP (ORDER BY dates DESC) FROM " + tableName);
+
+ assertTrue(rs.next());
+ Assert.assertArrayEquals(new int[]{9, 7}, (int[]) rs.getArray(1).getArray());
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void rowLessThanOffsetNoGroupBy() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL PRIMARY KEY, page_id UNSIGNED_LONG,"
+ + " dates INTEGER, val INTEGER)";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (2, 8, 1, 7)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (3, 8, 2, 9)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT FIRST_VALUES(val, 3) WITHIN GROUP (ORDER BY dates DESC) FROM " + tableName);
+
+ assertTrue(rs.next());
+ Assert.assertArrayEquals(new int[]{9, 7}, (int[]) rs.getArray(1).getArray());
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void offsetValueDescOrder() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL PRIMARY KEY, page_id UNSIGNED_LONG,"
+ + " \"DATE\" INTEGER, \"value\" UNSIGNED_LONG)";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (1, 8, 0, 300)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (2, 8, 1, 7)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (3, 8, 2, 9)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (4, 8, 3, 4)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (5, 8, 4, 2)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (6, 8, 5, 150)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT FIRST_VALUES(\"value\", 2) WITHIN GROUP (ORDER BY \"DATE\" DESC) FROM "
+ + tableName + " GROUP BY page_id");
+
+ assertTrue(rs.next());
+ Assert.assertArrayEquals(new long[]{150, 2}, (long[]) rs.getArray(1).getArray());
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void offsetValueSubAggregation() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL PRIMARY KEY, page_id UNSIGNED_LONG,"
+ + " \"DATE\" INTEGER, \"value\" UNSIGNED_LONG)";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (1, 8, 0, 300)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (2, 8, 1, 7)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (3, 9, 2, 9)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (4, 9, 3, 4)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (5, 10, 4, 2)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (6, 10, 5, 150)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT FIRST_VALUES(SUM_VALUE, 2) WITHIN GROUP (ORDER BY MIN_DATE ASC) FROM (" +
+ "SELECT MIN(\"DATE\") AS MIN_DATE, SUM(\"value\") AS SUM_VALUE FROM "
+ + tableName + " GROUP BY page_id) x");
+
+ assertTrue(rs.next());
+ Assert.assertArrayEquals(new long[]{307, 13}, (long[]) rs.getArray(1).getArray());
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void offsetValueLastMismatchByColumn() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL PRIMARY KEY, page_id UNSIGNED_LONG,"
+ + " \"DATE\" INTEGER, \"value\" UNSIGNED_LONG)";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (1, 8, 5, 8)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (2, 8, 2, 7)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (3, 8, 1, 9)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (4, 8, 4, 4)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (5, 8, 3, 2)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (6, 8, 0, 1)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT FIRST_VALUES(\"value\", 2) WITHIN GROUP (ORDER BY \"DATE\" DESC) FROM "
+ + tableName + " GROUP BY page_id");
+
+ assertTrue(rs.next());
+ Assert.assertArrayEquals(new long[]{8, 4}, (long[]) rs.getArray(1).getArray());
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void testSortOrderInDataColWithOffset() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL, page_id UNSIGNED_LONG,"
+ + " dates BIGINT NOT NULL, \"value\" BIGINT NOT NULL CONSTRAINT pk PRIMARY KEY (id, dates, \"value\" DESC))";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute("UPSERT INTO " + tableName
+ + " (id, page_id, dates, \"value\") VALUES (1, 8, 1, 3)");
+ conn.createStatement().execute("UPSERT INTO " + tableName
+ + " (id, page_id, dates, \"value\") VALUES (2, 8, 2, 7)");
+ conn.createStatement().execute("UPSERT INTO " + tableName
+ + " (id, page_id, dates, \"value\") VALUES (3, 8, 3, 9)");
+ conn.createStatement().execute("UPSERT INTO " + tableName
+ + " (id, page_id, dates, \"value\") VALUES (5, 8, 5, 158)");
+ conn.createStatement().execute("UPSERT INTO " + tableName
+ + " (id, page_id, dates, \"value\") VALUES (4, 8, 4, 5)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT FIRST_VALUES(\"value\", 2) WITHIN GROUP (ORDER BY dates ASC) FROM "
+ + tableName + " GROUP BY page_id");
+
+ assertTrue(rs.next());
+ Assert.assertArrayEquals(new long[]{3, 7}, (long[]) rs.getArray(1).getArray());
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void nonUniqueValuesInOrderByAsc() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL PRIMARY KEY, page_id UNSIGNED_LONG,"
+ + " dates INTEGER, val INTEGER)";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (2, 8, 1, 7)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (3, 8, 2, 9)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (4, 8, 2, 4)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (5, 8, 2, 2)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (6, 8, 3, 3)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT FIRST_VALUES(val, 3) WITHIN GROUP (ORDER BY dates ASC) FROM " + tableName
+ + " GROUP BY page_id");
+
+ int[] result = null;
+ assertTrue(rs.next());
+ result = (int[]) rs.getArray(1).getArray();
+ assertEquals(result[0], 7);
+ assertInIntArray(new int[]{2, 4, 9}, result[1]);
+ assertInIntArray(new int[]{2, 4, 9}, result[2]);
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void nonUniqueValuesInOrderByAscSkipDuplicit() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL PRIMARY KEY, page_id UNSIGNED_LONG,"
+ + " dates INTEGER, val INTEGER)";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (2, 8, 1, 7)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (3, 8, 2, 9)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (4, 8, 2, 4)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (5, 8, 2, 2)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (6, 8, 3, 3)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT FIRST_VALUES(val, 5) WITHIN GROUP (ORDER BY dates ASC) FROM " + tableName
+ + " GROUP BY page_id");
+
+ int[] result = null;
+ assertTrue(rs.next());
+ result = (int[]) rs.getArray(1).getArray();
+ assertEquals(result[0], 7);
+ assertInIntArray(new int[]{2, 4, 9}, result[1]);
+ assertInIntArray(new int[]{2, 4, 9}, result[2]);
+ assertInIntArray(new int[]{2, 4, 9}, result[3]);
+ assertEquals(result[4], 3);
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void nonUniqueValuesInOrderByDesc() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL PRIMARY KEY, page_id UNSIGNED_LONG,"
+ + " dates INTEGER, val INTEGER)";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (2, 8, 1, 7)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (3, 8, 2, 9)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (4, 8, 2, 4)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (5, 8, 2, 2)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (6, 8, 3, 3)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT FIRST_VALUES(val, 3) WITHIN GROUP (ORDER BY dates DESC) FROM " + tableName
+ + " GROUP BY page_id");
+
+ int[] result = null;
+ assertTrue(rs.next());
+ result = (int[]) rs.getArray(1).getArray();
+ assertEquals(result[0], 3);
+ assertInIntArray(new int[]{2, 4, 9}, result[1]);
+ assertInIntArray(new int[]{2, 4, 9}, result[2]);
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void nonUniqueValuesInOrderNextValueDesc() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL PRIMARY KEY, page_id UNSIGNED_LONG,"
+ + " dates INTEGER, val INTEGER)";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (2, 8, 0, 7)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (3, 8, 1, 9)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (4, 8, 2, 4)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (5, 8, 2, 2)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (6, 8, 3, 3)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (7, 8, 3, 5)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT FIRST_VALUES(val, 2) WITHIN GROUP (ORDER BY dates DESC) FROM " + tableName
+ + " GROUP BY page_id");
+
+ int[] result = null;
+ assertTrue(rs.next());
+ result = (int[]) rs.getArray(1).getArray();
+ assertInIntArray(new int[]{3, 5}, result[0]);
+ assertInIntArray(new int[]{3, 5}, result[1]);
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void nonUniqueValuesInOrderNextValueAsc() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL PRIMARY KEY, page_id UNSIGNED_LONG,"
+ + " dates INTEGER, val INTEGER)";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (2, 8, 0, 7)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (3, 8, 1, 9)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (4, 8, 2, 4)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (5, 8, 2, 2)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (6, 8, 3, 3)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (7, 8, 3, 5)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT FIRST_VALUES(val, 5) WITHIN GROUP (ORDER BY dates ASC) FROM " + tableName
+ + " GROUP BY page_id");
+
+ int[] result = null;
+ assertTrue(rs.next());
+ result = (int[]) rs.getArray(1).getArray();
+ assertEquals(result[0], 7);
+ assertEquals(result[1], 9);
+ assertInIntArray(new int[]{2, 4}, result[2]);
+ assertInIntArray(new int[]{2, 4}, result[3]);
+ assertInIntArray(new int[]{3, 5}, result[4]);
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void ignoreNullValues() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL, page_id UNSIGNED_LONG,"
+ + " dates BIGINT NOT NULL, \"value\" BIGINT NULL CONSTRAINT pk PRIMARY KEY (id, dates))";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute("UPSERT INTO " + tableName
+ + " (id, page_id, dates, \"value\") VALUES (1, 8, 1, 1)");
+ conn.createStatement().execute("UPSERT INTO " + tableName
+ + " (id, page_id, dates, \"value\") VALUES (2, 8, 2, NULL)");
+ conn.createStatement().execute("UPSERT INTO " + tableName
+ + " (id, page_id, dates, \"value\") VALUES (3, 8, 3, NULL)");
+ conn.createStatement().execute("UPSERT INTO " + tableName
+ + " (id, page_id, dates, \"value\") VALUES (5, 8, 4, 4)");
+ conn.createStatement().execute("UPSERT INTO " + tableName
+ + " (id, page_id, dates, \"value\") VALUES (4, 8, 5, 5)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT FIRST_VALUES(\"value\", 2) WITHIN GROUP (ORDER BY dates ASC) FROM "
+ + tableName + " GROUP BY page_id");
+
+ assertTrue(rs.next());
+ Assert.assertArrayEquals(new long[]{1, 4}, (long[]) rs.getArray(1).getArray());
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void rowLessThanOffsetWithGroupBy() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL PRIMARY KEY, page_id UNSIGNED_LONG,"
+ + " dates INTEGER, val INTEGER)";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (2, 8, 0, 7)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (3, 8, 1, 9)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (4, 8, 2, 4)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (5, 8, 2, 2)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (6, 8, 3, 3)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (7, 9, 3, 5)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT FIRST_VALUES(val, 2) WITHIN GROUP (ORDER BY dates ASC) FROM "
+ + tableName + " GROUP BY page_id");
+
+ assertTrue(rs.next());
+ Assert.assertArrayEquals(new int[]{7, 9}, (int[]) rs.getArray(1).getArray());
+ assertTrue(rs.next());
+ Assert.assertArrayEquals(new int[]{5}, (int[]) rs.getArray(1).getArray());
+ assertFalse(rs.next());
+ }
+
+ private void assertInIntArray(int[] should, int actualValue) {
+ ArrayList<Integer> shouldList = new ArrayList<Integer>();
+ for (int i: should) {
+ shouldList.add(i);
+ }
+ assertTrue(shouldList.contains(actualValue));
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/3733a4ef/phoenix-core/src/it/java/org/apache/phoenix/end2end/LastValuesFunctionIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/LastValuesFunctionIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/LastValuesFunctionIT.java
new file mode 100644
index 0000000..f48e4df
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/LastValuesFunctionIT.java
@@ -0,0 +1,640 @@
+/*
+ * 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.phoenix.end2end;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.util.ArrayList;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+
+public class LastValuesFunctionIT extends ParallelStatsDisabledIT {
+
+ @Test
+ public void simpleTest() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL PRIMARY KEY, page_id UNSIGNED_LONG,"
+ + " dates INTEGER, val INTEGER)";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (2, 8, 1, 7)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (3, 8, 2, 9)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (4, 8, 3, 4)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (5, 8, 4, 2)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (6, 9, 5, 10)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (7, 9, 6, 13)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (8, 9, 7, 8)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (9, 9, 8, 6)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT LAST_VALUES(val, 3) WITHIN GROUP (ORDER BY dates DESC) FROM " + tableName
+ + " GROUP BY page_id");
+
+ assertTrue(rs.next());
+ Assert.assertArrayEquals(new int[]{7, 9, 4}, (int[]) rs.getArray(1).getArray());
+ assertTrue(rs.next());
+ Assert.assertArrayEquals(new int[]{10, 13, 8}, (int[]) rs.getArray(1).getArray());
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void varcharDatatypeSimpleTest() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String table_name = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + table_name + " "
+ + "(id INTEGER NOT NULL PRIMARY KEY, page_id UNSIGNED_LONG, "
+ + "\"DATE\" VARCHAR(3), \"value\" VARCHAR(3))";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + table_name + " (id, page_id, \"DATE\", \"value\") VALUES (1, 8, '1', '3')");
+ conn.createStatement().execute(
+ "UPSERT INTO " + table_name + " (id, page_id, \"DATE\", \"value\") VALUES (2, 8, '2', '7')");
+ conn.createStatement().execute(
+ "UPSERT INTO " + table_name + " (id, page_id, \"DATE\", \"value\") VALUES (3, 8, '3', '9')");
+ conn.createStatement().execute(
+ "UPSERT INTO " + table_name + " (id, page_id, \"DATE\", \"value\") VALUES (5, 8, '4', '2')");
+ conn.createStatement().execute(
+ "UPSERT INTO " + table_name + " (id, page_id, \"DATE\", \"value\") VALUES (4, 8, '5', '4')");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT LAST_VALUES(\"value\", 2) WITHIN GROUP (ORDER BY \"DATE\" ASC) FROM " + table_name
+ + " GROUP BY page_id");
+
+ assertTrue(rs.next());
+ Assert.assertArrayEquals(new String[]{"4", "2"}, (String[]) rs.getArray(1).getArray());
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void floatDatatypeSimpleTest() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL PRIMARY KEY, page_id UNSIGNED_LONG,"
+ + " \"DATE\" FLOAT, \"value\" FLOAT)";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (1, 8, 1, 300)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (2, 8, 2, 7)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (3, 8, 3, 9)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (5, 8, 4, 2)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (4, 8, 5, 400)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT LAST_VALUES(\"value\", 2) WITHIN GROUP (ORDER BY \"DATE\" ASC) FROM " + tableName
+ + " GROUP BY page_id");
+
+ assertTrue(rs.next());
+ assertEquals(((float[])rs.getArray(1).getArray())[0], 400, 0.000001);
+ assertEquals(((float[])rs.getArray(1).getArray())[1], 2, 0.000001);
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void doubleDatatypeSimpleTest() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL PRIMARY KEY, page_id UNSIGNED_LONG, "
+ + "\"DATE\" DOUBLE, \"value\" DOUBLE)";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (1, 8, 1, 300)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (2, 8, 2, 7)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (3, 8, 3, 9)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (5, 8, 4, 2)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (4, 8, 5, 400)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT LAST_VALUES(\"value\", 2) WITHIN GROUP (ORDER BY \"DATE\" ASC) FROM " + tableName
+ + " GROUP BY page_id");
+
+ assertTrue(rs.next());
+ assertEquals(((double[])rs.getArray(1).getArray())[0], 400, 0.000001);
+ assertEquals(((double[])rs.getArray(1).getArray())[1], 2, 0.000001);
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void offsetValueAscOrder() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL PRIMARY KEY, page_id UNSIGNED_LONG,"
+ + " \"DATE\" INTEGER, \"value\" UNSIGNED_LONG)";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (1, 8, 0, 300)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (2, 8, 1, 7)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (3, 8, 2, 9)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (4, 8, 3, 4)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (5, 8, 4, 2)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (6, 8, 5, 150)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT LAST_VALUES(\"value\", 2) WITHIN GROUP (ORDER BY \"DATE\" ASC) FROM "
+ + tableName + " GROUP BY page_id");
+
+ assertTrue(rs.next());
+ Assert.assertArrayEquals(new long[]{150, 2}, (long[]) rs.getArray(1).getArray());
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void simpleTestNoGroupBy() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL PRIMARY KEY, page_id UNSIGNED_LONG,"
+ + " dates INTEGER, val INTEGER)";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (2, 8, 1, 7)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (3, 8, 2, 9)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT LAST_VALUES(val, 2) WITHIN GROUP (ORDER BY dates DESC) FROM " + tableName);
+
+ assertTrue(rs.next());
+ Assert.assertArrayEquals(new int[]{7, 9}, (int[]) rs.getArray(1).getArray());
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void rowLessThanOffsetNoGroupBy() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL PRIMARY KEY, page_id UNSIGNED_LONG,"
+ + " dates INTEGER, val INTEGER)";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (2, 8, 1, 7)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (3, 8, 2, 9)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT LAST_VALUES(val, 3) WITHIN GROUP (ORDER BY dates DESC) FROM " + tableName);
+
+ assertTrue(rs.next());
+ Assert.assertArrayEquals(new int[]{7, 9}, (int[]) rs.getArray(1).getArray());
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void offsetValueDescOrder() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL PRIMARY KEY, page_id UNSIGNED_LONG,"
+ + " \"DATE\" INTEGER, \"value\" UNSIGNED_LONG)";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (1, 8, 0, 300)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (2, 8, 1, 7)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (3, 8, 2, 9)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (4, 8, 3, 4)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (5, 8, 4, 2)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (6, 8, 5, 150)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT LAST_VALUES(\"value\", 2) WITHIN GROUP (ORDER BY \"DATE\" DESC) FROM "
+ + tableName + " GROUP BY page_id");
+
+ assertTrue(rs.next());
+ Assert.assertArrayEquals(new long[]{300, 7}, (long[]) rs.getArray(1).getArray());
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void offsetValueSubAggregation() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL PRIMARY KEY, page_id UNSIGNED_LONG,"
+ + " \"DATE\" INTEGER, \"value\" UNSIGNED_LONG)";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (1, 8, 0, 300)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (2, 8, 1, 7)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (3, 9, 2, 9)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (4, 9, 3, 4)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (5, 10, 4, 2)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (6, 10, 5, 150)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT LAST_VALUES(SUM_VALUE, 2) WITHIN GROUP (ORDER BY MIN_DATE ASC) FROM (" +
+ "SELECT MIN(\"DATE\") AS MIN_DATE, SUM(\"value\") AS SUM_VALUE FROM "
+ + tableName + " GROUP BY page_id) x");
+
+ assertTrue(rs.next());
+ Assert.assertArrayEquals(new long[]{152, 13}, (long[]) rs.getArray(1).getArray());
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void offsetValueLastMismatchByColumn() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL PRIMARY KEY, page_id UNSIGNED_LONG,"
+ + " \"DATE\" INTEGER, \"value\" UNSIGNED_LONG)";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (1, 8, 5, 8)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (2, 8, 2, 7)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (3, 8, 1, 9)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (4, 8, 4, 4)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (5, 8, 3, 2)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, \"DATE\", \"value\") VALUES (6, 8, 0, 1)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT LAST_VALUES(\"value\", 2) WITHIN GROUP (ORDER BY \"DATE\" DESC) FROM "
+ + tableName + " GROUP BY page_id");
+
+ assertTrue(rs.next());
+ Assert.assertArrayEquals(new long[]{1, 9}, (long[]) rs.getArray(1).getArray());
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void testSortOrderInDataColWithOffset() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL, page_id UNSIGNED_LONG,"
+ + " dates BIGINT NOT NULL, \"value\" BIGINT NOT NULL CONSTRAINT pk PRIMARY KEY (id, dates, \"value\" DESC))";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute("UPSERT INTO " + tableName
+ + " (id, page_id, dates, \"value\") VALUES (1, 8, 1, 3)");
+ conn.createStatement().execute("UPSERT INTO " + tableName
+ + " (id, page_id, dates, \"value\") VALUES (2, 8, 2, 7)");
+ conn.createStatement().execute("UPSERT INTO " + tableName
+ + " (id, page_id, dates, \"value\") VALUES (3, 8, 3, 9)");
+ conn.createStatement().execute("UPSERT INTO " + tableName
+ + " (id, page_id, dates, \"value\") VALUES (5, 8, 5, 158)");
+ conn.createStatement().execute("UPSERT INTO " + tableName
+ + " (id, page_id, dates, \"value\") VALUES (4, 8, 4, 5)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT LAST_VALUES(\"value\", 2) WITHIN GROUP (ORDER BY dates ASC) FROM "
+ + tableName + " GROUP BY page_id");
+
+ assertTrue(rs.next());
+ Assert.assertArrayEquals(new long[]{158, 5}, (long[]) rs.getArray(1).getArray());
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void nonUniqueValuesInOrderByAsc() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL PRIMARY KEY, page_id UNSIGNED_LONG,"
+ + " dates INTEGER, val INTEGER)";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (2, 8, 1, 7)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (3, 8, 2, 9)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (4, 8, 2, 4)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (5, 8, 2, 2)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (6, 8, 3, 3)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT LAST_VALUES(val, 3) WITHIN GROUP (ORDER BY dates ASC) FROM " + tableName
+ + " GROUP BY page_id");
+
+ int[] result = null;
+ assertTrue(rs.next());
+ result = (int[]) rs.getArray(1).getArray();
+ assertEquals(result[0], 3);
+ assertInIntArray(new int[]{2, 4, 9}, result[1]);
+ assertInIntArray(new int[]{2, 4, 9}, result[2]);
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void nonUniqueValuesInOrderByAscSkipDuplicit() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL PRIMARY KEY, page_id UNSIGNED_LONG,"
+ + " dates INTEGER, val INTEGER)";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (2, 8, 1, 7)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (3, 8, 2, 9)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (4, 8, 2, 4)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (5, 8, 2, 2)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (6, 8, 3, 3)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT LAST_VALUES(val, 5) WITHIN GROUP (ORDER BY dates ASC) FROM " + tableName
+ + " GROUP BY page_id");
+
+ int[] result = null;
+ assertTrue(rs.next());
+ result = (int[]) rs.getArray(1).getArray();
+ assertEquals(result[0], 3);
+ assertInIntArray(new int[]{2, 4, 9}, result[1]);
+ assertInIntArray(new int[]{2, 4, 9}, result[2]);
+ assertInIntArray(new int[]{2, 4, 9}, result[3]);
+ assertEquals(result[4], 7);
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void nonUniqueValuesInOrderByDesc() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL PRIMARY KEY, page_id UNSIGNED_LONG,"
+ + " dates INTEGER, val INTEGER)";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (2, 8, 1, 7)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (3, 8, 2, 9)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (4, 8, 2, 4)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (5, 8, 2, 2)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (6, 8, 3, 3)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT LAST_VALUES(val, 3) WITHIN GROUP (ORDER BY dates DESC) FROM " + tableName
+ + " GROUP BY page_id");
+
+ int[] result = null;
+ assertTrue(rs.next());
+ result = (int[]) rs.getArray(1).getArray();
+ assertEquals(result[0], 7);
+ assertInIntArray(new int[]{2, 4, 9}, result[1]);
+ assertInIntArray(new int[]{2, 4, 9}, result[2]);
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void nonUniqueValuesInOrderNextValueDesc() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL PRIMARY KEY, page_id UNSIGNED_LONG,"
+ + " dates INTEGER, val INTEGER)";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (2, 8, 0, 7)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (3, 8, 1, 9)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (4, 8, 2, 4)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (5, 8, 2, 2)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (6, 8, 3, 3)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (7, 8, 3, 5)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT LAST_VALUES(val, 2) WITHIN GROUP (ORDER BY dates DESC) FROM " + tableName
+ + " GROUP BY page_id");
+
+ int[] result = null;
+ assertTrue(rs.next());
+ result = (int[]) rs.getArray(1).getArray();
+ assertEquals(result[0], 7);
+ assertEquals(result[1], 9);
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void nonUniqueValuesInOrderNextValueAsc() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL PRIMARY KEY, page_id UNSIGNED_LONG,"
+ + " dates INTEGER, val INTEGER)";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (2, 8, 0, 7)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (3, 8, 1, 9)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (4, 8, 2, 4)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (5, 8, 2, 2)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (6, 8, 3, 3)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (7, 8, 3, 5)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT LAST_VALUES(val, 5) WITHIN GROUP (ORDER BY dates ASC) FROM " + tableName
+ + " GROUP BY page_id");
+
+ int[] result = null;
+ assertTrue(rs.next());
+ result = (int[]) rs.getArray(1).getArray();
+ assertInIntArray(new int[]{3, 5}, result[0]);
+ assertInIntArray(new int[]{3, 5}, result[1]);
+ assertInIntArray(new int[]{2, 4}, result[2]);
+ assertInIntArray(new int[]{2, 4}, result[3]);
+ assertEquals(result[4], 9);
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void ignoreNullValues() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL, page_id UNSIGNED_LONG,"
+ + " dates BIGINT NOT NULL, \"value\" BIGINT NULL CONSTRAINT pk PRIMARY KEY (id, dates))";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute("UPSERT INTO " + tableName
+ + " (id, page_id, dates, \"value\") VALUES (1, 8, 1, 1)");
+ conn.createStatement().execute("UPSERT INTO " + tableName
+ + " (id, page_id, dates, \"value\") VALUES (2, 8, 2, NULL)");
+ conn.createStatement().execute("UPSERT INTO " + tableName
+ + " (id, page_id, dates, \"value\") VALUES (3, 8, 3, NULL)");
+ conn.createStatement().execute("UPSERT INTO " + tableName
+ + " (id, page_id, dates, \"value\") VALUES (5, 8, 4, NULL)");
+ conn.createStatement().execute("UPSERT INTO " + tableName
+ + " (id, page_id, dates, \"value\") VALUES (4, 8, 5, 5)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT LAST_VALUES(\"value\", 2) WITHIN GROUP (ORDER BY dates ASC) FROM "
+ + tableName + " GROUP BY page_id");
+
+ assertTrue(rs.next());
+ Assert.assertArrayEquals(new long[]{5, 1}, (long[]) rs.getArray(1).getArray());
+ assertFalse(rs.next());
+ }
+
+ @Test
+ public void rowLessThanOffsetWithGroupBy() throws Exception {
+ Connection conn = DriverManager.getConnection(getUrl());
+
+ String tableName = generateUniqueName();
+ String ddl = "CREATE TABLE IF NOT EXISTS " + tableName + " "
+ + "(id INTEGER NOT NULL PRIMARY KEY, page_id UNSIGNED_LONG,"
+ + " dates INTEGER, val INTEGER)";
+ conn.createStatement().execute(ddl);
+
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (2, 8, 0, 7)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (3, 8, 1, 9)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (4, 8, 2, 4)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (5, 8, 2, 2)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (6, 8, 3, 3)");
+ conn.createStatement().execute(
+ "UPSERT INTO " + tableName + " (id, page_id, dates, val) VALUES (7, 9, 3, 5)");
+ conn.commit();
+
+ ResultSet rs = conn.createStatement().executeQuery(
+ "SELECT LAST_VALUES(val, 2) WITHIN GROUP (ORDER BY dates ASC) FROM "
+ + tableName + " GROUP BY page_id");
+
+ assertTrue(rs.next());
+ int[] result = (int[]) rs.getArray(1).getArray();
+ assertEquals(result[0], 3);
+ assertInIntArray(new int[]{2, 4}, result[1]);
+ assertTrue(rs.next());
+ Assert.assertArrayEquals(new int[]{5}, (int[]) rs.getArray(1).getArray());
+ assertFalse(rs.next());
+ }
+
+ private void assertInIntArray(int[] should, int actualValue) {
+ ArrayList<Integer> shouldList = new ArrayList<Integer>();
+ for (int i: should) {
+ shouldList.add(i);
+ }
+ assertTrue(shouldList.contains(actualValue));
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/3733a4ef/phoenix-core/src/main/java/org/apache/phoenix/expression/ExpressionType.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/ExpressionType.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/ExpressionType.java
index 7f95da6..3c42559 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/expression/ExpressionType.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/ExpressionType.java
@@ -19,7 +19,117 @@ package org.apache.phoenix.expression;
import java.util.Map;
-import org.apache.phoenix.expression.function.*;
+import org.apache.phoenix.expression.function.AbsFunction;
+import org.apache.phoenix.expression.function.ArrayAllComparisonExpression;
+import org.apache.phoenix.expression.function.ArrayAnyComparisonExpression;
+import org.apache.phoenix.expression.function.ArrayAppendFunction;
+import org.apache.phoenix.expression.function.ArrayConcatFunction;
+import org.apache.phoenix.expression.function.ArrayElemRefExpression;
+import org.apache.phoenix.expression.function.ArrayFillFunction;
+import org.apache.phoenix.expression.function.ArrayIndexFunction;
+import org.apache.phoenix.expression.function.ArrayLengthFunction;
+import org.apache.phoenix.expression.function.ArrayPrependFunction;
+import org.apache.phoenix.expression.function.ArrayToStringFunction;
+import org.apache.phoenix.expression.function.ByteBasedRegexpReplaceFunction;
+import org.apache.phoenix.expression.function.ByteBasedRegexpSplitFunction;
+import org.apache.phoenix.expression.function.ByteBasedRegexpSubstrFunction;
+import org.apache.phoenix.expression.function.CbrtFunction;
+import org.apache.phoenix.expression.function.CeilDateExpression;
+import org.apache.phoenix.expression.function.CeilDecimalExpression;
+import org.apache.phoenix.expression.function.CeilFunction;
+import org.apache.phoenix.expression.function.CeilMonthExpression;
+import org.apache.phoenix.expression.function.CeilTimestampExpression;
+import org.apache.phoenix.expression.function.CeilWeekExpression;
+import org.apache.phoenix.expression.function.CeilYearExpression;
+import org.apache.phoenix.expression.function.CoalesceFunction;
+import org.apache.phoenix.expression.function.ConvertTimezoneFunction;
+import org.apache.phoenix.expression.function.CountAggregateFunction;
+import org.apache.phoenix.expression.function.DayOfMonthFunction;
+import org.apache.phoenix.expression.function.DecodeFunction;
+import org.apache.phoenix.expression.function.DefaultValueExpression;
+import org.apache.phoenix.expression.function.DistinctCountAggregateFunction;
+import org.apache.phoenix.expression.function.DistinctValueAggregateFunction;
+import org.apache.phoenix.expression.function.EncodeFunction;
+import org.apache.phoenix.expression.function.ExpFunction;
+import org.apache.phoenix.expression.function.ExternalSqlTypeIdFunction;
+import org.apache.phoenix.expression.function.FirstValueFunction;
+import org.apache.phoenix.expression.function.FirstValuesFunction;
+import org.apache.phoenix.expression.function.FloorDateExpression;
+import org.apache.phoenix.expression.function.FloorDecimalExpression;
+import org.apache.phoenix.expression.function.FloorFunction;
+import org.apache.phoenix.expression.function.FloorMonthExpression;
+import org.apache.phoenix.expression.function.FloorWeekExpression;
+import org.apache.phoenix.expression.function.FloorYearExpression;
+import org.apache.phoenix.expression.function.GetBitFunction;
+import org.apache.phoenix.expression.function.GetByteFunction;
+import org.apache.phoenix.expression.function.HourFunction;
+import org.apache.phoenix.expression.function.IndexStateNameFunction;
+import org.apache.phoenix.expression.function.InstrFunction;
+import org.apache.phoenix.expression.function.InvertFunction;
+import org.apache.phoenix.expression.function.LTrimFunction;
+import org.apache.phoenix.expression.function.LastValueFunction;
+import org.apache.phoenix.expression.function.LastValuesFunction;
+import org.apache.phoenix.expression.function.LengthFunction;
+import org.apache.phoenix.expression.function.LnFunction;
+import org.apache.phoenix.expression.function.LogFunction;
+import org.apache.phoenix.expression.function.LowerFunction;
+import org.apache.phoenix.expression.function.LpadFunction;
+import org.apache.phoenix.expression.function.MD5Function;
+import org.apache.phoenix.expression.function.MaxAggregateFunction;
+import org.apache.phoenix.expression.function.MinAggregateFunction;
+import org.apache.phoenix.expression.function.MinuteFunction;
+import org.apache.phoenix.expression.function.MonthFunction;
+import org.apache.phoenix.expression.function.NowFunction;
+import org.apache.phoenix.expression.function.NthValueFunction;
+import org.apache.phoenix.expression.function.OctetLengthFunction;
+import org.apache.phoenix.expression.function.PercentRankAggregateFunction;
+import org.apache.phoenix.expression.function.PercentileContAggregateFunction;
+import org.apache.phoenix.expression.function.PercentileDiscAggregateFunction;
+import org.apache.phoenix.expression.function.PowerFunction;
+import org.apache.phoenix.expression.function.RTrimFunction;
+import org.apache.phoenix.expression.function.RandomFunction;
+import org.apache.phoenix.expression.function.RegexpReplaceFunction;
+import org.apache.phoenix.expression.function.RegexpSplitFunction;
+import org.apache.phoenix.expression.function.RegexpSubstrFunction;
+import org.apache.phoenix.expression.function.ReverseFunction;
+import org.apache.phoenix.expression.function.RoundDateExpression;
+import org.apache.phoenix.expression.function.RoundDecimalExpression;
+import org.apache.phoenix.expression.function.RoundFunction;
+import org.apache.phoenix.expression.function.RoundMonthExpression;
+import org.apache.phoenix.expression.function.RoundTimestampExpression;
+import org.apache.phoenix.expression.function.RoundWeekExpression;
+import org.apache.phoenix.expression.function.RoundYearExpression;
+import org.apache.phoenix.expression.function.SQLIndexTypeFunction;
+import org.apache.phoenix.expression.function.SQLTableTypeFunction;
+import org.apache.phoenix.expression.function.SQLViewTypeFunction;
+import org.apache.phoenix.expression.function.SecondFunction;
+import org.apache.phoenix.expression.function.SetBitFunction;
+import org.apache.phoenix.expression.function.SetByteFunction;
+import org.apache.phoenix.expression.function.SignFunction;
+import org.apache.phoenix.expression.function.SqlTypeNameFunction;
+import org.apache.phoenix.expression.function.SqrtFunction;
+import org.apache.phoenix.expression.function.StddevPopFunction;
+import org.apache.phoenix.expression.function.StddevSampFunction;
+import org.apache.phoenix.expression.function.StringBasedRegexpReplaceFunction;
+import org.apache.phoenix.expression.function.StringBasedRegexpSplitFunction;
+import org.apache.phoenix.expression.function.StringBasedRegexpSubstrFunction;
+import org.apache.phoenix.expression.function.StringToArrayFunction;
+import org.apache.phoenix.expression.function.SubstrFunction;
+import org.apache.phoenix.expression.function.SumAggregateFunction;
+import org.apache.phoenix.expression.function.TimezoneOffsetFunction;
+import org.apache.phoenix.expression.function.ToCharFunction;
+import org.apache.phoenix.expression.function.ToDateFunction;
+import org.apache.phoenix.expression.function.ToNumberFunction;
+import org.apache.phoenix.expression.function.ToTimeFunction;
+import org.apache.phoenix.expression.function.ToTimestampFunction;
+import org.apache.phoenix.expression.function.TrimFunction;
+import org.apache.phoenix.expression.function.TruncFunction;
+import org.apache.phoenix.expression.function.UDFExpression;
+import org.apache.phoenix.expression.function.UpperFunction;
+import org.apache.phoenix.expression.function.WeekFunction;
+import org.apache.phoenix.expression.function.YearFunction;
+import org.apache.phoenix.expression.function.DayOfWeekFunction;
+import org.apache.phoenix.expression.function.DayOfYearFunction;
import com.google.common.collect.Maps;
@@ -179,7 +289,10 @@ public enum ExpressionType {
DayOfWeekFunction(DayOfWeekFunction.class),
DayOfYearFunction(DayOfYearFunction.class),
DefaultValueExpression(DefaultValueExpression.class),
- ArrayColumnExpression(SingleCellColumnExpression.class);
+ ArrayColumnExpression(SingleCellColumnExpression.class),
+ FirstValuesFunction(FirstValuesFunction.class),
+ LastValuesFunction(LastValuesFunction.class);
+
ExpressionType(Class<? extends Expression> clazz) {
this.clazz = clazz;
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/3733a4ef/phoenix-core/src/main/java/org/apache/phoenix/expression/aggregator/FirstLastValueBaseClientAggregator.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/aggregator/FirstLastValueBaseClientAggregator.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/aggregator/FirstLastValueBaseClientAggregator.java
index 804c38e..5c49312 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/expression/aggregator/FirstLastValueBaseClientAggregator.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/aggregator/FirstLastValueBaseClientAggregator.java
@@ -27,16 +27,17 @@ import java.util.TreeMap;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.Bytes.ByteArrayComparator;
-import org.apache.phoenix.schema.types.PDataType;
-import org.apache.phoenix.schema.types.PVarbinary;
import org.apache.phoenix.schema.SortOrder;
import org.apache.phoenix.schema.tuple.SingleKeyValueTuple;
import org.apache.phoenix.schema.tuple.Tuple;
+import org.apache.phoenix.schema.types.PArrayDataType;
+import org.apache.phoenix.schema.types.PDataType;
+import org.apache.phoenix.schema.types.PVarbinary;
import org.apache.phoenix.util.ByteUtil;
import org.apache.phoenix.util.FirstLastNthValueDataContainer;
/**
- * Base client aggregator for (FIRST|LAST|NTH)_VALUE functions
+ * Base client aggregator for (FIRST|LAST|NTH)_VALUE and (FIRST|LAST)_VALUES functions
*
*/
public class FirstLastValueBaseClientAggregator extends BaseAggregator {
@@ -47,9 +48,19 @@ public class FirstLastValueBaseClientAggregator extends BaseAggregator {
protected byte[] topValue = null;
protected TreeMap<byte[], LinkedList<byte[]>> topValues = new TreeMap<byte[], LinkedList<byte[]>>(new ByteArrayComparator());
protected boolean isAscending;
+ protected PDataType dataType;
+
+ // Set to true for retrieving multiple top values for FIRST_VALUES or LAST_VALUES
+ protected boolean isArrayReturnType = false;
public FirstLastValueBaseClientAggregator() {
super(SortOrder.getDefault());
+ this.dataType = PVarbinary.INSTANCE;
+ }
+
+ public FirstLastValueBaseClientAggregator(PDataType type) {
+ super(SortOrder.getDefault());
+ this.dataType = (type == null) ? PVarbinary.INSTANCE : type;
}
@Override
@@ -73,20 +84,44 @@ public class FirstLastValueBaseClientAggregator extends BaseAggregator {
entrySet = topValues.descendingMap().entrySet();
}
- int counter = offset;
+ int counter = 0;
+ ImmutableBytesWritable arrPtr = new ImmutableBytesWritable(ByteUtil.EMPTY_BYTE_ARRAY);
for (Map.Entry<byte[], LinkedList<byte[]>> entry : entrySet) {
ListIterator<byte[]> it = entry.getValue().listIterator();
while (it.hasNext()) {
- if (--counter == 0) {
- ptr.set(it.next());
- return true;
+ if (isArrayReturnType) {
+ ImmutableBytesWritable newArrPtr = new ImmutableBytesWritable(it.next());
+ PArrayDataType.appendItemToArray(
+ newArrPtr,
+ arrPtr.getLength(),
+ arrPtr.getOffset(),
+ arrPtr.get(),
+ PDataType.fromTypeId(dataType.getSqlType() - PDataType.ARRAY_TYPE_BASE),
+ counter,
+ null,
+ sortOrder);
+ arrPtr = newArrPtr;
+
+ if (++counter == offset) {
+ break;
+ }
+ } else {
+ if (++counter == offset) {
+ ptr.set(it.next());
+ return true;
+ }
+ it.next();
}
- it.next();
}
}
- //not enought values to return Nth
- return false;
+ if (isArrayReturnType) {
+ ptr.set(arrPtr.get());
+ return true;
+ }
+
+ ptr.set(ByteUtil.EMPTY_BYTE_ARRAY);
+ return true;
}
if (topValue == null) {
@@ -100,7 +135,7 @@ public class FirstLastValueBaseClientAggregator extends BaseAggregator {
@Override
public void aggregate(Tuple tuple, ImmutableBytesWritable ptr) {
- //if is called cause aggregation in ORDER BY clausule
+ //if is called cause aggregation in ORDER BY clause
if (tuple instanceof SingleKeyValueTuple) {
topValue = ptr.copyBytes();
return;
@@ -143,13 +178,15 @@ public class FirstLastValueBaseClientAggregator extends BaseAggregator {
@Override
public PDataType getDataType() {
- return PVarbinary.INSTANCE;
+ return dataType;
}
- public void init(int offset) {
- if (offset != 0) {
+ public void init(int offset, boolean isArrayReturnType) {
+ if (offset > 0) {
useOffset = true;
this.offset = offset;
}
+
+ this.isArrayReturnType = isArrayReturnType;
}
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/3733a4ef/phoenix-core/src/main/java/org/apache/phoenix/expression/aggregator/FirstLastValueServerAggregator.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/aggregator/FirstLastValueServerAggregator.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/aggregator/FirstLastValueServerAggregator.java
index 66b38c6..b41c6c6 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/expression/aggregator/FirstLastValueServerAggregator.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/aggregator/FirstLastValueServerAggregator.java
@@ -17,21 +17,22 @@
*/
package org.apache.phoenix.expression.aggregator;
-import org.apache.phoenix.expression.Expression;
-import org.apache.phoenix.schema.types.PDataType;
-import org.apache.phoenix.schema.types.PVarbinary;
-import org.apache.phoenix.schema.tuple.Tuple;
-import org.apache.phoenix.util.ByteUtil;
-import org.apache.phoenix.util.SizedUtil;
import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.TreeMap;
+
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.phoenix.expression.Expression;
import org.apache.phoenix.schema.SortOrder;
+import org.apache.phoenix.schema.tuple.Tuple;
+import org.apache.phoenix.schema.types.PDataType;
+import org.apache.phoenix.schema.types.PVarbinary;
+import org.apache.phoenix.util.ByteUtil;
import org.apache.phoenix.util.FirstLastNthValueDataContainer;
+import org.apache.phoenix.util.SizedUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
http://git-wip-us.apache.org/repos/asf/phoenix/blob/3733a4ef/phoenix-core/src/main/java/org/apache/phoenix/expression/function/FirstValueFunction.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/function/FirstValueFunction.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/function/FirstValueFunction.java
index bed3e7a..420deba 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/expression/function/FirstValueFunction.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/function/FirstValueFunction.java
@@ -66,7 +66,7 @@ public class FirstValueFunction extends FirstLastValueBaseFunction {
@Override
public Aggregator newClientAggregator() {
FirstLastValueBaseClientAggregator aggregator = new FirstLastValueBaseClientAggregator();
- aggregator.init(0);
+ aggregator.init(0, false);
return aggregator;
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/3733a4ef/phoenix-core/src/main/java/org/apache/phoenix/expression/function/FirstValuesFunction.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/function/FirstValuesFunction.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/function/FirstValuesFunction.java
new file mode 100644
index 0000000..fec2eaf
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/function/FirstValuesFunction.java
@@ -0,0 +1,97 @@
+/*
+ * 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.phoenix.expression.function;
+
+import java.util.List;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.phoenix.expression.Expression;
+import org.apache.phoenix.expression.LiteralExpression;
+import org.apache.phoenix.expression.aggregator.Aggregator;
+import org.apache.phoenix.expression.aggregator.FirstLastValueBaseClientAggregator;
+import org.apache.phoenix.expression.aggregator.FirstLastValueServerAggregator;
+import org.apache.phoenix.parse.FirstValuesAggregateParseNode;
+import org.apache.phoenix.parse.FunctionParseNode;
+import org.apache.phoenix.schema.types.PArrayDataType;
+import org.apache.phoenix.schema.types.PBoolean;
+import org.apache.phoenix.schema.types.PDataType;
+import org.apache.phoenix.schema.types.PInteger;
+
+/**
+ * Built-in function for FIRST_VALUES(<expression>, <expression>) WITHIN GROUP (ORDER BY <expression> ASC/DESC) aggregate
+ * function
+ *
+ */
+@FunctionParseNode.BuiltInFunction(name = FirstValuesFunction.NAME, nodeClass = FirstValuesAggregateParseNode.class, args = {
+ @FunctionParseNode.Argument(),
+ @FunctionParseNode.Argument(allowedTypes = { PBoolean.class }, isConstant = true),
+ @FunctionParseNode.Argument(),
+ @FunctionParseNode.Argument(allowedTypes = { PInteger.class }, isConstant = true)})
+public class FirstValuesFunction extends FirstLastValueBaseFunction {
+ public static final String NAME = "FIRST_VALUES";
+ private int offset;
+
+ public FirstValuesFunction() {
+ }
+
+ public FirstValuesFunction(List<Expression> childExpressions) {
+ this(childExpressions, null);
+ }
+
+ public FirstValuesFunction(List<Expression> childExpressions, CountAggregateFunction delegate) {
+ super(childExpressions, delegate);
+ }
+
+ @Override
+ public Aggregator newServerAggregator(Configuration conf) {
+ FirstLastValueServerAggregator aggregator = new FirstLastValueServerAggregator();
+
+ offset = ((Number) ((LiteralExpression) children.get(3)).getValue()).intValue();
+ boolean order = (Boolean) ((LiteralExpression) children.get(1)).getValue();
+
+ aggregator.init(children, order, offset);
+
+ return aggregator;
+ }
+
+ @Override
+ public Aggregator newClientAggregator() {
+ FirstLastValueBaseClientAggregator aggregator = new FirstLastValueBaseClientAggregator(getDataType());
+
+ if (children.size() < 3) {
+ aggregator.init(offset, true);
+ } else {
+ aggregator.init(((Number) ((LiteralExpression) children.get(3)).getValue()).intValue(), true);
+ }
+
+ return aggregator;
+ }
+
+ @Override
+ public String getName() {
+ return NAME;
+ }
+
+ @Override
+ public PDataType getDataType() {
+ if (children.size() < 3) {
+ return null;
+ }
+ return PDataType.fromTypeId(children.get(2).getDataType().getSqlType() + PArrayDataType.ARRAY_TYPE_BASE);
+ }
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/3733a4ef/phoenix-core/src/main/java/org/apache/phoenix/expression/function/LastValueFunction.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/expression/function/LastValueFunction.java b/phoenix-core/src/main/java/org/apache/phoenix/expression/function/LastValueFunction.java
index fdd6898..f5b3193 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/expression/function/LastValueFunction.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/expression/function/LastValueFunction.java
@@ -68,7 +68,7 @@ public class LastValueFunction extends FirstLastValueBaseFunction {
public Aggregator newClientAggregator() {
FirstLastValueBaseClientAggregator aggregator = new FirstLastValueBaseClientAggregator();
- aggregator.init(0);
+ aggregator.init(0, false);
return aggregator;
}