@Test public void testGetData() throws Exception { CommandOutput<JdbcCommand, ResultSet> output = new JdbcCommandOutput(); output.put(new JdbcCommand(), buildMockResultSet()); State state = new WorkUnitState(); state.setId("id"); JdbcExtractor jdbcExtractor = new MysqlExtractor((WorkUnitState) state); List<String> columnNames = Lists.newArrayListWithCapacity(COLUMNS.size()); for (MockJdbcColumn mockJdbcColumn:COLUMNS) { columnNames.add(mockJdbcColumn.getColumnName()); } jdbcExtractor.setHeaderRecord(columnNames); Iterator<JsonElement> itr = jdbcExtractor.getData(output); assertTrue(itr.hasNext()); JsonObject obj = itr.next().getAsJsonObject(); for (MockJdbcColumn column : COLUMNS) { assertEquals(obj.get(column.getColumnName()).getAsString(), column.getValue()); } }
@Override public Iterator<JsonElement> getData(CommandOutput<?, ?> response) throws DataRecordException, IOException { this.log.debug("Extract data records from resultset"); RecordSetList<JsonElement> recordSet = this.getNewRecordSetList(); if (response == null || !this.hasNextRecord()) { return recordSet.iterator(); } ResultSet resultset = null; Iterator<ResultSet> itr = (Iterator<ResultSet>) response.getResults().values().iterator(); if (itr.hasNext()) { resultset = itr.next(); } else { throw new DataRecordException("Failed to get source record count from database - Resultset has no records"); } try { final ResultSetMetaData resultsetMetadata = resultset.getMetaData(); int batchSize = this.workUnitState.getPropAsInt(ConfigurationKeys.SOURCE_QUERYBASED_FETCH_SIZE, 0); batchSize = (batchSize == 0 ? ConfigurationKeys.DEFAULT_SOURCE_FETCH_SIZE : batchSize); int recordCount = 0; while (resultset.next()) { final int numColumns = resultsetMetadata.getColumnCount(); JsonObject jsonObject = new JsonObject(); for (int i = 1; i < numColumns + 1; i++) { final String columnName = this.getHeaderRecord().get(i - 1); jsonObject.addProperty(columnName, parseColumnAsString(resultset, resultsetMetadata, i)); } recordSet.add(jsonObject); recordCount++; this.totalRecordCount++; if (recordCount >= batchSize) { this.log.info("Total number of records processed so far: " + this.totalRecordCount); return recordSet.iterator(); } } this.setNextRecord(false); this.log.info("Total number of records processed so far: " + this.totalRecordCount); return recordSet.iterator(); } catch (Exception e) { throw new DataRecordException("Failed to get records from database; error - " + e.getMessage(), e); } }
JdbcExtractor extends QueryBasedExtractor<JsonArray, JsonElement> implements SourceSpecificLayer<JsonArray, JsonElement>, JdbcSpecificLayer { @Override public Iterator<JsonElement> getData(CommandOutput<?, ?> response) throws DataRecordException, IOException { this.log.debug("Extract data records from resultset"); RecordSetList<JsonElement> recordSet = this.getNewRecordSetList(); if (response == null || !this.hasNextRecord()) { return recordSet.iterator(); } ResultSet resultset = null; Iterator<ResultSet> itr = (Iterator<ResultSet>) response.getResults().values().iterator(); if (itr.hasNext()) { resultset = itr.next(); } else { throw new DataRecordException("Failed to get source record count from database - Resultset has no records"); } try { final ResultSetMetaData resultsetMetadata = resultset.getMetaData(); int batchSize = this.workUnitState.getPropAsInt(ConfigurationKeys.SOURCE_QUERYBASED_FETCH_SIZE, 0); batchSize = (batchSize == 0 ? ConfigurationKeys.DEFAULT_SOURCE_FETCH_SIZE : batchSize); int recordCount = 0; while (resultset.next()) { final int numColumns = resultsetMetadata.getColumnCount(); JsonObject jsonObject = new JsonObject(); for (int i = 1; i < numColumns + 1; i++) { final String columnName = this.getHeaderRecord().get(i - 1); jsonObject.addProperty(columnName, parseColumnAsString(resultset, resultsetMetadata, i)); } recordSet.add(jsonObject); recordCount++; this.totalRecordCount++; if (recordCount >= batchSize) { this.log.info("Total number of records processed so far: " + this.totalRecordCount); return recordSet.iterator(); } } this.setNextRecord(false); this.log.info("Total number of records processed so far: " + this.totalRecordCount); return recordSet.iterator(); } catch (Exception e) { throw new DataRecordException("Failed to get records from database; error - " + e.getMessage(), e); } } }
JdbcExtractor extends QueryBasedExtractor<JsonArray, JsonElement> implements SourceSpecificLayer<JsonArray, JsonElement>, JdbcSpecificLayer { @Override public Iterator<JsonElement> getData(CommandOutput<?, ?> response) throws DataRecordException, IOException { this.log.debug("Extract data records from resultset"); RecordSetList<JsonElement> recordSet = this.getNewRecordSetList(); if (response == null || !this.hasNextRecord()) { return recordSet.iterator(); } ResultSet resultset = null; Iterator<ResultSet> itr = (Iterator<ResultSet>) response.getResults().values().iterator(); if (itr.hasNext()) { resultset = itr.next(); } else { throw new DataRecordException("Failed to get source record count from database - Resultset has no records"); } try { final ResultSetMetaData resultsetMetadata = resultset.getMetaData(); int batchSize = this.workUnitState.getPropAsInt(ConfigurationKeys.SOURCE_QUERYBASED_FETCH_SIZE, 0); batchSize = (batchSize == 0 ? ConfigurationKeys.DEFAULT_SOURCE_FETCH_SIZE : batchSize); int recordCount = 0; while (resultset.next()) { final int numColumns = resultsetMetadata.getColumnCount(); JsonObject jsonObject = new JsonObject(); for (int i = 1; i < numColumns + 1; i++) { final String columnName = this.getHeaderRecord().get(i - 1); jsonObject.addProperty(columnName, parseColumnAsString(resultset, resultsetMetadata, i)); } recordSet.add(jsonObject); recordCount++; this.totalRecordCount++; if (recordCount >= batchSize) { this.log.info("Total number of records processed so far: " + this.totalRecordCount); return recordSet.iterator(); } } this.setNextRecord(false); this.log.info("Total number of records processed so far: " + this.totalRecordCount); return recordSet.iterator(); } catch (Exception e) { throw new DataRecordException("Failed to get records from database; error - " + e.getMessage(), e); } } JdbcExtractor(WorkUnitState workUnitState); }
JdbcExtractor extends QueryBasedExtractor<JsonArray, JsonElement> implements SourceSpecificLayer<JsonArray, JsonElement>, JdbcSpecificLayer { @Override public Iterator<JsonElement> getData(CommandOutput<?, ?> response) throws DataRecordException, IOException { this.log.debug("Extract data records from resultset"); RecordSetList<JsonElement> recordSet = this.getNewRecordSetList(); if (response == null || !this.hasNextRecord()) { return recordSet.iterator(); } ResultSet resultset = null; Iterator<ResultSet> itr = (Iterator<ResultSet>) response.getResults().values().iterator(); if (itr.hasNext()) { resultset = itr.next(); } else { throw new DataRecordException("Failed to get source record count from database - Resultset has no records"); } try { final ResultSetMetaData resultsetMetadata = resultset.getMetaData(); int batchSize = this.workUnitState.getPropAsInt(ConfigurationKeys.SOURCE_QUERYBASED_FETCH_SIZE, 0); batchSize = (batchSize == 0 ? ConfigurationKeys.DEFAULT_SOURCE_FETCH_SIZE : batchSize); int recordCount = 0; while (resultset.next()) { final int numColumns = resultsetMetadata.getColumnCount(); JsonObject jsonObject = new JsonObject(); for (int i = 1; i < numColumns + 1; i++) { final String columnName = this.getHeaderRecord().get(i - 1); jsonObject.addProperty(columnName, parseColumnAsString(resultset, resultsetMetadata, i)); } recordSet.add(jsonObject); recordCount++; this.totalRecordCount++; if (recordCount >= batchSize) { this.log.info("Total number of records processed so far: " + this.totalRecordCount); return recordSet.iterator(); } } this.setNextRecord(false); this.log.info("Total number of records processed so far: " + this.totalRecordCount); return recordSet.iterator(); } catch (Exception e) { throw new DataRecordException("Failed to get records from database; error - " + e.getMessage(), e); } } JdbcExtractor(WorkUnitState workUnitState); Map<String, Schema> getMetadataColumnMap(); void setMetadataColumnMap(Map<String, Schema> metadataColumnMap); List<String> getMetadataColumnList(); void setMetadataColumnList(List<String> metadataColumnList); long getSampleRecordCount(); void setSampleRecordCount(long sampleRecordCount); String getExtractSql(); void setExtractSql(String extractSql); String getOutputColumnProjection(); void setOutputColumnProjection(String outputColumnProjection); String getInputColumnProjection(); void setInputColumnProjection(String inputColumnProjection); List<ColumnAttributes> getColumnAliasMap(); void addToColumnAliasMap(ColumnAttributes columnAliasMap); boolean isFirstPull(); void setFirstPull(boolean firstPull); int getTimeOut(); boolean hasNextRecord(); void setNextRecord(boolean nextRecord); @Override void setTimeOut(int timeOut); Gson getGson(); @Override void extractMetadata(String schema, String entity, WorkUnit workUnit); @Override long getMaxWatermark(String schema, String entity, String watermarkColumn, List<Predicate> predicateList,
      String watermarkSourceFormat); @Override long getSourceCount(String schema, String entity, WorkUnit workUnit, List<Predicate> predicateList); @Override Iterator<JsonElement> getRecordSet(String schema, String entity, WorkUnit workUnit,
      List<Predicate> predicateList); @Override JsonArray getSchema(CommandOutput<?, ?> response); @Override long getHighWatermark(CommandOutput<?, ?> response, String watermarkColumn, String watermarkColumnFormat); @Override long getCount(CommandOutput<?, ?> response); @Override Iterator<JsonElement> getData(CommandOutput<?, ?> response); static boolean hasJoinOperation(String selectQuery); String getLeftDelimitedIdentifier(); String getRightDelimitedIdentifier(); @Override void closeConnection(); }
