/**
 * FILE: CombineShapeReader.java
 * PATH: org.datasyslab.geospark.formatMapper.shapefileParser.shapes.CombineShapeReader.java
 * Copyright (c) 2015-2017 GeoSpark Development Team
 * All rights reserved.
 */
package org.datasyslab.geospark.formatMapper.shapefileParser.shapes;

import org.apache.commons.io.FilenameUtils;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.CombineFileSplit;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.datasyslab.geospark.formatMapper.shapefileParser.parseUtils.shp.ShapeType;

import java.io.IOException;
import java.util.List;

public class CombineShapeReader extends RecordReader<ShapeKey, PrimitiveShape> {

    /** id of input path of .shp file */
    private FileSplit shpSplit = null;

    /** id of input path of .shx file */
    private FileSplit shxSplit = null;

    /** id of input path of .dbf file */
    private FileSplit dbfSplit = null;

    /** RecordReader for .shp file */
    private ShapeFileReader shapeFileReader = null;

    /** RecordReader for .dbf file */
    private DbfFileReader dbfFileReader = null;

    /** suffix of attribute file */
    private final static String DBF_SUFFIX = "dbf";

    /** suffix of shape record file */
    private final static String SHP_SUFFIX = "shp";

    /** suffix of index file */
    private final static String SHX_SUFFIX = "shx";

    /** value of current record */
    private PrimitiveShape value = null;

    /** flag of whether .dbf exists */
    private boolean hasDbf = false;

    /** flag of whether having next .dbf record */
    private boolean hasNextDbf = false;

    /**
     * cut the combined split into FileSplit for .shp, .shx and .dbf
     * @param split
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    public void initialize(InputSplit split, TaskAttemptContext context)
            throws IOException, InterruptedException
    {
        CombineFileSplit fileSplit = (CombineFileSplit)split;
        Path[] paths = fileSplit.getPaths();
        for(int i = 0;i < paths.length; ++i){
            String suffix = FilenameUtils.getExtension(paths[i].toString());
            if(suffix.equals(SHP_SUFFIX)) shpSplit = new FileSplit(paths[i], fileSplit.getOffset(i), fileSplit.getLength(i), fileSplit.getLocations());
            else if(suffix.equals(SHX_SUFFIX)) shxSplit = new FileSplit(paths[i], fileSplit.getOffset(i), fileSplit.getLength(i), fileSplit.getLocations());
            else if(suffix.equals(DBF_SUFFIX)) dbfSplit = new FileSplit(paths[i], fileSplit.getOffset(i), fileSplit.getLength(i), fileSplit.getLocations());
        }
        // if shape file doesn't exists, throw an IOException
        if(shpSplit == null) throw new IOException("Can't find .shp file.");
        else{
            shapeFileReader = new ShapeFileReader();
            shapeFileReader.initialize(shpSplit, context);
        }
        if(dbfSplit != null){
            dbfFileReader = new DbfFileReader();
            dbfFileReader.initialize(dbfSplit, context);
            hasDbf = true;
        }else hasDbf = false;

    }

    public boolean nextKeyValue() throws IOException, InterruptedException {

        boolean hasNextShp = shapeFileReader.nextKeyValue();
        if(hasDbf) hasNextDbf = dbfFileReader.nextKeyValue();
        int curShapeType = shapeFileReader.getCurrentValue().getTypeID();
        while(curShapeType == ShapeType.NULL.getId() || curShapeType == ShapeType.UNDEFINED.getId()){
            hasNextShp = shapeFileReader.nextKeyValue();
            if(hasDbf) hasNextDbf = dbfFileReader.nextKeyValue();
            curShapeType = shapeFileReader.getCurrentValue().getTypeID();
        }
        // check if records match in .shp and .dbf
        if(hasNextShp && !hasNextDbf){
            Exception e = new Exception("shape record loses attributes in .dbf file at ID=" + shapeFileReader.getCurrentKey().getIndex());
            e.printStackTrace();
        }else if(!hasNextShp && hasNextDbf){
            Exception e = new Exception("Redundant attributes in .dbf exists");
            e.printStackTrace();
        }
        return hasNextShp;
    }

    public ShapeKey getCurrentKey() throws IOException, InterruptedException {
        return shapeFileReader.getCurrentKey();
    }

    public PrimitiveShape getCurrentValue() throws IOException, InterruptedException {
        value = new PrimitiveShape();
        value.setPrimitiveRecord(shapeFileReader.getCurrentValue());
        if(hasDbf && hasNextDbf) value.setPrimitiveAttribute(dbfFileReader.getCurrentValue());
        return value;
    }

    public float getProgress() throws IOException, InterruptedException {
        return shapeFileReader.getProgress();
    }

    public void close() throws IOException {
        shapeFileReader.close();
    }
}

@@ -614,32 +614,64 @@
 nput.FileSplit;%0A
+import org.apache.log4j.Logger;%0A
 import org.datas
@@ -1803,24 +1803,126 @@
 f = false;%0A%0A
+    /** dubug logger */%0A    final static Logger logger = Logger.getLogger(CombineShapeReader.class);%0A%0A
     /**%0A    
@@ -4044,24 +4044,48 @@
 hp and .dbf%0A
+        if(hasDbf)%7B%0A    
         if(h
@@ -4103,32 +4103,36 @@
 & !hasNextDbf)%7B%0A
+    
             Exce
@@ -4258,32 +4258,36 @@
 ));%0A            
+ 
+   
 e.printStackTrac
@@ -4283,32 +4283,36 @@
 ntStackTrace();%0A
+    
         %7Delse if
@@ -4344,32 +4344,36 @@
 f)%7B%0A            
+ 
+   
 Exception e = ne
@@ -4420,24 +4420,28 @@
 f exists%22);%0A
+    
             
@@ -4453,32 +4453,46 @@
 ntStackTrace();%0A
+            %7D%0A
         %7D%0A      

add check for hasDbf when doing hasNextKey() in shapereader