FilterTask implements Task { @Override public ExecutionResult execute(ExecutionResult previousTaskResult) throws SQLException { try { Predicate<DataRow> predicate = condition != null ? dr -> expressionEvaluator.evaluateAs(condition, dr, Boolean.class) : dr -> true; DataContainer newData = new DataContainer( previousTaskResult.getData(), rows -> rows.stream().filter(predicate).collect(Collectors.toList()) ); return new ExecutionResult(){ { setCount(newData.getRows().size()); setData(newData); } }; } catch (Exception e) { throw new SQLException(e); } }  @Override ExecutionResult execute(ExecutionResult previousTaskResult); void setCondition(BooleanExpression condition); Optional<BooleanExpression> getCondition(); @Override String toString();  }
@Test public void testExecute() throws Exception { FilterTask filterTask = applicationContext.getBean(FilterTask.class); BooleanExpression condition = new BinaryBooleanExpression( new SimpleBooleanExpression(new ColumnExpression(FIRST_COLUMN), BooleanRelation.EQUAL, "two"), BinaryBooleanOperator.AND, new SimpleBooleanExpression(new ColumnExpression(SECOND_COLUMN), BooleanRelation.LESS_THAN_EQUAL, 2) ); filterTask.setCondition(condition); ExecutionResult output = filterTask.execute(getInput()); assertThat(output.getCount(), equalTo(1)); assertThat(output.getData().getRows().size(), equalTo(1)); DataRow data = output.getData().getRows().get(0); assertThat(data.get(FIRST_COLUMN), equalTo("two")); assertThat(data.get(SECOND_COLUMN), equalTo(2)); }