PHOENIX-1556 Base hash versus sort merge join decision on cost
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/6b40a36b Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/6b40a36b Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/6b40a36b Branch: refs/heads/4.x-HBase-1.1 Commit: 6b40a36b51eb92361dc2cfdca291f4391c7fcc01 Parents: 56cdcb9 Author: maryannxue <maryann....@gmail.com> Authored: Mon Feb 12 14:07:30 2018 -0800 Committer: maryannxue <maryann....@gmail.com> Committed: Tue Mar 13 16:48:01 2018 -0700 ---------------------------------------------------------------------- .../phoenix/end2end/CostBasedDecisionIT.java | 420 ++++++++++++----- .../apache/phoenix/compile/JoinCompiler.java | 43 ++ .../phoenix/compile/ListJarsQueryPlan.java | 8 +- .../apache/phoenix/compile/QueryCompiler.java | 449 ++++++++++--------- .../org/apache/phoenix/compile/QueryPlan.java | 2 + .../apache/phoenix/compile/TraceQueryPlan.java | 6 + .../apache/phoenix/execute/AggregatePlan.java | 41 +- .../phoenix/execute/ClientAggregatePlan.java | 46 +- .../phoenix/execute/ClientProcessingPlan.java | 4 + .../apache/phoenix/execute/ClientScanPlan.java | 22 +- .../apache/phoenix/execute/CorrelatePlan.java | 26 +- .../apache/phoenix/execute/CursorFetchPlan.java | 6 + .../apache/phoenix/execute/HashJoinPlan.java | 128 ++++-- .../execute/LiteralResultIterationPlan.java | 6 + .../org/apache/phoenix/execute/ScanPlan.java | 14 +- .../phoenix/execute/SortMergeJoinPlan.java | 20 +- .../phoenix/execute/TupleProjectionPlan.java | 6 + .../org/apache/phoenix/execute/UnionPlan.java | 12 +- .../apache/phoenix/execute/UnnestArrayPlan.java | 6 + .../execute/visitor/AvgRowWidthVisitor.java | 205 +++++++++ .../execute/visitor/ByteCountVisitor.java | 125 ++++++ .../execute/visitor/QueryPlanVisitor.java | 46 ++ .../execute/visitor/RowCountVisitor.java | 335 ++++++++++++++ .../apache/phoenix/jdbc/PhoenixStatement.java | 6 + .../java/org/apache/phoenix/util/CostUtil.java | 61 +-- .../query/ParallelIteratorsSplitTest.java | 6 + 26 files changed, 1612 insertions(+), 437 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/phoenix/blob/6b40a36b/phoenix-core/src/it/java/org/apache/phoenix/end2end/CostBasedDecisionIT.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/CostBasedDecisionIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/CostBasedDecisionIT.java index a3584ce..493855a 100644 --- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/CostBasedDecisionIT.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/CostBasedDecisionIT.java @@ -32,12 +32,16 @@ import org.apache.phoenix.query.QueryServices; import org.apache.phoenix.util.PropertiesUtil; import org.apache.phoenix.util.QueryUtil; import org.apache.phoenix.util.ReadOnlyProps; + import org.junit.BeforeClass; import org.junit.Test; import com.google.common.collect.Maps; public class CostBasedDecisionIT extends BaseUniqueNamesOwnClusterIT { + private final String testTable500; + private final String testTable990; + private final String testTable1000; @BeforeClass public static void doSetup() throws Exception { @@ -46,9 +50,16 @@ public class CostBasedDecisionIT extends BaseUniqueNamesOwnClusterIT { props.put(QueryServices.STATS_UPDATE_FREQ_MS_ATTRIB, Long.toString(5)); props.put(QueryServices.USE_STATS_FOR_PARALLELIZATION, Boolean.toString(true)); props.put(QueryServices.COST_BASED_OPTIMIZER_ENABLED, Boolean.toString(true)); + props.put(QueryServices.MAX_SERVER_CACHE_SIZE_ATTRIB, Long.toString(150000)); setUpTestDriver(new ReadOnlyProps(props.entrySet().iterator())); } + public CostBasedDecisionIT() throws Exception { + testTable500 = initTestTableValues(500); + testTable990 = initTestTableValues(990); + testTable1000 = initTestTableValues(1000); + } + @Test public void testCostOverridesStaticPlanOrdering1() throws Exception { Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES); @@ -64,10 +75,7 @@ public class CostBasedDecisionIT extends BaseUniqueNamesOwnClusterIT { String query = "SELECT rowkey, c1, c2 FROM " + tableName + " where c1 LIKE 'X0%' ORDER BY rowkey"; // Use the data table plan that opts out order-by when stats are not available. - ResultSet rs = conn.createStatement().executeQuery("explain " + query); - String plan = QueryUtil.getExplainPlan(rs); - assertTrue("Expected 'FULL SCAN' in the plan:\n" + plan + ".", - plan.contains("FULL SCAN")); + verifyQueryPlan(query, "FULL SCAN"); PreparedStatement stmt = conn.prepareStatement("UPSERT INTO " + tableName + " (rowkey, c1, c2) VALUES (?, ?, ?)"); for (int i = 0; i < 10000; i++) { @@ -81,10 +89,7 @@ public class CostBasedDecisionIT extends BaseUniqueNamesOwnClusterIT { conn.createStatement().execute("UPDATE STATISTICS " + tableName); // Use the index table plan that has a lower cost when stats become available. - rs = conn.createStatement().executeQuery("explain " + query); - plan = QueryUtil.getExplainPlan(rs); - assertTrue("Expected 'RANGE SCAN' in the plan:\n" + plan + ".", - plan.contains("RANGE SCAN")); + verifyQueryPlan(query, "RANGE SCAN"); } finally { conn.close(); } @@ -103,12 +108,12 @@ public class CostBasedDecisionIT extends BaseUniqueNamesOwnClusterIT { "c2 VARCHAR)"); conn.createStatement().execute("CREATE LOCAL INDEX " + tableName + "_idx ON " + tableName + " (c1)"); - String query = "SELECT rowkey, max(c1), max(c2) FROM " + tableName + " where c1 LIKE 'X%' GROUP BY rowkey"; + String query = "SELECT c1, max(rowkey), max(c2) FROM " + tableName + " where rowkey <= 'z' GROUP BY c1"; // Use the index table plan that opts out order-by when stats are not available. - ResultSet rs = conn.createStatement().executeQuery("explain " + query); - String plan = QueryUtil.getExplainPlan(rs); - assertTrue("Expected 'RANGE SCAN' in the plan:\n" + plan + ".", - plan.contains("RANGE SCAN")); + verifyQueryPlan(query, + "CLIENT PARALLEL 1-WAY RANGE SCAN OVER " + tableName + " [*] - ['z']\n" + + " SERVER AGGREGATE INTO DISTINCT ROWS BY [C1]\n" + + "CLIENT MERGE SORT"); PreparedStatement stmt = conn.prepareStatement("UPSERT INTO " + tableName + " (rowkey, c1, c2) VALUES (?, ?, ?)"); for (int i = 0; i < 10000; i++) { @@ -124,10 +129,11 @@ public class CostBasedDecisionIT extends BaseUniqueNamesOwnClusterIT { // Given that the range on C1 is meaningless and group-by becomes // order-preserving if using the data table, the data table plan should // come out as the best plan based on the costs. - rs = conn.createStatement().executeQuery("explain " + query); - plan = QueryUtil.getExplainPlan(rs); - assertTrue("Expected 'FULL SCAN' in the plan:\n" + plan + ".", - plan.contains("FULL SCAN")); + verifyQueryPlan(query, + "CLIENT PARALLEL 1-WAY RANGE SCAN OVER " + tableName + " [1]\n" + + " SERVER FILTER BY FIRST KEY ONLY AND \"ROWKEY\" <= 'z'\n" + + " SERVER AGGREGATE INTO ORDERED DISTINCT ROWS BY [\"C1\"]\n" + + "CLIENT MERGE SORT"); } finally { conn.close(); } @@ -150,14 +156,10 @@ public class CostBasedDecisionIT extends BaseUniqueNamesOwnClusterIT { String query = "SELECT * FROM " + tableName + " where c1 BETWEEN 10 AND 20 AND c2 < 9000 AND C3 < 5000"; // Use the idx2 plan with a wider PK slot span when stats are not available. - ResultSet rs = conn.createStatement().executeQuery("explain " + query); - String plan = QueryUtil.getExplainPlan(rs); - String indexPlan = + verifyQueryPlan(query, "CLIENT PARALLEL 1-WAY RANGE SCAN OVER " + tableName + " [2,*] - [2,9,000]\n" + " SERVER FILTER BY ((\"C1\" >= 10 AND \"C1\" <= 20) AND TO_INTEGER(\"C3\") < 5000)\n" + - "CLIENT MERGE SORT"; - assertTrue("Expected '" + indexPlan + "' in the plan:\n" + plan + ".", - plan.contains(indexPlan)); + "CLIENT MERGE SORT"); PreparedStatement stmt = conn.prepareStatement("UPSERT INTO " + tableName + " (rowkey, c1, c2, c3) VALUES (?, ?, ?, ?)"); for (int i = 0; i < 10000; i++) { @@ -171,14 +173,10 @@ public class CostBasedDecisionIT extends BaseUniqueNamesOwnClusterIT { conn.createStatement().execute("UPDATE STATISTICS " + tableName); // Use the idx2 plan that scans less data when stats become available. - rs = conn.createStatement().executeQuery("explain " + query); - plan = QueryUtil.getExplainPlan(rs); - String dataPlan = + verifyQueryPlan(query, "CLIENT PARALLEL 1-WAY RANGE SCAN OVER " + tableName + " [1,10] - [1,20]\n" + " SERVER FILTER BY (\"C2\" < 9000 AND \"C3\" < 5000)\n" + - "CLIENT MERGE SORT"; - assertTrue("Expected '" + dataPlan + "' in the plan:\n" + plan + ".", - plan.contains(dataPlan)); + "CLIENT MERGE SORT"); } finally { conn.close(); } @@ -201,15 +199,11 @@ public class CostBasedDecisionIT extends BaseUniqueNamesOwnClusterIT { String query = "UPSERT INTO " + tableName + " SELECT * FROM " + tableName + " where c1 BETWEEN 10 AND 20 AND c2 < 9000 AND C3 < 5000"; // Use the idx2 plan with a wider PK slot span when stats are not available. - ResultSet rs = conn.createStatement().executeQuery("explain " + query); - String plan = QueryUtil.getExplainPlan(rs); - String indexPlan = + verifyQueryPlan(query, "UPSERT SELECT\n" + "CLIENT PARALLEL 1-WAY RANGE SCAN OVER " + tableName + " [2,*] - [2,9,000]\n" + " SERVER FILTER BY ((\"C1\" >= 10 AND \"C1\" <= 20) AND TO_INTEGER(\"C3\") < 5000)\n" + - "CLIENT MERGE SORT"; - assertTrue("Expected '" + indexPlan + "' in the plan:\n" + plan + ".", - plan.contains(indexPlan)); + "CLIENT MERGE SORT"); PreparedStatement stmt = conn.prepareStatement("UPSERT INTO " + tableName + " (rowkey, c1, c2, c3) VALUES (?, ?, ?, ?)"); for (int i = 0; i < 10000; i++) { @@ -223,15 +217,11 @@ public class CostBasedDecisionIT extends BaseUniqueNamesOwnClusterIT { conn.createStatement().execute("UPDATE STATISTICS " + tableName); // Use the idx2 plan that scans less data when stats become available. - rs = conn.createStatement().executeQuery("explain " + query); - plan = QueryUtil.getExplainPlan(rs); - String dataPlan = + verifyQueryPlan(query, "UPSERT SELECT\n" + "CLIENT PARALLEL 1-WAY RANGE SCAN OVER " + tableName + " [1,10] - [1,20]\n" + " SERVER FILTER BY (\"C2\" < 9000 AND \"C3\" < 5000)\n" + - "CLIENT MERGE SORT"; - assertTrue("Expected '" + dataPlan + "' in the plan:\n" + plan + ".", - plan.contains(dataPlan)); + "CLIENT MERGE SORT"); } finally { conn.close(); } @@ -254,15 +244,11 @@ public class CostBasedDecisionIT extends BaseUniqueNamesOwnClusterIT { String query = "DELETE FROM " + tableName + " where c1 BETWEEN 10 AND 20 AND c2 < 9000 AND C3 < 5000"; // Use the idx2 plan with a wider PK slot span when stats are not available. - ResultSet rs = conn.createStatement().executeQuery("explain " + query); - String plan = QueryUtil.getExplainPlan(rs); - String indexPlan = + verifyQueryPlan(query, "DELETE ROWS\n" + "CLIENT PARALLEL 1-WAY RANGE SCAN OVER " + tableName + " [2,*] - [2,9,000]\n" + " SERVER FILTER BY ((\"C1\" >= 10 AND \"C1\" <= 20) AND TO_INTEGER(\"C3\") < 5000)\n" + - "CLIENT MERGE SORT"; - assertTrue("Expected '" + indexPlan + "' in the plan:\n" + plan + ".", - plan.contains(indexPlan)); + "CLIENT MERGE SORT"); PreparedStatement stmt = conn.prepareStatement("UPSERT INTO " + tableName + " (rowkey, c1, c2, c3) VALUES (?, ?, ?, ?)"); for (int i = 0; i < 10000; i++) { @@ -276,15 +262,11 @@ public class CostBasedDecisionIT extends BaseUniqueNamesOwnClusterIT { conn.createStatement().execute("UPDATE STATISTICS " + tableName); // Use the idx2 plan that scans less data when stats become available. - rs = conn.createStatement().executeQuery("explain " + query); - plan = QueryUtil.getExplainPlan(rs); - String dataPlan = + verifyQueryPlan(query, "DELETE ROWS\n" + "CLIENT PARALLEL 1-WAY RANGE SCAN OVER " + tableName + " [1,10] - [1,20]\n" + " SERVER FILTER BY (\"C2\" < 9000 AND \"C3\" < 5000)\n" + - "CLIENT MERGE SORT"; - assertTrue("Expected '" + dataPlan + "' in the plan:\n" + plan + ".", - plan.contains(dataPlan)); + "CLIENT MERGE SORT"); } finally { conn.close(); } @@ -303,22 +285,17 @@ public class CostBasedDecisionIT extends BaseUniqueNamesOwnClusterIT { "c2 VARCHAR)"); conn.createStatement().execute("CREATE LOCAL INDEX " + tableName + "_idx ON " + tableName + " (c1)"); - String query = "SELECT c1, max(rowkey), max(c2) FROM " + tableName + " where rowkey LIKE 'k%' GROUP BY c1 " - + "UNION ALL SELECT rowkey, max(c1), max(c2) FROM " + tableName + " where c1 LIKE 'X%' GROUP BY rowkey"; + String query = "SELECT c1, max(rowkey), max(c2) FROM " + tableName + " where rowkey <= 'z' GROUP BY c1 " + + "UNION ALL SELECT c1, max(rowkey), max(c2) FROM " + tableName + " where rowkey >= 'a' GROUP BY c1"; // Use the default plan when stats are not available. - ResultSet rs = conn.createStatement().executeQuery("explain " + query); - String plan = QueryUtil.getExplainPlan(rs); - String defaultPlan = + verifyQueryPlan(query, "UNION ALL OVER 2 QUERIES\n" + - " CLIENT PARALLEL 1-WAY RANGE SCAN OVER " + tableName + " ['k'] - ['l']\n" + + " CLIENT PARALLEL 1-WAY RANGE SCAN OVER " + tableName + " [*] - ['z']\n" + " SERVER AGGREGATE INTO DISTINCT ROWS BY [C1]\n" + " CLIENT MERGE SORT\n" + - " CLIENT PARALLEL 1-WAY RANGE SCAN OVER " + tableName + " [1,'X'] - [1,'Y']\n" + - " SERVER FILTER BY FIRST KEY ONLY\n" + - " SERVER AGGREGATE INTO DISTINCT ROWS BY [\"ROWKEY\"]\n" + - " CLIENT MERGE SORT"; - assertTrue("Expected '" + defaultPlan + "' in the plan:\n" + plan + ".", - plan.contains(defaultPlan)); + " CLIENT PARALLEL 1-WAY RANGE SCAN OVER " + tableName + " ['a'] - [*]\n" + + " SERVER AGGREGATE INTO DISTINCT ROWS BY [C1]\n" + + " CLIENT MERGE SORT"); PreparedStatement stmt = conn.prepareStatement("UPSERT INTO " + tableName + " (rowkey, c1, c2) VALUES (?, ?, ?)"); for (int i = 0; i < 10000; i++) { @@ -332,19 +309,16 @@ public class CostBasedDecisionIT extends BaseUniqueNamesOwnClusterIT { conn.createStatement().execute("UPDATE STATISTICS " + tableName); // Use the optimal plan based on cost when stats become available. - rs = conn.createStatement().executeQuery("explain " + query); - plan = QueryUtil.getExplainPlan(rs); - String optimizedPlan = + verifyQueryPlan(query, "UNION ALL OVER 2 QUERIES\n" + " CLIENT PARALLEL 1-WAY RANGE SCAN OVER " + tableName + " [1]\n" + - " SERVER FILTER BY FIRST KEY ONLY AND \"ROWKEY\" LIKE 'k%'\n" + + " SERVER FILTER BY FIRST KEY ONLY AND \"ROWKEY\" <= 'z'\n" + " SERVER AGGREGATE INTO ORDERED DISTINCT ROWS BY [\"C1\"]\n" + " CLIENT MERGE SORT\n" + - " CLIENT PARALLEL 1-WAY FULL SCAN OVER " + tableName + "\n" + - " SERVER FILTER BY C1 LIKE 'X%'\n" + - " SERVER AGGREGATE INTO ORDERED DISTINCT ROWS BY [ROWKEY]"; - assertTrue("Expected '" + optimizedPlan + "' in the plan:\n" + plan + ".", - plan.contains(optimizedPlan)); + " CLIENT PARALLEL 1-WAY RANGE SCAN OVER " + tableName + " [1]\n" + + " SERVER FILTER BY FIRST KEY ONLY AND \"ROWKEY\" >= 'a'\n" + + " SERVER AGGREGATE INTO ORDERED DISTINCT ROWS BY [\"C1\"]\n" + + " CLIENT MERGE SORT"); } finally { conn.close(); } @@ -363,23 +337,18 @@ public class CostBasedDecisionIT extends BaseUniqueNamesOwnClusterIT { "c2 VARCHAR)"); conn.createStatement().execute("CREATE LOCAL INDEX " + tableName + "_idx ON " + tableName + " (c1)"); - String query = "SELECT t1.rowkey, t1.c1, t1.c2, mc1, mc2 FROM " + tableName + " t1 " - + "JOIN (SELECT rowkey, max(c1) mc1, max(c2) mc2 FROM " + tableName + " where c1 LIKE 'X%' GROUP BY rowkey) t2 " - + "ON t1.rowkey = t2.rowkey WHERE t1.c1 LIKE 'X0%' ORDER BY t1.rowkey"; + String query = "SELECT t1.rowkey, t1.c1, t1.c2, t2.c1, mc2 FROM " + tableName + " t1 " + + "JOIN (SELECT c1, max(rowkey) mrk, max(c2) mc2 FROM " + tableName + " where rowkey <= 'z' GROUP BY c1) t2 " + + "ON t1.rowkey = t2.mrk WHERE t1.c1 LIKE 'X0%' ORDER BY t1.rowkey"; // Use the default plan when stats are not available. - ResultSet rs = conn.createStatement().executeQuery("explain " + query); - String plan = QueryUtil.getExplainPlan(rs); - String defaultPlan = + verifyQueryPlan(query, "CLIENT PARALLEL 1-WAY FULL SCAN OVER " + tableName + "\n" + " SERVER FILTER BY C1 LIKE 'X0%'\n" + " PARALLEL INNER-JOIN TABLE 0\n" + - " CLIENT PARALLEL 1-WAY RANGE SCAN OVER " + tableName + " [1,'X'] - [1,'Y']\n" + - " SERVER FILTER BY FIRST KEY ONLY\n" + - " SERVER AGGREGATE INTO DISTINCT ROWS BY [\"ROWKEY\"]\n" + + " CLIENT PARALLEL 1-WAY RANGE SCAN OVER " + tableName + " [*] - ['z']\n" + + " SERVER AGGREGATE INTO DISTINCT ROWS BY [C1]\n" + " CLIENT MERGE SORT\n" + - " DYNAMIC SERVER FILTER BY T1.ROWKEY IN (T2.ROWKEY)"; - assertTrue("Expected '" + defaultPlan + "' in the plan:\n" + plan + ".", - plan.contains(defaultPlan)); + " DYNAMIC SERVER FILTER BY T1.ROWKEY IN (T2.MRK)"); PreparedStatement stmt = conn.prepareStatement("UPSERT INTO " + tableName + " (rowkey, c1, c2) VALUES (?, ?, ?)"); for (int i = 0; i < 10000; i++) { @@ -393,20 +362,17 @@ public class CostBasedDecisionIT extends BaseUniqueNamesOwnClusterIT { conn.createStatement().execute("UPDATE STATISTICS " + tableName); // Use the optimal plan based on cost when stats become available. - rs = conn.createStatement().executeQuery("explain " + query); - plan = QueryUtil.getExplainPlan(rs); - String optimizedPlan = + verifyQueryPlan(query, "CLIENT PARALLEL 626-WAY RANGE SCAN OVER " + tableName + " [1,'X0'] - [1,'X1']\n" + " SERVER FILTER BY FIRST KEY ONLY\n" + " SERVER SORTED BY [\"T1.:ROWKEY\"]\n" + "CLIENT MERGE SORT\n" + " PARALLEL INNER-JOIN TABLE 0\n" + - " CLIENT PARALLEL 1-WAY FULL SCAN OVER " + tableName + "\n" + - " SERVER FILTER BY C1 LIKE 'X%'\n" + - " SERVER AGGREGATE INTO ORDERED DISTINCT ROWS BY [ROWKEY]\n" + - " DYNAMIC SERVER FILTER BY \"T1.:ROWKEY\" IN (T2.ROWKEY)"; - assertTrue("Expected '" + optimizedPlan + "' in the plan:\n" + plan + ".", - plan.contains(optimizedPlan)); + " CLIENT PARALLEL 1-WAY RANGE SCAN OVER " + tableName + " [1]\n" + + " SERVER FILTER BY FIRST KEY ONLY AND \"ROWKEY\" <= 'z'\n" + + " SERVER AGGREGATE INTO ORDERED DISTINCT ROWS BY [\"C1\"]\n" + + " CLIENT MERGE SORT\n" + + " DYNAMIC SERVER FILTER BY \"T1.:ROWKEY\" IN (T2.MRK)"); } finally { conn.close(); } @@ -432,10 +398,7 @@ public class CostBasedDecisionIT extends BaseUniqueNamesOwnClusterIT { String indexPlan = "SERVER FILTER BY FIRST KEY ONLY AND (\"ROWKEY\" >= 1 AND \"ROWKEY\" <= 10)"; // Use the index table plan that opts out order-by when stats are not available. - ResultSet rs = conn.createStatement().executeQuery("explain " + query); - String plan = QueryUtil.getExplainPlan(rs); - assertTrue("Expected '" + indexPlan + "' in the plan:\n" + plan + ".", - plan.contains(indexPlan)); + verifyQueryPlan(query, indexPlan); PreparedStatement stmt = conn.prepareStatement("UPSERT INTO " + tableName + " (rowkey, c1, c2) VALUES (?, ?, ?)"); for (int i = 0; i < 10000; i++) { @@ -449,18 +412,261 @@ public class CostBasedDecisionIT extends BaseUniqueNamesOwnClusterIT { conn.createStatement().execute("UPDATE STATISTICS " + tableName); // Use the data table plan that has a lower cost when stats are available. - rs = conn.createStatement().executeQuery("explain " + query); - plan = QueryUtil.getExplainPlan(rs); - assertTrue("Expected '" + dataPlan + "' in the plan:\n" + plan + ".", - plan.contains(dataPlan)); + verifyQueryPlan(query, dataPlan); // Use the index table plan as has been hinted. - rs = conn.createStatement().executeQuery("explain " + hintedQuery); - plan = QueryUtil.getExplainPlan(rs); - assertTrue("Expected '" + indexPlan + "' in the plan:\n" + plan + ".", - plan.contains(indexPlan)); + verifyQueryPlan(hintedQuery, indexPlan); } finally { conn.close(); } } + + /** Sort-merge-join w/ both children ordered wins over hash-join. */ + @Test + public void testJoinStrategy() throws Exception { + String q = "SELECT *\n" + + "FROM " + testTable500 + " t1 JOIN " + testTable1000 + " t2\n" + + "ON t1.ID = t2.ID"; + String expected = + "SORT-MERGE-JOIN (INNER) TABLES\n" + + " CLIENT PARALLEL 1-WAY FULL SCAN OVER " + testTable500 + "\n" + + "AND\n" + + " CLIENT PARALLEL 1-WAY FULL SCAN OVER " + testTable1000; + verifyQueryPlan(q, expected); + } + + /** Sort-merge-join w/ both children ordered wins over hash-join in an un-grouped aggregate query. */ + @Test + public void testJoinStrategy2() throws Exception { + String q = "SELECT count(*)\n" + + "FROM " + testTable500 + " t1 JOIN " + testTable1000 + " t2\n" + + "ON t1.ID = t2.ID\n" + + "WHERE t1.COL1 < 200"; + String expected = + "SORT-MERGE-JOIN (INNER) TABLES\n" + + " CLIENT PARALLEL 1-WAY FULL SCAN OVER " + testTable500 + "\n" + + " SERVER FILTER BY COL1 < 200\n" + + "AND (SKIP MERGE)\n" + + " CLIENT PARALLEL 1-WAY FULL SCAN OVER " + testTable1000 + "\n" + + " SERVER FILTER BY FIRST KEY ONLY\n" + + "CLIENT AGGREGATE INTO SINGLE ROW"; + verifyQueryPlan(q, expected); + } + + /** Hash-join w/ PK/FK optimization wins over sort-merge-join w/ larger side ordered. */ + @Test + public void testJoinStrategy3() throws Exception { + String q = "SELECT *\n" + + "FROM " + testTable500 + " t1 JOIN " + testTable1000 + " t2\n" + + "ON t1.COL1 = t2.ID\n" + + "WHERE t1.ID > 200"; + String expected = + "CLIENT PARALLEL 1-WAY FULL SCAN OVER " + testTable1000 + "\n" + + " PARALLEL INNER-JOIN TABLE 0\n" + + " CLIENT PARALLEL 1-WAY RANGE SCAN OVER " + testTable500 + " [201] - [*]\n" + + " DYNAMIC SERVER FILTER BY T2.ID IN (T1.COL1)"; + verifyQueryPlan(q, expected); + } + + /** Hash-join w/ PK/FK optimization wins over hash-join w/o PK/FK optimization when two sides are close in size. */ + @Test + public void testJoinStrategy4() throws Exception { + String q = "SELECT *\n" + + "FROM " + testTable990 + " t1 JOIN " + testTable1000 + " t2\n" + + "ON t1.ID = t2.COL1"; + String expected = + "CLIENT PARALLEL 1-WAY FULL SCAN OVER " + testTable990 + "\n" + + " PARALLEL INNER-JOIN TABLE 0\n" + + " CLIENT PARALLEL 1-WAY FULL SCAN OVER " + testTable1000 + "\n" + + " DYNAMIC SERVER FILTER BY T1.ID IN (T2.COL1)"; + verifyQueryPlan(q, expected); + } + + /** Hash-join wins over sort-merge-join w/ smaller side ordered. */ + @Test + public void testJoinStrategy5() throws Exception { + String q = "SELECT *\n" + + "FROM " + testTable500 + " t1 JOIN " + testTable1000 + " t2\n" + + "ON t1.ID = t2.COL1\n" + + "WHERE t1.ID > 200"; + String expected = + "CLIENT PARALLEL 1-WAY FULL SCAN OVER " + testTable1000 + "\n" + + " PARALLEL INNER-JOIN TABLE 0\n" + + " CLIENT PARALLEL 1-WAY RANGE SCAN OVER " + testTable500 + " [201] - [*]"; + verifyQueryPlan(q, expected); + } + + /** Hash-join wins over sort-merge-join w/o any side ordered. */ + @Test + public void testJoinStrategy6() throws Exception { + String q = "SELECT *\n" + + "FROM " + testTable500 + " t1 JOIN " + testTable1000 + " t2\n" + + "ON t1.COL1 = t2.COL1\n" + + "WHERE t1.ID > 200"; + String expected = + "CLIENT PARALLEL 1-WAY FULL SCAN OVER " + testTable1000 + "\n" + + " PARALLEL INNER-JOIN TABLE 0\n" + + " CLIENT PARALLEL 1-WAY RANGE SCAN OVER " + testTable500 + " [201] - [*]"; + verifyQueryPlan(q, expected); + } + + /** + * Hash-join wins over sort-merge-join w/ both sides ordered in an order-by query. + * This is because order-by can only be done on the client side after sort-merge-join + * and order-by w/o limit on the client side is very expensive. + */ + @Test + public void testJoinStrategy7() throws Exception { + String q = "SELECT *\n" + + "FROM " + testTable500 + " t1 JOIN " + testTable1000 + " t2\n" + + "ON t1.ID = t2.ID\n" + + "ORDER BY t1.COL1"; + String expected = + "CLIENT PARALLEL 1001-WAY FULL SCAN OVER " + testTable1000 + "\n" + + " SERVER SORTED BY [T1.COL1]\n" + + "CLIENT MERGE SORT\n" + + " PARALLEL INNER-JOIN TABLE 0\n" + + " CLIENT PARALLEL 1-WAY FULL SCAN OVER " + testTable500 + "\n" + + " DYNAMIC SERVER FILTER BY T2.ID IN (T1.ID)"; + verifyQueryPlan(q, expected); + } + + /** + * Sort-merge-join w/ both sides ordered wins over hash-join in an order-by limit query. + * This is because order-by can only be done on the client side after sort-merge-join + * but order-by w/ limit on the client side is less expensive. + */ + @Test + public void testJoinStrategy8() throws Exception { + String q = "SELECT *\n" + + "FROM " + testTable500 + " t1 JOIN " + testTable1000 + " t2\n" + + "ON t1.ID = t2.ID\n" + + "ORDER BY t1.COL1 LIMIT 5"; + String expected = + "SORT-MERGE-JOIN (INNER) TABLES\n" + + " CLIENT PARALLEL 1-WAY FULL SCAN OVER " + testTable500 + "\n" + + "AND\n" + + " CLIENT PARALLEL 1-WAY FULL SCAN OVER " + testTable1000 + "\n" + + "CLIENT TOP 5 ROWS SORTED BY [T1.COL1]"; + verifyQueryPlan(q, expected); + } + + /** + * Multi-table join: sort-merge-join chosen since all join keys are PK. + */ + @Test + public void testJoinStrategy9() throws Exception { + String q = "SELECT *\n" + + "FROM " + testTable1000 + " t1 LEFT JOIN " + testTable500 + " t2\n" + + "ON t1.ID = t2.ID AND t2.ID > 200\n" + + "LEFT JOIN " + testTable990 + " t3\n" + + "ON t1.ID = t3.ID AND t3.ID < 100"; + String expected = + "SORT-MERGE-JOIN (LEFT) TABLES\n" + + " SORT-MERGE-JOIN (LEFT) TABLES\n" + + " CLIENT PARALLEL 1-WAY FULL SCAN OVER " + testTable1000 + "\n" + + " AND\n" + + " CLIENT PARALLEL 1-WAY RANGE SCAN OVER " + testTable500 + " [201] - [*]\n" + + "AND\n" + + " CLIENT PARALLEL 1-WAY RANGE SCAN OVER " + testTable990 + " [*] - [100]"; + verifyQueryPlan(q, expected); + } + + /** + * Multi-table join: a mix of join strategies chosen based on cost. + */ + @Test + public void testJoinStrategy10() throws Exception { + String q = "SELECT *\n" + + "FROM " + testTable1000 + " t1 JOIN " + testTable500 + " t2\n" + + "ON t1.ID = t2.COL1 AND t2.ID > 200\n" + + "JOIN " + testTable990 + " t3\n" + + "ON t1.ID = t3.ID AND t3.ID < 100"; + String expected = + "SORT-MERGE-JOIN (INNER) TABLES\n" + + " CLIENT PARALLEL 1-WAY FULL SCAN OVER " + testTable1000 + "\n" + + " PARALLEL INNER-JOIN TABLE 0\n" + + " CLIENT PARALLEL 1-WAY RANGE SCAN OVER " + testTable500 + " [201] - [*]\n" + + " DYNAMIC SERVER FILTER BY T1.ID IN (T2.COL1)\n" + + "AND\n" + + " CLIENT PARALLEL 1-WAY RANGE SCAN OVER " + testTable990 + " [*] - [100]"; + verifyQueryPlan(q, expected); + } + + /** + * Multi-table join: hash-join two tables in parallel since two RHS tables are both small + * and can fit in memory at the same time. + */ + @Test + public void testJoinStrategy11() throws Exception { + String q = "SELECT *\n" + + "FROM " + testTable1000 + " t1 JOIN " + testTable500 + " t2\n" + + "ON t1.COL2 = t2.COL1 AND t2.ID > 200\n" + + "JOIN " + testTable990 + " t3\n" + + "ON t1.COL1 = t3.COL2 AND t3.ID < 100"; + String expected = + "CLIENT PARALLEL 1-WAY FULL SCAN OVER " + testTable1000 + "\n" + + " PARALLEL INNER-JOIN TABLE 0\n" + + " CLIENT PARALLEL 1-WAY RANGE SCAN OVER " + testTable500 + " [201] - [*]\n" + + " PARALLEL INNER-JOIN TABLE 1\n" + + " CLIENT PARALLEL 1-WAY RANGE SCAN OVER " + testTable990 + " [*] - [100]"; + verifyQueryPlan(q, expected); + } + + /** + * Multi-table join: similar to {@link this#testJoinStrategy11()}, but the two RHS + * tables cannot fit in memory at the same time, and thus a mix of join strategies + * is chosen based on cost. + */ + @Test + public void testJoinStrategy12() throws Exception { + String q = "SELECT *\n" + + "FROM " + testTable1000 + " t1 JOIN " + testTable990 + " t2\n" + + "ON t1.COL2 = t2.COL1\n" + + "JOIN " + testTable990 + " t3\n" + + "ON t1.COL1 = t3.COL2"; + String expected = + "SORT-MERGE-JOIN (INNER) TABLES\n" + + " CLIENT PARALLEL 1001-WAY FULL SCAN OVER " + testTable1000 + "\n" + + " SERVER SORTED BY [T1.COL1]\n" + + " CLIENT MERGE SORT\n" + + " PARALLEL INNER-JOIN TABLE 0\n" + + " CLIENT PARALLEL 1-WAY FULL SCAN OVER " + testTable990 + "\n" + + "AND\n" + + " CLIENT PARALLEL 991-WAY FULL SCAN OVER " + testTable990 + "\n" + + " SERVER SORTED BY [T3.COL2]\n" + + " CLIENT MERGE SORT"; + verifyQueryPlan(q, expected); + } + + private static void verifyQueryPlan(String query, String expected) throws Exception { + Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES); + Connection conn = DriverManager.getConnection(getUrl(), props); + ResultSet rs = conn.createStatement().executeQuery("explain " + query); + String plan = QueryUtil.getExplainPlan(rs); + assertTrue("Expected '" + expected + "' in the plan:\n" + plan + ".", + plan.contains(expected)); + } + + private static String initTestTableValues(int rows) throws Exception { + Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES); + try (Connection conn = DriverManager.getConnection(getUrl(), props)) { + String tableName = generateUniqueName(); + conn.createStatement().execute("CREATE TABLE " + tableName + " (\n" + + "ID INTEGER NOT NULL PRIMARY KEY,\n" + + "COL1 INTEGER," + + "COL2 INTEGER)"); + PreparedStatement stmt = conn.prepareStatement( + "UPSERT INTO " + tableName + " VALUES(?, ?, ?)"); + for (int i = 0; i < rows; i++) { + stmt.setInt(1, i + 1); + stmt.setInt(2, rows - i); + stmt.setInt(3, rows + i); + stmt.execute(); + } + conn.commit(); + conn.createStatement().execute("UPDATE STATISTICS " + tableName); + return tableName; + } + } } http://git-wip-us.apache.org/repos/asf/phoenix/blob/6b40a36b/phoenix-core/src/main/java/org/apache/phoenix/compile/JoinCompiler.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/JoinCompiler.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/JoinCompiler.java index f3c4c24..f5a7e39 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/compile/JoinCompiler.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/JoinCompiler.java @@ -71,6 +71,8 @@ import org.apache.phoenix.parse.TableNodeVisitor; import org.apache.phoenix.parse.TableWildcardParseNode; import org.apache.phoenix.parse.UDFParseNode; import org.apache.phoenix.parse.WildcardParseNode; +import org.apache.phoenix.query.QueryServices; +import org.apache.phoenix.query.QueryServicesOptions; import org.apache.phoenix.schema.ColumnNotFoundException; import org.apache.phoenix.schema.ColumnRef; import org.apache.phoenix.schema.LocalIndexDataColumnRef; @@ -124,6 +126,8 @@ public class JoinCompiler { private final ColumnResolver origResolver; private final boolean useStarJoin; private final Map<ColumnRef, ColumnRefType> columnRefs; + private final boolean useSortMergeJoin; + private final boolean costBased; private JoinCompiler(PhoenixStatement statement, SelectStatement select, ColumnResolver resolver) { @@ -132,6 +136,9 @@ public class JoinCompiler { this.origResolver = resolver; this.useStarJoin = !select.getHint().hasHint(Hint.NO_STAR_JOIN); this.columnRefs = new HashMap<ColumnRef, ColumnRefType>(); + this.useSortMergeJoin = select.getHint().hasHint(Hint.USE_SORT_MERGE_JOIN); + this.costBased = statement.getConnection().getQueryServices().getProps().getBoolean( + QueryServices.COST_BASED_OPTIMIZER_ENABLED, QueryServicesOptions.DEFAULT_COST_BASED_OPTIMIZER_ENABLED); } public static JoinTable compile(PhoenixStatement statement, SelectStatement select, ColumnResolver resolver) throws SQLException { @@ -365,6 +372,42 @@ public class JoinCompiler { } /** + * Return a list of all applicable join strategies. The order of the strategies in the + * returned list is based on the static rule below. However, the caller can decide on + * an optimal join strategy by evaluating and comparing the costs. + * 1. If hint USE_SORT_MERGE_JOIN is specified, + * return a singleton list containing only SORT_MERGE. + * 2. If 1) matches pattern "A LEFT/INNER/SEMI/ANTI JOIN B"; or + * 2) matches pattern "A LEFT/INNER/SEMI/ANTI JOIN B (LEFT/INNER/SEMI/ANTI JOIN C)+" + * and hint NO_STAR_JOIN is not specified, + * add BUILD_RIGHT to the returned list. + * 3. If matches pattern "A RIGHT/INNER JOIN B", where B is either a named table reference + * or a flat sub-query, + * add BUILD_LEFT to the returned list. + * 4. add SORT_MERGE to the returned list. + */ + public List<Strategy> getApplicableJoinStrategies() { + List<Strategy> strategies = Lists.newArrayList(); + if (useSortMergeJoin) { + strategies.add(Strategy.SORT_MERGE); + } else { + if (getStarJoinVector() != null) { + strategies.add(Strategy.HASH_BUILD_RIGHT); + } + JoinSpec lastJoinSpec = joinSpecs.get(joinSpecs.size() - 1); + JoinType type = lastJoinSpec.getType(); + if ((type == JoinType.Right || type == JoinType.Inner) + && lastJoinSpec.getJoinTable().getJoinSpecs().isEmpty() + && lastJoinSpec.getJoinTable().getTable().isFlat()) { + strategies.add(Strategy.HASH_BUILD_LEFT); + } + strategies.add(Strategy.SORT_MERGE); + } + + return strategies; + } + + /** * Returns a boolean vector indicating whether the evaluation of join expressions * can be evaluated at an early stage if the input JoinSpec can be taken as a * star join. Otherwise returns null. http://git-wip-us.apache.org/repos/asf/phoenix/blob/6b40a36b/phoenix-core/src/main/java/org/apache/phoenix/compile/ListJarsQueryPlan.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/ListJarsQueryPlan.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/ListJarsQueryPlan.java index 0688b94..e3ed110 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/compile/ListJarsQueryPlan.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/ListJarsQueryPlan.java @@ -39,6 +39,7 @@ import org.apache.hadoop.hbase.client.Scan; import org.apache.hadoop.hbase.io.ImmutableBytesWritable; import org.apache.phoenix.compile.GroupByCompiler.GroupBy; import org.apache.phoenix.compile.OrderByCompiler.OrderBy; +import org.apache.phoenix.execute.visitor.QueryPlanVisitor; import org.apache.phoenix.expression.Determinism; import org.apache.phoenix.expression.Expression; import org.apache.phoenix.expression.LiteralExpression; @@ -251,7 +252,12 @@ public class ListJarsQueryPlan implements QueryPlan { return false; } - @Override + @Override + public <T> T accept(QueryPlanVisitor<T> visitor) { + return visitor.visit(this); + } + + @Override public Set<TableRef> getSourceRefs() { return Collections.<TableRef>emptySet(); } http://git-wip-us.apache.org/repos/asf/phoenix/blob/6b40a36b/phoenix-core/src/main/java/org/apache/phoenix/compile/QueryCompiler.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/QueryCompiler.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/QueryCompiler.java index 3b14850..243f03e 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/compile/QueryCompiler.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/QueryCompiler.java @@ -24,6 +24,7 @@ import java.util.Collections; import java.util.List; import java.util.Set; +import org.apache.hadoop.hbase.client.Query; import org.apache.hadoop.hbase.client.Scan; import org.apache.hadoop.hbase.util.Pair; import org.apache.phoenix.compile.GroupByCompiler.GroupBy; @@ -52,6 +53,7 @@ import org.apache.phoenix.jdbc.PhoenixConnection; import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData; import org.apache.phoenix.jdbc.PhoenixStatement; import org.apache.phoenix.join.HashJoinInfo; +import org.apache.phoenix.optimize.Cost; import org.apache.phoenix.parse.AliasedNode; import org.apache.phoenix.parse.EqualParseNode; import org.apache.phoenix.parse.HintNode.Hint; @@ -63,7 +65,10 @@ import org.apache.phoenix.parse.SQLParser; import org.apache.phoenix.parse.SelectStatement; import org.apache.phoenix.parse.SubqueryParseNode; import org.apache.phoenix.parse.TableNode; +import org.apache.phoenix.query.ConnectionQueryServices; import org.apache.phoenix.query.QueryConstants; +import org.apache.phoenix.query.QueryServices; +import org.apache.phoenix.query.QueryServicesOptions; import org.apache.phoenix.schema.AmbiguousColumnException; import org.apache.phoenix.schema.ColumnNotFoundException; import org.apache.phoenix.schema.PDatum; @@ -102,9 +107,9 @@ public class QueryCompiler { private final ParallelIteratorFactory parallelIteratorFactory; private final SequenceManager sequenceManager; private final boolean projectTuples; - private final boolean useSortMergeJoin; private final boolean noChildParentJoinOptimization; private final QueryPlan dataPlan; + private final boolean costBased; public QueryCompiler(PhoenixStatement statement, SelectStatement select, ColumnResolver resolver, boolean projectTuples, QueryPlan dataPlan) throws SQLException { this(statement, select, resolver, Collections.<PDatum>emptyList(), null, new SequenceManager(statement), projectTuples, dataPlan); @@ -119,9 +124,10 @@ public class QueryCompiler { this.parallelIteratorFactory = parallelIteratorFactory; this.sequenceManager = sequenceManager; this.projectTuples = projectTuples; - this.useSortMergeJoin = select.getHint().hasHint(Hint.USE_SORT_MERGE_JOIN); this.noChildParentJoinOptimization = select.getHint().hasHint(Hint.NO_CHILD_PARENT_JOIN_OPTIMIZATION); - if (statement.getConnection().getQueryServices().getLowestClusterHBaseVersion() >= PhoenixDatabaseMetaData.ESSENTIAL_FAMILY_VERSION_THRESHOLD) { + ConnectionQueryServices services = statement.getConnection().getQueryServices(); + this.costBased = services.getProps().getBoolean(QueryServices.COST_BASED_OPTIMIZER_ENABLED, QueryServicesOptions.DEFAULT_COST_BASED_OPTIMIZER_ENABLED); + if (services.getLowestClusterHBaseVersion() >= PhoenixDatabaseMetaData.ESSENTIAL_FAMILY_VERSION_THRESHOLD) { this.scan.setAttribute(LOAD_COLUMN_FAMILIES_ON_DEMAND_ATTR, QueryConstants.TRUE); } if (select.getHint().hasHint(Hint.NO_CACHE)) { @@ -201,41 +207,17 @@ public class QueryCompiler { } } - /* + /** * Call compileJoinQuery() for join queries recursively down to the leaf JoinTable nodes. - * This matches the input JoinTable node against patterns in the following order: - * 1. A (leaf JoinTable node, which can be a named table reference or a subquery of any kind.) - * Returns the compilation result of a single table scan or of an independent subquery. - * 2. Matching either of (when hint USE_SORT_MERGE_JOIN not specified): - * 1) A LEFT/INNER JOIN B - * 2) A LEFT/INNER JOIN B (LEFT/INNER JOIN C)+, if hint NO_STAR_JOIN not specified - * where A can be a named table reference or a flat subquery, and B, C, ... can be a named - * table reference, a sub-join or a subquery of any kind. - * Returns a HashJoinPlan{scan: A, hash: B, C, ...}. - * 3. Matching pattern: - * A RIGHT/INNER JOIN B (when hint USE_SORT_MERGE_JOIN not specified) - * where B can be a named table reference or a flat subquery, and A can be a named table - * reference, a sub-join or a subquery of any kind. - * Returns a HashJoinPlan{scan: B, hash: A}. - * NOTE that "A LEFT/RIGHT/INNER/FULL JOIN B RIGHT/INNER JOIN C" is viewed as - * "(A LEFT/RIGHT/INNER/FULL JOIN B) RIGHT/INNER JOIN C" here, which means the left part in the - * parenthesis is considered a sub-join. - * viewed as a sub-join. - * 4. All the rest that do not qualify for previous patterns or conditions, including FULL joins. - * Returns a SortMergeJoinPlan, the sorting part of which is pushed down to the JoinTable nodes - * of both sides as order-by clauses. - * NOTE that SEMI or ANTI joins are treated the same way as LEFT joins in JoinTable pattern matching. - * - * If no join algorithm hint is provided, according to the above compilation process, a join query - * plan can probably consist of both HashJoinPlan and SortMergeJoinPlan which may enclose each other. - * TODO 1) Use table statistics to guide the choice of join plans. - * 2) Make it possible to hint a certain join algorithm for a specific join step. + * If it is a leaf node, call compileSingleFlatQuery() or compileSubquery(), otherwise: + * 1) If option COST_BASED_OPTIMIZER_ENABLED is on and stats are available, return the + * join plan with the best cost. Note that the "best" plan is only locally optimal, + * and might or might not be globally optimal. + * 2) Otherwise, return the join plan compiled with the default strategy. + * @see JoinCompiler.JoinTable#getApplicableJoinStrategies() */ - @SuppressWarnings("unchecked") protected QueryPlan compileJoinQuery(StatementContext context, List<Object> binds, JoinTable joinTable, boolean asSubquery, boolean projectPKColumns, List<OrderByNode> orderBy) throws SQLException { - byte[] emptyByteArray = new byte[0]; - List<JoinSpec> joinSpecs = joinTable.getJoinSpecs(); - if (joinSpecs.isEmpty()) { + if (joinTable.getJoinSpecs().isEmpty()) { Table table = joinTable.getTable(); SelectStatement subquery = table.getAsSubquery(orderBy); if (!table.isSubselect()) { @@ -253,198 +235,229 @@ public class QueryCompiler { return new TupleProjectionPlan(plan, new TupleProjector(plan.getProjector()), table.compilePostFilterExpression(context)); } - boolean[] starJoinVector; - if (!this.useSortMergeJoin && (starJoinVector = joinTable.getStarJoinVector()) != null) { - Table table = joinTable.getTable(); - PTable initialProjectedTable; - TableRef tableRef; - SelectStatement query; - TupleProjector tupleProjector; - if (!table.isSubselect()) { - context.setCurrentTable(table.getTableRef()); - initialProjectedTable = table.createProjectedTable(!projectPKColumns, context); - tableRef = table.getTableRef(); - table.projectColumns(context.getScan()); - query = joinTable.getAsSingleSubquery(table.getAsSubquery(orderBy), asSubquery); - tupleProjector = new TupleProjector(initialProjectedTable); - } else { - SelectStatement subquery = table.getAsSubquery(orderBy); - QueryPlan plan = compileSubquery(subquery, false); - initialProjectedTable = table.createProjectedTable(plan.getProjector()); - tableRef = plan.getTableRef(); - context.getScan().setFamilyMap(plan.getContext().getScan().getFamilyMap()); - query = joinTable.getAsSingleSubquery((SelectStatement) plan.getStatement(), asSubquery); - tupleProjector = new TupleProjector(plan.getProjector()); + List<JoinCompiler.Strategy> strategies = joinTable.getApplicableJoinStrategies(); + assert strategies.size() > 0; + if (!costBased || strategies.size() == 1) { + return compileJoinQuery( + strategies.get(0), context, binds, joinTable, asSubquery, projectPKColumns, orderBy); + } + + QueryPlan bestPlan = null; + Cost bestCost = null; + for (JoinCompiler.Strategy strategy : strategies) { + StatementContext newContext = new StatementContext( + context.getStatement(), context.getResolver(), new Scan(), context.getSequenceManager()); + QueryPlan plan = compileJoinQuery( + strategy, newContext, binds, joinTable, asSubquery, projectPKColumns, orderBy); + Cost cost = plan.getCost(); + if (bestPlan == null || cost.compareTo(bestCost) < 0) { + bestPlan = plan; + bestCost = cost; } - context.setCurrentTable(tableRef); - PTable projectedTable = initialProjectedTable; - int count = joinSpecs.size(); - ImmutableBytesPtr[] joinIds = new ImmutableBytesPtr[count]; - List<Expression>[] joinExpressions = new List[count]; - JoinType[] joinTypes = new JoinType[count]; - PTable[] tables = new PTable[count]; - int[] fieldPositions = new int[count]; - StatementContext[] subContexts = new StatementContext[count]; - QueryPlan[] subPlans = new QueryPlan[count]; - HashSubPlan[] hashPlans = new HashSubPlan[count]; - fieldPositions[0] = projectedTable.getColumns().size() - projectedTable.getPKColumns().size(); - for (int i = 0; i < count; i++) { - JoinSpec joinSpec = joinSpecs.get(i); - Scan subScan = ScanUtil.newScan(originalScan); - subContexts[i] = new StatementContext(statement, context.getResolver(), subScan, new SequenceManager(statement)); - subPlans[i] = compileJoinQuery(subContexts[i], binds, joinSpec.getJoinTable(), true, true, null); - boolean hasPostReference = joinSpec.getJoinTable().hasPostReference(); - if (hasPostReference) { - tables[i] = subContexts[i].getResolver().getTables().get(0).getTable(); - projectedTable = JoinCompiler.joinProjectedTables(projectedTable, tables[i], joinSpec.getType()); + } + context.setResolver(bestPlan.getContext().getResolver()); + context.setCurrentTable(bestPlan.getContext().getCurrentTable()); + return bestPlan; + } + + protected QueryPlan compileJoinQuery(JoinCompiler.Strategy strategy, StatementContext context, List<Object> binds, JoinTable joinTable, boolean asSubquery, boolean projectPKColumns, List<OrderByNode> orderBy) throws SQLException { + byte[] emptyByteArray = new byte[0]; + List<JoinSpec> joinSpecs = joinTable.getJoinSpecs(); + switch (strategy) { + case HASH_BUILD_RIGHT: { + boolean[] starJoinVector = joinTable.getStarJoinVector(); + Table table = joinTable.getTable(); + PTable initialProjectedTable; + TableRef tableRef; + SelectStatement query; + TupleProjector tupleProjector; + if (!table.isSubselect()) { + context.setCurrentTable(table.getTableRef()); + initialProjectedTable = table.createProjectedTable(!projectPKColumns, context); + tableRef = table.getTableRef(); + table.projectColumns(context.getScan()); + query = joinTable.getAsSingleSubquery(table.getAsSubquery(orderBy), asSubquery); + tupleProjector = new TupleProjector(initialProjectedTable); } else { - tables[i] = null; + SelectStatement subquery = table.getAsSubquery(orderBy); + QueryPlan plan = compileSubquery(subquery, false); + initialProjectedTable = table.createProjectedTable(plan.getProjector()); + tableRef = plan.getTableRef(); + context.getScan().setFamilyMap(plan.getContext().getScan().getFamilyMap()); + query = joinTable.getAsSingleSubquery((SelectStatement) plan.getStatement(), asSubquery); + tupleProjector = new TupleProjector(plan.getProjector()); } - } - for (int i = 0; i < count; i++) { - JoinSpec joinSpec = joinSpecs.get(i); - context.setResolver(FromCompiler.getResolverForProjectedTable(projectedTable, context.getConnection(), query.getUdfParseNodes())); - joinIds[i] = new ImmutableBytesPtr(emptyByteArray); // place-holder - Pair<List<Expression>, List<Expression>> joinConditions = joinSpec.compileJoinConditions(context, subContexts[i], JoinCompiler.Strategy.HASH_BUILD_RIGHT); - joinExpressions[i] = joinConditions.getFirst(); - List<Expression> hashExpressions = joinConditions.getSecond(); - Pair<Expression, Expression> keyRangeExpressions = new Pair<Expression, Expression>(null, null); - boolean optimized = getKeyExpressionCombinations(keyRangeExpressions, context, joinTable.getStatement(), tableRef, joinSpec.getType(), joinExpressions[i], hashExpressions); - Expression keyRangeLhsExpression = keyRangeExpressions.getFirst(); - Expression keyRangeRhsExpression = keyRangeExpressions.getSecond(); - joinTypes[i] = joinSpec.getType(); - if (i < count - 1) { - fieldPositions[i + 1] = fieldPositions[i] + (tables[i] == null ? 0 : (tables[i].getColumns().size() - tables[i].getPKColumns().size())); + context.setCurrentTable(tableRef); + PTable projectedTable = initialProjectedTable; + int count = joinSpecs.size(); + ImmutableBytesPtr[] joinIds = new ImmutableBytesPtr[count]; + List<Expression>[] joinExpressions = new List[count]; + JoinType[] joinTypes = new JoinType[count]; + PTable[] tables = new PTable[count]; + int[] fieldPositions = new int[count]; + StatementContext[] subContexts = new StatementContext[count]; + QueryPlan[] subPlans = new QueryPlan[count]; + HashSubPlan[] hashPlans = new HashSubPlan[count]; + fieldPositions[0] = projectedTable.getColumns().size() - projectedTable.getPKColumns().size(); + for (int i = 0; i < count; i++) { + JoinSpec joinSpec = joinSpecs.get(i); + Scan subScan = ScanUtil.newScan(originalScan); + subContexts[i] = new StatementContext(statement, context.getResolver(), subScan, new SequenceManager(statement)); + subPlans[i] = compileJoinQuery(subContexts[i], binds, joinSpec.getJoinTable(), true, true, null); + boolean hasPostReference = joinSpec.getJoinTable().hasPostReference(); + if (hasPostReference) { + tables[i] = subContexts[i].getResolver().getTables().get(0).getTable(); + projectedTable = JoinCompiler.joinProjectedTables(projectedTable, tables[i], joinSpec.getType()); + } else { + tables[i] = null; + } + } + for (int i = 0; i < count; i++) { + JoinSpec joinSpec = joinSpecs.get(i); + context.setResolver(FromCompiler.getResolverForProjectedTable(projectedTable, context.getConnection(), query.getUdfParseNodes())); + joinIds[i] = new ImmutableBytesPtr(emptyByteArray); // place-holder + Pair<List<Expression>, List<Expression>> joinConditions = joinSpec.compileJoinConditions(context, subContexts[i], strategy); + joinExpressions[i] = joinConditions.getFirst(); + List<Expression> hashExpressions = joinConditions.getSecond(); + Pair<Expression, Expression> keyRangeExpressions = new Pair<Expression, Expression>(null, null); + boolean optimized = getKeyExpressionCombinations(keyRangeExpressions, context, joinTable.getStatement(), tableRef, joinSpec.getType(), joinExpressions[i], hashExpressions); + Expression keyRangeLhsExpression = keyRangeExpressions.getFirst(); + Expression keyRangeRhsExpression = keyRangeExpressions.getSecond(); + joinTypes[i] = joinSpec.getType(); + if (i < count - 1) { + fieldPositions[i + 1] = fieldPositions[i] + (tables[i] == null ? 0 : (tables[i].getColumns().size() - tables[i].getPKColumns().size())); + } + hashPlans[i] = new HashSubPlan(i, subPlans[i], optimized ? null : hashExpressions, joinSpec.isSingleValueOnly(), keyRangeLhsExpression, keyRangeRhsExpression); + } + TupleProjector.serializeProjectorIntoScan(context.getScan(), tupleProjector); + QueryPlan plan = compileSingleFlatQuery(context, query, binds, asSubquery, !asSubquery && joinTable.isAllLeftJoin(), null, !table.isSubselect() && projectPKColumns ? tupleProjector : null, true); + Expression postJoinFilterExpression = joinTable.compilePostFilterExpression(context, table); + Integer limit = null; + Integer offset = null; + if (!query.isAggregate() && !query.isDistinct() && query.getOrderBy().isEmpty()) { + limit = plan.getLimit(); + offset = plan.getOffset(); } - hashPlans[i] = new HashSubPlan(i, subPlans[i], optimized ? null : hashExpressions, joinSpec.isSingleValueOnly(), keyRangeLhsExpression, keyRangeRhsExpression); + HashJoinInfo joinInfo = new HashJoinInfo(projectedTable, joinIds, joinExpressions, joinTypes, + starJoinVector, tables, fieldPositions, postJoinFilterExpression, QueryUtil.getOffsetLimit(limit, offset)); + return HashJoinPlan.create(joinTable.getStatement(), plan, joinInfo, hashPlans); } - TupleProjector.serializeProjectorIntoScan(context.getScan(), tupleProjector); - QueryPlan plan = compileSingleFlatQuery(context, query, binds, asSubquery, !asSubquery && joinTable.isAllLeftJoin(), null, !table.isSubselect() && projectPKColumns ? tupleProjector : null, true); - Expression postJoinFilterExpression = joinTable.compilePostFilterExpression(context, table); - Integer limit = null; - Integer offset = null; - if (!query.isAggregate() && !query.isDistinct() && query.getOrderBy().isEmpty()) { - limit = plan.getLimit(); - offset = plan.getOffset(); + case HASH_BUILD_LEFT: { + JoinSpec lastJoinSpec = joinSpecs.get(joinSpecs.size() - 1); + JoinType type = lastJoinSpec.getType(); + JoinTable rhsJoinTable = lastJoinSpec.getJoinTable(); + Table rhsTable = rhsJoinTable.getTable(); + JoinTable lhsJoin = joinTable.getSubJoinTableWithoutPostFilters(); + Scan subScan = ScanUtil.newScan(originalScan); + StatementContext lhsCtx = new StatementContext(statement, context.getResolver(), subScan, new SequenceManager(statement)); + QueryPlan lhsPlan = compileJoinQuery(lhsCtx, binds, lhsJoin, true, true, null); + PTable rhsProjTable; + TableRef rhsTableRef; + SelectStatement rhs; + TupleProjector tupleProjector; + if (!rhsTable.isSubselect()) { + context.setCurrentTable(rhsTable.getTableRef()); + rhsProjTable = rhsTable.createProjectedTable(!projectPKColumns, context); + rhsTableRef = rhsTable.getTableRef(); + rhsTable.projectColumns(context.getScan()); + rhs = rhsJoinTable.getAsSingleSubquery(rhsTable.getAsSubquery(orderBy), asSubquery); + tupleProjector = new TupleProjector(rhsProjTable); + } else { + SelectStatement subquery = rhsTable.getAsSubquery(orderBy); + QueryPlan plan = compileSubquery(subquery, false); + rhsProjTable = rhsTable.createProjectedTable(plan.getProjector()); + rhsTableRef = plan.getTableRef(); + context.getScan().setFamilyMap(plan.getContext().getScan().getFamilyMap()); + rhs = rhsJoinTable.getAsSingleSubquery((SelectStatement) plan.getStatement(), asSubquery); + tupleProjector = new TupleProjector(plan.getProjector()); + } + context.setCurrentTable(rhsTableRef); + context.setResolver(FromCompiler.getResolverForProjectedTable(rhsProjTable, context.getConnection(), rhs.getUdfParseNodes())); + ImmutableBytesPtr[] joinIds = new ImmutableBytesPtr[]{new ImmutableBytesPtr(emptyByteArray)}; + Pair<List<Expression>, List<Expression>> joinConditions = lastJoinSpec.compileJoinConditions(lhsCtx, context, strategy); + List<Expression> joinExpressions = joinConditions.getSecond(); + List<Expression> hashExpressions = joinConditions.getFirst(); + boolean needsMerge = lhsJoin.hasPostReference(); + PTable lhsTable = needsMerge ? lhsCtx.getResolver().getTables().get(0).getTable() : null; + int fieldPosition = needsMerge ? rhsProjTable.getColumns().size() - rhsProjTable.getPKColumns().size() : 0; + PTable projectedTable = needsMerge ? JoinCompiler.joinProjectedTables(rhsProjTable, lhsTable, type == JoinType.Right ? JoinType.Left : type) : rhsProjTable; + TupleProjector.serializeProjectorIntoScan(context.getScan(), tupleProjector); + context.setResolver(FromCompiler.getResolverForProjectedTable(projectedTable, context.getConnection(), rhs.getUdfParseNodes())); + QueryPlan rhsPlan = compileSingleFlatQuery(context, rhs, binds, asSubquery, !asSubquery && type == JoinType.Right, null, !rhsTable.isSubselect() && projectPKColumns ? tupleProjector : null, true); + Expression postJoinFilterExpression = joinTable.compilePostFilterExpression(context, rhsTable); + Integer limit = null; + Integer offset = null; + if (!rhs.isAggregate() && !rhs.isDistinct() && rhs.getOrderBy().isEmpty()) { + limit = rhsPlan.getLimit(); + offset = rhsPlan.getOffset(); + } + HashJoinInfo joinInfo = new HashJoinInfo(projectedTable, joinIds, new List[]{joinExpressions}, + new JoinType[]{type == JoinType.Right ? JoinType.Left : type}, new boolean[]{true}, + new PTable[]{lhsTable}, new int[]{fieldPosition}, postJoinFilterExpression, QueryUtil.getOffsetLimit(limit, offset)); + Pair<Expression, Expression> keyRangeExpressions = new Pair<Expression, Expression>(null, null); + getKeyExpressionCombinations(keyRangeExpressions, context, joinTable.getStatement(), rhsTableRef, type, joinExpressions, hashExpressions); + return HashJoinPlan.create(joinTable.getStatement(), rhsPlan, joinInfo, new HashSubPlan[]{new HashSubPlan(0, lhsPlan, hashExpressions, false, keyRangeExpressions.getFirst(), keyRangeExpressions.getSecond())}); } - HashJoinInfo joinInfo = new HashJoinInfo(projectedTable, joinIds, joinExpressions, joinTypes, - starJoinVector, tables, fieldPositions, postJoinFilterExpression, QueryUtil.getOffsetLimit(limit, offset)); - return HashJoinPlan.create(joinTable.getStatement(), plan, joinInfo, hashPlans); - } + case SORT_MERGE: { + JoinTable lhsJoin = joinTable.getSubJoinTableWithoutPostFilters(); + JoinSpec lastJoinSpec = joinSpecs.get(joinSpecs.size() - 1); + JoinType type = lastJoinSpec.getType(); + JoinTable rhsJoin = lastJoinSpec.getJoinTable(); + if (type == JoinType.Right) { + JoinTable temp = lhsJoin; + lhsJoin = rhsJoin; + rhsJoin = temp; + } - JoinSpec lastJoinSpec = joinSpecs.get(joinSpecs.size() - 1); - JoinType type = lastJoinSpec.getType(); - if (!this.useSortMergeJoin - && (type == JoinType.Right || type == JoinType.Inner) - && lastJoinSpec.getJoinTable().getJoinSpecs().isEmpty() - && lastJoinSpec.getJoinTable().getTable().isFlat()) { - JoinTable rhsJoinTable = lastJoinSpec.getJoinTable(); - Table rhsTable = rhsJoinTable.getTable(); - JoinTable lhsJoin = joinTable.getSubJoinTableWithoutPostFilters(); - Scan subScan = ScanUtil.newScan(originalScan); - StatementContext lhsCtx = new StatementContext(statement, context.getResolver(), subScan, new SequenceManager(statement)); - QueryPlan lhsPlan = compileJoinQuery(lhsCtx, binds, lhsJoin, true, true, null); - PTable rhsProjTable; - TableRef rhsTableRef; - SelectStatement rhs; - TupleProjector tupleProjector; - if (!rhsTable.isSubselect()) { - context.setCurrentTable(rhsTable.getTableRef()); - rhsProjTable = rhsTable.createProjectedTable(!projectPKColumns, context); - rhsTableRef = rhsTable.getTableRef(); - rhsTable.projectColumns(context.getScan()); - rhs = rhsJoinTable.getAsSingleSubquery(rhsTable.getAsSubquery(orderBy), asSubquery); - tupleProjector = new TupleProjector(rhsProjTable); - } else { - SelectStatement subquery = rhsTable.getAsSubquery(orderBy); - QueryPlan plan = compileSubquery(subquery, false); - rhsProjTable = rhsTable.createProjectedTable(plan.getProjector()); - rhsTableRef = plan.getTableRef(); - context.getScan().setFamilyMap(plan.getContext().getScan().getFamilyMap()); - rhs = rhsJoinTable.getAsSingleSubquery((SelectStatement) plan.getStatement(), asSubquery); - tupleProjector = new TupleProjector(plan.getProjector()); - } - context.setCurrentTable(rhsTableRef); - context.setResolver(FromCompiler.getResolverForProjectedTable(rhsProjTable, context.getConnection(), rhs.getUdfParseNodes())); - ImmutableBytesPtr[] joinIds = new ImmutableBytesPtr[] {new ImmutableBytesPtr(emptyByteArray)}; - Pair<List<Expression>, List<Expression>> joinConditions = lastJoinSpec.compileJoinConditions(lhsCtx, context, JoinCompiler.Strategy.HASH_BUILD_LEFT); - List<Expression> joinExpressions = joinConditions.getSecond(); - List<Expression> hashExpressions = joinConditions.getFirst(); - boolean needsMerge = lhsJoin.hasPostReference(); - PTable lhsTable = needsMerge ? lhsCtx.getResolver().getTables().get(0).getTable() : null; - int fieldPosition = needsMerge ? rhsProjTable.getColumns().size() - rhsProjTable.getPKColumns().size() : 0; - PTable projectedTable = needsMerge ? JoinCompiler.joinProjectedTables(rhsProjTable, lhsTable, type == JoinType.Right ? JoinType.Left : type) : rhsProjTable; - TupleProjector.serializeProjectorIntoScan(context.getScan(), tupleProjector); - context.setResolver(FromCompiler.getResolverForProjectedTable(projectedTable, context.getConnection(), rhs.getUdfParseNodes())); - QueryPlan rhsPlan = compileSingleFlatQuery(context, rhs, binds, asSubquery, !asSubquery && type == JoinType.Right, null, !rhsTable.isSubselect() && projectPKColumns ? tupleProjector : null, true); - Expression postJoinFilterExpression = joinTable.compilePostFilterExpression(context, rhsTable); - Integer limit = null; - Integer offset = null; - if (!rhs.isAggregate() && !rhs.isDistinct() && rhs.getOrderBy().isEmpty()) { - limit = rhsPlan.getLimit(); - offset = rhsPlan.getOffset(); - } - HashJoinInfo joinInfo = new HashJoinInfo(projectedTable, joinIds, new List[] { joinExpressions }, - new JoinType[] { type == JoinType.Right ? JoinType.Left : type }, new boolean[] { true }, - new PTable[] { lhsTable }, new int[] { fieldPosition }, postJoinFilterExpression, QueryUtil.getOffsetLimit(limit, offset)); - Pair<Expression, Expression> keyRangeExpressions = new Pair<Expression, Expression>(null, null); - getKeyExpressionCombinations(keyRangeExpressions, context, joinTable.getStatement(), rhsTableRef, type, joinExpressions, hashExpressions); - return HashJoinPlan.create(joinTable.getStatement(), rhsPlan, joinInfo, new HashSubPlan[] {new HashSubPlan(0, lhsPlan, hashExpressions, false, keyRangeExpressions.getFirst(), keyRangeExpressions.getSecond())}); - } + List<EqualParseNode> joinConditionNodes = lastJoinSpec.getOnConditions(); + List<OrderByNode> lhsOrderBy = Lists.<OrderByNode>newArrayListWithExpectedSize(joinConditionNodes.size()); + List<OrderByNode> rhsOrderBy = Lists.<OrderByNode>newArrayListWithExpectedSize(joinConditionNodes.size()); + for (EqualParseNode condition : joinConditionNodes) { + lhsOrderBy.add(NODE_FACTORY.orderBy(type == JoinType.Right ? condition.getRHS() : condition.getLHS(), false, true)); + rhsOrderBy.add(NODE_FACTORY.orderBy(type == JoinType.Right ? condition.getLHS() : condition.getRHS(), false, true)); + } - JoinTable lhsJoin = joinTable.getSubJoinTableWithoutPostFilters(); - JoinTable rhsJoin = lastJoinSpec.getJoinTable(); - if (type == JoinType.Right) { - JoinTable temp = lhsJoin; - lhsJoin = rhsJoin; - rhsJoin = temp; - } - - List<EqualParseNode> joinConditionNodes = lastJoinSpec.getOnConditions(); - List<OrderByNode> lhsOrderBy = Lists.<OrderByNode> newArrayListWithExpectedSize(joinConditionNodes.size()); - List<OrderByNode> rhsOrderBy = Lists.<OrderByNode> newArrayListWithExpectedSize(joinConditionNodes.size()); - for (EqualParseNode condition : joinConditionNodes) { - lhsOrderBy.add(NODE_FACTORY.orderBy(type == JoinType.Right ? condition.getRHS() : condition.getLHS(), false, true)); - rhsOrderBy.add(NODE_FACTORY.orderBy(type == JoinType.Right ? condition.getLHS() : condition.getRHS(), false, true)); - } - - Scan lhsScan = ScanUtil.newScan(originalScan); - StatementContext lhsCtx = new StatementContext(statement, context.getResolver(), lhsScan, new SequenceManager(statement)); - boolean preserveRowkey = !projectPKColumns && type != JoinType.Full; - QueryPlan lhsPlan = compileJoinQuery(lhsCtx, binds, lhsJoin, true, !preserveRowkey, lhsOrderBy); - PTable lhsProjTable = lhsCtx.getResolver().getTables().get(0).getTable(); - boolean isInRowKeyOrder = preserveRowkey && lhsPlan.getOrderBy().getOrderByExpressions().isEmpty(); - - Scan rhsScan = ScanUtil.newScan(originalScan); - StatementContext rhsCtx = new StatementContext(statement, context.getResolver(), rhsScan, new SequenceManager(statement)); - QueryPlan rhsPlan = compileJoinQuery(rhsCtx, binds, rhsJoin, true, true, rhsOrderBy); - PTable rhsProjTable = rhsCtx.getResolver().getTables().get(0).getTable(); - - Pair<List<Expression>, List<Expression>> joinConditions = lastJoinSpec.compileJoinConditions(type == JoinType.Right ? rhsCtx : lhsCtx, type == JoinType.Right ? lhsCtx : rhsCtx, JoinCompiler.Strategy.SORT_MERGE); - List<Expression> lhsKeyExpressions = type == JoinType.Right ? joinConditions.getSecond() : joinConditions.getFirst(); - List<Expression> rhsKeyExpressions = type == JoinType.Right ? joinConditions.getFirst() : joinConditions.getSecond(); - - boolean needsMerge = rhsJoin.hasPostReference(); - int fieldPosition = needsMerge ? lhsProjTable.getColumns().size() - lhsProjTable.getPKColumns().size() : 0; - PTable projectedTable = needsMerge ? JoinCompiler.joinProjectedTables(lhsProjTable, rhsProjTable, type == JoinType.Right ? JoinType.Left : type) : lhsProjTable; - - ColumnResolver resolver = FromCompiler.getResolverForProjectedTable(projectedTable, context.getConnection(), joinTable.getStatement().getUdfParseNodes()); - TableRef tableRef = resolver.getTables().get(0); - StatementContext subCtx = new StatementContext(statement, resolver, ScanUtil.newScan(originalScan), new SequenceManager(statement)); - subCtx.setCurrentTable(tableRef); - QueryPlan innerPlan = new SortMergeJoinPlan(subCtx, joinTable.getStatement(), tableRef, type == JoinType.Right ? JoinType.Left : type, lhsPlan, rhsPlan, lhsKeyExpressions, rhsKeyExpressions, projectedTable, lhsProjTable, needsMerge ? rhsProjTable : null, fieldPosition, lastJoinSpec.isSingleValueOnly()); - context.setCurrentTable(tableRef); - context.setResolver(resolver); - TableNode from = NODE_FACTORY.namedTable(tableRef.getTableAlias(), NODE_FACTORY.table(tableRef.getTable().getSchemaName().getString(), tableRef.getTable().getTableName().getString())); - ParseNode where = joinTable.getPostFiltersCombined(); - SelectStatement select = asSubquery - ? NODE_FACTORY.select(from, joinTable.getStatement().getHint(), false, - Collections.<AliasedNode> emptyList(), where, null, null, orderBy, null, null, 0, false, - joinTable.getStatement().hasSequence(), Collections.<SelectStatement> emptyList(), + Scan lhsScan = ScanUtil.newScan(originalScan); + StatementContext lhsCtx = new StatementContext(statement, context.getResolver(), lhsScan, new SequenceManager(statement)); + boolean preserveRowkey = !projectPKColumns && type != JoinType.Full; + QueryPlan lhsPlan = compileJoinQuery(lhsCtx, binds, lhsJoin, true, !preserveRowkey, lhsOrderBy); + PTable lhsProjTable = lhsCtx.getResolver().getTables().get(0).getTable(); + boolean isInRowKeyOrder = preserveRowkey && lhsPlan.getOrderBy().getOrderByExpressions().isEmpty(); + + Scan rhsScan = ScanUtil.newScan(originalScan); + StatementContext rhsCtx = new StatementContext(statement, context.getResolver(), rhsScan, new SequenceManager(statement)); + QueryPlan rhsPlan = compileJoinQuery(rhsCtx, binds, rhsJoin, true, true, rhsOrderBy); + PTable rhsProjTable = rhsCtx.getResolver().getTables().get(0).getTable(); + + Pair<List<Expression>, List<Expression>> joinConditions = lastJoinSpec.compileJoinConditions(type == JoinType.Right ? rhsCtx : lhsCtx, type == JoinType.Right ? lhsCtx : rhsCtx, strategy); + List<Expression> lhsKeyExpressions = type == JoinType.Right ? joinConditions.getSecond() : joinConditions.getFirst(); + List<Expression> rhsKeyExpressions = type == JoinType.Right ? joinConditions.getFirst() : joinConditions.getSecond(); + + boolean needsMerge = rhsJoin.hasPostReference(); + int fieldPosition = needsMerge ? lhsProjTable.getColumns().size() - lhsProjTable.getPKColumns().size() : 0; + PTable projectedTable = needsMerge ? JoinCompiler.joinProjectedTables(lhsProjTable, rhsProjTable, type == JoinType.Right ? JoinType.Left : type) : lhsProjTable; + + ColumnResolver resolver = FromCompiler.getResolverForProjectedTable(projectedTable, context.getConnection(), joinTable.getStatement().getUdfParseNodes()); + TableRef tableRef = resolver.getTables().get(0); + StatementContext subCtx = new StatementContext(statement, resolver, ScanUtil.newScan(originalScan), new SequenceManager(statement)); + subCtx.setCurrentTable(tableRef); + QueryPlan innerPlan = new SortMergeJoinPlan(subCtx, joinTable.getStatement(), tableRef, type == JoinType.Right ? JoinType.Left : type, lhsPlan, rhsPlan, lhsKeyExpressions, rhsKeyExpressions, projectedTable, lhsProjTable, needsMerge ? rhsProjTable : null, fieldPosition, lastJoinSpec.isSingleValueOnly()); + context.setCurrentTable(tableRef); + context.setResolver(resolver); + TableNode from = NODE_FACTORY.namedTable(tableRef.getTableAlias(), NODE_FACTORY.table(tableRef.getTable().getSchemaName().getString(), tableRef.getTable().getTableName().getString())); + ParseNode where = joinTable.getPostFiltersCombined(); + SelectStatement select = asSubquery + ? NODE_FACTORY.select(from, joinTable.getStatement().getHint(), false, + Collections.<AliasedNode>emptyList(), where, null, null, orderBy, null, null, 0, false, + joinTable.getStatement().hasSequence(), Collections.<SelectStatement>emptyList(), joinTable.getStatement().getUdfParseNodes()) - : NODE_FACTORY.select(joinTable.getStatement(), from, where); - - return compileSingleFlatQuery(context, select, binds, asSubquery, false, innerPlan, null, isInRowKeyOrder); + : NODE_FACTORY.select(joinTable.getStatement(), from, where); + + return compileSingleFlatQuery(context, select, binds, asSubquery, false, innerPlan, null, isInRowKeyOrder); + } + default: + throw new IllegalArgumentException("Invalid join strategy '" + strategy + "'"); + } } private boolean getKeyExpressionCombinations(Pair<Expression, Expression> combination, StatementContext context, SelectStatement select, TableRef table, JoinType type, final List<Expression> joinExpressions, final List<Expression> hashExpressions) throws SQLException { http://git-wip-us.apache.org/repos/asf/phoenix/blob/6b40a36b/phoenix-core/src/main/java/org/apache/phoenix/compile/QueryPlan.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/QueryPlan.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/QueryPlan.java index ca88984..c2edaf3 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/compile/QueryPlan.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/QueryPlan.java @@ -24,6 +24,7 @@ import org.apache.hadoop.hbase.client.ResultScanner; import org.apache.hadoop.hbase.client.Scan; import org.apache.phoenix.compile.GroupByCompiler.GroupBy; import org.apache.phoenix.compile.OrderByCompiler.OrderBy; +import org.apache.phoenix.execute.visitor.QueryPlanVisitor; import org.apache.phoenix.iterate.ParallelScanGrouper; import org.apache.phoenix.iterate.ResultIterator; import org.apache.phoenix.optimize.Cost; @@ -90,4 +91,5 @@ public interface QueryPlan extends StatementPlan { */ public boolean useRoundRobinIterator() throws SQLException; + <T> T accept(QueryPlanVisitor<T> visitor); } http://git-wip-us.apache.org/repos/asf/phoenix/blob/6b40a36b/phoenix-core/src/main/java/org/apache/phoenix/compile/TraceQueryPlan.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/TraceQueryPlan.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/TraceQueryPlan.java index 2714858..02aadc5 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/compile/TraceQueryPlan.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/TraceQueryPlan.java @@ -35,6 +35,7 @@ import org.apache.htrace.Sampler; import org.apache.htrace.TraceScope; import org.apache.phoenix.compile.GroupByCompiler.GroupBy; import org.apache.phoenix.compile.OrderByCompiler.OrderBy; +import org.apache.phoenix.execute.visitor.QueryPlanVisitor; import org.apache.phoenix.expression.Determinism; import org.apache.phoenix.expression.Expression; import org.apache.phoenix.expression.LiteralExpression; @@ -270,6 +271,11 @@ public class TraceQueryPlan implements QueryPlan { } @Override + public <T> T accept(QueryPlanVisitor<T> visitor) { + return visitor.visit(this); + } + + @Override public Long getEstimatedRowsToScan() { return 0l; } http://git-wip-us.apache.org/repos/asf/phoenix/blob/6b40a36b/phoenix-core/src/main/java/org/apache/phoenix/execute/AggregatePlan.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/execute/AggregatePlan.java b/phoenix-core/src/main/java/org/apache/phoenix/execute/AggregatePlan.java index 2e042e7..0c8e8dc 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/execute/AggregatePlan.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/execute/AggregatePlan.java @@ -33,6 +33,10 @@ import org.apache.phoenix.compile.StatementContext; import org.apache.phoenix.coprocessor.BaseScannerRegionObserver; import org.apache.phoenix.coprocessor.GroupedAggregateRegionObserver; import org.apache.phoenix.coprocessor.UngroupedAggregateRegionObserver; +import org.apache.phoenix.execute.visitor.AvgRowWidthVisitor; +import org.apache.phoenix.execute.visitor.ByteCountVisitor; +import org.apache.phoenix.execute.visitor.QueryPlanVisitor; +import org.apache.phoenix.execute.visitor.RowCountVisitor; import org.apache.phoenix.expression.Expression; import org.apache.phoenix.expression.OrderByExpression; import org.apache.phoenix.expression.RowKeyExpression; @@ -117,25 +121,39 @@ public class AggregatePlan extends BaseQueryPlan { @Override public Cost getCost() { - Long byteCount = null; + Double outputBytes = this.accept(new ByteCountVisitor()); + Double rowWidth = this.accept(new AvgRowWidthVisitor()); + Long inputRows = null; try { - byteCount = getEstimatedBytesToScan(); + inputRows = getEstimatedRowsToScan(); } catch (SQLException e) { // ignored. } - - if (byteCount == null) { + if (inputRows == null || outputBytes == null || rowWidth == null) { return Cost.UNKNOWN; } + double inputBytes = inputRows * rowWidth; + double rowsBeforeHaving = RowCountVisitor.aggregate( + RowCountVisitor.filter( + inputRows.doubleValue(), + RowCountVisitor.stripSkipScanFilter( + context.getScan().getFilter())), + groupBy); + double rowsAfterHaving = RowCountVisitor.filter(rowsBeforeHaving, having); + double bytesBeforeHaving = rowWidth * rowsBeforeHaving; + double bytesAfterHaving = rowWidth * rowsAfterHaving; int parallelLevel = CostUtil.estimateParallelLevel( true, context.getConnection().getQueryServices()); - Cost cost = CostUtil.estimateAggregateCost(byteCount, - groupBy, aggregators.getEstimatedByteSize(), parallelLevel); + Cost cost = new Cost(0, 0, inputBytes); + Cost aggCost = CostUtil.estimateAggregateCost( + inputBytes, bytesBeforeHaving, groupBy, parallelLevel); + cost = cost.plus(aggCost); if (!orderBy.getOrderByExpressions().isEmpty()) { - double outputBytes = CostUtil.estimateAggregateOutputBytes( - byteCount, groupBy, aggregators.getEstimatedByteSize()); - Cost orderByCost = CostUtil.estimateOrderByCost(outputBytes, parallelLevel); + parallelLevel = CostUtil.estimateParallelLevel( + false, context.getConnection().getQueryServices()); + Cost orderByCost = CostUtil.estimateOrderByCost( + bytesAfterHaving, outputBytes, parallelLevel); cost = cost.plus(orderByCost); } return cost; @@ -304,4 +322,9 @@ public class AggregatePlan extends BaseQueryPlan { return false; } + @Override + public <T> T accept(QueryPlanVisitor<T> visitor) { + return visitor.visit(this); + } + }
