geode-core/src/test/java/org/apache/geode/cache/query/internal/NWayMergeResultsJUnitTest.java - geode - Git at Google

 /*
  * 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.geode.cache.query.internal;

 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.fail;

 import java.lang.reflect.Method;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.Iterator;
 import java.util.List;
 import java.util.NoSuchElementException;
 import java.util.SortedSet;
 import java.util.TreeSet;

 import org.junit.Test;

 import org.apache.geode.cache.query.Struct;
 import org.apache.geode.cache.query.internal.types.ObjectTypeImpl;
 import org.apache.geode.cache.query.internal.types.StructTypeImpl;
 import org.apache.geode.cache.query.types.ObjectType;

 public class NWayMergeResultsJUnitTest {

   @Test
   public void testNonDistinct() throws Exception {
     final int numSortedLists = 40;
     Collection<List<Integer>> listOfSortedLists = new ArrayList<List<Integer>>();
     for (int i = 0; i < numSortedLists; ++i) {
       listOfSortedLists.add(new ArrayList<Integer>());
     }
     int step = 0;
     for (List<Integer> list : listOfSortedLists) {
       step = step + 1;
       for (int i = -500; i < 500; i = i + step) {
         list.add(i);
       }
     }
     int totalElements = 0;
     for (List<Integer> list : listOfSortedLists) {
       totalElements += list.size();
     }

     int combinedArray[] = new int[totalElements];
     int i = 0;
     for (List<Integer> list : listOfSortedLists) {
       for (Integer num : list) {
         combinedArray[i++] = num.intValue();
       }
     }
     Arrays.sort(combinedArray);

     NWayMergeResults<Integer> mergedResults =
         createSingleFieldMergedResult(listOfSortedLists, false, -1);
     Iterator<Integer> iter = mergedResults.iterator();
     for (int elem : combinedArray) {
       assertEquals(elem, iter.next().intValue());
     }
     assertFalse(iter.hasNext());
     try {
       iter.next();
       fail("next should have thrown NoSuchElementException");
     } catch (NoSuchElementException nsee) {
       // Ok.
     }

     assertEquals(combinedArray.length, mergedResults.size());
   }

   @Test
   public void testDistinct() throws Exception {
     final int numSortedLists = 40;
     Collection<List<Integer>> listOfSortedLists = new ArrayList<List<Integer>>();
     for (int i = 0; i < numSortedLists; ++i) {
       listOfSortedLists.add(new ArrayList<Integer>());
     }
     int step = 0;
     for (List<Integer> list : listOfSortedLists) {
       step = step + 1;
       for (int i = -500; i < 500; i = i + step) {
         list.add(i);
       }
     }

     SortedSet<Integer> sortedSet = new TreeSet<Integer>();

     for (List<Integer> list : listOfSortedLists) {
       for (Integer num : list) {
         sortedSet.add(num);
       }
     }

     NWayMergeResults<Integer> mergedResults =
         createSingleFieldMergedResult(listOfSortedLists, true, -1);

     Iterator<Integer> iter = mergedResults.iterator();
     for (int elem : sortedSet) {
       assertEquals(elem, iter.next().intValue());
     }
     assertFalse(iter.hasNext());

     try {
       iter.next();
       fail("next should have thrown NoSuchElementException");
     } catch (NoSuchElementException nsee) {
       // Ok.
     }

     assertEquals(sortedSet.size(), mergedResults.size());
   }

   @Test
   public void testLimitNoDistinct() throws Exception {
     final int numSortedLists = 40;
     final int limit = 53;
     Collection<List<Integer>> listOfSortedLists = new ArrayList<List<Integer>>();
     for (int i = 0; i < numSortedLists; ++i) {
       listOfSortedLists.add(new ArrayList<Integer>());
     }
     int step = 0;
     for (List<Integer> list : listOfSortedLists) {
       step = step + 1;
       for (int i = -500; i < 500; i = i + step) {
         list.add(i);
       }
     }
     int totalElements = 0;
     for (List<Integer> list : listOfSortedLists) {
       totalElements += list.size();
     }

     int combinedArray[] = new int[totalElements];
     int i = 0;
     for (List<Integer> list : listOfSortedLists) {
       for (Integer num : list) {
         combinedArray[i++] = num.intValue();
       }
     }
     Arrays.sort(combinedArray);

     NWayMergeResults<Integer> mergedResults =
         createSingleFieldMergedResult(listOfSortedLists, false, limit);

     Iterator<Integer> iter = mergedResults.iterator();
     int count = 0;
     for (int elem : combinedArray) {
       if (count == limit) {
         break;
       }
       assertEquals(elem, iter.next().intValue());
       ++count;
     }
     assertFalse(iter.hasNext());
     try {
       iter.next();
       fail("next should have thrown NoSuchElementException");
     } catch (NoSuchElementException nsee) {
       // Ok.
     }

     assertEquals(limit, mergedResults.size());
   }

   @Test
   public void testLimitDistinct() throws Exception {
     final int numSortedLists = 40;
     final int limit = 53;
     Collection<List<Integer>> listOfSortedLists = new ArrayList<List<Integer>>();
     for (int i = 0; i < numSortedLists; ++i) {
       listOfSortedLists.add(new ArrayList<Integer>());
     }
     int step = 0;
     for (List<Integer> list : listOfSortedLists) {
       step = step + 1;
       for (int i = -500; i < 500; i = i + step) {
         list.add(i);
       }
     }

     SortedSet<Integer> sortedSet = new TreeSet<Integer>();

     for (List<Integer> list : listOfSortedLists) {
       for (Integer num : list) {
         sortedSet.add(num);
       }
     }

     NWayMergeResults<Integer> mergedResults =
         createSingleFieldMergedResult(listOfSortedLists, true, limit);

     Iterator<Integer> iter = mergedResults.iterator();

     int count = 0;

     for (int elem : sortedSet) {
       if (count == limit) {
         break;
       }
       assertEquals(elem, iter.next().intValue());
       ++count;
     }
     assertFalse(iter.hasNext());

     try {
       iter.next();
       fail("next should have thrown NoSuchElementException");
     } catch (NoSuchElementException nsee) {
       // Ok.
     }
     assertEquals(limit, mergedResults.size());
   }

   @Test
   public void testNonDistinctStruct() throws Exception {
     final int numSortedLists = 40;
     StructTypeImpl structType = new StructTypeImpl(new String[] {"a", "b"},
         new ObjectType[] {new ObjectTypeImpl(Integer.TYPE), new ObjectTypeImpl(Integer.TYPE)});
     Collection<List<Struct>> listOfSortedLists = new ArrayList<List<Struct>>();
     for (int i = 0; i < numSortedLists; ++i) {
       listOfSortedLists.add(new ArrayList<Struct>());
     }
     int step = 0;
     for (List<Struct> list : listOfSortedLists) {
       step = step + 1;
       int j = 1000;
       for (int i = -500; i < 500; i = i + step) {
         Struct struct = new StructImpl(structType,
             new Object[] {Integer.valueOf(i), Integer.valueOf(j - step)});
         list.add(struct);
       }
     }
     int totalElements = 0;
     for (List<Struct> list : listOfSortedLists) {
       totalElements += list.size();
     }

     Struct combinedArray[] = new Struct[totalElements];
     int i = 0;
     for (List<Struct> list : listOfSortedLists) {
       for (Struct struct : list) {
         combinedArray[i++] = struct;
       }
     }
     Arrays.sort(combinedArray, new Comparator<Struct>() {
       @Override
       public int compare(Struct o1, Struct o2) {
         Object[] fields_1 = o1.getFieldValues();
         Object[] fields_2 = o2.getFieldValues();
         int compare = ((Comparable) fields_1[0]).compareTo((Comparable) fields_2[0]);
         if (compare == 0) {
           // second field is descending
           compare = ((Comparable) fields_2[1]).compareTo((Comparable) fields_1[1]);
         }
         return compare;
       }
     });

     NWayMergeResults<Struct> mergedResults =
         createStructFieldMergedResult(listOfSortedLists, false, -1, structType);

     Iterator<Struct> iter = mergedResults.iterator();
     for (Struct elem : combinedArray) {
       assertEquals(elem, iter.next());
     }
     assertFalse(iter.hasNext());
     try {
       iter.next();
       fail("next should have thrown NoSuchElementException");
     } catch (NoSuchElementException nsee) {
       // Ok.
     }
   }

   @Test
   public void testDistinctStruct() throws Exception {
     final int numSortedLists = 40;
     StructTypeImpl structType = new StructTypeImpl(new String[] {"a", "b"},
         new ObjectType[] {new ObjectTypeImpl(Integer.TYPE), new ObjectTypeImpl(Integer.TYPE)});
     Collection<List<Struct>> listOfSortedLists = new ArrayList<List<Struct>>();
     for (int i = 0; i < numSortedLists; ++i) {
       listOfSortedLists.add(new ArrayList<Struct>());
     }
     int step = 0;
     for (List<Struct> list : listOfSortedLists) {
       step = step + 1;
       int j = 1000;
       for (int i = -500; i < 500; i = i + step) {
         Struct struct = new StructImpl(structType,
             new Object[] {Integer.valueOf(i), Integer.valueOf(j - step)});
         list.add(struct);
       }
     }

     SortedSet<Struct> sortedSet = new TreeSet<Struct>(new Comparator<Struct>() {
       @Override
       public int compare(Struct o1, Struct o2) {
         Object[] fields_1 = o1.getFieldValues();
         Object[] fields_2 = o2.getFieldValues();
         int compare = ((Comparable) fields_1[0]).compareTo((Comparable) fields_2[0]);
         if (compare == 0) {
           // second field is descending
           compare = ((Comparable) fields_2[1]).compareTo((Comparable) fields_1[1]);
         }
         return compare;
       }
     });

     int i = 0;
     for (List<Struct> list : listOfSortedLists) {
       for (Struct struct : list) {
         sortedSet.add(struct);
       }
     }

     NWayMergeResults<Struct> mergedResults =
         createStructFieldMergedResult(listOfSortedLists, true, -1, structType);

     Iterator<Struct> iter = mergedResults.iterator();
     for (Struct elem : sortedSet) {
       assertEquals(elem, iter.next());
     }
     assertFalse(iter.hasNext());
     try {
       iter.next();
       fail("next should have thrown NoSuchElementException");
     } catch (NoSuchElementException nsee) {
       // Ok.
     }
   }

   @Test
   public void testOccurrenceNonDistinct() throws Exception {
     final int numSortedLists = 40;
     Collection<List<Integer>> listOfSortedLists = new ArrayList<List<Integer>>();
     for (int i = 0; i < numSortedLists; ++i) {
       listOfSortedLists.add(new ArrayList<Integer>());
     }
     int step = 0;
     for (List<Integer> list : listOfSortedLists) {
       step = step + 1;
       for (int i = -500; i < 500; i = i + step) {
         list.add(i);
       }
     }
     int totalElements = 0;
     for (List<Integer> list : listOfSortedLists) {
       totalElements += list.size();
     }

     int combinedArray[] = new int[totalElements];
     int i = 0;
     for (List<Integer> list : listOfSortedLists) {
       for (Integer num : list) {
         combinedArray[i++] = num.intValue();
       }
     }
     Arrays.sort(combinedArray);
     // count occurrence of 70, 72,73 , 75
     int num70 = 0, num72 = 0, num73 = 0, num75 = 0;
     for (int num : combinedArray) {
       if (num == 70) {
         ++num70;
       } else if (num == 72) {
         ++num72;
       } else if (num == 73) {
         ++num73;
       } else if (num == 75) {
         ++num75;
       }
     }
     NWayMergeResults<Integer> mergedResults =
         createSingleFieldMergedResult(listOfSortedLists, false, -1);
     assertEquals(num70, mergedResults.occurrences(Integer.valueOf(70)));
     assertEquals(num72, mergedResults.occurrences(Integer.valueOf(72)));
     assertEquals(num73, mergedResults.occurrences(Integer.valueOf(73)));
     assertEquals(num75, mergedResults.occurrences(Integer.valueOf(75)));
   }

   @Test
   public void testOccurrenceDistinct() throws Exception {
     final int numSortedLists = 40;
     Collection<List<Integer>> listOfSortedLists = new ArrayList<List<Integer>>();
     for (int i = 0; i < numSortedLists; ++i) {
       listOfSortedLists.add(new ArrayList<Integer>());
     }
     int step = 0;
     for (List<Integer> list : listOfSortedLists) {
       step = step + 1;
       for (int i = -500; i < 500; i = i + step) {
         list.add(i);
       }
     }

     SortedSet<Integer> sortedSet = new TreeSet<Integer>();

     for (List<Integer> list : listOfSortedLists) {
       for (Integer num : list) {
         sortedSet.add(num);
       }
     }

     NWayMergeResults<Integer> mergedResults =
         createSingleFieldMergedResult(listOfSortedLists, true, -1);

     assertEquals(1, mergedResults.occurrences(Integer.valueOf(70)));
     assertEquals(1, mergedResults.occurrences(Integer.valueOf(72)));
     assertEquals(1, mergedResults.occurrences(Integer.valueOf(73)));
     assertEquals(1, mergedResults.occurrences(Integer.valueOf(75)));

   }

   private <E> NWayMergeResults<E> createSingleFieldMergedResult(
       Collection<? extends Collection<E>> sortedResults, boolean isDistinct, int limit)
       throws Exception {
     CompiledSortCriterion csc = new CompiledSortCriterion(false,
         CompiledSortCriterion.ProjectionField.getProjectionField());
     Method method = CompiledSortCriterion.class
         .getDeclaredMethod("substituteExpressionWithProjectionField", Integer.TYPE);
     method.setAccessible(true);
     method.invoke(csc, 0);
     List<CompiledSortCriterion> orderByAttribs = new ArrayList<CompiledSortCriterion>();
     orderByAttribs.add(csc);
     ExecutionContext context = new ExecutionContext(null, null);
     ObjectType elementType = new ObjectTypeImpl(Object.class);

     return new NWayMergeResults<E>(sortedResults, isDistinct, limit, orderByAttribs, context,
         elementType);
   }

   private NWayMergeResults<Struct> createStructFieldMergedResult(
       Collection<? extends Collection<Struct>> sortedResults, boolean isDistinct, int limit,
       StructTypeImpl structType) throws Exception {
     CompiledSortCriterion csc1 = new CompiledSortCriterion(false,
         CompiledSortCriterion.ProjectionField.getProjectionField());
     CompiledSortCriterion csc2 =
         new CompiledSortCriterion(true, CompiledSortCriterion.ProjectionField.getProjectionField());
     Method method = CompiledSortCriterion.class
         .getDeclaredMethod("substituteExpressionWithProjectionField", Integer.TYPE);
     method.setAccessible(true);
     method.invoke(csc1, 0);
     method.invoke(csc2, 1);
     List<CompiledSortCriterion> orderByAttribs = new ArrayList<CompiledSortCriterion>();
     orderByAttribs.add(csc1);
     orderByAttribs.add(csc2);
     ExecutionContext context = new ExecutionContext(null, null);

     return new NWayMergeResults<Struct>(sortedResults, false, -1, orderByAttribs, context,
         structType);
   }

   @Test
   public void testCombination() throws Exception {
     List<String> results1 = new ArrayList<String>();
     results1.add("IBM");
     results1.add("YHOO");

     List<String> results2 = new ArrayList<String>();
     results2.add("APPL");
     results2.add("ORCL");

     List<String> results3 = new ArrayList<String>();
     results3.add("DELL");
     results3.add("RHAT");

     List<String> results4 = new ArrayList<String>();
     results4.add("ORCL");
     results4.add("SAP");

     List<String> results5 = Collections.emptyList();
     List<String> results6 = Collections.emptyList();
     List<String> results7 = Collections.emptyList();

     List<Collection<String>> sortedLists = new ArrayList<Collection<String>>();
     sortedLists.add(results1);
     sortedLists.add(results2);
     sortedLists.add(results3);
     sortedLists.add(results4);
     sortedLists.add(results5);
     sortedLists.add(results6);
     sortedLists.add(results7);

     NWayMergeResults<String> mergedResults = createSingleFieldMergedResult(sortedLists, true, -1);

     List<String> results8 = new ArrayList<String>();
     results8.add("DELL");
     results8.add("GOOG");
     results8.add("HP");
     results8.add("MSFT");
     results8.add("SAP");
     results8.add("SUN");

     List<String> results9 = new ArrayList<String>();
     results9.add("AOL");
     results9.add("APPL");
     results9.add("GOOG");
     results9.add("IBM");
     results9.add("SUN");
     results9.add("YHOO");

     List<Collection<String>> sortedLists1 = new ArrayList<Collection<String>>();
     sortedLists1.add(mergedResults);
     sortedLists1.add(results8);
     sortedLists1.add(results9);

     NWayMergeResults<String> netMergedResults =
         createSingleFieldMergedResult(sortedLists1, true, -1);

     assertEquals(12, netMergedResults.size());
   }

 }
	/*
	* 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.geode.cache.query.internal;

	import static org.junit.Assert.assertEquals;
	import static org.junit.Assert.assertFalse;
	import static org.junit.Assert.fail;

	import java.lang.reflect.Method;
	import java.util.ArrayList;
	import java.util.Arrays;
	import java.util.Collection;
	import java.util.Collections;
	import java.util.Comparator;
	import java.util.Iterator;
	import java.util.List;
	import java.util.NoSuchElementException;
	import java.util.SortedSet;
	import java.util.TreeSet;

	import org.junit.Test;

	import org.apache.geode.cache.query.Struct;
	import org.apache.geode.cache.query.internal.types.ObjectTypeImpl;
	import org.apache.geode.cache.query.internal.types.StructTypeImpl;
	import org.apache.geode.cache.query.types.ObjectType;

	public class NWayMergeResultsJUnitTest {

	@Test
	public void testNonDistinct() throws Exception {
	final int numSortedLists = 40;
	Collection<List<Integer>> listOfSortedLists = new ArrayList<List<Integer>>();
	for (int i = 0; i < numSortedLists; ++i) {
	listOfSortedLists.add(new ArrayList<Integer>());
	}
	int step = 0;
	for (List<Integer> list : listOfSortedLists) {
	step = step + 1;
	for (int i = -500; i < 500; i = i + step) {
	list.add(i);
	}
	}
	int totalElements = 0;
	for (List<Integer> list : listOfSortedLists) {
	totalElements += list.size();
	}

	int combinedArray[] = new int[totalElements];
	int i = 0;
	for (List<Integer> list : listOfSortedLists) {
	for (Integer num : list) {
	combinedArray[i++] = num.intValue();
	}
	}
	Arrays.sort(combinedArray);

	NWayMergeResults<Integer> mergedResults =
	createSingleFieldMergedResult(listOfSortedLists, false, -1);
	Iterator<Integer> iter = mergedResults.iterator();
	for (int elem : combinedArray) {
	assertEquals(elem, iter.next().intValue());
	}
	assertFalse(iter.hasNext());
	try {
	iter.next();
	fail("next should have thrown NoSuchElementException");
	} catch (NoSuchElementException nsee) {
	// Ok.
	}

	assertEquals(combinedArray.length, mergedResults.size());
	}

	@Test
	public void testDistinct() throws Exception {
	final int numSortedLists = 40;
	Collection<List<Integer>> listOfSortedLists = new ArrayList<List<Integer>>();
	for (int i = 0; i < numSortedLists; ++i) {
	listOfSortedLists.add(new ArrayList<Integer>());
	}
	int step = 0;
	for (List<Integer> list : listOfSortedLists) {
	step = step + 1;
	for (int i = -500; i < 500; i = i + step) {
	list.add(i);
	}
	}

	SortedSet<Integer> sortedSet = new TreeSet<Integer>();

	for (List<Integer> list : listOfSortedLists) {
	for (Integer num : list) {
	sortedSet.add(num);
	}
	}

	NWayMergeResults<Integer> mergedResults =
	createSingleFieldMergedResult(listOfSortedLists, true, -1);

	Iterator<Integer> iter = mergedResults.iterator();
	for (int elem : sortedSet) {
	assertEquals(elem, iter.next().intValue());
	}
	assertFalse(iter.hasNext());

	try {
	iter.next();
	fail("next should have thrown NoSuchElementException");
	} catch (NoSuchElementException nsee) {
	// Ok.
	}

	assertEquals(sortedSet.size(), mergedResults.size());
	}

	@Test
	public void testLimitNoDistinct() throws Exception {
	final int numSortedLists = 40;
	final int limit = 53;
	Collection<List<Integer>> listOfSortedLists = new ArrayList<List<Integer>>();
	for (int i = 0; i < numSortedLists; ++i) {
	listOfSortedLists.add(new ArrayList<Integer>());
	}
	int step = 0;
	for (List<Integer> list : listOfSortedLists) {
	step = step + 1;
	for (int i = -500; i < 500; i = i + step) {
	list.add(i);
	}
	}
	int totalElements = 0;
	for (List<Integer> list : listOfSortedLists) {
	totalElements += list.size();
	}

	int combinedArray[] = new int[totalElements];
	int i = 0;
	for (List<Integer> list : listOfSortedLists) {
	for (Integer num : list) {
	combinedArray[i++] = num.intValue();
	}
	}
	Arrays.sort(combinedArray);

	NWayMergeResults<Integer> mergedResults =
	createSingleFieldMergedResult(listOfSortedLists, false, limit);

	Iterator<Integer> iter = mergedResults.iterator();
	int count = 0;
	for (int elem : combinedArray) {
	if (count == limit) {
	break;
	}
	assertEquals(elem, iter.next().intValue());
	++count;
	}
	assertFalse(iter.hasNext());
	try {
	iter.next();
	fail("next should have thrown NoSuchElementException");
	} catch (NoSuchElementException nsee) {
	// Ok.
	}

	assertEquals(limit, mergedResults.size());
	}

	@Test
	public void testLimitDistinct() throws Exception {
	final int numSortedLists = 40;
	final int limit = 53;
	Collection<List<Integer>> listOfSortedLists = new ArrayList<List<Integer>>();
	for (int i = 0; i < numSortedLists; ++i) {
	listOfSortedLists.add(new ArrayList<Integer>());
	}
	int step = 0;
	for (List<Integer> list : listOfSortedLists) {
	step = step + 1;
	for (int i = -500; i < 500; i = i + step) {
	list.add(i);
	}
	}

	SortedSet<Integer> sortedSet = new TreeSet<Integer>();

	for (List<Integer> list : listOfSortedLists) {
	for (Integer num : list) {
	sortedSet.add(num);
	}
	}

	NWayMergeResults<Integer> mergedResults =
	createSingleFieldMergedResult(listOfSortedLists, true, limit);

	Iterator<Integer> iter = mergedResults.iterator();

	int count = 0;

	for (int elem : sortedSet) {
	if (count == limit) {
	break;
	}
	assertEquals(elem, iter.next().intValue());
	++count;
	}
	assertFalse(iter.hasNext());

	try {
	iter.next();
	fail("next should have thrown NoSuchElementException");
	} catch (NoSuchElementException nsee) {
	// Ok.
	}
	assertEquals(limit, mergedResults.size());
	}

	@Test
	public void testNonDistinctStruct() throws Exception {
	final int numSortedLists = 40;
	StructTypeImpl structType = new StructTypeImpl(new String[] {"a", "b"},
	new ObjectType[] {new ObjectTypeImpl(Integer.TYPE), new ObjectTypeImpl(Integer.TYPE)});
	Collection<List<Struct>> listOfSortedLists = new ArrayList<List<Struct>>();
	for (int i = 0; i < numSortedLists; ++i) {
	listOfSortedLists.add(new ArrayList<Struct>());
	}
	int step = 0;
	for (List<Struct> list : listOfSortedLists) {
	step = step + 1;
	int j = 1000;
	for (int i = -500; i < 500; i = i + step) {
	Struct struct = new StructImpl(structType,
	new Object[] {Integer.valueOf(i), Integer.valueOf(j - step)});
	list.add(struct);
	}
	}
	int totalElements = 0;
	for (List<Struct> list : listOfSortedLists) {
	totalElements += list.size();
	}

	Struct combinedArray[] = new Struct[totalElements];
	int i = 0;
	for (List<Struct> list : listOfSortedLists) {
	for (Struct struct : list) {
	combinedArray[i++] = struct;
	}
	}
	Arrays.sort(combinedArray, new Comparator<Struct>() {
	@Override
	public int compare(Struct o1, Struct o2) {
	Object[] fields_1 = o1.getFieldValues();
	Object[] fields_2 = o2.getFieldValues();
	int compare = ((Comparable) fields_1[0]).compareTo((Comparable) fields_2[0]);
	if (compare == 0) {
	// second field is descending
	compare = ((Comparable) fields_2[1]).compareTo((Comparable) fields_1[1]);
	}
	return compare;
	}
	});

	NWayMergeResults<Struct> mergedResults =
	createStructFieldMergedResult(listOfSortedLists, false, -1, structType);

	Iterator<Struct> iter = mergedResults.iterator();
	for (Struct elem : combinedArray) {
	assertEquals(elem, iter.next());
	}
	assertFalse(iter.hasNext());
	try {
	iter.next();
	fail("next should have thrown NoSuchElementException");
	} catch (NoSuchElementException nsee) {
	// Ok.
	}
	}

	@Test
	public void testDistinctStruct() throws Exception {
	final int numSortedLists = 40;
	StructTypeImpl structType = new StructTypeImpl(new String[] {"a", "b"},
	new ObjectType[] {new ObjectTypeImpl(Integer.TYPE), new ObjectTypeImpl(Integer.TYPE)});
	Collection<List<Struct>> listOfSortedLists = new ArrayList<List<Struct>>();
	for (int i = 0; i < numSortedLists; ++i) {
	listOfSortedLists.add(new ArrayList<Struct>());
	}
	int step = 0;
	for (List<Struct> list : listOfSortedLists) {
	step = step + 1;
	int j = 1000;
	for (int i = -500; i < 500; i = i + step) {
	Struct struct = new StructImpl(structType,
	new Object[] {Integer.valueOf(i), Integer.valueOf(j - step)});
	list.add(struct);
	}
	}

	SortedSet<Struct> sortedSet = new TreeSet<Struct>(new Comparator<Struct>() {
	@Override
	public int compare(Struct o1, Struct o2) {
	Object[] fields_1 = o1.getFieldValues();
	Object[] fields_2 = o2.getFieldValues();
	int compare = ((Comparable) fields_1[0]).compareTo((Comparable) fields_2[0]);
	if (compare == 0) {
	// second field is descending
	compare = ((Comparable) fields_2[1]).compareTo((Comparable) fields_1[1]);
	}
	return compare;
	}
	});

	int i = 0;
	for (List<Struct> list : listOfSortedLists) {
	for (Struct struct : list) {
	sortedSet.add(struct);
	}
	}

	NWayMergeResults<Struct> mergedResults =
	createStructFieldMergedResult(listOfSortedLists, true, -1, structType);

	Iterator<Struct> iter = mergedResults.iterator();
	for (Struct elem : sortedSet) {
	assertEquals(elem, iter.next());
	}
	assertFalse(iter.hasNext());
	try {
	iter.next();
	fail("next should have thrown NoSuchElementException");
	} catch (NoSuchElementException nsee) {
	// Ok.
	}
	}

	@Test
	public void testOccurrenceNonDistinct() throws Exception {
	final int numSortedLists = 40;
	Collection<List<Integer>> listOfSortedLists = new ArrayList<List<Integer>>();
	for (int i = 0; i < numSortedLists; ++i) {
	listOfSortedLists.add(new ArrayList<Integer>());
	}
	int step = 0;
	for (List<Integer> list : listOfSortedLists) {
	step = step + 1;
	for (int i = -500; i < 500; i = i + step) {
	list.add(i);
	}
	}
	int totalElements = 0;
	for (List<Integer> list : listOfSortedLists) {
	totalElements += list.size();
	}

	int combinedArray[] = new int[totalElements];
	int i = 0;
	for (List<Integer> list : listOfSortedLists) {
	for (Integer num : list) {
	combinedArray[i++] = num.intValue();
	}
	}
	Arrays.sort(combinedArray);
	// count occurrence of 70, 72,73 , 75
	int num70 = 0, num72 = 0, num73 = 0, num75 = 0;
	for (int num : combinedArray) {
	if (num == 70) {
	++num70;
	} else if (num == 72) {
	++num72;
	} else if (num == 73) {
	++num73;
	} else if (num == 75) {
	++num75;
	}
	}
	NWayMergeResults<Integer> mergedResults =
	createSingleFieldMergedResult(listOfSortedLists, false, -1);
	assertEquals(num70, mergedResults.occurrences(Integer.valueOf(70)));
	assertEquals(num72, mergedResults.occurrences(Integer.valueOf(72)));
	assertEquals(num73, mergedResults.occurrences(Integer.valueOf(73)));
	assertEquals(num75, mergedResults.occurrences(Integer.valueOf(75)));
	}

	@Test
	public void testOccurrenceDistinct() throws Exception {
	final int numSortedLists = 40;
	Collection<List<Integer>> listOfSortedLists = new ArrayList<List<Integer>>();
	for (int i = 0; i < numSortedLists; ++i) {
	listOfSortedLists.add(new ArrayList<Integer>());
	}
	int step = 0;
	for (List<Integer> list : listOfSortedLists) {
	step = step + 1;
	for (int i = -500; i < 500; i = i + step) {
	list.add(i);
	}
	}

	SortedSet<Integer> sortedSet = new TreeSet<Integer>();

	for (List<Integer> list : listOfSortedLists) {
	for (Integer num : list) {
	sortedSet.add(num);
	}
	}

	NWayMergeResults<Integer> mergedResults =
	createSingleFieldMergedResult(listOfSortedLists, true, -1);

	assertEquals(1, mergedResults.occurrences(Integer.valueOf(70)));
	assertEquals(1, mergedResults.occurrences(Integer.valueOf(72)));
	assertEquals(1, mergedResults.occurrences(Integer.valueOf(73)));
	assertEquals(1, mergedResults.occurrences(Integer.valueOf(75)));

	}

	private <E> NWayMergeResults<E> createSingleFieldMergedResult(
	Collection<? extends Collection<E>> sortedResults, boolean isDistinct, int limit)
	throws Exception {
	CompiledSortCriterion csc = new CompiledSortCriterion(false,
	CompiledSortCriterion.ProjectionField.getProjectionField());
	Method method = CompiledSortCriterion.class
	.getDeclaredMethod("substituteExpressionWithProjectionField", Integer.TYPE);
	method.setAccessible(true);
	method.invoke(csc, 0);
	List<CompiledSortCriterion> orderByAttribs = new ArrayList<CompiledSortCriterion>();
	orderByAttribs.add(csc);
	ExecutionContext context = new ExecutionContext(null, null);
	ObjectType elementType = new ObjectTypeImpl(Object.class);

	return new NWayMergeResults<E>(sortedResults, isDistinct, limit, orderByAttribs, context,
	elementType);
	}

	private NWayMergeResults<Struct> createStructFieldMergedResult(
	Collection<? extends Collection<Struct>> sortedResults, boolean isDistinct, int limit,
	StructTypeImpl structType) throws Exception {
	CompiledSortCriterion csc1 = new CompiledSortCriterion(false,
	CompiledSortCriterion.ProjectionField.getProjectionField());
	CompiledSortCriterion csc2 =
	new CompiledSortCriterion(true, CompiledSortCriterion.ProjectionField.getProjectionField());
	Method method = CompiledSortCriterion.class
	.getDeclaredMethod("substituteExpressionWithProjectionField", Integer.TYPE);
	method.setAccessible(true);
	method.invoke(csc1, 0);
	method.invoke(csc2, 1);
	List<CompiledSortCriterion> orderByAttribs = new ArrayList<CompiledSortCriterion>();
	orderByAttribs.add(csc1);
	orderByAttribs.add(csc2);
	ExecutionContext context = new ExecutionContext(null, null);

	return new NWayMergeResults<Struct>(sortedResults, false, -1, orderByAttribs, context,
	structType);
	}

	@Test
	public void testCombination() throws Exception {
	List<String> results1 = new ArrayList<String>();
	results1.add("IBM");
	results1.add("YHOO");

	List<String> results2 = new ArrayList<String>();
	results2.add("APPL");
	results2.add("ORCL");

	List<String> results3 = new ArrayList<String>();
	results3.add("DELL");
	results3.add("RHAT");

	List<String> results4 = new ArrayList<String>();
	results4.add("ORCL");
	results4.add("SAP");

	List<String> results5 = Collections.emptyList();
	List<String> results6 = Collections.emptyList();
	List<String> results7 = Collections.emptyList();

	List<Collection<String>> sortedLists = new ArrayList<Collection<String>>();
	sortedLists.add(results1);
	sortedLists.add(results2);
	sortedLists.add(results3);
	sortedLists.add(results4);
	sortedLists.add(results5);
	sortedLists.add(results6);
	sortedLists.add(results7);

	NWayMergeResults<String> mergedResults = createSingleFieldMergedResult(sortedLists, true, -1);

	List<String> results8 = new ArrayList<String>();
	results8.add("DELL");
	results8.add("GOOG");
	results8.add("HP");
	results8.add("MSFT");
	results8.add("SAP");
	results8.add("SUN");

	List<String> results9 = new ArrayList<String>();
	results9.add("AOL");
	results9.add("APPL");
	results9.add("GOOG");
	results9.add("IBM");
	results9.add("SUN");
	results9.add("YHOO");

	List<Collection<String>> sortedLists1 = new ArrayList<Collection<String>>();
	sortedLists1.add(mergedResults);
	sortedLists1.add(results8);
	sortedLists1.add(results9);

	NWayMergeResults<String> netMergedResults =
	createSingleFieldMergedResult(sortedLists1, true, -1);

	assertEquals(12, netMergedResults.size());
	}

	}