/**
    * Splits a source scan src into many runs.
    * @param src the source scan
    * @param runs the number of runs
    * @return the list of created runs
    */
   protected List<TempTable> splitIntoRuns(Scan src, int runs) {
	      List<TempTable> temps = new ArrayList<TempTable>();
	      src.beforeFirst();
	      boolean next = src.next();
	      if (!next)
	         return temps;
	      
	      /* k k-block runs */
	      int k = runs;
	      
	      // p is the underlying plan
	      int recordsPerRun = p.recordsOutput() / k;
	      int copiedRecords=0, createdRuns=0;
	      	      	      
	      while (createdRuns<k) {
	     
		      /* create a target run */
		      TempTable currenttemp = new TempTable(sch, tx);
		      temps.add(currenttemp);
		      UpdateScan currentscan = currenttemp.open();
		      copiedRecords=0;
		      	    	  
		      /* copy each run */
		      while ((copiedRecords<recordsPerRun || createdRuns==k-1) && next) {
		    		  next = copy(src, currentscan);
		    		  copiedRecords++;
		      }
		    
		      /* sort the run */
		      boolean anySwap=true;		      
		      while(anySwap) {
		    	  anySwap=false;
			      currentscan.beforeFirst();
			      RID curRid = null, nextRid = null;
			      while(currentscan.next()) {
			    	  curRid = currentscan.getRid();
			    	  if(currentscan.next()) {
			    		  nextRid=currentscan.getRid();
			    		  if(comp.compare(currentscan, curRid, nextRid)>0) {
			    			  swap(currentscan, curRid, nextRid);
			    			  anySwap=true;
			    		  }
			    	  }
		    		  currentscan.moveToRid(curRid);
				  }
		      }
		      currentscan.close();
		      createdRuns++;
	      }
	      return temps;
   }
   
   /**
    * Swaps the content of two records in UpdateScan given the two RIDS
    * @param s the UpdateScan
    * @param r1 the first RID
    * @param r2 the second RID
    */
   private void swap(UpdateScan s, RID r1, RID r2) {
	   RID currentRid = s.getRid();
	   
	   s.moveToRid(r1);
	   Map<String,Constant> values1 = new HashMap<String,Constant>();
	   for(String fldName : p.schema().fields()) {
		   values1.put(fldName, s.getVal(fldName));
	   }
	   
	   s.moveToRid(r2);
	   Map<String,Constant> values2 = new HashMap<String,Constant>();
	   for(String fldName : p.schema().fields()) {
		   values2.put(fldName, s.getVal(fldName));
		   s.setVal(fldName, values1.get(fldName));
	   }
	   
	   s.moveToRid(r1);
	   for(String fldName : p.schema().fields()) {
		   s.setVal(fldName, values2.get(fldName));
	   }
	   
	   s.moveToRid(currentRid);  
   }

  /**
    * Compares two records at two specified RIDS of a scan.
    * The sort fields are considered in turn.
    * When a field is encountered for which the records have
    * different values, those values are used as the result
    * of the comparison.
    * If the two records have the same values for all
    * sort fields, then the method returns 0.
    * @param s the scan
    * @param r1 the first RID
    * @param r2 the second RID
    * @return the result of comparing the two records according to the field list
    */
   public int compare(UpdateScan s, RID r1, RID r2) {
	   RID currentRid = s.getRid();
	   for (String fldname : fields) {
		   	 s.moveToRid(r1);
	         Constant val1 = s.getVal(fldname);
		   	 s.moveToRid(r2);
	         Constant val2 = s.getVal(fldname);
	         int result = val1.compareTo(val2);
	         if (result != 0) {
			   	 s.moveToRid(currentRid);
	             return result;
	         }
	      }
	   	  s.moveToRid(currentRid);
	      return 0;
   }