@Test public void testCreatePartitionedInputSplits() throws Exception { OperatorID operatorID = OperatorIDGenerator.fromUid("uid"); OperatorSubtaskState state = createOperatorSubtaskState(new StreamFlatMap<>(new StatefulFunction())); OperatorState operatorState = new OperatorState(operatorID, 1, 128); operatorState.putState(0, state); KeyedStateInputFormat<?, ?> format = new KeyedStateInputFormat<>(operatorState, new MemoryStateBackend(), Types.INT, new ReaderFunction()); KeyGroupRangeInputSplit[] splits = format.createInputSplits(4); Assert.assertEquals("Failed to properly partition operator state into input splits", 4, splits.length); }
@Override public KeyGroupRangeInputSplit[] createInputSplits(int minNumSplits) throws IOException { final int maxParallelism = operatorState.getMaxParallelism(); final List<KeyGroupRange> keyGroups = sortedKeyGroupRanges(minNumSplits, maxParallelism); return CollectionUtil.mapWithIndex( keyGroups, (keyGroupRange, index) -> createKeyGroupRangeInputSplit( operatorState, maxParallelism, keyGroupRange, index) ).toArray(KeyGroupRangeInputSplit[]::new); }
KeyedStateInputFormat extends RichInputFormat<OUT, KeyGroupRangeInputSplit> implements KeyContext { @Override public KeyGroupRangeInputSplit[] createInputSplits(int minNumSplits) throws IOException { final int maxParallelism = operatorState.getMaxParallelism(); final List<KeyGroupRange> keyGroups = sortedKeyGroupRanges(minNumSplits, maxParallelism); return CollectionUtil.mapWithIndex( keyGroups, (keyGroupRange, index) -> createKeyGroupRangeInputSplit( operatorState, maxParallelism, keyGroupRange, index) ).toArray(KeyGroupRangeInputSplit[]::new); } }
KeyedStateInputFormat extends RichInputFormat<OUT, KeyGroupRangeInputSplit> implements KeyContext { @Override public KeyGroupRangeInputSplit[] createInputSplits(int minNumSplits) throws IOException { final int maxParallelism = operatorState.getMaxParallelism(); final List<KeyGroupRange> keyGroups = sortedKeyGroupRanges(minNumSplits, maxParallelism); return CollectionUtil.mapWithIndex( keyGroups, (keyGroupRange, index) -> createKeyGroupRangeInputSplit( operatorState, maxParallelism, keyGroupRange, index) ).toArray(KeyGroupRangeInputSplit[]::new); } KeyedStateInputFormat(
		OperatorState operatorState,
		StateBackend stateBackend,
		TypeInformation<K> keyType,
		KeyedStateReaderFunction<K, OUT> userFunction); }
KeyedStateInputFormat extends RichInputFormat<OUT, KeyGroupRangeInputSplit> implements KeyContext { @Override public KeyGroupRangeInputSplit[] createInputSplits(int minNumSplits) throws IOException { final int maxParallelism = operatorState.getMaxParallelism(); final List<KeyGroupRange> keyGroups = sortedKeyGroupRanges(minNumSplits, maxParallelism); return CollectionUtil.mapWithIndex( keyGroups, (keyGroupRange, index) -> createKeyGroupRangeInputSplit( operatorState, maxParallelism, keyGroupRange, index) ).toArray(KeyGroupRangeInputSplit[]::new); } KeyedStateInputFormat(
		OperatorState operatorState,
		StateBackend stateBackend,
		TypeInformation<K> keyType,
		KeyedStateReaderFunction<K, OUT> userFunction); @Override void configure(Configuration parameters); @Override InputSplitAssigner getInputSplitAssigner(KeyGroupRangeInputSplit[] inputSplits); @Override BaseStatistics getStatistics(BaseStatistics cachedStatistics); @Override KeyGroupRangeInputSplit[] createInputSplits(int minNumSplits); @Override void openInputFormat(); @Override @SuppressWarnings("unchecked") void open(KeyGroupRangeInputSplit split); @Override void close(); @Override boolean reachedEnd(); @Override OUT nextRecord(OUT reuse); @Override @SuppressWarnings("unchecked") void setCurrentKey(Object key); @Override Object getCurrentKey(); }
KeyedStateInputFormat extends RichInputFormat<OUT, KeyGroupRangeInputSplit> implements KeyContext { @Override public KeyGroupRangeInputSplit[] createInputSplits(int minNumSplits) throws IOException { final int maxParallelism = operatorState.getMaxParallelism(); final List<KeyGroupRange> keyGroups = sortedKeyGroupRanges(minNumSplits, maxParallelism); return CollectionUtil.mapWithIndex( keyGroups, (keyGroupRange, index) -> createKeyGroupRangeInputSplit( operatorState, maxParallelism, keyGroupRange, index) ).toArray(KeyGroupRangeInputSplit[]::new); } KeyedStateInputFormat(
		OperatorState operatorState,
		StateBackend stateBackend,
		TypeInformation<K> keyType,
		KeyedStateReaderFunction<K, OUT> userFunction); @Override void configure(Configuration parameters); @Override InputSplitAssigner getInputSplitAssigner(KeyGroupRangeInputSplit[] inputSplits); @Override BaseStatistics getStatistics(BaseStatistics cachedStatistics); @Override KeyGroupRangeInputSplit[] createInputSplits(int minNumSplits); @Override void openInputFormat(); @Override @SuppressWarnings("unchecked") void open(KeyGroupRangeInputSplit split); @Override void close(); @Override boolean reachedEnd(); @Override OUT nextRecord(OUT reuse); @Override @SuppressWarnings("unchecked") void setCurrentKey(Object key); @Override Object getCurrentKey();  }