SQLIngester extends DataIngester { @Override public RowSet getRows(String baseQuery, Map<String, String> preds, int limit, int offset) throws SQLException { initializeConnection(); 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); } SQLIngester(MacroBaseConf conf);  SQLIngester(MacroBaseConf conf, Connection connection); abstract String getDriverClass(); abstract String getJDBCUrlPrefix(); void connect(); Schema getSchema(String baseQuery); String getRowsSql(String baseQuery,
                             Map<String, String> preds,
                             int limit,
                             int offset); @Override RowSet getRows(String baseQuery,
                          Map<String, String> preds,
                          int limit,
                          int offset); @Override MBStream<Datum> getStream();  }
@Test public void testGetRows() throws Exception { JDBCMockObjectFactory factory = new JDBCMockObjectFactory(); factory.registerMockDriver(); MockConnection connection = factory.getMockConnection(); StatementResultSetHandler statementHandler = connection.getStatementResultSetHandler(); MockResultSet result = statementHandler.createResultSet(); MockResultSetMetaData metaData = new MockResultSetMetaData(); List<Integer> fakeData = new ArrayList<>(); final int NUM_ROWS = 100; final int NUM_ATTRS = 5; final int NUM_HIGH = 2; final int NUM_LOW = 3; final int NUM_AUXILIARY = 1; final int DIMENSION = NUM_ATTRS + NUM_HIGH + NUM_LOW + NUM_AUXILIARY; Integer val = 1; for(int rno = 0; rno < NUM_ROWS; ++rno) { List<Object> rowString = new ArrayList<>(); for(Integer i = 0; i < DIMENSION; ++i) { rowString.add(val.toString()); val++; } result.addRow(rowString); } int column = 1; List<String> attributes = new ArrayList<>(); for(int i = 0; i < NUM_ATTRS; ++i) { String attrName = String.format("attr%d", i); metaData.setColumnName(column, attrName); attributes.add(attrName); column++; } List<String> lowMetrics = new ArrayList<>(); for(int i = 0; i < NUM_LOW; ++i) { String metricName = String.format("lowMetric%d", i); metaData.setColumnName(column, metricName); lowMetrics.add(metricName); column++; } List<String> highMetrics = new ArrayList<>(); for(int i = 0; i < NUM_HIGH; ++i) { String metricName = String.format("highMetric%d", i); metaData.setColumnName(column, metricName); highMetrics.add(metricName); column++; } metaData.setColumnCount(DIMENSION); result.setResultSetMetaData(metaData); statementHandler.prepareGlobalResultSet(result); List<String> allMetrics = new ArrayList<>(); allMetrics.addAll(lowMetrics); allMetrics.addAll(highMetrics); MacroBaseConf conf = new MacroBaseConf(); conf.set(MacroBaseConf.ATTRIBUTES, attributes); conf.set(MacroBaseConf.METRICS, allMetrics); conf.set(MacroBaseConf.BASE_QUERY, "SELECT * FROM test LIMIT 10000;"); SQLIngester loader = new TestSQLIngester(conf, connection); loader.connect(); Map<String, String> preds = new HashMap<>(); preds.put("c1", "v1"); RowSet rs = loader.getRows(conf.getString(MacroBaseConf.BASE_QUERY), preds, 100, 1000); assertEquals(NUM_ROWS, rs.getRows().size()); int curVal = 1; for (RowSet.Row r : rs.getRows()) { for(ColumnValue cv : r.getColumnValues()) { assertEquals(curVal, Integer.parseInt(cv.getValue())); curVal++; } } }