@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() ); }
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 ); }
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 ); } }
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 ); }
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 ); }
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 );  }