PHOENIX-3547 Supporting more number of indices per table.

dev/test-patch.properties

@@ -27,7 +27,7 @@ MAX_LINE_LENGTH=100
 # All supported branches for testing with precommit build
 # be sure to consider branch name prefixes in the order, ie, 4.x should appear
 # before 4 since the latter is a prefix
-BRANCH_NAMES="4.x-HBase-0.98 4.x-HBase-1.1 4.x-HBase-1.2 master"
+BRANCH_NAMES="4.x-HBase-1.2 4.x-HBase-1.3 4.x-HBase-1.4 master"
 
 
 # All supported Hadoop versions that we want to test the compilation with

phoenix-core/src/it/java/org/apache/phoenix/end2end/AlterTableWithViewsIT.java

@@ -530,7 +530,7 @@ public void testAddExistingViewPkColumnToBaseTableWithMultipleViews() throws Exc
 
             try {
                 // should fail because there are two view with different pk columns
-                conn.createStatement().execute("ALTER TABLE " + tableName + " ADD VIEW_COL1 DECIMAL PRIMARY KEY, VIEW_COL2 VARCHAR PRIMARY KEY");
+                conn.createStatement().execute("ALTER TABLE " + tableName + " ADD VIEW_COL1 DECIMAL(10,2) PRIMARY KEY, VIEW_COL2 VARCHAR(256) PRIMARY KEY");
                 fail();
             }
             catch (SQLException e) {

phoenix-core/src/it/java/org/apache/phoenix/end2end/BasePermissionsIT.java

@@ -16,8 +16,28 @@
  */
 package org.apache.phoenix.end2end;
 
-import com.google.common.base.Joiner;
-import com.google.common.base.Throwables;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.io.IOException;
+import java.lang.reflect.UndeclaredThrowableException;
+import java.security.PrivilegedExceptionAction;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -38,34 +58,13 @@
 import org.apache.phoenix.query.QueryConstants;
 import org.apache.phoenix.query.QueryServices;
 import org.apache.phoenix.util.PhoenixRuntime;
-import org.apache.phoenix.util.QueryUtil;
 import org.junit.After;
 import org.junit.BeforeClass;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 
-import java.io.IOException;
-import java.lang.reflect.UndeclaredThrowableException;
-import java.security.PrivilegedExceptionAction;
-import java.sql.Connection;
-import java.sql.DriverManager;
-import java.sql.PreparedStatement;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.sql.Statement;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Properties;
-import java.util.Set;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
+import com.google.common.base.Joiner;
+import com.google.common.base.Throwables;
 
 @RunWith(Parameterized.class)
 public class BasePermissionsIT extends BaseTest {
@@ -75,17 +74,23 @@ public class BasePermissionsIT extends BaseTest {
     static String SUPERUSER;
 
     static HBaseTestingUtility testUtil;
-    static final Set<String> PHOENIX_SYSTEM_TABLES = new HashSet<>(Arrays.asList(
-            "SYSTEM.CATALOG", "SYSTEM.SEQUENCE", "SYSTEM.STATS", "SYSTEM.FUNCTION"));
+    static final Set<String> PHOENIX_SYSTEM_TABLES =
+            new HashSet<>(Arrays.asList("SYSTEM.CATALOG", "SYSTEM.SEQUENCE", "SYSTEM.STATS",
+                "SYSTEM.FUNCTION", "SYSTEM.MUTEX"));
 
-    static final Set<String> PHOENIX_SYSTEM_TABLES_IDENTIFIERS = new HashSet<>(Arrays.asList(
-            "SYSTEM.\"CATALOG\"", "SYSTEM.\"SEQUENCE\"", "SYSTEM.\"STATS\"", "SYSTEM.\"FUNCTION\""));
+    static final Set<String> PHOENIX_SYSTEM_TABLES_IDENTIFIERS =
+            new HashSet<>(Arrays.asList("SYSTEM.\"CATALOG\"", "SYSTEM.\"SEQUENCE\"",
+                "SYSTEM.\"STATS\"", "SYSTEM.\"FUNCTION\"", "SYSTEM.\"MUTEX\""));
 
     static final String SYSTEM_SEQUENCE_IDENTIFIER =
             QueryConstants.SYSTEM_SCHEMA_NAME + "." + "\"" + PhoenixDatabaseMetaData.SYSTEM_SEQUENCE_TABLE+ "\"";
 
+    static final String SYSTEM_MUTEX_IDENTIFIER =
+            QueryConstants.SYSTEM_SCHEMA_NAME + "." + "\""
+                    + PhoenixDatabaseMetaData.SYSTEM_MUTEX_TABLE_NAME + "\"";
+
     static final Set<String> PHOENIX_NAMESPACE_MAPPED_SYSTEM_TABLES = new HashSet<>(Arrays.asList(
-            "SYSTEM:CATALOG", "SYSTEM:SEQUENCE", "SYSTEM:STATS", "SYSTEM:FUNCTION"));
+            "SYSTEM:CATALOG", "SYSTEM:SEQUENCE", "SYSTEM:STATS", "SYSTEM:FUNCTION", "SYSTEM:MUTEX"));
 
     // Create Multiple users so that we can use Hadoop UGI to run tasks as various users
     // Permissions can be granted or revoke by superusers and admins only

phoenix-core/src/it/java/org/apache/phoenix/end2end/BaseTenantSpecificViewIndexIT.java

@@ -148,9 +148,9 @@ private void createAndVerifyIndex(Connection conn, String viewName, String table
                             + "CLIENT MERGE SORT", QueryUtil.getExplainPlan(rs));
         } else {
             String expected = saltBuckets == null ? 
-                    "CLIENT PARALLEL 1-WAY RANGE SCAN OVER _IDX_" + tableName + " [-32768,'" + tenantId + "','" + valuePrefix + "v2-1']\n"
+                    "CLIENT PARALLEL 1-WAY RANGE SCAN OVER _IDX_" + tableName + " [-9223372036854775808,'" + tenantId + "','" + valuePrefix + "v2-1']\n"
                             + "    SERVER FILTER BY FIRST KEY ONLY" :
-                    "CLIENT PARALLEL 3-WAY RANGE SCAN OVER _IDX_" + tableName + " [0,-32768,'" + tenantId + "','" + valuePrefix + "v2-1'] - ["+(saltBuckets.intValue()-1)+",-32768,'" + tenantId + "','" + valuePrefix + "v2-1']\n"
+                    "CLIENT PARALLEL 3-WAY RANGE SCAN OVER _IDX_" + tableName + " [0,-9223372036854775808,'" + tenantId + "','" + valuePrefix + "v2-1'] - ["+(saltBuckets.intValue()-1)+",-9223372036854775808,'" + tenantId + "','" + valuePrefix + "v2-1']\n"
 
                   + "    SERVER FILTER BY FIRST KEY ONLY\n"
                   + "CLIENT MERGE SORT";

phoenix-core/src/it/java/org/apache/phoenix/end2end/BaseViewIT.java

@@ -196,8 +196,8 @@ protected Pair<String,Scan> testUpdatableViewIndex(Integer saltBuckets, boolean
                 queryPlan);
         } else {
             assertEquals(saltBuckets == null
-                    ? "CLIENT PARALLEL 1-WAY RANGE SCAN OVER " + viewIndexPhysicalName +" [" + Short.MIN_VALUE + ",51]"
-                            : "CLIENT PARALLEL " + saltBuckets + "-WAY RANGE SCAN OVER " + viewIndexPhysicalName + " [0," + Short.MIN_VALUE + ",51] - ["+(saltBuckets.intValue()-1)+"," + Short.MIN_VALUE + ",51]\nCLIENT MERGE SORT",
+                    ? "CLIENT PARALLEL 1-WAY RANGE SCAN OVER " + viewIndexPhysicalName +" [" + Long.MIN_VALUE + ",51]"
+                            : "CLIENT PARALLEL " + saltBuckets + "-WAY RANGE SCAN OVER " + viewIndexPhysicalName + " [0," + Long.MIN_VALUE + ",51] - ["+(saltBuckets.intValue()-1)+"," + Long.MIN_VALUE + ",51]\nCLIENT MERGE SORT",
                             queryPlan);
         }
 
