!1086 Sort aggregation support for unsorted partitioned table

Merge pull request !1086 from NeerajUnnikrishnan/sortAggr

Author: it-is-a-robot

presto-benchto-benchmarks/src/main/resources/sql/presto/queries/q_partition2Colu.sort_agg.plan.txt

@@ -0,0 +1,6 @@
+remote exchange (GATHER, SINGLE, [])
+    final hashaggregation over (wr_account_credit, wr_net_loss)
+        local exchange (GATHER, SINGLE, [])
+            remote exchange (REPARTITION, HASH, ["wr_account_credit", "wr_net_loss"])
+                partial sortaggregate over (wr_account_credit, wr_net_loss)
+                    scan web_returns_partiotion_netloss_returneddatesk

presto-benchto-benchmarks/src/main/resources/sql/presto/queries/q_partition2Colu.sql

@@ -0,0 +1,3 @@
+select count(${database}.${schema}.web_returns_partiotion_netloss_returneddatesk.wr_net_loss),${database}.${schema}.web_returns_partiotion_netloss_returneddatesk.wr_net_loss
+from ${database}.${schema}.web_returns_partiotion_netloss_returneddatesk
+group by ${database}.${schema}.web_returns_partiotion_netloss_returneddatesk.wr_account_credit, ${database}.${schema}.web_returns_partiotion_netloss_returneddatesk.wr_net_loss

presto-benchto-benchmarks/src/main/resources/sql/presto/queries/q_partition2ColuBucket8.sort_agg.plan.txt

@@ -0,0 +1,6 @@
+remote exchange (GATHER, SINGLE, [])
+    final hashaggregation over (wr_account_credit, wr_net_loss)
+        local exchange (GATHER, SINGLE, [])
+            remote exchange (REPARTITION, HASH, ["wr_account_credit", "wr_net_loss"])
+                partial sortaggregate over (wr_account_credit, wr_net_loss)
+                    scan web_returns_partiotion_netloss_returneddatesk_bucket8

presto-benchto-benchmarks/src/main/resources/sql/presto/queries/q_partition2ColuBucket8.sql

@@ -0,0 +1,3 @@
+select count(${database}.${schema}.web_returns_partiotion_netloss_returneddatesk_bucket8.wr_net_loss),${database}.${schema}.web_returns_partiotion_netloss_returneddatesk_bucket8.wr_net_loss
+from ${database}.${schema}.web_returns_partiotion_netloss_returneddatesk_bucket8
+group by ${database}.${schema}.web_returns_partiotion_netloss_returneddatesk_bucket8.wr_account_credit, ${database}.${schema}.web_returns_partiotion_netloss_returneddatesk_bucket8.wr_net_loss

presto-benchto-benchmarks/src/main/resources/sql/presto/queries/q_PartitionBucketCount1.sort_agg.plan.txt presto-benchto-benchmarks/src/main/resources/sql/presto/queries/q_sortedPartitionBucketCount1.sort_agg.plan.txt

@@ -0,0 +1,6 @@
+remote exchange (GATHER, SINGLE, [])
+    final hashaggregation over (wr_account_credit, wr_net_loss)
+        local exchange (GATHER, SINGLE, [])
+            remote exchange (REPARTITION, HASH, ["wr_account_credit", "wr_net_loss"])
+                partial sortaggregate over (wr_account_credit, wr_net_loss)
+                    scan web_returns_partiotion_netloss_returneddatesk_bucket1_sort

presto-benchto-benchmarks/src/main/resources/sql/presto/queries/q_PartitionBucketCount1.sql presto-benchto-benchmarks/src/main/resources/sql/presto/queries/q_sortedPartitionBucketCount1.sql

