Query Fails when distinct and limit operators are present

[vinay-kl]

trino> select distinct(business_unit) from bidb.dim_style LIMIT 1001;

Query 20220404_120710_31324_mzjme, FAILED, 45 nodes
Splits: 81 total, 29 done (35.80%)
CPU Time: 0.1s total, 15.6K rows/s, 7.41MB/s, 17% active
Per Node: 0.0 parallelism, 152 rows/s, 74.1KB/s
Parallelism: 0.4
Peak Memory: 899KB
0.30 [2.05K rows, 994KB] [6.87K rows/s, 3.26MB/s]

Query 20220404_120710_31324_mzjme failed: Internal error
com.google.common.base.VerifyException
    at com.google.common.base.Verify.verify(Verify.java:102)
    at io.trino.operator.DistinctLimitOperator.getOutput(DistinctLimitOperator.java:205)
    at io.trino.operator.Driver.processInternal(Driver.java:388)
    at io.trino.operator.Driver.lambda$processFor$9(Driver.java:292)
    at io.trino.operator.Driver.tryWithLock(Driver.java:693)
    at io.trino.operator.Driver.processFor(Driver.java:285)
    at io.trino.execution.SqlTaskExecution$DriverSplitRunner.processFor(SqlTaskExecution.java:1092)
    at io.trino.execution.executor.PrioritizedSplitRunner.process(PrioritizedSplitRunner.java:163)
    at io.trino.execution.executor.TaskExecutor$TaskRunner.run(TaskExecutor.java:488)
    at io.trino.$gen.Trino_374____20220403_075730_2.run(Unknown Source)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
    at java.base/java.lang.Thread.run(Thread.java:829)

`2022-04-04T12:07:11.230Z    ERROR    transaction-finishing-497    io.trino.transaction.InMemoryTransactionManager    Connector threw exception on abort
java.lang.IllegalStateException: Tried to rollback buffered metastore operations after transaction has been committed/aborted
    at io.trino.plugin.hive.metastore.SemiTransactionalHiveMetastore.rollback(SemiTransactionalHiveMetastore.java:1252)
    at io.trino.plugin.hive.HiveMetadata.rollback(HiveMetadata.java:3422)
    at io.trino.plugin.hive.HiveTransactionManager.lambda$rollback$1(HiveTransactionManager.java:70)
    at java.base/java.util.Optional.ifPresent(Optional.java:183)
    at io.trino.plugin.hive.HiveTransactionManager.rollback(HiveTransactionManager.java:68)
    at io.trino.plugin.hive.HiveConnector.rollback(HiveConnector.java:213)
    at io.trino.transaction.InMemoryTransactionManager$TransactionMetadata$ConnectorTransactionMetadata.abort(InMemoryTransactionManager.java:630)
    at io.trino.transaction.InMemoryTransactionManager$TransactionMetadata.safeAbort(InMemoryTransactionManager.java:545)
    at io.trino.transaction.InMemoryTransactionManager$TransactionMetadata.lambda$abortInternal$9(InMemoryTransactionManager.java:536)
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
    at com.google.common.util.concurrent.TrustedListenableFutureTask$TrustedFutureInterruptibleTask.runInterruptibly(TrustedListenableFutureTask.java:131)
    at com.google.common.util.concurrent.InterruptibleTask.run(InterruptibleTask.java:74)
    at com.google.common.util.concurrent.TrustedListenableFutureTask.run(TrustedListenableFutureTask.java:82)
    at io.airlift.concurrent.BoundedExecutor.drainQueue(BoundedExecutor.java:80)
    at io.trino.$gen.Trino_374____20220403_075729_2.run(Unknown Source)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
    at java.base/java.lang.Thread.run(Thread.java:829)

[vinay-kl]

slack thread link https://trinodb.slack.com/archives/CGB0QHWSW/p1648895226302229

[pettyjamesm]

Can you provide the EXPLAIN output for the query? Does this failure occur reliably or does it sometimes succeed?

[pettyjamesm]

Based on the stacktrace and doing some digging, it seems like this failure mode would require GroupByIdBlock to report an incorrect value for getGroupCount(). It looks like NoChannelGroupByHash could report a group count of 0 even if it previously reported 1, which seems incorrect to me, but shouldn't be able to trigger this particular assertion to fail and doesn't appear to be used for distinct limits anyway.

More likely, something in this commit from #10965 which introduced the low cardinality dictionary handling (and was introduced in Trino 373) is causing the behavioral inconsistency with regards to GroupByIdBlock#getGroupCount() but I'm not sure exactly how that's happening. The code is a little tricky to follow. cc @skrzypo987

[vinay-kl]

trino> explain select distinct(business_unit) from bidb.dim_style LIMIT 1001;
                                                                         Query Plan                                                                          
-------------------------------------------------------------------------------------------------------------------------------------------------------------
 Fragment 0 [SINGLE]                                                                                                                                         
     Output layout: [business_unit]                                                                                                                          
     Output partitioning: SINGLE []                                                                                                                          
     Stage Execution Strategy: UNGROUPED_EXECUTION                                                                                                           
     Output[business_unit]                                                                                                                                   
     │   Layout: [business_unit:varchar]                                                                                                                     
     │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}                                                                                             
     └─ Project[]                                                                                                                                            
        │   Layout: [business_unit:varchar]                                                                                                                  
        │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}                                                                                          
        └─ DistinctLimit[1001][$hashvalue]                                                                                                                   
           │   Layout: [business_unit:varchar, $hashvalue:bigint]                                                                                            
           └─ LocalExchange[SINGLE] ()                                                                                                                       
              │   Layout: [business_unit:varchar, $hashvalue:bigint]                                                                                         
              │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}                                                                                    
              └─ RemoteSource[1]                                                                                                                             
                     Layout: [business_unit:varchar, $hashvalue_0:bigint]                                                                                    
                                                                                                                                                             
 Fragment 1 [SOURCE]                                                                                                                                         
     Output layout: [business_unit, $hashvalue_1]                                                                                                            
     Output partitioning: SINGLE []                                                                                                                          
     Stage Execution Strategy: UNGROUPED_EXECUTION                                                                                                           
     DistinctLimitPartial[1001][$hashvalue_1]                                                                                                                
     │   Layout: [business_unit:varchar, $hashvalue_1:bigint]                                                                                                
     └─ ScanProject[table = hive:bidb:dim_style buckets=12, grouped = false]                                                                                 
            Layout: [business_unit:varchar, $hashvalue_1:bigint]                                                                                             
            Estimates: {rows: 12165183 (332.12MB), cpu: 227.71M, memory: 0B, network: 0B}/{rows: 12165183 (332.12MB), cpu: 559.83M, memory: 0B, network: 0B} 
            $hashvalue_1 := combine_hash(bigint '0', COALESCE("$operator$hash_code"("business_unit"), 0))                                                    
            business_unit := business_unit:string:REGULAR

[findepi]

cc @sopel39 @lukasz-stec

[sopel39]

I think I know what the problem is.
Previously (without dictionary support) positions in input blocks were assigned group id incrementally. Now it's not necessarily the case, e.g:

input | new_group_id
a     | 3
b     | 1
c     | 2 

might happen. This break assumption in DistinctLimitOperator (where it simply fails), but I also think causes potential correctness issue in MarkDistinctHash unfortunately.

[sopel39]

I'm labeling that as release blocker as it's correctness issue