#PARSER PQL
#METADATA TimeInterval
#METADATA Latency
#DROPALLQUERIES
#DROPALLSINKS
#DROPALLSOURCES
#PARALLELIZATION INTER_OPERATOR 4 AUTO

#DEFINE QUERY1
#DEFINE QUERY2

///#DEFINE RUNTEST 1

#IFDEF RUNTEST
#IF RUNTEST == "2"
#UNDEF QUERY2
#ENDIF 

#IF RUNTEST == "3"
#UNDEF QUERY1
#ENDIF 
#ENDIF

#INCLUDE ${WORKSPACEPROJECT}/GC_BASE_CONSTANTS.qry

#IFDEF RUNTEST
#DEFINE OutputPath O:/${NOW}_TEST_${RUNTEST}/
#ELSE
#DEFINE OutputPath O:/${NOW}/
#ENDIF

#INCLUDE ${WORKSPACEPROJECT}/Source.qry

#IFDEF QUERY1
#QNAME QUERY1
#INCLUDE ${WORKSPACEPROJECT}/Query1Part1.pql

top10_pre = TOPK({
            	k = 10,
            	scoringfunction = 'count',
            	TIEWITHTIMESTAMP = true                                                  
          	},
          	num_trips_per_cellpair
        )

/// Merge all top k with same time stamp
top10 = CHANGEDETECT({ATTR = ['TimeInterval.start','TimeInterval.end'], SENDLASTOFSAMEOBJECTS = true}, top10_pre)
        
mapped10 = MAP({
            expressions = [
            	['format(elementAt(trigger,2),"YYYY-MM-DD HH:mm:ss")','pickup_datetime'],
            	['format(elementAt(trigger,3),"YYYY-MM-DD HH:mm:ss"))','dropoff_datetime']
				#LOOP i 0 upto 9
              		,['concat(concat(toInteger(elementAt(topk[${i}],0)/${ID_LEFTSHIFT}), "."), toInteger(elementAt(topk[${i}],0)%${ID_LEFTSHIFT}))','start_cell_id_${i}']
             		,['concat(concat(toInteger(elementAt(topk[${i}],1)/${ID_LEFTSHIFT}), "."), toInteger(elementAt(topk[${i}],1)%${ID_LEFTSHIFT}))','end_cell_id_${i}']
				#ENDLOOP
				,['Latency.latency', 'delay']
				,['format(toDate(TimeInterval.start),"YYYY-MM-DD HH:mm:ss")','start']
				,['format(toDate(TimeInterval.end),"YYYY-MM-DD HH:mm:ss")','end']
            ],
                ALLOWNULL = true,
                SUPPRESSERRORS = true        
          }, 
          CALCLATENCY(top10)
          )
	        
out = CSVFILESINK({FILENAME = '${OUTPUTPATH}/Query1_res.csv', SINK = 'Q1', NULLVALUETEXT = 'null', OPTIONS = [['WRITEDELAYSIZE', '${BUFFERSIZE}'],['createdir','true']]}, 
        mapped10
       )
        
#ENDIF
#IFDEF QUERY2
#QNAME QUERY2

#INCLUDE ${WORKSPACEPROJECT}/Query2Part1.pql

top10 = TOPK({	k = 10,
        	    	scoringfunction = 'profitability',
            		TIEWITHTIMESTAMP = true
          		},
          		profitability
        	)
        
	    

mapedtop10 = MAP({
            expressions = [
            	['format(elementAt(trigger,0),"YYYY-MM-DD HH:mm:ss"))','pickup_datetime'],
            	['format(elementAt(trigger,1),"YYYY-MM-DD HH:mm:ss"))','dropoff_datetime']
				#LOOP i 0 upto 9
					,['concat(concat(toInteger(elementAt(topk[${i}],2)/${ID_LEFTSHIFT}), "."), toInteger(elementAt(topk[${i}],2)%${ID_LEFTSHIFT}))','profitable_cell_id_${i}']
              		,['toInteger(elementAt(topk[${i}],3))','empty_taxies_in_cell_${i}']
              		,['toDouble(elementAt(topk[${i}],4))','median_profit_in_cell_${i}']
              		,['toDouble(elementAt(topk[${i}],5))','profitability_of_cell_${i}']
				#ENDLOOP
				,['Latency.latency', 'delay']
            ],
                ALLOWNULL = true,
                SUPPRESSERRORS = true        
          },
          CALCLATENCY(top10)
         )

out = CSVFILESINK({FILENAME = '${OUTPUTPATH}/Query2_res.csv', SINK = 'Q2', NULLVALUETEXT = 'null', OPTIONS = [['WRITEDELAYSIZE', '${BUFFERSIZE}'],['createdir','true']]}, mapedtop10)

#ENDIF

#IFDEF QUERY1
#STARTQUERY QUERY1
#ENDIF

#IFDEF QUERY2
#STARTQUERY QUERY2
#ENDIF