@@ -0,0 +1,3 @@
+select count(${database}.${schema}.web_returns_partiotion_netloss_returneddatesk_bucket1_sort.wr_net_loss),${database}.${schema}.web_returns_partiotion_netloss_returneddatesk_bucket1_sort.wr_net_loss
+from ${database}.${schema}.web_returns_partiotion_netloss_returneddatesk_bucket1_sort
+group by ${database}.${schema}.web_returns_partiotion_netloss_returneddatesk_bucket1_sort.wr_account_credit, ${database}.${schema}.web_returns_partiotion_netloss_returneddatesk_bucket1_sort.wr_net_loss

presto-benchto-benchmarks/src/main/resources/sql/presto/queries/q_PartitionBucketCount32.sort_agg.plan.txt presto-benchto-benchmarks/src/main/resources/sql/presto/queries/q_sortedPartitionBucketCount32.sort_agg.plan.txt

@@ -0,0 +1,6 @@
+remote exchange (GATHER, SINGLE, [])
+    final hashaggregation over (wr_account_credit, wr_net_loss)
+        local exchange (GATHER, SINGLE, [])
+            remote exchange (REPARTITION, HASH, ["wr_account_credit", "wr_net_loss"])
+                partial sortaggregate over (wr_account_credit, wr_net_loss)
+                    scan web_returns_partiotion_netloss_returneddatesk_bucket8_sort

presto-benchto-benchmarks/src/main/resources/sql/presto/queries/q_PartitionBucketCount32.sql presto-benchto-benchmarks/src/main/resources/sql/presto/queries/q_sortedPartitionBucketCount32.sql

@@ -0,0 +1,3 @@
+select count(${database}.${schema}.web_returns_partiotion_netloss_returneddatesk_bucket8_sort.wr_net_loss),${database}.${schema}.web_returns_partiotion_netloss_returneddatesk_bucket8_sort.wr_net_loss
+from ${database}.${schema}.web_returns_partiotion_netloss_returneddatesk_bucket8_sort
+group by ${database}.${schema}.web_returns_partiotion_netloss_returneddatesk_bucket8_sort.wr_account_credit, ${database}.${schema}.web_returns_partiotion_netloss_returneddatesk_bucket8_sort.wr_net_loss

presto-benchto-benchmarks/src/main/resources/sql/presto/queries/q_useOnlyPartition2ColuBucket1OnSortTable.sort_agg.plan.txt

@@ -77,6 +77,7 @@ public TestSortBasedAggregationPlan()
 
             Map<String, String> hiveCatalogConfig = ImmutableMap.<String, String>builder()
                     .put("hive.orc-predicate-pushdown-enabled", "false")
+                    .put("hive.max-partitions-per-writers", "9999")
                     .build();
 
             queryRunner.createCatalog(catalog, hiveConnectorFactory, hiveCatalogConfig);
@@ -118,14 +119,38 @@ public TestSortBasedAggregationPlan()
                     "bucketed_by=array['wr_return_quantity'], bucket_count = 1, " +
                     "partitioned_by = ARRAY['wr_net_loss'], " +
                     "sorted_by = ARRAY['wr_return_quantity','wr_returned_time_sk']) " +
