@Test public void shouldReassignRegionIdsBasedOnTopologicalSort () { final OperatorDef stateful1 = OperatorDefBuilder.newInstance( "stateful1", StatefulOperator0.class ).build(); final OperatorDef stateful2 = OperatorDefBuilder.newInstance( "stateful2", StatefulOperator0.class ).build(); final OperatorDef stateful3 = OperatorDefBuilder.newInstance( "stateful3", StatefulOperator0.class ).build(); final OperatorDef stateful4 = OperatorDefBuilder.newInstance( "stateful4", StatefulOperator2.class ).build(); final OperatorDef stateful5 = OperatorDefBuilder.newInstance( "stateful5", StatefulOperator2.class ).build(); final OperatorDef stateful6 = OperatorDefBuilder.newInstance( "stateful6", StatefulOperator2.class ).build(); final FlowDef flowDef = new FlowDefBuilder().add( stateful1 ) .add( stateful2 ) .add( stateful3 ) .add( stateful4 ) .add( stateful5 ) .add( stateful6 ) .connect( "stateful1", "stateful4" ) .connect( "stateful2", "stateful4" ) .connect( "stateful4", "stateful5" ) .connect( "stateful3", "stateful5" ) .connect( "stateful5", "stateful6" ) .build(); final List<RegionDef> regions = new ArrayList<>(); regions.add( new RegionDef( 6, STATEFUL, emptyList(), singletonList( stateful1 ) ) ); regions.add( new RegionDef( 5, STATEFUL, emptyList(), singletonList( stateful2 ) ) ); regions.add( new RegionDef( 4, STATEFUL, emptyList(), singletonList( stateful3 ) ) ); regions.add( new RegionDef( 3, STATEFUL, emptyList(), singletonList( stateful4 ) ) ); regions.add( new RegionDef( 2, STATEFUL, emptyList(), singletonList( stateful5 ) ) ); regions.add( new RegionDef( 1, STATEFUL, emptyList(), singletonList( stateful6 ) ) ); final List<RegionDef> sortedRegions = flowOptimizer.reassignRegionIds( flowDef.getOperatorsMap(), flowDef.getConnections(), regions ); assertEquals( new RegionDef( 1, STATEFUL, emptyList(), singletonList( stateful1 ) ), sortedRegions.get( 0 ) ); assertEquals( new RegionDef( 2, STATEFUL, emptyList(), singletonList( stateful2 ) ), sortedRegions.get( 1 ) ); assertEquals( new RegionDef( 3, STATEFUL, emptyList(), singletonList( stateful3 ) ), sortedRegions.get( 2 ) ); assertEquals( new RegionDef( 4, STATEFUL, emptyList(), singletonList( stateful4 ) ), sortedRegions.get( 3 ) ); assertEquals( new RegionDef( 5, STATEFUL, emptyList(), singletonList( stateful5 ) ), sortedRegions.get( 4 ) ); assertEquals( new RegionDef( 6, STATEFUL, emptyList(), singletonList( stateful6 ) ), sortedRegions.get( 5 ) ); }
List<RegionDef> reassignRegionIds ( final Map<String, OperatorDef> operators, final Collection<Entry<Port, Port>> connections, final List<RegionDef> regions ) { final List<RegionDef> sortedRegions = sortTopologically( operators, connections, regions ); final List<Integer> regionIds = regions.stream().map( RegionDef::getRegionId ).sorted( Integer::compareTo ).collect( toList() ); final List<RegionDef> reassignedRegions = new ArrayList<>(); for ( int i = 0; i < regions.size(); i++ ) { final RegionDef regionDef = sortedRegions.get( i ); final int regionId = regionIds.get( i ); final RegionDef regionDefWithNewId = new RegionDef( regionId, regionDef.getRegionType(), regionDef.getPartitionFieldNames(), regionDef.getOperators() ); reassignedRegions.add( regionDefWithNewId ); } return reassignedRegions; }
FlowDefOptimizerImpl implements FlowDefOptimizer { List<RegionDef> reassignRegionIds ( final Map<String, OperatorDef> operators, final Collection<Entry<Port, Port>> connections, final List<RegionDef> regions ) { final List<RegionDef> sortedRegions = sortTopologically( operators, connections, regions ); final List<Integer> regionIds = regions.stream().map( RegionDef::getRegionId ).sorted( Integer::compareTo ).collect( toList() ); final List<RegionDef> reassignedRegions = new ArrayList<>(); for ( int i = 0; i < regions.size(); i++ ) { final RegionDef regionDef = sortedRegions.get( i ); final int regionId = regionIds.get( i ); final RegionDef regionDefWithNewId = new RegionDef( regionId, regionDef.getRegionType(), regionDef.getPartitionFieldNames(), regionDef.getOperators() ); reassignedRegions.add( regionDefWithNewId ); } return reassignedRegions; } }
FlowDefOptimizerImpl implements FlowDefOptimizer { List<RegionDef> reassignRegionIds ( final Map<String, OperatorDef> operators, final Collection<Entry<Port, Port>> connections, final List<RegionDef> regions ) { final List<RegionDef> sortedRegions = sortTopologically( operators, connections, regions ); final List<Integer> regionIds = regions.stream().map( RegionDef::getRegionId ).sorted( Integer::compareTo ).collect( toList() ); final List<RegionDef> reassignedRegions = new ArrayList<>(); for ( int i = 0; i < regions.size(); i++ ) { final RegionDef regionDef = sortedRegions.get( i ); final int regionId = regionIds.get( i ); final RegionDef regionDefWithNewId = new RegionDef( regionId, regionDef.getRegionType(), regionDef.getPartitionFieldNames(), regionDef.getOperators() ); reassignedRegions.add( regionDefWithNewId ); } return reassignedRegions; } @Inject  FlowDefOptimizerImpl( final JokerConfig jokerConfig, final IdGenerator idGenerator ); }
FlowDefOptimizerImpl implements FlowDefOptimizer { List<RegionDef> reassignRegionIds ( final Map<String, OperatorDef> operators, final Collection<Entry<Port, Port>> connections, final List<RegionDef> regions ) { final List<RegionDef> sortedRegions = sortTopologically( operators, connections, regions ); final List<Integer> regionIds = regions.stream().map( RegionDef::getRegionId ).sorted( Integer::compareTo ).collect( toList() ); final List<RegionDef> reassignedRegions = new ArrayList<>(); for ( int i = 0; i < regions.size(); i++ ) { final RegionDef regionDef = sortedRegions.get( i ); final int regionId = regionIds.get( i ); final RegionDef regionDefWithNewId = new RegionDef( regionId, regionDef.getRegionType(), regionDef.getPartitionFieldNames(), regionDef.getOperators() ); reassignedRegions.add( regionDefWithNewId ); } return reassignedRegions; } @Inject  FlowDefOptimizerImpl( final JokerConfig jokerConfig, final IdGenerator idGenerator ); @Override Pair<FlowDef, List<RegionDef>> optimize( final FlowDef flow, final List<RegionDef> regions ); }
FlowDefOptimizerImpl implements FlowDefOptimizer { List<RegionDef> reassignRegionIds ( final Map<String, OperatorDef> operators, final Collection<Entry<Port, Port>> connections, final List<RegionDef> regions ) { final List<RegionDef> sortedRegions = sortTopologically( operators, connections, regions ); final List<Integer> regionIds = regions.stream().map( RegionDef::getRegionId ).sorted( Integer::compareTo ).collect( toList() ); final List<RegionDef> reassignedRegions = new ArrayList<>(); for ( int i = 0; i < regions.size(); i++ ) { final RegionDef regionDef = sortedRegions.get( i ); final int regionId = regionIds.get( i ); final RegionDef regionDefWithNewId = new RegionDef( regionId, regionDef.getRegionType(), regionDef.getPartitionFieldNames(), regionDef.getOperators() ); reassignedRegions.add( regionDefWithNewId ); } return reassignedRegions; } @Inject  FlowDefOptimizerImpl( final JokerConfig jokerConfig, final IdGenerator idGenerator ); @Override Pair<FlowDef, List<RegionDef>> optimize( final FlowDef flow, final List<RegionDef> regions );  }