public InputSplit[] getSplits(JobConf jobConf, int numSplits) throws IOException {
        
        // Get the splitsize (number of rows), defaults to 100,000 as much larger seems to 
        // screw up getting accurate rows
        int splitSize = Integer.parseInt(jobConf.get(SimpleDBInputFormat.SIMPLEDB_SPLIT_SIZE, String.valueOf(MAX_SPLIT_SIZE)));
        if (splitSize > MAX_SPLIT_SIZE) {
            splitSize = MAX_SPLIT_SIZE;
        }
        
        // Get total number of rows to calculate number of splits required
        SimpleDBDAO sdb = new SimpleDBDAO(jobConf);
        long totalItems;
        if (sdb.getWhereQuery() != null) {
            totalItems = sdb.getCount();
        } else {
            totalItems = sdb.getTotalItemCount();
        }

        long totalSplits = 1;
        if (splitSize < totalItems) {
            totalSplits = totalItems / splitSize;
        }

        if (LOG.isDebugEnabled()) {
            LOG.debug("Total Rows:" + String.valueOf(totalItems));
            LOG.debug("Total Splits:" + String.valueOf(totalSplits));
        }
        
        // Array to hold splits
        ArrayList<SimpleDBInputSplit> splits = new ArrayList<SimpleDBInputSplit>();
       
        // Create Splits
        for (int i = 0; i  < totalSplits; i++) {
            
            String splitToken = sdb.getSplitToken(i, splitSize);
            long startRow = i * splitSize;
            long endRow = startRow + splitSize;
            if (endRow > totalItems) {
                endRow = totalItems;
            }
            
            SimpleDBInputSplit split = new SimpleDBInputSplit(startRow, endRow, splitToken);
            splits.add(split);
            
            if (LOG.isDebugEnabled()) {
                LOG.debug("Created Split: " + split.toString());
            }
        }
        
        // Return array of splits
        return splits.toArray(new SimpleDBInputSplit[splits.size()]);
    }