package edu.umn.cs.spatialHadoop.mapReduce;

import java.io.IOException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.spatial.GridInfo;


/**
 * Performs a spatial join between two files. If both files are of the same grid,
 * SJRO is performed rightaway. If they are of different grids, the smaller file
 * is repartitioned according to the grid used in the larger file. After this,
 * SJRO is performed.
 * @author aseldawy
 *
 */
public class SJRepartitionMapReduce {
	public static final Log LOG = LogFactory.getLog(SJRepartitionMapReduce.class);

	public static void SJRO(JobConf conf, Path[] inputFiles, Path outputPath) throws IOException {
	  
    FileSystem fileSystem = FileSystem.get(conf);
    
    // Find grid info for largest file
    GridInfo gridInfo = null;
    long largestFileSize = 0;
    for (Path inputFile : inputFiles) {
      FileStatus fileStatus = fileSystem.getFileStatus(inputFile);
      if (gridInfo == null
          || (fileStatus.getLen() > largestFileSize && fileStatus.getGridInfo() != null)) {
        largestFileSize = fileStatus.getLen();
        gridInfo = fileStatus.getGridInfo();
      }
    }
    LOG.info("Repartitioning according to the grid: "+gridInfo);

    // Now, repartition all files that are not of this grid
    for (int i = 0; i < inputFiles.length; i++) {
      FileStatus fileStatus = fileSystem.getFileStatus(inputFiles[i]);
      if (fileStatus.getGridInfo() == null || !fileStatus.getGridInfo().equals(gridInfo)) {
        LOG.info("Going to repartition "+inputFiles[i]);
        Path repartitioned = new Path(inputFiles[i].toUri().getPath()+".grid");
        RepartitionMapReduce.repartition(conf, inputFiles[i], repartitioned, gridInfo);
        // Use the repartitioned file instead of original file
        inputFiles[i] = repartitioned;
      }
    }
    
    // Currently, all files are of the same grid, go ahead to SpatialJoin
    SJROMapReduce.SJRO(conf, inputFiles, outputPath);
	}
	
	/**
	 * Entry point to the file.
	 * Params <input filenames> <output filename>
	 * input filenames: A list of paths to input files in HDFS
	 * output filename: A path to an output file in HDFS
	 * @param args
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
    JobConf conf = new JobConf(RepartitionMapReduce.class);
    Path[] inputFiles = {new Path(args[0]), new Path(args[1])};
    Path outputPath = new Path(args[2]);
    
    SJRO(conf, inputFiles, outputPath);
	}
}

@@ -350,16 +350,149 @@
 idInfo;%0A
+import org.apache.hadoop.spatial.Rectangle;%0Aimport org.apache.hadoop.spatial.WriteGridFile;%0A%0Aimport edu.umn.cs.CommandLineArguments;%0A
 %0A%0A/**%0A *
@@ -919,20 +919,38 @@
 ic void 
-SJRO
+spatialJoinRepartition
 (JobConf
@@ -1171,16 +1171,67 @@
 ze = 0;%0A
+    Rectangle mbr = null;%0A    long totalSizes = 0;%0A
     for 
@@ -1329,16 +1329,198 @@
 tFile);%0A
+      Rectangle fileMBR = WriteGridFile.getMBR(fileSystem, inputFile);%0A      mbr = (Rectangle) (mbr == null ? fileMBR : mbr.union(fileMBR));%0A      totalSizes += fileStatus.getLen();%0A
       if
@@ -1732,24 +1732,301 @@
     %7D%0A    %7D%0A
+    if (!gridInfo.getMBR().contains(mbr)) %7B%0A      gridInfo.xOrigin = mbr.x;%0A      gridInfo.yOrigin = mbr.y;%0A      gridInfo.gridWidth= mbr.width;%0A      gridInfo.gridHeight = mbr.height;%0A      gridInfo.calculateCellDimensions(totalSizes, fileSystem.getDefaultBlockSize());%0A    %7D%0A
     LOG.info
@@ -3196,122 +3196,186 @@
     
-Path%5B%5D inputFiles
+CommandLineArguments cla
  = 
-%7B
 new 
-Path(args%5B0%5D), new Path(args%5B1%5D)%7D;%0A    Path outputPath = new Path(args%5B2%5D);%0A    %0A    SJRO
+CommandLineArguments(args);%0A    Path%5B%5D inputFiles = cla.getInputPaths();%0A    Path outputPath = cla.getOutputPath();%0A    %0A    spatialJoinRepartition
 (con

Refactor SJReprtitionMapReduce changed method name