protected void initLocalStrategies(int numInputs) throws Exception {

		final MemoryManager memMan = getMemoryManager();
		final IOManager ioMan = getIOManager();

		this.localStrategies = new CloseableInputProvider[numInputs];
		this.inputs = new MutableObjectIterator[numInputs];
		this.excludeFromReset = new boolean[numInputs];
		this.inputIsCached = new boolean[numInputs];
		this.inputIsAsyncMaterialized = new boolean[numInputs];
		this.materializationMemory = new int[numInputs];

		// set up the local strategies first, such that the can work before any temp barrier is created
		for (int i = 0; i < numInputs; i++) {
			initInputLocalStrategy(i);
		}

		// we do another loop over the inputs, because we want to instantiate all
		// sorters, etc before requesting the first input (as this call may block)

		// we have two types of materialized inputs, and both are replayable (can act as a cache)
		// The first variant materializes in a different thread and hence
		// acts as a pipeline breaker. this one should only be there, if a pipeline breaker is needed.
		// the second variant spills to the side and will not read unless the result is also consumed
		// in a pipelined fashion.
		this.resettableInputs = new SpillingResettableMutableObjectIterator[numInputs];
		this.tempBarriers = new TempBarrier[numInputs];

		for (int i = 0; i < numInputs; i++) {
			final int memoryPages;
			final boolean async = this.config.isInputAsynchronouslyMaterialized(i);
			final boolean cached =  this.config.isInputCached(i);

			this.inputIsAsyncMaterialized[i] = async;
			this.inputIsCached[i] = cached;

			if (async || cached) {
				memoryPages = memMan.computeNumberOfPages(this.config.getInputMaterializationMemory(i));
				if (memoryPages <= 0) {
					throw new Exception("Input marked as materialized/cached, but no memory for materialization provided.");
				}
				this.materializationMemory[i] = memoryPages;
			} else {
				memoryPages = 0;
			}

			if (async) {
				@SuppressWarnings({ "unchecked", "rawtypes" })
				TempBarrier<?> barrier = new TempBarrier(this, getInput(i), this.inputSerializers[i], memMan, ioMan, memoryPages);
				barrier.startReading();
				this.tempBarriers[i] = barrier;
				this.inputs[i] = null;
			} else if (cached) {
				@SuppressWarnings({ "unchecked", "rawtypes" })
				SpillingResettableMutableObjectIterator<?> iter = new SpillingResettableMutableObjectIterator(
					getInput(i), this.inputSerializers[i].getSerializer(), getMemoryManager(), getIOManager(), memoryPages, this);
				this.resettableInputs[i] = iter;
				this.inputs[i] = iter;
			}
		}
	}
protected void initLocalStrategies ( int numInputs ) throws Exception { final MemoryManager memMan = getMemoryManager ( ) ; final IOManager ioMan = getIOManager ( ) ; this . localStrategies = new CloseableInputProvider [ numInputs ] ; this . inputs = new MutableObjectIterator [ numInputs ] ; this . excludeFromReset = new boolean [ numInputs ] ; this . inputIsCached = new boolean [ numInputs ] ; this . inputIsAsyncMaterialized = new boolean [ numInputs ] ; this . materializationMemory = new int [ numInputs ] ; for ( int i = 0 ; i < numInputs ; i ++ ) { initInputLocalStrategy ( i ) ; } this . resettableInputs = new SpillingResettableMutableObjectIterator [ numInputs ] ; this . tempBarriers = new TempBarrier [ numInputs ] ; for ( int i = 0 ; i < numInputs ; i ++ ) { final int memoryPages ; final boolean async = this . config . isInputAsynchronouslyMaterialized ( i ) ; final boolean cached = this . config . isInputCached ( i ) ; this . inputIsAsyncMaterialized [ i ] = async ; this . inputIsCached [ i ] = cached ; if ( async || cached ) { memoryPages = memMan . computeNumberOfPages ( this . config . getInputMaterializationMemory ( i ) ) ; if ( memoryPages <= 0 ) { throw new Exception ( "Input marked as materialized/cached, but no memory for materialization provided." ) ; } this . materializationMemory [ i ] = memoryPages ; } else { memoryPages = 0 ; } if ( async ) { @ SuppressWarnings ( { "unchecked" , "rawtypes" } ) TempBarrier < ? > barrier = new TempBarrier ( this , getInput ( i ) , this . inputSerializers [ i ] , memMan , ioMan , memoryPages ) ; barrier . startReading ( ) ; this . tempBarriers [ i ] = barrier ; this . inputs [ i ] = null ; } else if ( cached ) { @ SuppressWarnings ( { "unchecked" , "rawtypes" } ) SpillingResettableMutableObjectIterator < ? > iter = new SpillingResettableMutableObjectIterator ( getInput ( i ) , this . inputSerializers [ i ] . getSerializer ( ) , getMemoryManager ( ) , getIOManager ( ) , memoryPages , this ) ; this . resettableInputs [ i ] = iter ; this . inputs [ i ] = iter ; } } }
NOTE: This method must be invoked after the invocation of {@code #initInputReaders()} and
{@code #initInputSerializersAndComparators(int)}!

@param numInputs