@@ -237,9 +237,9 @@ protected Pair<String,Scan> testUpdatableViewIndex(Integer saltBuckets, boolean
         } else {
             physicalTableName = viewIndexPhysicalName;
             assertEquals(saltBuckets == null
-                    ? "CLIENT PARALLEL 1-WAY RANGE SCAN OVER " + viewIndexPhysicalName +" [" + (Short.MIN_VALUE+1) + ",'foo']\n"
+                    ? "CLIENT PARALLEL 1-WAY RANGE SCAN OVER " + viewIndexPhysicalName +" [" + (Long.MIN_VALUE+1) + ",'foo']\n"
                             + "    SERVER FILTER BY FIRST KEY ONLY"
-                            : "CLIENT PARALLEL " + saltBuckets + "-WAY RANGE SCAN OVER " + viewIndexPhysicalName + " [0," + (Short.MIN_VALUE+1) + ",'foo'] - ["+(saltBuckets.intValue()-1)+"," + (Short.MIN_VALUE+1) + ",'foo']\n"
+                            : "CLIENT PARALLEL " + saltBuckets + "-WAY RANGE SCAN OVER " + viewIndexPhysicalName + " [0," + (Long.MIN_VALUE+1) + ",'foo'] - ["+(saltBuckets.intValue()-1)+"," + (Long.MIN_VALUE+1) + ",'foo']\n"
                                     + "    SERVER FILTER BY FIRST KEY ONLY\n"
                                     + "CLIENT MERGE SORT",
                             QueryUtil.getExplainPlan(rs));

phoenix-core/src/it/java/org/apache/phoenix/end2end/ChangePermissionsIT.java

@@ -57,7 +57,8 @@ private void grantSystemTableAccess(User superUser, User... users) throws Except
             } else {
                 verifyAllowed(grantPermissions("RX", user, PHOENIX_SYSTEM_TABLES_IDENTIFIERS, false), superUser);
             }
