TableInputParameterGeneration implements ParameterGenerationService { @Override public void pushDown( Condition condition, ParameterGeneration parameterGeneration, StepInterface stepInterface ) throws PushDownOptimizationException { DatabaseWrapper db = getDatabaseWrapper( stepInterface ); verifyDbConnection( db ); dbMeta = db.getDatabaseMeta(); StringBuilder sqlFragment = new StringBuilder(); RowMeta paramsMeta = new RowMeta(); List<Object> params = new LinkedList<Object>(); convertCondition( condition, sqlFragment, paramsMeta, params ); String fragmentId = db.createRuntimePushDown( sqlFragment.toString(), paramsMeta, params, getParameterDefault() ); stepInterface.setVariable( parameterGeneration.getParameterName(), fragmentId ); TableInputData tableInput = getTableInputData( stepInterface ); tableInput.db = db; } TableInputParameterGeneration( ValueMetaResolver resolver ); @Override String getParameterDefault(); @Override OptimizationImpactInfo preview( Condition pushDownCondition, ParameterGeneration parameterGeneration, StepInterface stepInterface ); @Override void pushDown( Condition condition, ParameterGeneration parameterGeneration, StepInterface stepInterface );  }
@Test public void testPushDown() throws Exception { String originalQuery = "SELECT DepartmentName, COUNT(*) as EmployeeCount " + "FROM Department, Employee " + "WHERE Employee.DepartmentId = Department.DepartmentId AND ${EMPLOYEE_FILTER} "; ParameterGeneration employeeFilterParamGen = factory.createPushDown(); employeeFilterParamGen.setParameterName( "EMPLOYEE_FILTER" ); Condition employeeFilter = newCondition( "Employee.Grade", "G7" ); service.pushDown( employeeFilter, employeeFilterParamGen, stepInterface ); assertThat( data.db, is( instanceOf( DatabaseWrapper.class ) ) ); final DatabaseWrapper databaseWrapper = (DatabaseWrapper) data.db; ArgumentCaptor<String> varCaptor = ArgumentCaptor.forClass( String.class ); verify( stepInterface ).setVariable( eq( employeeFilterParamGen.getParameterName() ), varCaptor.capture() ); final List<String> fragmentIds = varCaptor.getAllValues(); assertThat( fragmentIds.size(), is( 1 ) ); assertTrue( databaseWrapper.pushDownMap.keySet().containsAll( fragmentIds ) ); Variables variables = new Variables(); variables.setVariable( employeeFilterParamGen.getParameterName(), fragmentIds.get( 0 ) ); String runtimeQuery = variables.environmentSubstitute( originalQuery ); for ( String fragment : fragmentIds ) { assertThat( runtimeQuery, containsString( fragment ) ); } RowMeta rowMeta = new RowMeta(); List<Object> values = new LinkedList<Object>(); String resultQuery = databaseWrapper.injectRuntime( databaseWrapper.pushDownMap, runtimeQuery, rowMeta, values ); String expectedQuery = "SELECT DepartmentName, COUNT(*) as EmployeeCount " + "FROM Department, Employee " + "WHERE Employee.DepartmentId = Department.DepartmentId AND Employee.Grade = ? "; assertThat( resultQuery, equalTo( expectedQuery ) ); assertThat( rowMeta.getValueMetaList(), equalTo( Arrays.asList( resolvedValueMeta ) ) ); assertThat( values, equalTo( Arrays.<Object>asList( "G7" ) ) ); }
@Test public void testFailures() throws Exception { try { service.pushDown( mock( Condition.class ), mock( ParameterGeneration.class ), mock( StepInterface.class ) ); fail(); } catch ( PushDownOptimizationException thrown ) { assertThat( thrown.getMessage(), notNullValue() ); } KettleDatabaseException expected = new KettleDatabaseException(); data.db = mock( DatabaseWrapper.class ); doThrow( expected ).when( data.db ).connect(); try { service.pushDown( mock( Condition.class ), mock( ParameterGeneration.class ), stepInterface ); fail(); } catch ( PushDownOptimizationException thrown ) { assertThat( thrown.getCause(), equalTo( (Throwable) expected ) ); } }