-                    "as select * from tpcds.tiny.web_returns limit 100");
+                    "as select * from tpcds.tiny.web_returns limit 500");
 
             queryRunner.execute("create table web_returns_partition_bucketCount32 with (" +
                     "bucketed_by=array['wr_return_quantity'], bucket_count = 1, " +
                     "partitioned_by = ARRAY['wr_net_loss'], " +
                     "sorted_by = ARRAY['wr_return_quantity','wr_returned_time_sk']) " +
-                    "as select * from tpcds.tiny.web_returns limit 100");
-
+                    "as select * from tpcds.tiny.web_returns limit 1000");
+
+            queryRunner.execute("create table web_returns_partiotion_netloss_returneddatesk with (" +
+                    "transactional = false, format='orc', " +
+                    "partitioned_by = ARRAY['wr_account_credit', 'wr_net_loss']) " +
+                    "as select * from tpcds.tiny.web_returns limit 500");
+
+            queryRunner.execute("create table web_returns_partiotion_netloss_returneddatesk_bucket8 with (" +
+                    "transactional = false, format='orc', " +
+                    "bucket_count = 8, bucketed_by = ARRAY['wr_return_quantity','wr_returned_time_sk'], " +
+                    "partitioned_by = ARRAY['wr_account_credit', 'wr_net_loss']) " +
+                    "as select * from tpcds.tiny.web_returns limit 500");
+
+            queryRunner.execute("create table web_returns_partiotion_netloss_returneddatesk_bucket8_sort with (" +
+                    "transactional = false, format='orc', " +
+                    "bucket_count = 8, bucketed_by = ARRAY['wr_return_quantity','wr_returned_time_sk'], " +
+                    "partitioned_by = ARRAY['wr_account_credit', 'wr_net_loss'], " +
+                    "sorted_by = ARRAY['wr_return_quantity','wr_returned_time_sk'])" +
+                    "as select * from tpcds.tiny.web_returns limit 500");
+
+            queryRunner.execute("create table web_returns_partiotion_netloss_returneddatesk_bucket1_sort with (" +
+                    "transactional = false, format='orc', " +
+                    "bucket_count = 1, bucketed_by = ARRAY['wr_return_quantity','wr_returned_time_sk'], " +
+                    "sorted_by = ARRAY['wr_return_quantity','wr_returned_time_sk'], " +
+                    "partitioned_by = ARRAY['wr_account_credit', 'wr_net_loss']) " +
+                    "as select * from tpcds.tiny.web_returns limit 500");
             return queryRunner;
         }, false, false, false, true);
     }