-            verifyAllowed(grantPermissions("W", user, SYSTEM_SEQUENCE_IDENTIFIER, false), superUser);
+            verifyAllowed(grantPermissions("RWX", user, SYSTEM_SEQUENCE_IDENTIFIER, false), superUser);
+            verifyAllowed(grantPermissions("RWX", user, SYSTEM_MUTEX_IDENTIFIER, false), superUser);
         }
     }
 
@@ -69,6 +70,7 @@ private void revokeSystemTableAccess(User superUser, User... users) throws Excep
                 verifyAllowed(revokePermissions(user, PHOENIX_SYSTEM_TABLES_IDENTIFIERS, false), superUser);
             }
             verifyAllowed(revokePermissions(user, SYSTEM_SEQUENCE_IDENTIFIER, false), superUser);
+            verifyAllowed(revokePermissions(user, SYSTEM_MUTEX_IDENTIFIER, false), superUser);
         }
     }
 

phoenix-core/src/it/java/org/apache/phoenix/end2end/ClientHashAggregateIT.java

@@ -0,0 +1,208 @@
+/*
+ * 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.apache.phoenix.util.TestUtil.TEST_PROPERTIES;
+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.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.util.Properties;
+
+import org.apache.phoenix.util.PropertiesUtil;
+import org.apache.phoenix.util.QueryUtil;
+import org.junit.Test;
+
+public class ClientHashAggregateIT extends ParallelStatsDisabledIT {
+
+    @Test
+    public void testSalted() throws Exception { 
+
+        Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+        Connection conn = DriverManager.getConnection(getUrl(), props);
+
+        try {
+            String table = createSalted(conn);
+            testTable(conn, table);
+        } finally {
+            conn.close();
+        }
+    }
+
+    @Test
+    public void testUnsalted() throws Exception { 
+
+        Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+        Connection conn = DriverManager.getConnection(getUrl(), props);
+
+        try {
+            String table = createUnsalted(conn);
+            testTable(conn, table);
+        } finally {
+            conn.close();
+        }
+    }
+
+    private void testTable(Connection conn, String table) throws Exception {
+        verifyExplain(conn, table, false, false);
+        verifyExplain(conn, table, false, true);
+        verifyExplain(conn, table, true, false);
+        verifyExplain(conn, table, true, true);
+
+        verifyResults(conn, table, 13, 0, false, false);
+        verifyResults(conn, table, 13, 0, false, true);
+        verifyResults(conn, table, 13, 0, true, false);
+        verifyResults(conn, table, 13, 0, true, true);
+
+        verifyResults(conn, table, 13, 17, false, true);
+        verifyResults(conn, table, 13, 17, true, true);
+
+        dropTable(conn, table);
+    }
+
+    private String createSalted(Connection conn) throws Exception {
+
+        String table = "SALTED_" + generateUniqueName();
+        String create = "CREATE TABLE " + table + " ("
+            + " keyA BIGINT NOT NULL,"
+            + " keyB BIGINT NOT NULL,"
+            + " val SMALLINT,"
+            + " CONSTRAINT pk PRIMARY KEY (keyA, keyB)"
+            + ") SALT_BUCKETS = 4";
+
+        conn.createStatement().execute(create);
+        return table;
+    }
+
+    private String createUnsalted(Connection conn) throws Exception {
+
+        String table = "UNSALTED_" + generateUniqueName();
+        String create = "CREATE TABLE " + table + " ("
+            + " keyA BIGINT NOT NULL,"
+            + " keyB BIGINT NOT NULL,"
+            + " val SMALLINT,"
+            + " CONSTRAINT pk PRIMARY KEY (keyA, keyB)"
+            + ")";
+
+        conn.createStatement().execute(create);
+        return table;
+    }
+
+    private String getQuery(String table, boolean hash, boolean swap, boolean sort) {
+
+        String query = "SELECT /*+ USE_SORT_MERGE_JOIN"
+            + (hash ? " HASH_AGGREGATE" : "") + " */"
+            + " t1.val v1, t2.val v2, COUNT(*) c"
+            + " FROM " + table + " t1 JOIN " + table + " t2"
+            + " ON (t1.keyB = t2.keyB)"
+            + " WHERE t1.keyA = 10 AND t2.keyA = 20"
+            + " GROUP BY "
+            + (swap ? "t2.val, t1.val" : "t1.val, t2.val")
+            + (sort ? " ORDER BY t1.val, t2.val" : "")
+            ;
+
+        return query;
+    }
+
+    private void verifyExplain(Connection conn, String table, boolean swap, boolean sort) throws Exception {
+
+        String query = "EXPLAIN " + getQuery(table, true, swap, sort);
+        Statement stmt = conn.createStatement();
+        ResultSet rs = stmt.executeQuery(query);
+        String plan = QueryUtil.getExplainPlan(rs);
+        rs.close();
+        assertTrue(plan != null && plan.contains("CLIENT HASH AGGREGATE"));
+        assertTrue(plan != null && (sort == plan.contains("CLIENT SORTED BY")));
+    }
+
+    private void verifyResults(Connection conn, String table, int c1, int c2, boolean swap, boolean sort) throws Exception {
+
+        String upsert = "UPSERT INTO " + table + "(keyA, keyB, val) VALUES(?, ?, ?)";
+        PreparedStatement upsertStmt = conn.prepareStatement(upsert);
+        for (int i = 0; i < c1; i++) {
+            upsertStmt.setInt(1, 10);
+            upsertStmt.setInt(2, 100+i);
+            upsertStmt.setInt(3, 1);
+            upsertStmt.execute();
+
+            upsertStmt.setInt(1, 20);
+            upsertStmt.setInt(2, 100+i);
+            upsertStmt.setInt(3, 2);
+            upsertStmt.execute();
+        }
+        for (int i = 0; i < c2; i++) {
+            upsertStmt.setInt(1, 10);
+            upsertStmt.setInt(2, 200+i);
+            upsertStmt.setInt(3, 2);
+            upsertStmt.execute();
+
+            upsertStmt.setInt(1, 20);
+            upsertStmt.setInt(2, 200+i);
+            upsertStmt.setInt(3, 1);
+            upsertStmt.execute();
+        }
+        conn.commit();
+
+        String hashQuery = getQuery(table, true, swap, sort);
+        String sortQuery = getQuery(table, false, swap, sort);
+        Statement stmt = conn.createStatement();
+        ResultSet hrs = stmt.executeQuery(hashQuery);
+        ResultSet srs = stmt.executeQuery(sortQuery);
+
+        try {
+            if (c1 > 0) {
+                assertTrue(hrs.next());
+                assertTrue(srs.next());
+                assertEquals(hrs.getInt("v1"), srs.getInt("v1"));
+                assertEquals(hrs.getInt("v2"), srs.getInt("v2"));
+                assertEquals(hrs.getInt("c"), srs.getInt("c"));
+                assertEquals(hrs.getInt("v1"), 1);
+                assertEquals(hrs.getInt("v2"), 2);
+                assertEquals(hrs.getInt("c"), c1);
+            }
+            if (c2 > 0) {
+                assertTrue(hrs.next());
+                assertTrue(srs.next());
+                assertEquals(hrs.getInt("v1"), srs.getInt("v1"));
+                assertEquals(hrs.getInt("v2"), srs.getInt("v2"));
+                assertEquals(hrs.getInt("c"), srs.getInt("c"));
+                assertEquals(hrs.getInt("v1"), 2);
+                assertEquals(hrs.getInt("v2"), 1);
+                assertEquals(hrs.getInt("c"), c2);
+            }
+            assertFalse(hrs.next());
+            assertFalse(srs.next());
+        } finally {
+            hrs.close();
+            srs.close();
+        }
+    }
+
+    private void dropTable(Connection conn, String table) throws Exception {
+
+        String drop = "DROP TABLE " + table;
+        Statement stmt = conn.createStatement();
+        stmt.execute(drop);
+        stmt.close();
+    }
+}