FlowDefOptimizerImpl implements FlowDefOptimizer { void mergeRegions ( final Map<String, OperatorDef> operators, final Set<Entry<Port, Port>> connections, final List<RegionDef> regions ) { int before, after = regions.size(); do { before = after; attemptToMergeRegions( operators, connections, regions ); after = regions.size(); } while ( before != after ); } @Inject  FlowDefOptimizerImpl( final JokerConfig jokerConfig, final IdGenerator idGenerator ); @Override Pair<FlowDef, List<RegionDef>> optimize( final FlowDef flow, final List<RegionDef> regions );  }
@Test public void shouldMergeStatelessAndStatefulRegions () { final OperatorDef stateless0 = OperatorDefBuilder.newInstance( "stateless0", StatelessOperator0.class ).build(); final OperatorDef stateless1 = OperatorDefBuilder.newInstance( "stateless1", StatelessOperator1.class ).build(); final OperatorDef stateful = OperatorDefBuilder.newInstance( "stateful", StatefulOperator2.class ).build(); final FlowDef flow = new FlowDefBuilder().add( stateless0 ) .add( stateless1 ) .add( stateful ) .connect( "stateless0", "stateless1" ) .connect( "stateless1", "stateful" ) .build(); final List<RegionDef> regions = regionDefFormer.createRegions( flow ); flowOptimizer.mergeRegions( flow.getOperatorsMap(), flow.getConnections(), regions ); assertEquals( 2, regions.size() ); final RegionDef region = regions.get( 1 ); assertEquals( STATEFUL, region.getRegionType() ); assertEquals( asList( stateless1, stateful ), region.getOperators() ); }
@Test public void shouldMergeStatefulAndStatelessRegions () { final OperatorDef stateful0 = OperatorDefBuilder.newInstance( "stateful0", StatefulOperator0.class ).build(); final OperatorDef stateful1 = OperatorDefBuilder.newInstance( "stateful1", StatefulOperator2.class ).build(); final OperatorDef stateless = OperatorDefBuilder.newInstance( "stateless", StatelessOperator1.class ).build(); final FlowDef flow = new FlowDefBuilder().add( stateful0 ) .add( stateful1 ) .add( stateless ) .connect( "stateful0", "stateful1" ) .connect( "stateful1", "stateless" ) .build(); final List<RegionDef> regions = regionDefFormer.createRegions( flow ); flowOptimizer.mergeRegions( flow.getOperatorsMap(), flow.getConnections(), regions ); assertEquals( 2, regions.size() ); final RegionDef region = regions.get( 1 ); assertEquals( STATEFUL, region.getRegionType() ); assertEquals( asList( stateful1, stateless ), region.getOperators() ); }
@Test public void shouldMergeMultipleStatefulAndStatelessRegions () { final OperatorDef stateful0 = OperatorDefBuilder.newInstance( "stateful0", StatefulOperator0.class ).build(); final OperatorDef stateful1 = OperatorDefBuilder.newInstance( "stateful1", StatefulOperator2.class ).build(); final OperatorDef stateful2 = OperatorDefBuilder.newInstance( "stateful2", StatefulOperator2.class ).build(); final OperatorDef stateful3 = OperatorDefBuilder.newInstance( "stateful3", StatefulOperator2.class ).build(); final OperatorDef stateless1 = OperatorDefBuilder.newInstance( "stateless1", StatelessOperator1.class ).build(); final OperatorDef stateless2 = OperatorDefBuilder.newInstance( "stateless2", StatelessOperator1.class ).build(); final FlowDef flow = new FlowDefBuilder().add( stateful0 ) .add( stateful1 ) .add( stateful2 ) .add( stateful3 ) .add( stateless1 ) .add( stateless2 ) .connect( "stateful0", "stateful1" ) .connect( "stateful1", "stateless1" ) .connect( "stateless1", "stateful2" ) .connect( "stateful2", "stateless2" ) .connect( "stateless2", "stateful3" ) .build(); final List<RegionDef> regions = regionDefFormer.createRegions( flow ); flowOptimizer.mergeRegions( flow.getOperatorsMap(), flow.getConnections(), regions ); assertEquals( 2, regions.size() ); final RegionDef region = regions.get( 1 ); assertEquals( STATEFUL, region.getRegionType() ); assertEquals( asList( stateful1, stateless1, stateful2, stateless2, stateful3 ), region.getOperators() ); }
@Test public void shouldMergeStatefulRegions () { final OperatorDef stateful0 = OperatorDefBuilder.newInstance( "stateful0", StatefulOperator0.class ).build(); final OperatorDef stateful1 = OperatorDefBuilder.newInstance( "stateful1", StatefulOperator2.class ).build(); final OperatorDef stateful2 = OperatorDefBuilder.newInstance( "stateful2", StatefulOperator2.class ).build(); final FlowDef flow = new FlowDefBuilder().add( stateful0 ) .add( stateful1 ) .add( stateful2 ) .connect( "stateful0", "stateful1" ) .connect( "stateful1", "stateful2" ) .build(); final List<RegionDef> regions = regionDefFormer.createRegions( flow ); flowOptimizer.mergeRegions( flow.getOperatorsMap(), flow.getConnections(), regions ); assertEquals( 2, regions.size() ); final RegionDef region = regions.get( 1 ); assertEquals( STATEFUL, region.getRegionType() ); assertEquals( asList( stateful1, stateful2 ), region.getOperators() ); }
@Test public void shouldMergeStatelessRegions () { final OperatorDef stateful = OperatorDefBuilder.newInstance( "stateful", StatefulOperator0.class ).build(); final OperatorDef stateless1 = OperatorDefBuilder.newInstance( "stateless1", StatelessOperator1.class ).build(); final OperatorDef stateless2 = OperatorDefBuilder.newInstance( "stateless2", StatelessOperator1.class ).build(); final FlowDef flow = new FlowDefBuilder().add( stateful ) .add( stateless1 ) .add( stateless2 ) .connect( "stateful", "stateless1" ) .connect( "stateless1", "stateless2" ) .build(); final List<RegionDef> regions = regionDefFormer.createRegions( flow ); flowOptimizer.mergeRegions( flow.getOperatorsMap(), flow.getConnections(), regions ); assertEquals( 2, regions.size() ); final RegionDef region = regions.get( 1 ); assertEquals( STATELESS, region.getRegionType() ); assertEquals( asList( stateless1, stateless2 ), region.getOperators() ); }
@Test public void shouldNotMergePartitionedStatefulAndStatelessRegionsWhenPartitionFieldNamesAreNotPresentInStatelessRegion () { final OperatorDef stateful = OperatorDefBuilder.newInstance( "stateful", StatefulOperator0.class ) .setExtendingSchema( schema0 ) .build(); final OperatorDef partitionedStateful = OperatorDefBuilder.newInstance( "partitionedStateful", PartitionedStatefulOperator.class ) .setExtendingSchema( schema1 ) .setPartitionFieldNames( partitionFieldNames ) .build(); final OperatorRuntimeSchemaBuilder statelessSchema = new OperatorRuntimeSchemaBuilder( 1, 1 ); final RuntimeSchemaField inputField = schema1.getInputPortSchemaBuilder( 0 ).getFields().iterator().next(); final RuntimeSchemaField outputField = schema1.getOutputPortSchemaBuilder( 0 ).getFields().iterator().next(); statelessSchema.addInputField( 0, inputField.getName(), inputField.getType() ); statelessSchema.addOutputField( 0, outputField.getName(), outputField.getType() ); final OperatorDef stateless = OperatorDefBuilder.newInstance( "stateless", StatelessOperator1.class ) .setExtendingSchema( statelessSchema ) .build(); final FlowDef flow = new FlowDefBuilder().add( stateful ) .add( partitionedStateful ) .add( stateless ) .connect( "stateful", "partitionedStateful" ) .connect( "partitionedStateful", "stateless" ) .build(); final List<RegionDef> regions = regionDefFormer.createRegions( flow ); flowOptimizer.mergeRegions( flow.getOperatorsMap(), flow.getConnections(), regions ); assertEquals( 3, regions.size() ); }