SQLIngester extends DataIngester { @Override public MBStream<Datum> getStream() throws Exception { if(!connected) { connect(); connected = true; } return output; } 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 testGetData() 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; Set<String> firstVals = new HashSet<>(); for(int rno = 0; rno < NUM_ROWS; ++rno) { List<Object> rowString = new ArrayList<>(); firstVals.add(val.toString()); 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;"); DatumEncoder encoder = conf.getEncoder(); SQLIngester ingester = new TestSQLIngester(conf, connection); List<Datum> data = ingester.getStream().drain(); for (Datum d : data) { String firstValString = encoder.getAttribute(d.attributes().get(0)).getValue(); Integer curValInt = Integer.parseInt(firstValString); assertTrue(firstVals.contains(firstValString)); firstVals.remove(firstValString); column = 0; for(int i = 0; i < NUM_ATTRS; ++i) { assertEquals(curValInt, (Integer) Integer.parseInt(encoder.getAttribute(d.attributes().get(i)).getValue())); curValInt++; column++; } for(int i = 0; i < NUM_LOW+NUM_HIGH; ++i) { assertEquals(curValInt, d.metrics().getEntry(i), 0); curValInt++; } } assertTrue(firstVals.isEmpty()); }
@Test public void testGetData2() 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 NUM_CONTEXTUAL_DISCRETE = 1; final int NUM_CONTEXTUAL_DOUBLE = 1; final int DIMENSION = NUM_ATTRS + NUM_HIGH + NUM_LOW + NUM_CONTEXTUAL_DISCRETE + NUM_CONTEXTUAL_DOUBLE + NUM_AUXILIARY; Integer val = 1; Set<String> firstVals = new HashSet<>(); for(int rno = 0; rno < NUM_ROWS; ++rno) { List<Object> rowString = new ArrayList<>(); firstVals.add(val.toString()); 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;"); DatumEncoder encoder = conf.getEncoder(); SQLIngester ingester = new TestSQLIngester(conf, connection); List<Datum> data = ingester.getStream().drain(); for (Datum d : data) { String firstValString = encoder.getAttribute(d.attributes().get(0)).getValue(); Integer curValInt = Integer.parseInt(firstValString); assertTrue(firstVals.contains(firstValString)); firstVals.remove(firstValString); column = 0; for(int i = 0; i < NUM_ATTRS; ++i) { assertEquals(curValInt, (Integer) Integer.parseInt(encoder.getAttribute(d.attributes().get(i)).getValue())); curValInt++; column++; } for(int i = 0; i < NUM_LOW+NUM_HIGH; ++i) { assertEquals(curValInt, d.metrics().getEntry(i), 0); curValInt++; } } assertTrue(firstVals.isEmpty()); }