JdbcExtractor extends QueryBasedExtractor<JsonArray, JsonElement> implements SourceSpecificLayer<JsonArray, JsonElement>, JdbcSpecificLayer { @Override public Iterator<JsonElement> getData(CommandOutput<?, ?> response) throws DataRecordException, IOException { this.log.debug("Extract data records from resultset"); RecordSetList<JsonElement> recordSet = this.getNewRecordSetList(); if (response == null || !this.hasNextRecord()) { return recordSet.iterator(); } ResultSet resultset = null; Iterator<ResultSet> itr = (Iterator<ResultSet>) response.getResults().values().iterator(); if (itr.hasNext()) { resultset = itr.next(); } else { throw new DataRecordException("Failed to get source record count from database - Resultset has no records"); } try { final ResultSetMetaData resultsetMetadata = resultset.getMetaData(); int batchSize = this.workUnitState.getPropAsInt(ConfigurationKeys.SOURCE_QUERYBASED_FETCH_SIZE, 0); batchSize = (batchSize == 0 ? ConfigurationKeys.DEFAULT_SOURCE_FETCH_SIZE : batchSize); int recordCount = 0; while (resultset.next()) { final int numColumns = resultsetMetadata.getColumnCount(); JsonObject jsonObject = new JsonObject(); for (int i = 1; i < numColumns + 1; i++) { final String columnName = this.getHeaderRecord().get(i - 1); jsonObject.addProperty(columnName, parseColumnAsString(resultset, resultsetMetadata, i)); } recordSet.add(jsonObject); recordCount++; this.totalRecordCount++; if (recordCount >= batchSize) { this.log.info("Total number of records processed so far: " + this.totalRecordCount); return recordSet.iterator(); } } this.setNextRecord(false); this.log.info("Total number of records processed so far: " + this.totalRecordCount); return recordSet.iterator(); } catch (Exception e) { throw new DataRecordException("Failed to get records from database; error - " + e.getMessage(), e); } } JdbcExtractor(WorkUnitState workUnitState); Map<String, Schema> getMetadataColumnMap(); void setMetadataColumnMap(Map<String, Schema> metadataColumnMap); List<String> getMetadataColumnList(); void setMetadataColumnList(List<String> metadataColumnList); long getSampleRecordCount(); void setSampleRecordCount(long sampleRecordCount); String getExtractSql(); void setExtractSql(String extractSql); String getOutputColumnProjection(); void setOutputColumnProjection(String outputColumnProjection); String getInputColumnProjection(); void setInputColumnProjection(String inputColumnProjection); List<ColumnAttributes> getColumnAliasMap(); void addToColumnAliasMap(ColumnAttributes columnAliasMap); boolean isFirstPull(); void setFirstPull(boolean firstPull); int getTimeOut(); boolean hasNextRecord(); void setNextRecord(boolean nextRecord); @Override void setTimeOut(int timeOut); Gson getGson(); @Override void extractMetadata(String schema, String entity, WorkUnit workUnit); @Override long getMaxWatermark(String schema, String entity, String watermarkColumn, List<Predicate> predicateList,
      String watermarkSourceFormat); @Override long getSourceCount(String schema, String entity, WorkUnit workUnit, List<Predicate> predicateList); @Override Iterator<JsonElement> getRecordSet(String schema, String entity, WorkUnit workUnit,
      List<Predicate> predicateList); @Override JsonArray getSchema(CommandOutput<?, ?> response); @Override long getHighWatermark(CommandOutput<?, ?> response, String watermarkColumn, String watermarkColumnFormat); @Override long getCount(CommandOutput<?, ?> response); @Override Iterator<JsonElement> getData(CommandOutput<?, ?> response); static boolean hasJoinOperation(String selectQuery); String getLeftDelimitedIdentifier(); String getRightDelimitedIdentifier(); @Override void closeConnection();  }