code: private <IN, OUT> Collection<Integer> transformOneInputTransform(OneInputTransformation<IN, OUT> transform) {
  
  		Collection<Integer> inputIds = transform(transform.getInput());
  
  		// the recursive call might have already transformed this
  		if (alreadyTransformed.containsKey(transform)) {
  			return alreadyTransformed.get(transform);
  		}
  
  		String slotSharingGroup = determineSlotSharingGroup(transform.getSlotSharingGroup(), inputIds);
  
  		streamGraph.addOperator(transform.getId(),
  				slotSharingGroup,
  				transform.getCoLocationGroupKey(),
  				transform.getOperator(),
  				transform.getInputType(),
  				transform.getOutputType(),
  				transform.getName());
  
  		if (transform.getStateKeySelector() != null) {
  			TypeSerializer<?> keySerializer = transform.getStateKeyType().createSerializer(env.getConfig());
  			streamGraph.setOneInputStateKey(transform.getId(), transform.getStateKeySelector(), keySerializer);
  		}
  
  		streamGraph.setParallelism(transform.getId(), transform.getParallelism());
  		streamGraph.setMaxParallelism(transform.getId(), transform.getMaxParallelism());
  
  		for (Integer inputId: inputIds) {
  			streamGraph.addEdge(inputId, transform.getId(), 0);
  		}
  
  		return Collections.singleton(transform.getId());
  	}
docs: Transforms a {@code OneInputTransformation}.
  
  <p>This recursively transforms the inputs, creates a new {@code StreamNode} in the graph and
  wired the inputs to this new node.