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

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.CombineFileInputFormat;

import java.io.IOException;

public class ShapeInputFormat extends CombineFileInputFormat<ShapeKey, PrimitiveShape> {
    public RecordReader<ShapeKey, PrimitiveShape> createRecordReader(InputSplit split, TaskAttemptContext context) throws IOException {
        return new CombineShapeReader();
    }

    @Override
    protected boolean isSplitable(JobContext context, Path file) {
        return false;
    }
}


@@ -270,16 +270,56 @@
 hapes;%0A%0A
+import org.apache.hadoop.fs.FileSystem;%0A
 import o
@@ -627,23 +627,122 @@
 java
-.io.IOException
+x.security.auth.login.Configuration;%0Aimport java.io.IOException;%0Aimport java.util.ArrayList;%0Aimport java.util.List
 ;%0A%0Ap
@@ -1121,12 +1121,475 @@
 e;%0A    %7D
+%0A%0A    @Override%0A    public List%3CInputSplit%3E getSplits(JobContext job) throws IOException %7B%0A        String path = job.getConfiguration().get(%22mapred.input.dir%22);%0A        String%5B%5D paths = path.split(%22,%22);%0A        List%3CInputSplit%3E splits = new ArrayList%3C%3E();%0A        for(int i = 0;i %3C paths.length; ++i)%7B%0A            job.getConfiguration().set(%22mapred.input.dir%22, paths%5Bi%5D);%0A            splits.add(super.getSplits(job).get(0));%0A        %7D%0A        return splits;%0A    %7D
 %0A%7D%0A%0A

add process for multiple small files