/*

  * 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 org.apache.giraph.block_app.reducers.map;

 import java.io.DataInput;

 import java.io.DataOutput;

 import java.io.IOException;

 import java.util.Iterator;

 import org.apache.commons.lang3.tuple.MutablePair;

 import org.apache.commons.lang3.tuple.Pair;

 import org.apache.giraph.block_app.framework.api.BlockMasterApi;

 import org.apache.giraph.block_app.framework.api.CreateReducersApi;

 import org.apache.giraph.block_app.framework.api.CreateReducersApi.CreateReducerFunctionApi;

 import org.apache.giraph.block_app.framework.piece.global_comm.BroadcastHandle;

 import org.apache.giraph.block_app.framework.piece.global_comm.ReducerHandle;

 import org.apache.giraph.block_app.framework.piece.global_comm.map.BroadcastMapHandle;

 import org.apache.giraph.block_app.framework.piece.global_comm.map.ReducerMapHandle;

 import org.apache.giraph.master.MasterGlobalCommUsage;

 import org.apache.giraph.reducers.ReduceOperation;

 import org.apache.giraph.types.ops.PrimitiveIdTypeOps;

 import org.apache.giraph.types.ops.PrimitiveTypeOps;

 import org.apache.giraph.types.ops.TypeOpsUtils;

 import org.apache.giraph.types.ops.collections.Basic2ObjectMap;

 import org.apache.giraph.types.ops.collections.WritableWriter;

 import org.apache.giraph.utils.WritableUtils;

 import org.apache.giraph.worker.WorkerBroadcastUsage;

 import org.apache.hadoop.io.Writable;

 import org.apache.hadoop.io.WritableComparable;

 /**

  * Efficient generic primitive map of values reduce operation.

  * (it is BasicMap Reduce, not to be confused with MapReduce)

  *

  * @param <K> Key type

  * @param <S> Single value type

  * @param <R> Reduced value type

  */

 public class BasicMapReduce<K extends WritableComparable, S,

     R extends Writable>

     implements ReduceOperation<Pair<K, S>, Basic2ObjectMap<K, R>> {

   private PrimitiveIdTypeOps<K> keyTypeOps;

   private PrimitiveTypeOps<R> typeOps;

   private ReduceOperation<S, R> elementReduceOp;

   private WritableWriter<R> writer;

   public BasicMapReduce() {

   }

   /**

    * Create ReduceOperation that reduces BasicMaps by reducing individual

    * elements corresponding to the same key.

    *

    * @param keyTypeOps TypeOps of keys

    * @param typeOps TypeOps of individual elements

    * @param elementReduceOp ReduceOperation for individual elements

    */

   public BasicMapReduce(

       PrimitiveIdTypeOps<K> keyTypeOps, PrimitiveTypeOps<R> typeOps,

       ReduceOperation<S, R> elementReduceOp) {

     this.keyTypeOps = keyTypeOps;

     this.typeOps = typeOps;

     this.elementReduceOp = elementReduceOp;

     init();

   }

   /**

    * Registers one new local reducer, that will reduce BasicMap,

    * by reducing individual elements corresponding to the same key

    * using {@code elementReduceOp}.

    *

    * This function will return ReducerMapHandle, by which

    * individual elements can be manipulated separately.

    *

    * @param keyTypeOps TypeOps of keys

    * @param typeOps TypeOps of individual elements

    * @param elementReduceOp ReduceOperation for individual elements

    * @param reduceApi API for creating reducers

    * @return Created ReducerMapHandle

    */

   public static <K extends WritableComparable, S, R extends Writable>

   ReducerMapHandle<K, S, R> createLocalMapHandles(

       PrimitiveIdTypeOps<K> keyTypeOps, PrimitiveTypeOps<R> typeOps,

       ReduceOperation<S, R> elementReduceOp,

       final CreateReducersApi reduceApi) {

     return createMapHandles(

         keyTypeOps, typeOps, elementReduceOp,

         new CreateReducerFunctionApi() {

           @Override

           public <S, R extends Writable> ReducerHandle<S, R> createReducer(

               ReduceOperation<S, R> reduceOp) {

             return reduceApi.createLocalReducer(reduceOp);

           }

         });

   }

   /**

    * Registers one new reducer, that will reduce BasicMap,

    * by reducing individual elements corresponding to the same key

    * using {@code elementReduceOp}.

    *

    * This function will return ReducerMapHandle, by which

    * individual elements can be manipulated separately.

    *

    * @param keyTypeOps TypeOps of keys

    * @param typeOps TypeOps of individual elements

    * @param elementReduceOp ReduceOperation for individual elements

    * @param createFunction Function for creating a reducer

    * @return Created ReducerMapHandle

    */

   public static <K extends WritableComparable, S, R extends Writable>

   ReducerMapHandle<K, S, R> createMapHandles(

       final PrimitiveIdTypeOps<K> keyTypeOps, final PrimitiveTypeOps<R> typeOps,

       ReduceOperation<S, R> elementReduceOp,

       CreateReducerFunctionApi createFunction) {

     final ReducerHandle<Pair<K, S>, Basic2ObjectMap<K, R>> reduceHandle =

       createFunction.createReducer(

           new BasicMapReduce<>(keyTypeOps, typeOps, elementReduceOp));

     final K curIndex = keyTypeOps.create();

     final R reusableValue = typeOps.create();

     final R initialValue = elementReduceOp.createInitialValue();

     final MutablePair<K, S> reusablePair = MutablePair.of(null, null);

     final ReducerHandle<S, R> elementReduceHandle = new ReducerHandle<S, R>() {

       @Override

       public R getReducedValue(MasterGlobalCommUsage master) {

         Basic2ObjectMap<K, R> result = reduceHandle.getReducedValue(master);

         R value = result.get(curIndex);

         if (value == null) {

           typeOps.set(reusableValue, initialValue);

         } else {

           typeOps.set(reusableValue, value);

         }

         return reusableValue;

       }

       @Override

       public void reduce(S valueToReduce) {

         reusablePair.setLeft(curIndex);

         reusablePair.setRight(valueToReduce);

         reduceHandle.reduce(reusablePair);

       }

       @Override

       public BroadcastHandle<R> broadcastValue(BlockMasterApi master) {

         throw new UnsupportedOperationException();

       }

     };

     return new ReducerMapHandle<K, S, R>() {

       @Override

       public ReducerHandle<S, R> get(K key) {

         keyTypeOps.set(curIndex, key);

         return elementReduceHandle;

       }

       @Override

       public int getReducedSize(BlockMasterApi master) {

         return reduceHandle.getReducedValue(master).size();

       }

       @Override

       public BroadcastMapHandle<K, R> broadcastValue(BlockMasterApi master) {

         final BroadcastHandle<Basic2ObjectMap<K, R>> broadcastHandle =

           reduceHandle.broadcastValue(master);

         final K curIndex = keyTypeOps.create();

         final R reusableValue = typeOps.create();

         final BroadcastHandle<R>

         elementBroadcastHandle = new BroadcastHandle<R>() {

           @Override

           public R getBroadcast(WorkerBroadcastUsage worker) {

             Basic2ObjectMap<K, R> result = broadcastHandle.getBroadcast(worker);

             R value = result.get(curIndex);

             if (value == null) {

               typeOps.set(reusableValue, initialValue);

             } else {

               typeOps.set(reusableValue, value);

             }

             return reusableValue;

           }

         };

         return new BroadcastMapHandle<K, R>() {

           @Override

           public BroadcastHandle<R> get(K key) {

             keyTypeOps.set(curIndex, key);

             return elementBroadcastHandle;

           }

           @Override

           public int getBroadcastedSize(WorkerBroadcastUsage worker) {

             return broadcastHandle.getBroadcast(worker).size();

           }

         };

       }

     };

   }

   private void init() {

     writer = new WritableWriter<R>() {

       @Override

       public void write(DataOutput out, R value) throws IOException {

         value.write(out);

       }

       @Override

       public R readFields(DataInput in) throws IOException {

         R result = typeOps.create();

         result.readFields(in);

         return result;

       }

     };

   }

   @Override

   public Basic2ObjectMap<K, R> createInitialValue() {

     return keyTypeOps.create2ObjectOpenHashMap(writer);

   }

   @Override

   public Basic2ObjectMap<K, R> reduce(

       Basic2ObjectMap<K, R> curValue, Pair<K, S> valueToReduce) {

     R result = curValue.get(valueToReduce.getLeft());

     if (result == null) {

       result = typeOps.create();

     }

     result = elementReduceOp.reduce(result, valueToReduce.getRight());

     curValue.put(valueToReduce.getLeft(), result);

     return curValue;

   }

   @Override

   public Basic2ObjectMap<K, R> reduceMerge(

       Basic2ObjectMap<K, R> curValue, Basic2ObjectMap<K, R> valueToReduce) {

     for (Iterator<K> iter = valueToReduce.fastKeyIterator(); iter.hasNext();) {

       K key = iter.next();

       R result = curValue.get(key);

       if (result == null) {

         result = typeOps.create();

       }

       result = elementReduceOp.reduceMerge(result, valueToReduce.get(key));

       curValue.put(key, result);

     }

     return curValue;

   }

   @Override

   public void write(DataOutput out) throws IOException {

     TypeOpsUtils.writeTypeOps(keyTypeOps, out);

     TypeOpsUtils.writeTypeOps(typeOps, out);

     WritableUtils.writeWritableObject(elementReduceOp, out);

   }

   @Override

   public void readFields(DataInput in) throws IOException {

     keyTypeOps = TypeOpsUtils.readTypeOps(in);

     typeOps = TypeOpsUtils.readTypeOps(in);

     elementReduceOp = WritableUtils.readWritableObject(in, null);

     init();

   }

 }