public BufferedKVExternalSorter(
		IOManager ioManager,
		BinaryRowSerializer keySerializer,
		BinaryRowSerializer valueSerializer,
		NormalizedKeyComputer nKeyComputer,
		RecordComparator comparator,
		int pageSize,
		Configuration conf) throws IOException {
	this.keySerializer = keySerializer;
	this.valueSerializer = valueSerializer;
	this.nKeyComputer = nKeyComputer;
	this.comparator = comparator;
	this.pageSize = pageSize;
	this.sorter = new QuickSort();
	this.maxNumFileHandles = conf.getInteger(ExecutionConfigOptions.TABLE_EXEC_SORT_MAX_NUM_FILE_HANDLES);
	this.compressionEnable = conf.getBoolean(ExecutionConfigOptions.TABLE_EXEC_SPILL_COMPRESSION_ENABLED);
	this.compressionCodecFactory = this.compressionEnable
		? BlockCompressionFactory.createBlockCompressionFactory(
			BlockCompressionFactory.CompressionFactoryName.LZ4.toString())
		: null;
	this.compressionBlockSize = (int) MemorySize.parse(
		conf.getString(ExecutionConfigOptions.TABLE_EXEC_SPILL_COMPRESSION_BLOCK_SIZE)).getBytes();
	this.ioManager = ioManager;
	this.enumerator = this.ioManager.createChannelEnumerator();
	this.channelManager = new SpillChannelManager();
	this.merger = new BinaryKVExternalMerger(
			ioManager, pageSize,
			maxNumFileHandles, channelManager,
			keySerializer, valueSerializer, comparator,
			compressionEnable,
		compressionCodecFactory,
			compressionBlockSize);
}
 

/**
 * Creates a new sorting thread.
 *
 * @param exceptionHandler The exception handler to call for all exceptions.
 * @param queues           The circularQueues used to pass buffers between the threads.
 */
public SortingThread(ExceptionHandler<IOException> exceptionHandler,
		CircularQueues queues) {
	super(exceptionHandler, "SortMerger sorting thread", queues);

	// members
	this.sorter = new QuickSort();
}
 

public BufferedKVExternalSorter(
		IOManager ioManager,
		BinaryRowDataSerializer keySerializer,
		BinaryRowDataSerializer valueSerializer,
		NormalizedKeyComputer nKeyComputer,
		RecordComparator comparator,
		int pageSize,
		Configuration conf) throws IOException {
	this.keySerializer = keySerializer;
	this.valueSerializer = valueSerializer;
	this.nKeyComputer = nKeyComputer;
	this.comparator = comparator;
	this.pageSize = pageSize;
	this.sorter = new QuickSort();
	this.maxNumFileHandles = conf.getInteger(ExecutionConfigOptions.TABLE_EXEC_SORT_MAX_NUM_FILE_HANDLES);
	this.compressionEnable = conf.getBoolean(ExecutionConfigOptions.TABLE_EXEC_SPILL_COMPRESSION_ENABLED);
	this.compressionCodecFactory = this.compressionEnable
		? BlockCompressionFactory.createBlockCompressionFactory(
			BlockCompressionFactory.CompressionFactoryName.LZ4.toString())
		: null;
	this.compressionBlockSize = (int) MemorySize.parse(
		conf.getString(ExecutionConfigOptions.TABLE_EXEC_SPILL_COMPRESSION_BLOCK_SIZE)).getBytes();
	this.ioManager = ioManager;
	this.enumerator = this.ioManager.createChannelEnumerator();
	this.channelManager = new SpillChannelManager();
	this.merger = new BinaryKVExternalMerger(
			ioManager, pageSize,
			maxNumFileHandles, channelManager,
			keySerializer, valueSerializer, comparator,
			compressionEnable,
		compressionCodecFactory,
			compressionBlockSize);
}
 

/**
 * Creates a new sorting thread.
 *
 * @param exceptionHandler The exception handler to call for all exceptions.
 * @param queues           The circularQueues used to pass buffers between the threads.
 */
public SortingThread(ExceptionHandler<IOException> exceptionHandler,
		CircularQueues queues) {
	super(exceptionHandler, "SortMerger sorting thread", queues);

	// members
	this.sorter = new QuickSort();
}