private Histogram getHistogramByDayBucketing ( SalesforceConnector connector , String entity , String watermarkColumn , Partition partition ) { Histogram histogram = new Histogram ( ) ; Calendar calendar = new GregorianCalendar ( ) ; Date startDate = Utils . toDate ( partition . getLowWatermark ( ) , Partitioner . WATERMARKTIMEFORMAT ) ; calendar . setTime ( startDate ) ; int startYear = calendar . get ( Calendar . YEAR ) ; String lowWatermarkDate = Utils . dateToString ( startDate , SalesforceExtractor . SALESFORCE_TIMESTAMP_FORMAT ) ; Date endDate = Utils . toDate ( partition . getHighWatermark ( ) , Partitioner . WATERMARKTIMEFORMAT ) ; calendar . setTime ( endDate ) ; int endYear = calendar . get ( Calendar . YEAR ) ; String highWatermarkDate = Utils . dateToString ( endDate , SalesforceExtractor . SALESFORCE_TIMESTAMP_FORMAT ) ; Map < String , String > values = new HashMap <> ( ) ; values . put ( "table" , entity ) ; values . put ( "column" , watermarkColumn ) ; StrSubstitutor sub = new StrSubstitutor ( values ) ; for ( int year = startYear ; year <= endYear ; year ++ ) { if ( year == startYear ) { values . put ( "start" , lowWatermarkDate ) ; values . put ( "greater" , partition . isLowWatermarkInclusive ( ) ? ">=" : ">" ) ; } else { values . put ( "start" , getDateString ( year ) ) ; values . put ( "greater" , ">=" ) ; } if ( year == endYear ) { values . put ( "end" , highWatermarkDate ) ; values . put ( "less" , partition . isHighWatermarkInclusive ( ) ? "<=" : "<" ) ; } else { values . put ( "end" , getDateString ( year + 1 ) ) ; values . put ( "less" , "<" ) ; } String query = sub . replace ( DAY_PARTITION_QUERY_TEMPLATE ) ; log . info ( "Histogram query: " + query ) ; histogram . add ( parseDayBucketingHistogram ( getRecordsForQuery ( connector , query ) ) ) ; } return histogram ; }
Get a histogram with day granularity buckets .