Make the order of complex grouping key consistent

[lukasz-stec]

Description

Additional context and related issues

Release notes

( X) This is not user-visible or is docs only, and no release notes are required.
( ) Release notes are required. Please propose a release note for me.
( ) Release notes are required, with the following suggested text:

# Section
* Fix some things. ({issue}`issuenumber`)

[martint]

Does this affect the order of expressions in the plan? If I recall correctly, this is just for doing some mapping later, but they are not used for driving the order in which things get added to the plan.

[sopel39]

There is

        // combine (cartesian product) with complex expressions
        List<List<Symbol>> groupingSets = sets.stream()
                .map(set -> ImmutableList.<Symbol>builder()
                        .addAll(set)
                        .addAll(complexExpressions.values())
                        .build())
                .collect(toImmutableList());

below.

[lukasz-stec]

yes, the code that @sopel39 pasted uses complexExpressions.values() which is not deterministict for HashSet

[lukasz-stec]

here is a sample query that shows the indeterminism:

explain
SELECT orderstatus || '1', orderstatus || '2', COUNT(DISTINCT orderkey), COUNT(DISTINCT custkey)
FROM orders
GROUP BY 1, 2;

the first run has
Aggregate[type = FINAL, keys = [concat$gid, concat$gid_1], hash = [$hashvalue]] in the Fragment 0
and the second one has
Aggregate[type = FINAL, keys = [concat$gid_1, concat$gid], hash = [$hashvalue]]

 Trino version: testversion                                                                                                                                                                                                          
 Fragment 0 [HASH]                                                                                                                                                                                                                   
     Output layout: [concat$gid, concat$gid_1, count, count_2]                                                                                                                                                                       
     Output partitioning: SINGLE []                                                                                                                                                                                                  
     Output[columnNames = [_col0, _col1, _col2, _col3]]                                                                                                                                                                              
     │   Layout: [concat$gid:varchar, concat$gid_1:varchar, count:bigint, count_2:bigint]                                                                                                                                            
     │   Estimates: {rows: ? (?), cpu: 0, memory: 0B, network: 0B}                                                                                                                                                                   
     │   _col0 := concat$gid                                                                                                                                                                                                         
     │   _col1 := concat$gid_1                                                                                                                                                                                                       
     │   _col2 := count                                                                                                                                                                                                              
     │   _col3 := count_2                                                                                                                                                                                                            
     └─ Project[]                                                                                                                                                                                                                    
        │   Layout: [concat$gid:varchar, concat$gid_1:varchar, count_2:bigint, count:bigint]                                                                                                                                         
        │   Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}                                                                                                                                                                
        └─ Aggregate[type = FINAL, keys = [concat$gid, concat$gid_1], hash = [$hashvalue]]                                                                                                                                           
           │   Layout: [concat$gid:varchar, concat$gid_1:varchar, $hashvalue:bigint, count_2:bigint, count:bigint]                                                                                                                   
           │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: 0B}                                                                                                                                                              
           │   count_2 := count("count_3")                                                                                                                                                                                           
           │   count := count("count_4")                                                                                                                                                                                             
           └─ LocalExchange[partitioning = HASH, hashColumn = [$hashvalue], arguments = ["concat$gid", "concat$gid_1"]]                                                                                                              
              │   Layout: [concat$gid:varchar, concat$gid_1:varchar, count_4:bigint, count_3:bigint, $hashvalue:bigint]                                                                                                              
              │   Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}                                                                                                                                                          
              └─ RemoteSource[sourceFragmentIds = [1]]                                                                                                                                                                               
                     Layout: [concat$gid:varchar, concat$gid_1:varchar, count_4:bigint, count_3:bigint, $hashvalue_5:bigint]                                                                                                         
                                                                                                                                                                                                                                     
 Fragment 1 [HASH]                                                                                                                                                                                                                   
     Output layout: [concat$gid, concat$gid_1, count_4, count_3, $hashvalue_6]                                                                                                                                                       
     Output partitioning: HASH [concat$gid, concat$gid_1][$hashvalue_6]                                                                                                                                                              
     Aggregate[type = PARTIAL, keys = [concat$gid, concat$gid_1], hash = [$hashvalue_6]]                                                                                                                                             
     │   Layout: [concat$gid:varchar, concat$gid_1:varchar, $hashvalue_6:bigint, count_4:bigint, count_3:bigint]                                                                                                                     
     │   count_4 := count("orderkey") (mask = orderkey$distinct)                                                                                                                                                                     
     │   count_3 := count("custkey") (mask = custkey$distinct)                                                                                                                                                                       
     └─ MarkDistinct[distinct = [concat$gid:varchar, concat$gid_1:varchar, custkey:bigint], marker = custkey$distinct, hash = [$hashvalue_7]]                                                                                        
        │   Layout: [orderkey:bigint, concat$gid:varchar, custkey:bigint, concat$gid_1:varchar, orderkey$distinct:boolean, $hashvalue_6:bigint, $hashvalue_7:bigint, custkey$distinct:boolean]                                       
        └─ LocalExchange[partitioning = HASH, hashColumn = [$hashvalue_7], arguments = ["concat$gid", "concat$gid_1", "custkey"]]                                                                                                    
           │   Layout: [orderkey:bigint, concat$gid:varchar, custkey:bigint, concat$gid_1:varchar, orderkey$distinct:boolean, $hashvalue_6:bigint, $hashvalue_7:bigint]                                                              
           │   Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}                                                                                                                                                             
           └─ RemoteSource[sourceFragmentIds = [2]]                                                                                                                                                                                  
                  Layout: [orderkey:bigint, concat$gid:varchar, custkey:bigint, concat$gid_1:varchar, orderkey$distinct:boolean, $hashvalue_8:bigint, $hashvalue_9:bigint]                                                           
                                                                                                                                                                                                                                     
 Fragment 2 [HASH]                                                                                                                                                                                                                   
     Output layout: [orderkey, concat$gid, custkey, concat$gid_1, orderkey$distinct, $hashvalue_10, $hashvalue_11]                                                                                                                   
     Output partitioning: HASH [concat$gid, concat$gid_1, custkey][$hashvalue_11]                                                                                                                                                    
     Project[]                                                                                                                                                                                                                       
     │   Layout: [orderkey:bigint, concat$gid:varchar, custkey:bigint, concat$gid_1:varchar, $hashvalue_10:bigint, $hashvalue_11:bigint, orderkey$distinct:boolean]                                                                  
     │   Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}                                                                                                                                                                   
     └─ MarkDistinct[distinct = [concat$gid:varchar, concat$gid_1:varchar, orderkey:bigint], marker = orderkey$distinct, hash = [$hashvalue_12]]                                                                                     
        │   Layout: [orderkey:bigint, concat$gid:varchar, custkey:bigint, concat$gid_1:varchar, $hashvalue_10:bigint, $hashvalue_11:bigint, $hashvalue_12:bigint, orderkey$distinct:boolean]                                         
        └─ LocalExchange[partitioning = HASH, hashColumn = [$hashvalue_12], arguments = ["concat$gid", "concat$gid_1", "orderkey"]]                                                                                                  
           │   Layout: [orderkey:bigint, concat$gid:varchar, custkey:bigint, concat$gid_1:varchar, $hashvalue_10:bigint, $hashvalue_11:bigint, $hashvalue_12:bigint]                                                                 
           │   Estimates: {rows: 15000000 (2.17GB), cpu: 2.17G, memory: 0B, network: 0B}                                                                                                                                             
           └─ RemoteSource[sourceFragmentIds = [3]]                                                                                                                                                                                  
                  Layout: [orderkey:bigint, concat$gid:varchar, custkey:bigint, concat$gid_1:varchar, $hashvalue_13:bigint, $hashvalue_14:bigint, $hashvalue_15:bigint]                                                              
                                                                                                                                                                                                                                     
 Fragment 3 [SOURCE]                                                                                                                                                                                                                 
     Output layout: [orderkey, concat$gid, custkey, concat$gid_1, $hashvalue_16, $hashvalue_17, $hashvalue_18]                                                                                                                       
     Output partitioning: HASH [concat$gid, concat$gid_1, orderkey][$hashvalue_18]                                                                                                                                                   
     Project[]                                                                                                                                                                                                                       
     │   Layout: [orderkey:bigint, concat$gid:varchar, custkey:bigint, concat$gid_1:varchar, $hashvalue_16:bigint, $hashvalue_17:bigint, $hashvalue_18:bigint]                                                                       
     │   Estimates: {rows: 15000000 (2.17GB), cpu: 2.17G, memory: 0B, network: 0B}                                                                                                                                                   
     │   $hashvalue_16 := combine_hash(combine_hash(bigint '0', COALESCE("$operator$hash_code"("concat$gid"), 0)), COALESCE("$operator$hash_code"("concat$gid_1"), 0))                                                               
     │   $hashvalue_17 := combine_hash(combine_hash(combine_hash(bigint '0', COALESCE("$operator$hash_code"("concat$gid"), 0)), COALESCE("$operator$hash_code"("concat$gid_1"), 0)), COALESCE("$operator$hash_code"("custkey"), 0))  
     │   $hashvalue_18 := combine_hash(combine_hash(combine_hash(bigint '0', COALESCE("$operator$hash_code"("concat$gid"), 0)), COALESCE("$operator$hash_code"("concat$gid_1"), 0)), COALESCE("$operator$hash_code"("orderkey"), 0)) 
     └─ ScanProject[table = hive:tpch:orders]                                                                                                                                                                                        
            Layout: [orderkey:bigint, concat$gid:varchar, custkey:bigint, concat$gid_1:varchar]                                                                                                                                      
            Estimates: {rows: 15000000 (1.79GB), cpu: 343.32M, memory: 0B, network: 0B}/{rows: 15000000 (1.79GB), cpu: 1.79G, memory: 0B, network: 0B}                                                                               
            concat$gid := concat(CAST("orderstatus" AS varchar), VARCHAR '1')                                                                                                                                                        
            concat$gid_1 := concat(CAST("orderstatus" AS varchar), VARCHAR '2')                                                                                                                                                      
            orderkey := orderkey:bigint:REGULAR                                                                                                                                                                                      
            orderstatus := orderstatus:varchar(1):REGULAR                                                                                                                                                                            
            custkey := custkey:bigint:REGULAR    

 Trino version: testversion                                                                                                                                                                                                          
 Fragment 0 [HASH]                                                                                                                                                                                                                   
     Output layout: [concat$gid, concat$gid_1, count, count_2]                                                                                                                                                                       
     Output partitioning: SINGLE []                                                                                                                                                                                                  
     Output[columnNames = [_col0, _col1, _col2, _col3]]                                                                                                                                                                              
     │   Layout: [concat$gid:varchar, concat$gid_1:varchar, count:bigint, count_2:bigint]                                                                                                                                            
     │   Estimates: {rows: ? (?), cpu: 0, memory: 0B, network: 0B}                                                                                                                                                                   
     │   _col0 := concat$gid                                                                                                                                                                                                         
     │   _col1 := concat$gid_1                                                                                                                                                                                                       
     │   _col2 := count                                                                                                                                                                                                              
     │   _col3 := count_2                                                                                                                                                                                                            
     └─ Project[]                                                                                                                                                                                                                    
        │   Layout: [concat$gid_1:varchar, concat$gid:varchar, count_2:bigint, count:bigint]                                                                                                                                         
        │   Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}                                                                                                                                                                
        └─ Aggregate[type = FINAL, keys = [concat$gid_1, concat$gid], hash = [$hashvalue]]                                                                                                                                           
           │   Layout: [concat$gid_1:varchar, concat$gid:varchar, $hashvalue:bigint, count_2:bigint, count:bigint]                                                                                                                   
           │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: 0B}                                                                                                                                                              
           │   count_2 := count("count_3")                                                                                                                                                                                           
           │   count := count("count_4")                                                                                                                                                                                             
           └─ LocalExchange[partitioning = HASH, hashColumn = [$hashvalue], arguments = ["concat$gid_1", "concat$gid"]]                                                                                                              
              │   Layout: [concat$gid_1:varchar, concat$gid:varchar, count_4:bigint, count_3:bigint, $hashvalue:bigint]                                                                                                              
              │   Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}                                                                                                                                                          
              └─ RemoteSource[sourceFragmentIds = [1]]                                                                                                                                                                               
                     Layout: [concat$gid_1:varchar, concat$gid:varchar, count_4:bigint, count_3:bigint, $hashvalue_5:bigint]                                                                                                         
                                                                                                                                                                                                                                     
 Fragment 1 [HASH]                                                                                                                                                                                                                   
     Output layout: [concat$gid_1, concat$gid, count_4, count_3, $hashvalue_6]                                                                                                                                                       
     Output partitioning: HASH [concat$gid_1, concat$gid][$hashvalue_6]                                                                                                                                                              
     Aggregate[type = PARTIAL, keys = [concat$gid_1, concat$gid], hash = [$hashvalue_6]]                                                                                                                                             
     │   Layout: [concat$gid_1:varchar, concat$gid:varchar, $hashvalue_6:bigint, count_4:bigint, count_3:bigint]                                                                                                                     
     │   count_4 := count("orderkey") (mask = orderkey$distinct)                                                                                                                                                                     
     │   count_3 := count("custkey") (mask = custkey$distinct)                                                                                                                                                                       
     └─ MarkDistinct[distinct = [concat$gid_1:varchar, concat$gid:varchar, custkey:bigint], marker = custkey$distinct, hash = [$hashvalue_7]]                                                                                        
        │   Layout: [orderkey:bigint, concat$gid:varchar, custkey:bigint, concat$gid_1:varchar, orderkey$distinct:boolean, $hashvalue_6:bigint, $hashvalue_7:bigint, custkey$distinct:boolean]                                       
        └─ LocalExchange[partitioning = HASH, hashColumn = [$hashvalue_7], arguments = ["concat$gid_1", "concat$gid", "custkey"]]                                                                                                    
           │   Layout: [orderkey:bigint, concat$gid:varchar, custkey:bigint, concat$gid_1:varchar, orderkey$distinct:boolean, $hashvalue_6:bigint, $hashvalue_7:bigint]                                                              
           │   Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}                                                                                                                                                             
           └─ RemoteSource[sourceFragmentIds = [2]]                                                                                                                                                                                  
                  Layout: [orderkey:bigint, concat$gid:varchar, custkey:bigint, concat$gid_1:varchar, orderkey$distinct:boolean, $hashvalue_8:bigint, $hashvalue_9:bigint]                                                           
                                                                                                                                                                                                                                     
 Fragment 2 [HASH]                                                                                                                                                                                                                   
     Output layout: [orderkey, concat$gid, custkey, concat$gid_1, orderkey$distinct, $hashvalue_10, $hashvalue_11]                                                                                                                   
     Output partitioning: HASH [concat$gid_1, concat$gid, custkey][$hashvalue_11]                                                                                                                                                    
     Project[]                                                                                                                                                                                                                       
     │   Layout: [orderkey:bigint, concat$gid:varchar, custkey:bigint, concat$gid_1:varchar, $hashvalue_10:bigint, $hashvalue_11:bigint, orderkey$distinct:boolean]                                                                  
     │   Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}                                                                                                                                                                   
     └─ MarkDistinct[distinct = [concat$gid_1:varchar, concat$gid:varchar, orderkey:bigint], marker = orderkey$distinct, hash = [$hashvalue_12]]                                                                                     
        │   Layout: [orderkey:bigint, concat$gid:varchar, custkey:bigint, concat$gid_1:varchar, $hashvalue_10:bigint, $hashvalue_11:bigint, $hashvalue_12:bigint, orderkey$distinct:boolean]                                         
        └─ LocalExchange[partitioning = HASH, hashColumn = [$hashvalue_12], arguments = ["concat$gid_1", "concat$gid", "orderkey"]]                                                                                                  
           │   Layout: [orderkey:bigint, concat$gid:varchar, custkey:bigint, concat$gid_1:varchar, $hashvalue_10:bigint, $hashvalue_11:bigint, $hashvalue_12:bigint]                                                                 
           │   Estimates: {rows: 15000000 (2.17GB), cpu: 2.17G, memory: 0B, network: 0B}                                                                                                                                             
           └─ RemoteSource[sourceFragmentIds = [3]]                                                                                                                                                                                  
                  Layout: [orderkey:bigint, concat$gid:varchar, custkey:bigint, concat$gid_1:varchar, $hashvalue_13:bigint, $hashvalue_14:bigint, $hashvalue_15:bigint]                                                              
                                                                                                                                                                                                                                     
 Fragment 3 [SOURCE]                                                                                                                                                                                                                 
     Output layout: [orderkey, concat$gid, custkey, concat$gid_1, $hashvalue_16, $hashvalue_17, $hashvalue_18]                                                                                                                       
     Output partitioning: HASH [concat$gid_1, concat$gid, orderkey][$hashvalue_18]                                                                                                                                                   
     Project[]                                                                                                                                                                                                                       
     │   Layout: [orderkey:bigint, concat$gid:varchar, custkey:bigint, concat$gid_1:varchar, $hashvalue_16:bigint, $hashvalue_17:bigint, $hashvalue_18:bigint]                                                                       
     │   Estimates: {rows: 15000000 (2.17GB), cpu: 2.17G, memory: 0B, network: 0B}                                                                                                                                                   
     │   $hashvalue_16 := combine_hash(combine_hash(bigint '0', COALESCE("$operator$hash_code"("concat$gid_1"), 0)), COALESCE("$operator$hash_code"("concat$gid"), 0))                                                               
     │   $hashvalue_17 := combine_hash(combine_hash(combine_hash(bigint '0', COALESCE("$operator$hash_code"("concat$gid_1"), 0)), COALESCE("$operator$hash_code"("concat$gid"), 0)), COALESCE("$operator$hash_code"("custkey"), 0))  
     │   $hashvalue_18 := combine_hash(combine_hash(combine_hash(bigint '0', COALESCE("$operator$hash_code"("concat$gid_1"), 0)), COALESCE("$operator$hash_code"("concat$gid"), 0)), COALESCE("$operator$hash_code"("orderkey"), 0)) 
     └─ ScanProject[table = hive:tpch:orders]                                                                                                                                                                                        
            Layout: [orderkey:bigint, concat$gid:varchar, custkey:bigint, concat$gid_1:varchar]                                                                                                                                      
            Estimates: {rows: 15000000 (1.79GB), cpu: 343.32M, memory: 0B, network: 0B}/{rows: 15000000 (1.79GB), cpu: 1.79G, memory: 0B, network: 0B}                                                                               
            concat$gid := concat(CAST("orderstatus" AS varchar), VARCHAR '1')                                                                                                                                                        
            concat$gid_1 := concat(CAST("orderstatus" AS varchar), VARCHAR '2')                                                                                                                                                      
            orderkey := orderkey:bigint:REGULAR                                                                                                                                                                                      
            orderstatus := orderstatus:varchar(1):REGULAR                                                                                                                                                                            
            custkey := custkey:bigint:REGULAR