docs: private Collection<Integer> transform(StreamTransformation<?> transform) {
  
  		if (alreadyTransformed.containsKey(transform)) {
  			return alreadyTransformed.get(transform);
  		}
  
  		LOG.debug("Transforming " + transform);
  
  		if (transform.getMaxParallelism() <= 0) {
  
  			// if the max parallelism hasn't been set, then first use the job wide max parallelism
  			// from the ExecutionConfig.
  			int globalMaxParallelismFromConfig = env.getConfig().getMaxParallelism();
  			if (globalMaxParallelismFromConfig > 0) {
  				transform.setMaxParallelism(globalMaxParallelismFromConfig);
  			}
  		}
  
  		// call at least once to trigger exceptions about MissingTypeInfo
  		transform.getOutputType();
  
  		Collection<Integer> transformedIds;
  		if (transform instanceof OneInputTransformation<?, ?>) {
  			transformedIds = transformOneInputTransform((OneInputTransformation<?, ?>) transform);
  		} else if (transform instanceof TwoInputTransformation<?, ?, ?>) {
  			transformedIds = transformTwoInputTransform((TwoInputTransformation<?, ?, ?>) transform);
  		} else if (transform instanceof SourceTransformation<?>) {
  			transformedIds = transformSource((SourceTransformation<?>) transform);
  		} else if (transform instanceof SinkTransformation<?>) {
  			transformedIds = transformSink((SinkTransformation<?>) transform);
  		} else if (transform instanceof UnionTransformation<?>) {
  			transformedIds = transformUnion((UnionTransformation<?>) transform);
  		} else if (transform instanceof SplitTransformation<?>) {
  			transformedIds = transformSplit((SplitTransformation<?>) transform);
  		} else if (transform instanceof SelectTransformation<?>) {
  			transformedIds = transformSelect((SelectTransformation<?>) transform);
  		} else if (transform instanceof FeedbackTransformation<?>) {
  			transformedIds = transformFeedback((FeedbackTransformation<?>) transform);
  		} else if (transform instanceof CoFeedbackTransformation<?>) {
  			transformedIds = transformCoFeedback((CoFeedbackTransformation<?>) transform);
  		} else if (transform instanceof PartitionTransformation<?>) {
  			transformedIds = transformPartition((PartitionTransformation<?>) transform);
  		} else if (transform instanceof SideOutputTransformation<?>) {
  			transformedIds = transformSideOutput((SideOutputTransformation<?>) transform);
  		} else {
  			throw new IllegalStateException("Unknown transformation: " + transform);
  		}
  
  		// need this check because the iterate transformation adds itself before
  		// transforming the feedback edges
  		if (!alreadyTransformed.containsKey(transform)) {
  			alreadyTransformed.put(transform, transformedIds);
  		}
  
  		if (transform.getBufferTimeout() >= 0) {
  			streamGraph.setBufferTimeout(transform.getId(), transform.getBufferTimeout());
  		}
  		if (transform.getUid() != null) {
  			streamGraph.setTransformationUID(transform.getId(), transform.getUid());
  		}
  		if (transform.getUserProvidedNodeHash() != null) {
  			streamGraph.setTransformationUserHash(transform.getId(), transform.getUserProvidedNodeHash());
  		}
  
  		if (!streamGraph.getExecutionConfig().hasAutoGeneratedUIDsEnabled()) {
  			if (transform.getUserProvidedNodeHash() == null && transform.getUid() == null) {
  				throw new IllegalStateException("Auto generated UIDs have been disabled " +
  					"but no UID or hash has been assigned to operator " + transform.getName());
  			}
  		}
  
  		if (transform.getMinResources() != null && transform.getPreferredResources() != null) {
  			streamGraph.setResources(transform.getId(), transform.getMinResources(), transform.getPreferredResources());
  		}
  
  		return transformedIds;
  	}