@Test public void testGetQuery() throws Exception { String query = processor.getQuery(dbAdapter, "myTable", null, null, null, null); assertEquals("SELECT * FROM myTable", query); query = processor.getQuery(dbAdapter, "myTable", "col1,col2", null, null, null); assertEquals("SELECT col1,col2 FROM myTable", query); query = processor.getQuery(dbAdapter, "myTable", null, Collections.singletonList("id"), null, null); assertEquals("SELECT * FROM myTable", query); Map<String, String> maxValues = new HashMap<>(); maxValues.put("id", "509"); StateManager stateManager = runner.getStateManager(); stateManager.setState(maxValues, Scope.CLUSTER); processor.putColumnType(processor.getStateKey("mytable", "id", dbAdapter), Types.INTEGER); query = processor.getQuery(dbAdapter, "myTable", null, Collections.singletonList("id"), null, stateManager.getState(Scope.CLUSTER).toMap()); assertEquals("SELECT * FROM myTable WHERE id > 509", query); maxValues.put("date_created", "2016-03-07 12:34:56"); stateManager.setState(maxValues, Scope.CLUSTER); processor.putColumnType(processor.getStateKey("mytable", "date_created", dbAdapter), Types.TIMESTAMP); query = processor.getQuery(dbAdapter, "myTable", null, Arrays.asList("id", "DATE_CREATED"), null, stateManager.getState(Scope.CLUSTER).toMap()); assertEquals("SELECT * FROM myTable WHERE id > 509 AND DATE_CREATED >= '2016-03-07 12:34:56'", query); maxValues.put("mytable@!@date-created", "2016-03-07 12:34:56"); stateManager.setState(maxValues, Scope.CLUSTER); processor.putColumnType(processor.getStateKey("\"myTable\"", "\"DATE-CREATED\"", dbAdapter), Types.TIMESTAMP); query = processor.getQuery(dbAdapter, "\"myTable\"", null, Arrays.asList("id", "\"DATE-CREATED\""), null, stateManager.getState(Scope.CLUSTER).toMap()); assertEquals("SELECT * FROM \"myTable\" WHERE id > 509 AND \"DATE-CREATED\" >= '2016-03-07 12:34:56'", query); dbAdapter = new MySQLDatabaseAdapter(); processor.putColumnType(processor.getStateKey("`myTable`", "`DATE-CREATED`", dbAdapter), Types.TIMESTAMP); query = processor.getQuery(dbAdapter, "`myTable`", null, Arrays.asList("id", "`DATE-CREATED`"), null, stateManager.getState(Scope.CLUSTER).toMap()); assertEquals("SELECT * FROM `myTable` WHERE id > 509 AND `DATE-CREATED` >= '2016-03-07 12:34:56'", query); dbAdapter = new MSSQLDatabaseAdapter(); processor.putColumnType(processor.getStateKey("[myTable]", "[DATE-CREATED]", dbAdapter), Types.TIMESTAMP); query = processor.getQuery(dbAdapter, "[myTable]", null, Arrays.asList("id", "[DATE-CREATED]"), null, stateManager.getState(Scope.CLUSTER).toMap()); assertEquals("SELECT * FROM [myTable] WHERE id > 509 AND [DATE-CREATED] >= '2016-03-07 12:34:56'", query); dbAdapter = new OracleDatabaseAdapter(); query = processor.getQuery(dbAdapter, "myTable", null, Arrays.asList("id", "DATE_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 (type = \"CUSTOMER\")", query); processor.putColumnType("mytable" + AbstractDatabaseFetchProcessor.NAMESPACE_DELIMITER + "time_created", Types.TIME); maxValues.clear(); maxValues.put("id", "509"); maxValues.put("time_created", "12:34:57"); maxValues.put("date_created", "2016-03-07 12:34:56"); stateManager = runner.getStateManager(); stateManager.clear(Scope.CLUSTER); stateManager.setState(maxValues, Scope.CLUSTER); 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 >= timestamp '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; }