@Test public void testGetQueryUsingPhoenixAdapter() throws Exception { Map<String, String> maxValues = new HashMap<>(); StateManager stateManager = runner.getStateManager(); processor.putColumnType("mytable" + AbstractDatabaseFetchProcessor.NAMESPACE_DELIMITER + "id", Types.INTEGER); processor.putColumnType("mytable" + AbstractDatabaseFetchProcessor.NAMESPACE_DELIMITER + "time_created", Types.TIME); processor.putColumnType("mytable" + AbstractDatabaseFetchProcessor.NAMESPACE_DELIMITER + "date_created", Types.TIMESTAMP); maxValues.put("id", "509"); maxValues.put("time_created", "12:34:57"); maxValues.put("date_created", "2016-03-07 12:34:56"); stateManager.setState(maxValues, Scope.CLUSTER); dbAdapter = new PhoenixDatabaseAdapter(); String query = processor.getQuery(dbAdapter, "myTable", null, Arrays.asList("id", "DATE_CREATED", "TIME_CREATED"), "type = \"CUSTOMER\"", stateManager.getState(Scope.CLUSTER).toMap()); assertEquals("SELECT * FROM myTable WHERE id > 509 AND DATE_CREATED >= timestamp '2016-03-07 12:34:56' AND TIME_CREATED >= time '12:34:57' AND (type = \"CUSTOMER\")", query); dbAdapter = new GenericDatabaseAdapter(); query = processor.getQuery(dbAdapter, "myTable", null, Arrays.asList("id", "DATE_CREATED", "TIME_CREATED"), "type = \"CUSTOMER\"", stateManager.getState(Scope.CLUSTER).toMap()); assertEquals("SELECT * FROM myTable WHERE id > 509 AND DATE_CREATED >= '2016-03-07 12:34:56' AND TIME_CREATED >= '12:34:57' AND (type = \"CUSTOMER\")", query); }
protected String getQuery(DatabaseAdapter dbAdapter, String tableName, String columnNames, List<String> maxValColumnNames, String customWhereClause, Map<String, String> stateMap) { return getQuery(dbAdapter, tableName, null, columnNames, maxValColumnNames, customWhereClause, stateMap); }
QueryDatabaseTable extends AbstractDatabaseFetchProcessor { protected String getQuery(DatabaseAdapter dbAdapter, String tableName, String columnNames, List<String> maxValColumnNames, String customWhereClause, Map<String, String> stateMap) { return getQuery(dbAdapter, tableName, null, columnNames, maxValColumnNames, customWhereClause, stateMap); } }
QueryDatabaseTable extends AbstractDatabaseFetchProcessor { protected String getQuery(DatabaseAdapter dbAdapter, String tableName, String columnNames, List<String> maxValColumnNames, String customWhereClause, Map<String, String> stateMap) { return getQuery(dbAdapter, tableName, null, columnNames, maxValColumnNames, customWhereClause, stateMap); } QueryDatabaseTable(); }
QueryDatabaseTable extends AbstractDatabaseFetchProcessor { protected String getQuery(DatabaseAdapter dbAdapter, String tableName, String columnNames, List<String> maxValColumnNames, String customWhereClause, Map<String, String> stateMap) { return getQuery(dbAdapter, tableName, null, columnNames, maxValColumnNames, customWhereClause, stateMap); } QueryDatabaseTable(); @Override Set<Relationship> getRelationships(); @OnScheduled void setup(final ProcessContext context); @OnStopped void stop(); @Override void onTrigger(final ProcessContext context, final ProcessSessionFactory sessionFactory); }
QueryDatabaseTable extends AbstractDatabaseFetchProcessor { protected String getQuery(DatabaseAdapter dbAdapter, String tableName, String columnNames, List<String> maxValColumnNames, String customWhereClause, Map<String, String> stateMap) { return getQuery(dbAdapter, tableName, null, columnNames, maxValColumnNames, customWhereClause, stateMap); } QueryDatabaseTable(); @Override Set<Relationship> getRelationships(); @OnScheduled void setup(final ProcessContext context); @OnStopped void stop(); @Override void onTrigger(final ProcessContext context, final ProcessSessionFactory sessionFactory); static final String RESULT_TABLENAME; static final String RESULT_ROW_COUNT; static final PropertyDescriptor FETCH_SIZE; static final PropertyDescriptor MAX_ROWS_PER_FLOW_FILE; static final PropertyDescriptor OUTPUT_BATCH_SIZE; static final PropertyDescriptor MAX_FRAGMENTS; }