content: /**
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you under the Apache License, Version 2.0 (the
   * "License"); you may not use this file except in compliance
   * with the License.  You may obtain a copy of the License at
   *
   *   http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing,
   * software distributed under the License is distributed on an
   * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
   * KIND, either express or implied.  See the License for the
   * specific language governing permissions and limitations
   * under the License.
   */
  package com.datatorrent.apps.logstream;
  
  import java.util.*;
  import java.util.Map.Entry;
  
  import javax.validation.ValidationException;
  import javax.validation.constraints.NotNull;
  
  import com.google.common.collect.Sets;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import org.apache.apex.malhar.lib.codec.KryoSerializableStreamCodec;
  import org.apache.apex.malhar.lib.logs.DimensionObject;
  import org.apache.commons.lang.mutable.MutableDouble;
  
  import com.datatorrent.api.*;
  import com.datatorrent.api.Context.OperatorContext;
  import com.datatorrent.api.Partitioner.PartitioningContext;
  import com.datatorrent.apps.logstream.LogstreamUtil.AggregateOperation;
  import com.datatorrent.apps.logstream.PropertyRegistry.LogstreamPropertyRegistry;
  import com.datatorrent.apps.logstream.PropertyRegistry.PropertyRegistry;
  import com.datatorrent.common.util.BaseOperator;
  import com.datatorrent.netlet.util.DTThrowable;
  
  /**
   * Partitionable dimension operator.
   * Computes specified operators on specified values for specified dimension combinations.
   * Each partition works on a filtered tuple of a log type.
   *
   * @since 0.9.4
   */
  public class DimensionOperator extends BaseOperator implements Partitioner<DimensionOperator>
  {
    /**
     * key: timebucket|timestamp|recordtype|filter|dimensionId|value.operationType
     * value: DimensionObject
     */
    public final transient DefaultOutputPort<Map<String, DimensionObject<String>>> aggregationsOutput = new DefaultOutputPort<Map<String, DimensionObject<String>>>()
    {
      @Override
      public Unifier<Map<String, DimensionObject<String>>> getUnifier()
      {
        DimensionOperatorUnifier unifier = new DimensionOperatorUnifier();
        return unifier;
      }
  
    };
    public final transient DefaultInputPort<Map<String, Object>> in = new DefaultInputPort<Map<String, Object>>()
    {
      @Override
      public void process(Map<String, Object> tuple)
      {
        processTuple(tuple);
      }
  
      @Override
      public StreamCodec<Map<String, Object>> getStreamCodec()
      {
        return new DimensionOperatorStreamCodec();
      }
  
    };
    @NotNull
    private PropertyRegistry<String> registry;
    private static final Logger logger = LoggerFactory.getLogger(DimensionOperator.class);
    private String timeKeyName;
    private long windowWidth = 500;
    private long currentWindowId;
    private transient TimeZone timeZone = TimeZone.getTimeZone("GMT");
    private transient Calendar calendar = new GregorianCalendar(timeZone);
    private int timeBucketFlags;
    private final Map<String, Map<String, Map<AggregateOperation, Number>>> cacheObject = new HashMap<String, Map<String, Map<AggregateOperation, Number>>>();
    private final HashMap<String, Number> recordType = new HashMap<String, Number>();
    private HashMap<Integer, HashMap<String, HashSet<AggregateOperation>>> valueOperations = new HashMap<Integer, HashMap<String, HashSet<AggregateOperation>>>();
    private HashMap<Integer, ArrayList<Integer>> dimensionCombinationList = new HashMap<Integer, ArrayList<Integer>>();
    private transient boolean firstTuple = true;
    private ArrayList<Integer> dimensionCombinations;
    private HashMap<String, HashSet<AggregateOperation>> valueOperationTypes;
    List<String> outTimeBuckets; // represents the list of timebuckets to be emitted when the time bucket changes
  
    @Override
    public void setup(OperatorContext context)
    {
      super.setup(context);
      if (context != null) {
        windowWidth = context.getValue(Context.DAGContext.STREAMING_WINDOW_SIZE_MILLIS);
      }
  
      LogstreamPropertyRegistry.setInstance(registry);
    }
  
    /**
     * Does dimensional computations for each incoming tuple and populates the cache with the computations
     *
     * @param tuple
     */
    protected void processTuple(Map<String, Object> tuple)
    {
      long time;
      if (timeKeyName != null) {
        time = (Long)tuple.get(timeKeyName);
      }
      else {
        time = LogstreamUtil.extractTime(currentWindowId, windowWidth);
      }
      List<String> timeBucketList = getTimeBucketList(time);
  
      if (firstTuple) {
        // populate record type
        extractType(tuple);
        outTimeBuckets = new ArrayList<String>(timeBucketList);
  
        // create all dimension combinations if not specified by user
        if (!dimensionCombinationList.containsKey(recordType.get(LogstreamUtil.LOG_TYPE))) {
          createAllDimensionCombinations();
        }
  
        dimensionCombinations = dimensionCombinationList.get(recordType.get(LogstreamUtil.LOG_TYPE));
        valueOperationTypes = valueOperations.get(recordType.get(LogstreamUtil.LOG_TYPE));
        firstTuple = false;
      }
  
      // temporary validation to ensure that unexpected records do not appear in any partition
      Number receivedLogType = (Number)tuple.get(LogstreamUtil.LOG_TYPE);
      Number receivedFilter = (Number)tuple.get(LogstreamUtil.FILTER);
  
      Number expectedLogType = recordType.get(LogstreamUtil.LOG_TYPE);
      Number expectedFilter = recordType.get(LogstreamUtil.FILTER);
  
      if (!receivedLogType.equals(expectedLogType) || !receivedFilter.equals(expectedFilter)) {
        logger.error("Unexpected tuple");
        logger.error("expected log type = {} received = {}", expectedLogType, receivedLogType);
        logger.error("expected filter = {} received = {}", expectedFilter, receivedFilter);
      }
      else {
        for (String timeBucket : timeBucketList) {
          for (Integer dimensionCombinationId : dimensionCombinations) {
            String dimensionCombination = registry.lookupValue(dimensionCombinationId);
            String[] dimensions = dimensionCombination.split(":");
  
            String dimValueName = new String();
            boolean isBadTuple = false;
            if (dimensions != null) {
              for (String dimension : dimensions) {
                Object dimVal = tuple.get(dimension);
                if (dimVal == null) {
                  logger.error("dimension \"{}\" not found in tuple", dimension);
                  isBadTuple = true;
                  continue;
                }
                if (!dimValueName.isEmpty()) {
                  dimValueName += ",";
                }
                dimValueName += tuple.get(dimension).toString();
              }
            }
  
            if (!isBadTuple) {
              for (Entry<String, HashSet<AggregateOperation>> entry : valueOperationTypes.entrySet()) {
                String valueKeyName = entry.getKey();
                Object value = tuple.get(valueKeyName);
                Number numberValue = LogstreamUtil.extractNumber(value);
                doComputations(timeBucket, dimensionCombinationId, dimValueName, valueKeyName, numberValue);
              }
            }
          }
        }
      }
    }
  
    /**
     * returns list of time buckets for the given timestamp
     *
     * @param time
     * @return
     */
    protected List<String> getTimeBucketList(long time)
    {
  
      calendar.setTimeInMillis(time);
  
      List<String> timeBucketList = new ArrayList<String>();
  
      if ((timeBucketFlags & LogstreamUtil.TIMEBUCKET_YEAR) != 0) {
        timeBucketList.add(String.format("Y|%04d", calendar.get(Calendar.YEAR)));
      }
      if ((timeBucketFlags & LogstreamUtil.TIMEBUCKET_MONTH) != 0) {
        timeBucketList.add(String.format("M|%04d%02d", calendar.get(Calendar.YEAR), calendar.get(Calendar.MONTH) + 1));
      }
      if ((timeBucketFlags & LogstreamUtil.TIMEBUCKET_WEEK) != 0) {
        timeBucketList.add(String.format("W|%04d%02d", calendar.get(Calendar.YEAR), calendar.get(Calendar.WEEK_OF_YEAR)));
      }
      if ((timeBucketFlags & LogstreamUtil.TIMEBUCKET_DAY) != 0) {
        timeBucketList.add(String.format("D|%04d%02d%02d", calendar.get(Calendar.YEAR), calendar.get(Calendar.MONTH) + 1, calendar.get(Calendar.DAY_OF_MONTH)));
      }
      if ((timeBucketFlags & LogstreamUtil.TIMEBUCKET_HOUR) != 0) {
        timeBucketList.add(String.format("h|%04d%02d%02d%02d", calendar.get(Calendar.YEAR), calendar.get(Calendar.MONTH) + 1, calendar.get(Calendar.DAY_OF_MONTH), calendar.get(Calendar.HOUR_OF_DAY)));
      }
      if ((timeBucketFlags & LogstreamUtil.TIMEBUCKET_MINUTE) != 0) {
        timeBucketList.add(String.format("m|%04d%02d%02d%02d%02d", calendar.get(Calendar.YEAR), calendar.get(Calendar.MONTH) + 1, calendar.get(Calendar.DAY_OF_MONTH), calendar.get(Calendar.HOUR_OF_DAY), calendar.get(Calendar.MINUTE)));
      }
      if ((timeBucketFlags & LogstreamUtil.TIMEBUCKET_SECOND) != 0) {
        timeBucketList.add(String.format("s|%04d%02d%02d%02d%02d%02d", calendar.get(Calendar.YEAR), calendar.get(Calendar.MONTH) + 1, calendar.get(Calendar.DAY_OF_MONTH), calendar.get(Calendar.HOUR_OF_DAY), calendar.get(Calendar.MINUTE), calendar.get(Calendar.SECOND)));
      }
  
      return timeBucketList;
    }
  
    /**
     * Does computations for the given dimension and its value names on the given value key name
     *
     * @param timeBucket time bucket
     * @param dimensionCombinationId id of dimension combination
     * @param dimValueName values of the dimension combination
     * @param valueKeyName name of the value key on which operations are performed
     * @param value value of the value key
     */
    private void doComputations(String timeBucket, Integer dimensionCombinationId, String dimValueName, String valueKeyName, Number value)
    {
      StringBuilder sb = new StringBuilder();
      sb.append(timeBucket).append("|").append(recordType.get(LogstreamUtil.LOG_TYPE)).append("|").append(recordType.get(LogstreamUtil.FILTER)).append("|").append(dimensionCombinationId).append("|").append(valueKeyName);
  
      //final key format --> timebucket|type|filter|dimId|val
      //eg: m|201311230108|1|4|10|bytes
      String key = sb.toString();
  
      Map<AggregateOperation, Number> aggregations;
  
      if (cacheObject.containsKey(key)) {
        Map<String, Map<AggregateOperation, Number>> dimValueNames = cacheObject.get(key);
        if (dimValueNames.containsKey(dimValueName)) {
          aggregations = dimValueNames.get(dimValueName);
        }
        else {
          aggregations = new HashMap<AggregateOperation, Number>();
          for (AggregateOperation aggregationType : valueOperationTypes.get(valueKeyName)) {
            aggregations.put(aggregationType, new MutableDouble(0));
          }
  
          dimValueNames.put(dimValueName, aggregations);
        }
      }
      else {
        Map<String, Map<AggregateOperation, Number>> newDimValueNames = new HashMap<String, Map<AggregateOperation, Number>>();
        aggregations = new HashMap<AggregateOperation, Number>();
        for (AggregateOperation aggregationType : valueOperationTypes.get(valueKeyName)) {
          aggregations.put(aggregationType, new MutableDouble(0));
        }
        newDimValueNames.put(dimValueName, aggregations);
        cacheObject.put(key, newDimValueNames);
      }
  
      if (aggregations.containsKey(AggregateOperation.SUM)) {
        MutableDouble aggrVal = (MutableDouble)aggregations.get(AggregateOperation.SUM);
        aggrVal.add(value);
      }
  
      if (aggregations.containsKey(AggregateOperation.COUNT)) {
        MutableDouble aggrVal = (MutableDouble)aggregations.get(AggregateOperation.COUNT);
        aggrVal.add(1);
      }
  
      if (aggregations.containsKey(AggregateOperation.AVERAGE)) {
        double avgVal = aggregations.get(AggregateOperation.AVERAGE).doubleValue();
        double countVal = aggregations.get(AggregateOperation.COUNT).doubleValue();
        double newAvg = ((avgVal * (countVal - 1)) + value.doubleValue()) / countVal;
        aggregations.put(AggregateOperation.AVERAGE, new MutableDouble(newAvg));
      }
  
    }
  
    @Override
    public void beginWindow(long windowId)
    {
      super.beginWindow(windowId);
      currentWindowId = windowId;
    }
  
    @Override
    public void endWindow()
    {
      if (outTimeBuckets == null || outTimeBuckets.isEmpty()) {
        return;
      }
  
      long time = LogstreamUtil.extractTime(currentWindowId, windowWidth);
  
      // get time buckets for current window id
      List<String> timeBucketList = getTimeBucketList(time);
  
      // get list of timebuckets to be emitted and replace them in outTimeBuckets with next time bucket to be emitted
      ArrayList<String> emitTimeBucketList = new ArrayList<String>();
      for (int i = 0; i < timeBucketList.size(); i++) {
        String timeBucket = timeBucketList.get(i);
        if (!timeBucket.equals(outTimeBuckets.get(i))) {
          emitTimeBucketList.add(outTimeBuckets.get(i));
          outTimeBuckets.set(i, timeBucket);
        }
      }
  
      // emit the computations for each time bucket in emitTimeBucketList and remove those buckets from the cache since they are now already processed
      if (!emitTimeBucketList.isEmpty()) {
        ArrayList<String> obsoleteKeys = new ArrayList<String>();
        for (String outTimeStr : emitTimeBucketList) {
          HashMap<String, DimensionObject<String>> outputAggregationsObject;
  
          for (Entry<String, Map<String, Map<AggregateOperation, Number>>> keys : cacheObject.entrySet()) {
            String key = keys.getKey();
            if (key.startsWith(outTimeStr)) {
              Map<String, Map<AggregateOperation, Number>> dimValues = keys.getValue();
  
              for (Entry<String, Map<AggregateOperation, Number>> dimValue : dimValues.entrySet()) {
                String dimValueName = dimValue.getKey();
                Map<AggregateOperation, Number> operations = dimValue.getValue();
  
                outputAggregationsObject = new HashMap<String, DimensionObject<String>>();
  
                for (Entry<AggregateOperation, Number> operation : operations.entrySet()) {
                  AggregateOperation aggrOperationType = operation.getKey();
                  Number aggr = operation.getValue();
  
                  String outKey = key + "." + aggrOperationType.name();
                  DimensionObject<String> outDimObj = new DimensionObject<String>((MutableDouble)aggr, dimValueName);
  
                  outputAggregationsObject.put(outKey, outDimObj);
  
                }
                aggregationsOutput.emit(outputAggregationsObject);
              }
  
              // remove emitted key
              obsoleteKeys.add(key);
            }
          }
  
          for (String key : obsoleteKeys) {
            cacheObject.remove(key);
          }
        }
      }
    }
  
    /**
     * supply the registry object which is used to store and retrieve meta information about each tuple
     *
     * @param registry
     */
    public void setRegistry(PropertyRegistry<String> registry)
    {
      this.registry = registry;
    }
  
    /**
     * Supply the properties to the operator.
     * The properties include type, timebucket, dimensioncombinations, values and operations on them
     * Input includes following properties:
     * type=logtype // input logtype for which the properties are to be set
     * timebucket= time bucket character //time bucket can be one of the following values s(for second)/m(for minute)/h(for hour)/D(for day)/W(for week)/M(for month)/Y(for year)
     * dimensions=a:b:c //colon separated dimension combination combination for which computations are expected
     * values=value.metric[:value.metric] //list of dot concatenated value name and metric separated by colon
     * eg: type=apache,timebucket=m,timebucket=h,dimensions=a:b:c,dimensions=b:c,dimensions=b,dimensions=d,values=x.sum:y.sum:y.avg
     *
     * @param properties
     */
    public void addPropertiesFromString(String[] properties)
    {
      try {
        ArrayList<Integer> dimCombinations = new ArrayList<Integer>();
        HashMap<String, HashSet<AggregateOperation>> valOpTypes = new HashMap<String, HashSet<AggregateOperation>>();
        String type = null;
        // user input example::
        // type=apache,timebucket=m,timebucket=h,dimensions=a:b:c,dimensions=b:c,dimensions=b,dimensions=d,values=x.sum:y.sum:y.avg
        for (String inputs : properties) {
          String[] split = inputs.split("=", 2);
          if (split[0].toLowerCase().equals("timebucket")) {
  
            int timeBucket = LogstreamUtil.extractTimeBucket(split[1]);
            if (timeBucket == 0) {
              logger.error("invalid time bucket", split[1]);
            }
            timeBucketFlags |= timeBucket;
          }
          else if (split[0].toLowerCase().equals("values")) {
            String[] values = split[1].split(":");
            for (String value : values) {
              String[] valueNames = value.split("\\.");
              String valueName = valueNames[0];
              String valueType = valueNames[1];
              if (valueType.toLowerCase().equals("sum")) {
                if (valOpTypes.containsKey(valueName)) {
                  valOpTypes.get(valueName).add(AggregateOperation.SUM);
                }
                else {
                  HashSet<AggregateOperation> valueTypeList = new HashSet<AggregateOperation>();
                  valueTypeList.add(AggregateOperation.SUM);
                  valOpTypes.put(valueName, valueTypeList);
                }
              }
              else if (valueType.equals("avg") || valueType.equals("average")) {
                if (valOpTypes.containsKey(valueName)) {
                  valOpTypes.get(valueName).add(AggregateOperation.AVERAGE);
                  valOpTypes.get(valueName).add(AggregateOperation.COUNT);
                }
                else {
                  HashSet<AggregateOperation> valueTypeList = new HashSet<AggregateOperation>();
                  valueTypeList.add(AggregateOperation.AVERAGE);
                  valueTypeList.add(AggregateOperation.COUNT);
                  valOpTypes.put(valueName, valueTypeList);
                }
              }
              else if (valueType.equals("count")) {
                if (valOpTypes.containsKey(valueName)) {
                  valOpTypes.get(valueName).add(AggregateOperation.COUNT);
                }
                else {
                  HashSet<AggregateOperation> valueTypeList = new HashSet<AggregateOperation>();
                  valueTypeList.add(AggregateOperation.COUNT);
                  valOpTypes.put(valueName, valueTypeList);
                }
              }
            }
          }
          else if (split[0].toLowerCase().equals("type")) {
            type = split[1];
          }
          else if (split[0].toLowerCase().equals("dimensions")) {
            // dimensions
            String dimensions = split[1];
            int dim = registry.bind("DIMENSION", dimensions);
            dimCombinations.add(dim);
          }
          else {
            throw new ValidationException("Invalid input property string " + Arrays.toString(properties));
          }
        }
  
        dimensionCombinationList.put(registry.getIndex(LogstreamUtil.LOG_TYPE, type), dimCombinations);
        valueOperations.put(registry.getIndex(LogstreamUtil.LOG_TYPE, type), valOpTypes);
  
      }
      catch (Exception e) {
        throw new RuntimeException(e);
      }
    }
  
    /**
     * <b>Note:</b> This partitioner does not support parallel partitioning.<br/><br/>
     * {@inheritDoc}
     */
    @Override
    public Collection<Partition<DimensionOperator>> definePartitions(Collection<Partition<DimensionOperator>> partitions, PartitioningContext context)
    {
      ArrayList<Partition<DimensionOperator>> newPartitions = new ArrayList<Partition<DimensionOperator>>();
      String[] filters = registry.list(LogstreamUtil.FILTER);
      int partitionSize;
  
      if (partitions.size() == 1) {
        // initial partitions; functional partitioning
        partitionSize = filters.length;
      }
      else {
        // redo partitions; double the partitions
        partitionSize = partitions.size() * 2;
      }
  
      for (int i = 0; i < partitionSize; i++) {
        try {
          DimensionOperator dimensionOperator = new DimensionOperator();
          dimensionOperator.registry = registry;
          dimensionOperator.timeBucketFlags = timeBucketFlags;
          dimensionOperator.valueOperations = new HashMap<Integer, HashMap<String, HashSet<AggregateOperation>>>(valueOperations);
          dimensionOperator.dimensionCombinationList = new HashMap<Integer, ArrayList<Integer>>(dimensionCombinationList);
  
          Partition<DimensionOperator> partition = new DefaultPartition<DimensionOperator>(dimensionOperator);
          newPartitions.add(partition);
        }
        catch (Throwable ex) {
          DTThrowable.rethrow(ex);
        }
      }
  
      int partitionBits = (Integer.numberOfLeadingZeros(0) - Integer.numberOfLeadingZeros(partitionSize / filters.length - 1));
      int partitionMask = 0;
      if (partitionBits > 0) {
        partitionMask = -1 >>> (Integer.numberOfLeadingZeros(-1)) - partitionBits;
      }
  
      partitionMask = (partitionMask << 16) | 0xffff; // right most 16 bits used for functional partitioning
  
      for (int i = 0; i < newPartitions.size(); i++) {
        Partition<DimensionOperator> partition = newPartitions.get(i);
        String partitionVal = filters[i % filters.length];
        int bits = i / filters.length;
        int filterId = registry.getIndex(LogstreamUtil.FILTER, partitionVal);
        filterId = 0xffff & filterId; // clear out first 16 bits
        int partitionKey = (bits << 16) | filterId; // first 16 bits for dynamic partitioning, last 16 bits for functional partitioning
        logger.debug("partitionKey = {} partitionMask = {}", Integer.toBinaryString(partitionKey), Integer.toBinaryString(partitionMask));
        partition.getPartitionKeys().put(in, new PartitionKeys(partitionMask, Sets.newHashSet(partitionKey)));
      }
  
      return newPartitions;
    }
  
    @Override
    public void partitioned(Map<Integer, Partition<DimensionOperator>> partitions)
    {
    }
  
    /**
     * Time key for dimension computations, if not provided then window timestamp is used
     *
     * @param timeKeyName
     */
    public void setTimeKeyName(String timeKeyName)
    {
      this.timeKeyName = timeKeyName;
    }
  
    /**
     * extracts the meta information about the tuple
     *
     * @param tuple
     */
    private void extractType(Map<String, Object> tuple)
    {
      recordType.put(LogstreamUtil.LOG_TYPE, (Number)tuple.get(LogstreamUtil.LOG_TYPE));
      recordType.put(LogstreamUtil.FILTER, (Number)tuple.get(LogstreamUtil.FILTER));
    }
  
    private void createAllDimensionCombinations()
    {
      logger.info("need to create all dimensions for type {}", recordType.get(LogstreamUtil.LOG_TYPE));
      //TODO create all dim combinations
      // temporary code to skip null pointer
      dimensionCombinationList.put((Integer)recordType.get(LogstreamUtil.LOG_TYPE), new ArrayList<Integer>());
    }
  
    public static class DimensionOperatorStreamCodec extends KryoSerializableStreamCodec<Map<String, Object>>
    {
      private static final Logger logger = LoggerFactory.getLogger(DimensionOperatorStreamCodec.class);
  
      @Override
      public int getPartition(Map<String, Object> o)
      {
        int ret = 0;
  
        int filterId = (Integer)o.get(LogstreamUtil.FILTER);
        int hashCode = o.hashCode();
  
        filterId = 0xffff & filterId; // clear out first 16 bits
  
        ret = (hashCode << 16) | filterId; // first 16 bits represent hashcode, last 16 bits represent filter type
  
        return ret;
  
      }
  
      private static final long serialVersionUID = 1L;
    }
  
  }
  