@@ -136,7 +161,8 @@ protected Stream<String> getQueryResourcePaths()
         return Stream.of("q_InnerJoin", "q_leftJoin", "q_rightjoin", "q_rightjoin_wrong_order", "q_Inner_LeftJoin", "q_sort_groupby_notsameOrder",
                 "q_sort_groupby_notsameOrder1", "q_groupByHavingMore", "q_InnerJoinWrongOrder", "q_InnerJoinLessCriterias",
                 "q_InnerJoinGroupLessCriterias", "q_bucketAndSortDifferentOrder", "BucketAndGroupSameSortIsMore", "BucketGroupAreDifferent",
-                "groupsAreMoreThanBucket", "ColNameEndWithInt", "q_PartitionBucketCount1", "q_PartitionBucketCount32")
+                "groupsAreMoreThanBucket", "ColNameEndWithInt", "q_sortedPartitionBucketCount1", "q_sortedPartitionBucketCount32",
+                "q_partition2Colu", "q_partition2ColuBucket8", "q_useOnlyPartition2ColuBucket8OnSortTable", "q_useOnlyPartition2ColuBucket1OnSortTable")
                 .flatMap(i -> {
                     String queryId = format("%s", i);
                     System.out.println("query ID: " + queryId);

presto-benchto-benchmarks/src/main/resources/sql/presto/queries/q_useOnlyPartition2ColuBucket1OnSortTable.sql

@@ -47,6 +47,7 @@
 import io.prestosql.plugin.hive.statistics.HiveStatisticsProvider;
 import io.prestosql.plugin.hive.util.ConfigurationUtils;
 import io.prestosql.plugin.hive.util.Statistics;
+import io.prestosql.spi.PartialAndFinalAggregationType;
 import io.prestosql.spi.PrestoException;
 import io.prestosql.spi.block.Block;
 import io.prestosql.spi.connector.ColumnHandle;
@@ -3129,36 +3130,44 @@ public List<ConnectorVacuumTableInfo> getTablesForVacuum()
     }
 
     @Override
-    public boolean canPerformSortBasedAggregation(ConnectorSession session, ConnectorTableHandle tableHandle, List<String> groupKeyNames)
+    public PartialAndFinalAggregationType validateAndGetSortAggregationType(ConnectorSession session, ConnectorTableHandle tableHandle, List<String> groupKeyNames)
     {
+        PartialAndFinalAggregationType partialAndFinalAggregationType = new PartialAndFinalAggregationType();
         ConnectorTableMetadata connectorTableMetadata = getTableMetadata(session, ((HiveTableHandle) tableHandle).getSchemaTableName());
         List<SortingColumn> sortingColumn = (List<SortingColumn>) connectorTableMetadata.getProperties().get(HiveTableProperties.SORTED_BY_PROPERTY);
-        if ((sortingColumn == null) || (sortingColumn.size() == 0)) {
-            return false;
-        }
+        boolean isSortingColumnsNotPresent = (sortingColumn == null) || (sortingColumn.size() == 0);
+
         List<String> partitionedBy = new ArrayList<>();
-        List<String> partitionedBytemp = (List<String>) connectorTableMetadata.getProperties().get(HiveTableProperties.PARTITIONED_BY_PROPERTY);
-        if (null != partitionedBytemp) {
-            partitionedBy.addAll(partitionedBytemp);
+        List<String> partitionedByTemp = (List<String>) connectorTableMetadata.getProperties().get(HiveTableProperties.PARTITIONED_BY_PROPERTY);
+        if ((partitionedByTemp != null) && (partitionedByTemp.size() != 0)) {
+            partitionedBy.addAll(partitionedByTemp);
+            if (isSortingColumnsNotPresent && (partitionedByTemp.size() != groupKeyNames.size())) {
+                return partialAndFinalAggregationType;
+            }
+        }
+        else if (isSortingColumnsNotPresent) {
+            return partialAndFinalAggregationType;
         }
         int bucketCount = 0;
-        List<String> bucketedColumns = (List<String>) connectorTableMetadata.getProperties().get(HiveTableProperties.BUCKETED_BY_PROPERTY);
-        if (null != bucketedColumns) {
-            bucketCount = (int) connectorTableMetadata.getProperties().get(HiveTableProperties.BUCKET_COUNT_PROPERTY);
+        List<String> bucketedColumns = new ArrayList<>();
+        if (!isSortingColumnsNotPresent) {
+            bucketedColumns.addAll((List<String>) connectorTableMetadata.getProperties().get(HiveTableProperties.BUCKETED_BY_PROPERTY));
+            if (null != bucketedColumns) {
+                bucketCount = (int) connectorTableMetadata.getProperties().get(HiveTableProperties.BUCKET_COUNT_PROPERTY);
+            }
         }
 
-        List<String> sortedColumnNames = sortingColumn.stream().map(column -> column.getColumnName()).collect(Collectors.toList());
-        List<String> partitionAndSortedColumnNames = new ArrayList<>();
-        // when there is partition by , sorted data is arranged 'partition by' followed by 'sorted by'
-        partitionAndSortedColumnNames.addAll(partitionedBy);
-        partitionAndSortedColumnNames.addAll(sortedColumnNames);
+        List<String> sortedColumnNames = new ArrayList<>();
+        if ((sortingColumn != null) && (sortingColumn.size() != 0)) {
+            sortedColumnNames.addAll(sortingColumn.stream().map(column -> column.getColumnName()).collect(Collectors.toList()));
+        }
 
         //grouping key should be sub set of sorted By and it should match all partition by columns
-        if ((sortedColumnNames.size() + partitionedBy.size() < groupKeyNames.size()) ||
+        if ((partitionedBy.size() + sortedColumnNames.size() < groupKeyNames.size()) ||
                 (partitionedBy.size() > groupKeyNames.size())) {
             //sorted columns are less than join criteria columns
             log.debug("number of sorted columns " + sortedColumnNames.size() + "are less join column size " + groupKeyNames.size());
-            return false;
+            return partialAndFinalAggregationType;
         }
 
         // bucketby columns and groupby Columns should be same.
@@ -3169,43 +3178,52 @@ public boolean canPerformSortBasedAggregation(ConnectorSession session, Connecto
                 (groupKeyNames.get(partitionedByCount).equals(bucketedColumns.get(0)))) || (bucketCount == 0));
 
         if ((bucketCount == 1) && (bucketedColumns.size() > 1)) {
-            boolean notMatching = false;
             int minSize = Math.min(groupKeyNames.size() - partitionedBy.size(), bucketedColumns.size());
             int partSize = partitionedBy.size();
             for (int keyIdx = 0; keyIdx < minSize; keyIdx++) {
                 if (!groupKeyNames.get(keyIdx + partSize).equals(bucketedColumns.get(keyIdx))) {
-                    notMatching = true;
-                    break;
+                    return partialAndFinalAggregationType;
                 }
             }
-            if (!notMatching) {
-                singleOrZeroBucketedColumn = true;
+            singleOrZeroBucketedColumn = true;
+        }
+
+        for (int numOfComparedKeys = 0; numOfComparedKeys < partitionedBy.size(); numOfComparedKeys++) {
+            if ((!groupKeyNames.get(numOfComparedKeys).equals(partitionedBy.get(numOfComparedKeys)))) {
+                return partialAndFinalAggregationType;
             }
         }
+
+        if (groupKeyNames.size() == partitionedBy.size()) {
+            partialAndFinalAggregationType.setPartialAsSortAndFinalAsHashAggregation(true);
+            return partialAndFinalAggregationType;
+        }
+
         if (singleOrZeroBucketedColumn || (groupKeyNames.size() == (bucketedColumns.size() + partitionedBy.size()))) {
-            for (int numOfComparedKeys = 0; numOfComparedKeys < groupKeyNames.size(); numOfComparedKeys++) {
-                boolean bucketedColumnsResult = numOfComparedKeys < partitionedBy.size() ? false : !singleOrZeroBucketedColumn &&
-                        (!groupKeyNames.get(numOfComparedKeys).equals(bucketedColumns.get(numOfComparedKeys - partitionedBy.size())));
-                if ((!groupKeyNames.get(numOfComparedKeys).equals(partitionAndSortedColumnNames.get(numOfComparedKeys))) ||
+            int numOfCmpKeysAfterPartitionedBy = partitionedBy.size();
+            for (int numOfComparedKeys = 0; numOfComparedKeys < groupKeyNames.size() - partitionedBy.size(); numOfComparedKeys++, numOfCmpKeysAfterPartitionedBy++) {
+                boolean bucketedColumnsResult = !singleOrZeroBucketedColumn && (!groupKeyNames.get(numOfComparedKeys).equals(bucketedColumns.get(numOfComparedKeys)));
+                if ((!groupKeyNames.get(numOfCmpKeysAfterPartitionedBy).equals(sortedColumnNames.get(numOfComparedKeys))) ||
                         (!singleOrZeroBucketedColumn && bucketedColumnsResult)) {
                     if (log.isDebugEnabled()) {
                         final String[] dbgGroupKeyNames = {new String("")};
                         groupKeyNames.stream().forEach(k -> dbgGroupKeyNames[0] = dbgGroupKeyNames[0].concat(k + " , "));
                         final String[] dbgSortedColumnNames = {new String("")};
-                        partitionAndSortedColumnNames.stream().forEach(k -> dbgSortedColumnNames[0] = dbgSortedColumnNames[0].concat(k + " , "));
+                        sortedColumnNames.stream().forEach(k -> dbgSortedColumnNames[0] = dbgSortedColumnNames[0].concat(k + " , "));
                         if ((null != bucketedColumns) && (bucketedColumns.size() > 0)) {
                             final String[] dbgbucketedColumns = {new String("")};
                             bucketedColumns.stream().forEach(k -> dbgbucketedColumns[0] = dbgbucketedColumns[0].concat(k + " , "));
                             log.debug("Not matching sortedColumnNames: " + dbgSortedColumnNames + " group columns name: " + dbgGroupKeyNames + " bucketedColumns :" + dbgbucketedColumns);
                         }
                         log.debug("Not matching sortedColumnNames: " + dbgSortedColumnNames + " group columns name: " + dbgGroupKeyNames);
                     }
-                    return false;
+                    return partialAndFinalAggregationType;
                 }
             }
-            return true;
+            partialAndFinalAggregationType.setSortAggregation(true);
+            return partialAndFinalAggregationType;
         }
-        return false;
+        return partialAndFinalAggregationType;
     }
 
     @Override

presto-benchto-benchmarks/src/main/resources/sql/presto/queries/q_useOnlyPartition2ColuBucket8OnSortTable.sort_agg.plan.txt

@@ -6609,6 +6609,52 @@ public void testCachedPlanForTablesWithSameName()
         assertUpdate(String.format("DROP TABLE %s.%s", schema, table));
     }
 
+    @Test
+    public void UnSortAggrePartitionBucketCount1()
+    {
+        initSortBasedAggregation();
+        computeActual("create table lineitem_partition_shipmode_comment_bucket1  with(transactional = false, " +
+                "format = 'ORC', partitioned_by = ARRAY['shipmode', 'comment'], bucketed_by=array['orderkey',  'partkey'], bucket_count=1)" +
+                "  as select * from tpch.tiny.lineitem limit 100");
+
+        String query = "select lineitem_partition_shipmode_comment_bucket1.shipmode, lineitem_partition_shipmode_comment_bucket1.comment " +
+                "from lineitem_partition_shipmode_comment_bucket1 " +
+                "group by shipmode, comment " +
+                "order by shipmode, comment ";
+
+        MaterializedResult sortResult = computeActual(testSessionSort, query);
+        MaterializedResult hashResult = computeActual(query);
+        assertEquals(sortResult.getMaterializedRows(), hashResult.getMaterializedRows());
+        sortResult = computeActual(testSessionSortPrcntDrv50, query);
+        assertEquals(sortResult.getMaterializedRows(), hashResult.getMaterializedRows());
+        sortResult = computeActual(testSessionSortPrcntDrv40, query);
+        assertEquals(sortResult.getMaterializedRows(), hashResult.getMaterializedRows());
+        assertUpdate("DROP TABLE lineitem_partition_shipmode_comment_bucket1");
+    }
+
+    @Test
+    public void SortAggreGroupOnlyPartitionColumns()
+    {
+        initSortBasedAggregation();
+        computeActual("create table lineitem_sort_partition_shipmode_comment_bucket1  with(transactional = false, " +
+                "format = 'ORC', partitioned_by = ARRAY['shipmode', 'comment'], bucketed_by=array['orderkey',  'partkey'], bucket_count=1, sorted_by = ARRAY['orderkey', 'partkey'])" +
+                "  as select * from tpch.tiny.lineitem limit 100");
+
+        String query = "select lineitem_sort_partition_shipmode_comment_bucket1.shipmode, lineitem_sort_partition_shipmode_comment_bucket1.comment " +
+                "from lineitem_sort_partition_shipmode_comment_bucket1 " +
+                "group by shipmode, comment " +
+                "order by shipmode, comment ";
+
+        MaterializedResult sortResult = computeActual(testSessionSort, query);
+        MaterializedResult hashResult = computeActual(query);
+        assertEquals(sortResult.getMaterializedRows(), hashResult.getMaterializedRows());
+        sortResult = computeActual(testSessionSortPrcntDrv50, query);
+        assertEquals(sortResult.getMaterializedRows(), hashResult.getMaterializedRows());
+        sortResult = computeActual(testSessionSortPrcntDrv40, query);
+        assertEquals(sortResult.getMaterializedRows(), hashResult.getMaterializedRows());
+        assertUpdate("DROP TABLE lineitem_sort_partition_shipmode_comment_bucket1");
+    }
+
     @Test
     public void testAcidFormatColumnNameConflict()
     {