library/src/main/java/com/datatorrent/lib/algo/MergeSort.java

/*
 * Copyright (c) 2013 DataTorrent, Inc. ALL Rights Reserved.
 *
 * Licensed 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.lib.algo;

import com.datatorrent.api.Context.OperatorContext;
import com.datatorrent.api.Operator.Unifier;
import com.datatorrent.api.DefaultInputPort;
import com.datatorrent.api.DefaultOutputPort;
import com.datatorrent.api.Operator;
import com.datatorrent.api.annotation.InputPortFieldAnnotation;
import com.datatorrent.api.annotation.OutputPortFieldAnnotation;

import java.util.ArrayList;

/**
 * <p>
 * Incoming sorted list is merged into already existing sorted list. The input list is expected to be sorted. <b>
 * At the end of the window, merged sorted list is emitted on sort output port. <br>
 * <br>
 * <b>Notes : </b> <br>
 * Get unifier instance must return instance of sub class itself, since merge operator
 * id unifier on output port. <br>
 * <br>
 *  <b>StateFull : Yes</b>, Sorted listed are merged over application window can be > 1. <br>
 *  <b>Partitions : Yes</b>, Operator itself is used as unfier on output port.
 * <br>
 * <b>Ports</b>:<br>
 * <b>data</b>: expects ArrayList&lt;K&gt;<br>
 * <b>sort</b>: emits ArrayList&lt;K&gt;<br>
 * <br>
 * <b>Abstract Methods: </b><br>
 * 1. compare : K type value compare criteria for sort.
 * 2. getUnifierInstance : Get unifier operator instance for output port, (must return self instance).
 *
 * @since 0.3.3
 */
public abstract class MergeSort<K>  implements Operator, Unifier<ArrayList<K>>
{
	/**
	 * Sorted merged list.
	 */
	private ArrayList<K> mergedList = null;
	
	/**
	 * Input port.
	 */
  @InputPortFieldAnnotation(name = "data")
  public final transient DefaultInputPort<ArrayList<K>> data = new DefaultInputPort<ArrayList<K>>()
  {
  	/**
  	 * Merge incoming tuple.
  	 */
    @Override
    public void process(ArrayList<K> tuple)
    {
    	mergedList = processMergeList(mergedList, tuple);
    }
  };
  
  /**
   * Sorted list output port.
   */
  @OutputPortFieldAnnotation(name = "sort")
  public final transient DefaultOutputPort<ArrayList<K>> sort = new DefaultOutputPort<ArrayList<K>>() {
  		@Override
  		public Unifier<ArrayList<K>> getUnifier()
  		{
  			return getUnifierInstance();
  		}
  };
  
	@Override
	public void setup(OperatorContext context)
	{
		// TODO Auto-generated method stub
		
	}

	@Override
	public void teardown()
	{
		// TODO Auto-generated method stub
		
	}
	@Override
	public void beginWindow(long windowId)
	{
		mergedList = null;
	}
	@Override
	public void endWindow()
	{
		sort.emit(mergedList);
		mergedList = null;
	}
	
	/**
	 *  Sorted parameter list are merged into sorted merged output list.
	 *  
	 * @param list1 sorted list aggregated by operator
	 * @param list2 Input port sorted list to be merged.
	 * @return sorted merged output list.
	 */
	protected ArrayList<K> processMergeList(ArrayList<K> list1,
			ArrayList<K> list2)
	{
		// null lists
		if (list1 == null) return list2;
		if (list2 == null) return list1;
		
		// Create output list   
		ArrayList<K> result = new ArrayList<K>();
		int index1 = 0;  
		int index2 = 0;
		boolean loop = true;
	  while (loop) {
	  	
	  	// list1 is exhausted
	  	if (index1 == list1.size()) {
	  		while(index2 < list2.size()) {
	  			result.add(list2.get(index2++));
	  		}
	  		break;
	  	}
	  	
	  	// list2 is exhausted  
	  	if (index2 == list2.size()) {
	  		while(index1 < list1.size()) {
	  			result.add(list1.get(index1++));
	  		}
	  		break;
	  	}
	  	
	  	// compare values   
	  	K val1 = list1.get(index1++);
	  	K val2 = list2.get(index2++);
	  	K[] vals = compare(val1, val2);
	  	result.add(vals[0]);
	  	if (vals[1] != null) result.add(vals[1]);
	  }
		
	  // done
    return result;
	}

	/**
	 * Unifier process function implementation.
	 */
	@Override
	public void process(ArrayList<K> tuple)
	{
		mergedList = processMergeList(mergedList, tuple);
	}
	
	/**
	 * abstract sort function to be implemented by sub class.
	 */
	abstract public  K[] compare(K val1, K val2);
	
	/**
	 *  Get output port unifier instance, sub class should return new instance of itself.
	 */
	abstract public Unifier<ArrayList<K>> getUnifierInstance();
}