class SQLIngester extends DataIngester {

    @Override
    public RowSet getRows(String baseQuery,
                          Map<String, String> preds,
                          int limit,
                          int offset) throws SQLException {
        initializeConnection();
        // TODO handle time column here

        Statement stmt = connection.createStatement();
        ResultSet rs = stmt.executeQuery(getRowsSql(baseQuery, preds, limit, offset));

        List<RowSet.Row> rows = Lists.newArrayList();
        while (rs.next()) {
            List<ColumnValue> columnValues = Lists.newArrayList();

            for (int i = 1; i <= rs.getMetaData().getColumnCount(); ++i) {
                columnValues.add(
                        new ColumnValue(rs.getMetaData().getColumnName(i),
                                rs.getString(i)));
            }
            rows.add(new RowSet.Row(columnValues));
        }

        return new RowSet(rows);
    }

    public  SQLIngester(MacroBaseConf conf);
    public  SQLIngester(MacroBaseConf conf, Connection connection);

}