Print splits count and distribution time in EXPLAIN ANALYZE

[lukasz-stec]

Description

Split distribution time can be a bottleneck in some scenarios. Having it in the explain makes it easier to diagnose without accessing query json. Split count is helpful to analyze query performance when there is either a big number of small splits or there is just one split like with basic jdbc connectors. Split count is visible also in the "Input rows distribution" metric but it is only available in the VERBOSE mode.

Sample output:
query

trino:tpch_sf10_snappy_parquet> EXPLAIN ANALYZE SELECT o.orderkey, c.name
                             -> FROM customer c
                             ->  JOIN orders o ON o.custkey = c.custkey
                             -> WHERE c.nationkey  = 0;

After (has Total split distribution time added to the source stages and and Splits added to the table scans)

                                                                                                 Query Plan                                                                                                  
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
 Trino version: testversion                                                                                                                                                                                  
 Queued: 466.67us, Analysis: 9.62ms, Planning: 38.51ms, Execution: 709.30ms                                                                                                                                  
 Fragment 1 [HASH]                                                                                                                                                                                           
     CPU: 730.14ms, Scheduled: 1.23s, Blocked 12.65s (Input: 4.65s, Output: 0.00ns), Input: 15059916 rows (259.32MB); per task: avg.: 5019972.00 std.dev.: 2112.29, Output: 598637 rows (20.55MB)            
     Peak Memory: 51.03MB, Tasks count: 3; per task: max: 19.34MB                                                                                                                                            
     Output layout: [orderkey, name]                                                                                                                                                                         
     Output partitioning: SINGLE []                                                                                                                                                                          
     InnerJoin[criteria = (custkey_0 = custkey), distribution = PARTITIONED]                                                                                                                                 
     │   Layout: [orderkey:bigint, name:varchar(25)]                                                                                                                                                         
     │   Estimates: {rows: 887411 (27.08MB), cpu: 286.40M, memory: 1.83MB, network: 0B}                                                                                                                      
     │   CPU: 692.00ms (50.96%), Scheduled: 1.01s (34.50%), Blocked: 5.35s (39.84%), Output: 598637 rows (20.55MB)                                                                                           
     │   Left (probe) Input avg.: 1250000.00 rows, Input std.dev.: 29.78%                                                                                                                                    
     │   Right (build) Input avg.: 4993.00 rows, Input std.dev.: 0.81%                                                                                                                                       
     │   Distribution: PARTITIONED                                                                                                                                                                           
     │   dynamicFilterAssignments = {custkey -> #df_451}                                                                                                                                                     
     ├─ RemoteSource[sourceFragmentIds = [2]]                                                                                                                                                                
     │      Layout: [orderkey:bigint, custkey_0:bigint]                                                                                                                                                      
     │      CPU: 24.00ms (1.77%), Scheduled: 207.00ms (7.08%), Blocked: 2.08s (15.44%), Output: 15000000 rows (257.49MB)                                                                                     
     │      Input avg.: 1250000.00 rows, Input std.dev.: 29.78%                                                                                                                                              
     └─ LocalExchange[partitioning = HASH, arguments = [custkey::bigint]]                                                                                                                                    
        │   Layout: [custkey:bigint, name:varchar(25)]                                                                                                                                                       
        │   Estimates: {rows: 60000 (1.83MB), cpu: 1.83M, memory: 0B, network: 0B}                                                                                                                           
        │   CPU: 6.00ms (0.44%), Scheduled: 6.00ms (0.21%), Blocked: 2.68s (19.97%), Output: 59916 rows (1.83MB)                                                                                             
        │   Input avg.: 4993.00 rows, Input std.dev.: 173.21%                                                                                                                                                
        └─ RemoteSource[sourceFragmentIds = [3]]                                                                                                                                                             
               Layout: [custkey:bigint, name:varchar(25)]                                                                                                                                                    
               CPU: 0.00ns (0.00%), Scheduled: 0.00ns (0.00%), Blocked: 2.57s (19.11%), Output: 59916 rows (1.83MB)                                                                                          
               Input avg.: 4993.00 rows, Input std.dev.: 173.21%                                                                                                                                             
                                                                                                                                                                                                             
 Fragment 2 [SOURCE]                                                                                                                                                                                         
     CPU: 567.68ms, Scheduled: 1.51s, Blocked 757.20ms (Input: 0.00ns, Output: 757.20ms), Input: 15000000 rows (257.49MB); per task: avg.: 5000000.00 std.dev.: 1850443.55, Output: 15000000 rows (257.49MB) 
     Peak Memory: 106.40MB, Tasks count: 3; per task: max: 62.60MB                                                                                                                                           
     Total split distribution time: 0.19ms                                                                                                                                                                   
     Output layout: [orderkey, custkey_0]                                                                                                                                                                    
     Output partitioning: HASH [custkey_0]                                                                                                                                                                   
     ScanFilter[table = hive:tpch_sf10_snappy_parquet:orders, dynamicFilters = {custkey_0 = #df_451}]                                                                                                        
         Layout: [orderkey:bigint, custkey_0:bigint]                                                                                                                                                         
         Estimates: {rows: 15000000 (257.49MB), cpu: 257.49M, memory: 0B, network: 0B}/{rows: 15000000 (257.49MB), cpu: 257.49M, memory: 0B, network: 0B}                                                    
         CPU: 564.00ms (41.53%), Scheduled: 1.51s (51.71%), Blocked: 757.00ms (5.63%), Output: 15000000 rows (257.49MB)                                                                                      
         Input avg.: 750000.00 rows, Input std.dev.: 161.05%                                                                                                                                                 
         custkey_0 := custkey:bigint:REGULAR                                                                                                                                                                 
         orderkey := orderkey:bigint:REGULAR                                                                                                                                                                 
         Input: 15000000 rows (257.49MB), Filtered: 0.00%, Physical input: 128.42MB, Physical input time: 283.51ms, Splits: 20                                                                               
         Dynamic filters:                                                                                                                                                                                    
             - df_451, [ SortedRangeSet[type=bigint, ranges=59916, {[29], ..., [1499984]}] ], collection time=321.59ms                                                                                       
                                                                                                                                                                                                             
 Fragment 3 [SOURCE]                                                                                                                                                                                         
     CPU: 72.17ms, Scheduled: 189.75ms, Blocked 0.00ns (Input: 0.00ns, Output: 0.00ns), Input: 1500000 rows (58.65MB); per task: avg.: 500000.00 std.dev.: 707106.78, Output: 59916 rows (1.83MB)            
     Peak Memory: 2.19MB, Tasks count: 3; per task: max: 15.23MB                                                                                                                                             
     Total split distribution time: 1.23ms                                                                                                                                                                   
     Amount of input data processed by the workers for this stage might be skewed                                                                                                                            
     Output layout: [custkey, name]                                                                                                                                                                          
     Output partitioning: HASH [custkey]                                                                                                                                                                     
     ScanFilterProject[table = hive:tpch_sf10_snappy_parquet:customer, filterPredicate = (nationkey = bigint '0')]                                                                                           
         Layout: [custkey:bigint, name:varchar(25)]                                                                                                                                                          
         Estimates: {rows: 1500000 (45.78MB), cpu: 58.65M, memory: 0B, network: 0B}/{rows: 60000 (1.83MB), cpu: 58.65M, memory: 0B, network: 0B}/{rows: 60000 (1.83MB), cpu: 1.83M, memory: 0B, network: 0B} 
         CPU: 72.00ms (5.30%), Scheduled: 190.00ms (6.50%), Blocked: 0.00ns (0.00%), Output: 59916 rows (1.83MB)                                                                                             
         Input avg.: 375000.00 rows, Input std.dev.: 173.21%                                                                                                                                                 
         nationkey := nationkey:bigint:REGULAR                                                                                                                                                               
         custkey := custkey:bigint:REGULAR                                                                                                                                                                   
         name := name:varchar(25):REGULAR                                                                                                                                                                    
         Input: 1500000 rows (58.65MB), Filtered: 96.01%, Physical input: 13.55MB, Physical input time: 24.44ms, Splits: 4  

Before

 Trino version: testversion                                                                                                                                                                                  
 Queued: 8.97ms, Analysis: 135.42ms, Planning: 428.38ms, Execution: 2.46s                                                                                                                                    
 Fragment 1 [HASH]                                                                                                                                                                                           
     CPU: 4.22s, Scheduled: 9.12s, Blocked 29.97s (Input: 11.29s, Output: 0.00ns), Input: 15059916 rows (259.32MB); per task: avg.: 5019972.00 std.dev.: 2112.29, Output: 598637 rows (20.55MB)              
     Peak Memory: 67.12MB, Tasks count: 3; per task: max: 23.94MB                                                                                                                                            
     Output layout: [orderkey, name]                                                                                                                                                                         
     Output partitioning: SINGLE []                                                                                                                                                                          
     InnerJoin[criteria = (custkey_0 = custkey), distribution = PARTITIONED]                                                                                                                                 
     │   Layout: [orderkey:bigint, name:varchar(25)]                                                                                                                                                         
     │   Estimates: {rows: 887411 (27.08MB), cpu: 286.40M, memory: 1.83MB, network: 0B}                                                                                                                      
     │   CPU: 4.12s (70.37%), Scheduled: 8.71s (58.15%), Blocked: 12.63s (37.81%), Output: 598637 rows (20.55MB)                                                                                             
     │   Left (probe) Input avg.: 1250000.00 rows, Input std.dev.: 22.46%                                                                                                                                    
     │   Right (build) Input avg.: 4993.00 rows, Input std.dev.: 0.81%                                                                                                                                       
     │   Distribution: PARTITIONED                                                                                                                                                                           
     │   dynamicFilterAssignments = {custkey -> #df_451}                                                                                                                                                     
     ├─ RemoteSource[sourceFragmentIds = [2]]                                                                                                                                                                
     │      Layout: [orderkey:bigint, custkey_0:bigint]                                                                                                                                                      
     │      CPU: 60.00ms (1.02%), Scheduled: 356.00ms (2.38%), Blocked: 3.92s (11.75%), Output: 15000000 rows (257.49MB)                                                                                     
     │      Input avg.: 1250000.00 rows, Input std.dev.: 22.46%                                                                                                                                              
     └─ LocalExchange[partitioning = HASH, arguments = [custkey::bigint]]                                                                                                                                    
        │   Layout: [custkey:bigint, name:varchar(25)]                                                                                                                                                       
        │   Estimates: {rows: 60000 (1.83MB), cpu: 1.83M, memory: 0B, network: 0B}                                                                                                                           
        │   CPU: 25.00ms (0.43%), Scheduled: 29.00ms (0.19%), Blocked: 6.28s (18.81%), Output: 59916 rows (1.83MB)                                                                                           
        │   Input avg.: 4993.00 rows, Input std.dev.: 173.21%                                                                                                                                                
        └─ RemoteSource[sourceFragmentIds = [3]]                                                                                                                                                             
               Layout: [custkey:bigint, name:varchar(25)]                                                                                                                                                    
               CPU: 2.00ms (0.03%), Scheduled: 2.00ms (0.01%), Blocked: 7.37s (22.06%), Output: 59916 rows (1.83MB)                                                                                          
               Input avg.: 4993.00 rows, Input std.dev.: 173.21%                                                                                                                                             
                                                                                                                                                                                                             
 Fragment 2 [SOURCE]                                                                                                                                                                                         
     CPU: 1.38s, Scheduled: 4.96s, Blocked 3.20s (Input: 0.00ns, Output: 3.20s), Input: 15000000 rows (257.49MB); per task: avg.: 5000000.00 std.dev.: 1850443.55, Output: 15000000 rows (257.49MB)          
     Peak Memory: 175.26MB, Tasks count: 3; per task: max: 71.39MB                                                                                                                                           
     Output layout: [orderkey, custkey_0]                                                                                                                                                                    
     Output partitioning: HASH [custkey_0]                                                                                                                                                                   
     ScanFilter[table = hive:tpch_sf10_snappy_parquet:orders, dynamicFilters = {custkey_0 = #df_451}]                                                                                                        
         Layout: [orderkey:bigint, custkey_0:bigint]                                                                                                                                                         
         Estimates: {rows: 15000000 (257.49MB), cpu: 257.49M, memory: 0B, network: 0B}/{rows: 15000000 (257.49MB), cpu: 257.49M, memory: 0B, network: 0B}                                                    
         CPU: 1.38s (23.57%), Scheduled: 4.96s (33.14%), Blocked: 3.20s (9.57%), Output: 15000000 rows (257.49MB)                                                                                            
         Input avg.: 750000.00 rows, Input std.dev.: 161.05%                                                                                                                                                 
         custkey_0 := custkey:bigint:REGULAR                                                                                                                                                                 
         orderkey := orderkey:bigint:REGULAR                                                                                                                                                                 
         Input: 15000000 rows (257.49MB), Filtered: 0.00%, Physical input: 128.42MB, Physical input time: 172.37ms                                                                                           
         Dynamic filters:                                                                                                                                                                                    
             - df_451, [ SortedRangeSet[type=bigint, ranges=59916, {[29], ..., [1499984]}] ], collection time=1.38s                                                                                          
                                                                                                                                                                                                             
 Fragment 3 [SOURCE]                                                                                                                                                                                         
     CPU: 268.10ms, Scheduled: 918.06ms, Blocked 0.00ns (Input: 0.00ns, Output: 0.00ns), Input: 1500000 rows (58.65MB); per task: avg.: 500000.00 std.dev.: 707106.78, Output: 59916 rows (1.83MB)           
     Peak Memory: 2.19MB, Tasks count: 3; per task: max: 14.25MB                                                                                                                                             
     Amount of input data processed by the workers for this stage might be skewed                                                                                                                            
     Output layout: [custkey, name]                                                                                                                                                                          
     Output partitioning: HASH [custkey]                                                                                                                                                                     
     ScanFilterProject[table = hive:tpch_sf10_snappy_parquet:customer, filterPredicate = (nationkey = bigint '0')]                                                                                           
         Layout: [custkey:bigint, name:varchar(25)]                                                                                                                                                          
         Estimates: {rows: 1500000 (45.78MB), cpu: 58.65M, memory: 0B, network: 0B}/{rows: 60000 (1.83MB), cpu: 58.65M, memory: 0B, network: 0B}/{rows: 60000 (1.83MB), cpu: 1.83M, memory: 0B, network: 0B} 
         CPU: 268.00ms (4.58%), Scheduled: 917.00ms (6.12%), Blocked: 0.00ns (0.00%), Output: 59916 rows (1.83MB)                                                                                            
         Input avg.: 375000.00 rows, Input std.dev.: 173.21%                                                                                                                                                 
         name := name:varchar(25):REGULAR                                                                                                                                                                    
         custkey := custkey:bigint:REGULAR                                                                                                                                                                   
         nationkey := nationkey:bigint:REGULAR                                                                                                                                                               
         Input: 1500000 rows (58.65MB), Filtered: 96.01%, Physical input: 13.55MB, Physical input time: 19.15ms  
``
<!-- Provide details that help an engineer who is unfamiliar with this part of the code. -->
## Additional context and related issues



<!-- Mark the appropriate option with an (x). Propose a release note if you can.
More info at https://trino.io/development/process#release-note -->
## Release notes

( ) 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.
(X ) Release notes are required, with the following suggested text:

```markdown
## Section
* Include split count and total split distribution time in the `EXPLAIN ANALYZE` output

[wendigo]

@lukasz-stec can you show output before and after?

[lukasz-stec]

@lukasz-stec can you show output before and after?

Sure, added to the desciption

[sopel39]

Where is splits count?

[lukasz-stec]

Where is splits count?

For table scans only at the end of the input row:
Input: 1500000 rows (58.65MB), ..., Splits: 4