#DEFINE NUM_FRAGMENTS_Q2_AGG 32
#DEFINE NUM_FRAGMENTS_Q2_JOIN 32
#DEFINE Q2_BUFFERSIZE 2000000

#DEFINE TAXIS_BUFFERSIZE 1000000
#INTEROPERATORPARALLELIZATION  profit ${NUM_FRAGMENTS_Q2_AGG} ${Q2_BUFFERSIZE} GroupedAggregateMultithreadedTransformationStrategy
#INTEROPERATORPARALLELIZATION  LAST_TAXI_EVENT ${NUM_FRAGMENTS_Q2_AGG} ${Q2_BUFFERSIZE} GroupedAggregateMultithreadedTransformationStrategy
///#INTEROPERATORPARALLELIZATION  EMPTY_TAXIS ${NUM_FRAGMENTS_Q2_AGG} ${Q2_BUFFERSIZE} GroupedAggregateMultithreadedTransformationStrategy
#INTEROPERATORPARALLELIZATION  join ${NUM_FRAGMENTS_Q2_JOIN} ${Q2_BUFFERSIZE} JoinMultithreadedTransformationStrategy
#ADDQUERY
query2_start = BUFFER({
                  name="Q2_BUFFER",
                  threaded = true,
                  drainatclose = false,
                  maxbuffersize = ${Q2_BUFFERSIZE}                  
                },
                taxi
              )

cell_id_calc = MAP({
                    expressions = [
						'medallion','pickup_datetime', 'dropoff_datetime',				                    						                      
						['toLong(pickup_cell_x * ${ID_LEFTSHIFT} + pickup_cell_y)', 'pickup_cell_id'],
                        ['toLong(dropoff_cell_x * ${ID_LEFTSHIFT} + dropoff_cell_y)', 'dropoff_cell_id'],				                    	
                        'fare_amount','tip_amount'
                    ]                
                  },
                  query2_start
                ) 
			
/// empty taxis
empty_taxis_windowed = TIMEWINDOW({
                            size = [30, 'MINUTES']                                                                                    
                          },
                          cell_id_calc
                        )
                      
last_taxi_event = AGGREGATE({
                      id = 'LAST_TAXI_EVENT',
                      name = 'LAST_TAXI_EVENT',
                      aggregations = [
                        ['LAST', 'pickup_datetime', 'pickup_datetime','Date'],
                        ['LAST', 'dropoff_datetime', 'dropoff_datetime','Date'],
                        ['LAST', 'dropoff_cell_id', 'l_dropoff_cell_id', 'Long']
                      ],
                      group_by = ['medallion']                                        
                    },
                    empty_taxis_windowed
                  )
        
empty_taxis_count = AGGREGATE({
                        id = 'EMPTY_TAXIS',
                        name = 'EMPTY_TAXIS',
                        aggregations = [
                          ['LAST', 'pickup_datetime', 'pickup_datetime','Date'],
                          ['LAST', 'dropoff_datetime', 'dropoff_datetime','Date'],
                          ['COUNT', 'medallion', 'empty_taxis', 'Long']
                        ]                        ,
                        group_by = ['l_dropoff_cell_id']                                                
                      },
                      last_taxi_event
                    )

/// profit
profit_windowed = TIMEWINDOW({
                      size = [15, 'MINUTES']                                        
                    },
                    cell_id_calc
                  )
                  
fare_above_zero =  SELECT({
                        predicate = 'fare_amount >= 0 && tip_amount >= 0'                            
                      },
                      profit_windowed
                    )  
              
profit_fare_plus_tip = MAP({
                            expressions = [
								'pickup_datetime', 'pickup_cell_id',								                              								                              
								['fare_amount + tip_amount', 'profit']
                            ]                                                    
                          },
                          fare_above_zero
                        )

profit_median = AGGREGATE({
                    id = 'profit',
                    name = 'profit',
                    aggregations = [['MEDIAN', 'profit', 'profit']],
                    group_by = ['pickup_cell_id'],
                    fastgrouping = true                                    
                  },
                  profit_fare_plus_tip
                )

/// together
profit_joined_empty_taxis = JOIN({
                                id = 'Join',
                                name= 'Join',
                                predicate = 'pickup_cell_id = l_dropoff_cell_id'                                                            
                              },
                              empty_taxis_count,
                              profit_median
                            )

profitability = MAP({
                    expressions = [
						'pickup_datetime', 'dropoff_datetime', 												
						['l_dropoff_cell_id','dropoff_cell_id'], 'empty_taxis', 'profit',						                      						                      
						['profit/empty_taxis', 'profitability']
                    ]                                    
                  },
                  profit_joined_empty_taxis
                )
