class CSVIngester extends DataIngester {

    @Override
    public RowSet getRows(String baseQuery,
                          Map<String, String> preds,
                          int limit,
                          int offset) throws Exception{

        filename = conf.getString(MacroBaseConf.CSV_INPUT_FILE);
        Compression compression = conf.getCsvCompression();

        if (compression == Compression.GZIP) {
            InputStream fileStream = new FileInputStream(filename);
            InputStream gzipStream = new GZIPInputStream(fileStream);
            Reader decoder = new InputStreamReader(gzipStream);
            csvParser = new CSVParser(decoder, CSVFormat.DEFAULT.withHeader());
        } else {
            File csvFile = new File(conf.getString(MacroBaseConf.CSV_INPUT_FILE));
            csvParser = CSVParser.parse(csvFile, Charset.defaultCharset(), CSVFormat.DEFAULT.withHeader());
        }
        schema = csvParser.getHeaderMap();
        Iterator<CSVRecord> rawIterator = csvParser.iterator();
        int rowCount = 0;

        List<RowSet.Row> rows = Lists.newArrayList();
        while (rawIterator.hasNext() && rowCount < limit) {
            CSVRecord record = rawIterator.next();
            List<ColumnValue> columnValues = Lists.newArrayList();

            if (includeRow(record, preds)) {
                for (Map.Entry<String, Integer> se : schema.entrySet()) {
                    columnValues.add(new ColumnValue(se.getKey(),record.get(se.getValue())));
                }

                rows.add(new RowSet.Row(columnValues));
                rowCount++;
            }
        }
         return new RowSet(rows);
    }

    public  CSVIngester(MacroBaseConf conf);

}