protected AbstractBlockResettableIterator(TypeSerializer<T> serializer, MemoryManager memoryManager,
		int numPages, AbstractInvokable ownerTask)
throws MemoryAllocationException
{
	if (numPages < 1) {
		throw new IllegalArgumentException("Block Resettable iterator requires at leat one page of memory");
	}
	
	this.memoryManager = memoryManager;
	this.serializer = serializer;
	
	this.emptySegments = new ArrayList<MemorySegment>(numPages);
	this.fullSegments = new ArrayList<MemorySegment>(numPages);
	memoryManager.allocatePages(ownerTask, emptySegments, numPages);
	
	this.collectingView = new SimpleCollectingOutputView(this.fullSegments, 
					new ListMemorySegmentSource(this.emptySegments), memoryManager.getPageSize());
	this.readView = new RandomAccessInputView(this.fullSegments, memoryManager.getPageSize());
	
	if (LOG.isDebugEnabled()) {
		LOG.debug("Iterator initialized using " + numPages + " memory buffers.");
	}
}
 

public RecordArea(int segmentSize) {
	int segmentSizeBits = MathUtils.log2strict(segmentSize);

	if ((segmentSize & (segmentSize - 1)) != 0) {
		throw new IllegalArgumentException("Segment size must be a power of 2!");
	}

	this.segmentSizeBits = segmentSizeBits;
	this.segmentSizeMask = segmentSize - 1;

	outView = new RecordAreaOutputView(segmentSize);
	try {
		addSegment();
	} catch (EOFException ex) {
		throw new RuntimeException("Bug in InPlaceMutableHashTable: we should have caught it earlier " +
			"that we don't have enough segments.");
	}
	inView = new RandomAccessInputView(segments, segmentSize);
}
 

private void testSerialize(
	BinaryRow row, MemorySegment[] memorySegments,
	Map<MemorySegment, Integer> msIndex, BinaryRowSerializer serializer, int position,
	int rowSize) throws IOException {
	RandomAccessOutputView out = new RandomAccessOutputView(memorySegments, 64);
	out.skipBytesToWrite(position);
	row.setTotalSize(rowSize);

	// this `row` contains random bytes, and now we're going to serialize `rowSize` bytes
	// (not including the row header) of the contents
	serializer.serializeToPages(row, out);

	// let's see how many segments we have written
	int segNumber = msIndex.get(out.getCurrentSegment()) + 1;
	int lastSegSize = out.getCurrentPositionInSegment();

	// now deserialize from the written segments
	ArrayList<MemorySegment> segments = new ArrayList<>(Arrays.asList(memorySegments).subList(0, segNumber));
	RandomAccessInputView input = new RandomAccessInputView(segments, 64, lastSegSize);
	input.skipBytesToRead(position);
	BinaryRow mapRow = serializer.mapFromPages(input);

	assertEquals(row, mapRow);
}
 

@Test
public void testSerStringToKryo() throws IOException {
	KryoSerializer<BinaryString> serializer = new KryoSerializer<>(
		BinaryString.class, new ExecutionConfig());

	BinaryString string = BinaryString.fromString("hahahahaha");
	RandomAccessOutputView out = new RandomAccessOutputView(
		new MemorySegment[]{MemorySegmentFactory.wrap(new byte[1024])}, 64);
	serializer.serialize(string, out);

	RandomAccessInputView input = new RandomAccessInputView(
		new ArrayList<>(Collections.singletonList(out.getCurrentSegment())), 64, 64);
	BinaryString newStr = serializer.deserialize(input);

	assertEquals(string, newStr);
}
 

protected AbstractBlockResettableIterator(TypeSerializer<T> serializer, MemoryManager memoryManager,
		int numPages, AbstractInvokable ownerTask)
throws MemoryAllocationException
{
	if (numPages < 1) {
		throw new IllegalArgumentException("Block Resettable iterator requires at leat one page of memory");
	}
	
	this.memoryManager = memoryManager;
	this.serializer = serializer;
	
	this.emptySegments = new ArrayList<MemorySegment>(numPages);
	this.fullSegments = new ArrayList<MemorySegment>(numPages);
	memoryManager.allocatePages(ownerTask, emptySegments, numPages);
	
	this.collectingView = new SimpleCollectingOutputView(this.fullSegments, 
					new ListMemorySegmentSource(this.emptySegments), memoryManager.getPageSize());
	this.readView = new RandomAccessInputView(this.fullSegments, memoryManager.getPageSize());
	
	if (LOG.isDebugEnabled()) {
		LOG.debug("Iterator initialized using " + numPages + " memory buffers.");
	}
}
 

public RecordArea(int segmentSize) {
	int segmentSizeBits = MathUtils.log2strict(segmentSize);

	if ((segmentSize & (segmentSize - 1)) != 0) {
		throw new IllegalArgumentException("Segment size must be a power of 2!");
	}

	this.segmentSizeBits = segmentSizeBits;
	this.segmentSizeMask = segmentSize - 1;

	outView = new RecordAreaOutputView(segmentSize);
	try {
		addSegment();
	} catch (EOFException ex) {
		throw new RuntimeException("Bug in InPlaceMutableHashTable: we should have caught it earlier " +
			"that we don't have enough segments.");
	}
	inView = new RandomAccessInputView(segments, segmentSize);
}
 

private void testSerialize(
	BinaryRowData row, MemorySegment[] memorySegments,
	Map<MemorySegment, Integer> msIndex, BinaryRowDataSerializer serializer, int position,
	int rowSize) throws IOException {
	RandomAccessOutputView out = new RandomAccessOutputView(memorySegments, 64);
	out.skipBytesToWrite(position);
	row.setTotalSize(rowSize);

	// this `row` contains random bytes, and now we're going to serialize `rowSize` bytes
	// (not including the row header) of the contents
	serializer.serializeToPages(row, out);

	// let's see how many segments we have written
	int segNumber = msIndex.get(out.getCurrentSegment()) + 1;
	int lastSegSize = out.getCurrentPositionInSegment();

	// now deserialize from the written segments
	ArrayList<MemorySegment> segments = new ArrayList<>(Arrays.asList(memorySegments).subList(0, segNumber));
	RandomAccessInputView input = new RandomAccessInputView(segments, 64, lastSegSize);
	input.skipBytesToRead(position);
	BinaryRowData mapRow = serializer.mapFromPages(input);

	assertEquals(row, mapRow);
}
 

@Test
public void testSerStringToKryo() throws IOException {
	KryoSerializer<BinaryStringData> serializer = new KryoSerializer<>(
		BinaryStringData.class, new ExecutionConfig());

	BinaryStringData string = BinaryStringData.fromString("hahahahaha");
	RandomAccessOutputView out = new RandomAccessOutputView(
		new MemorySegment[]{MemorySegmentFactory.wrap(new byte[1024])}, 64);
	serializer.serialize(string, out);

	RandomAccessInputView input = new RandomAccessInputView(
		new ArrayList<>(Collections.singletonList(out.getCurrentSegment())), 64, 64);
	StringData newStr = serializer.deserialize(input);

	assertEquals(string, newStr);
}
 

protected AbstractBlockResettableIterator(TypeSerializer<T> serializer, MemoryManager memoryManager,
		int numPages, AbstractInvokable ownerTask)
throws MemoryAllocationException
{
	if (numPages < 1) {
		throw new IllegalArgumentException("Block Resettable iterator requires at leat one page of memory");
	}
	
	this.memoryManager = memoryManager;
	this.serializer = serializer;
	
	this.emptySegments = new ArrayList<MemorySegment>(numPages);
	this.fullSegments = new ArrayList<MemorySegment>(numPages);
	memoryManager.allocatePages(ownerTask, emptySegments, numPages);
	
	this.collectingView = new SimpleCollectingOutputView(this.fullSegments, 
					new ListMemorySegmentSource(this.emptySegments), memoryManager.getPageSize());
	this.readView = new RandomAccessInputView(this.fullSegments, memoryManager.getPageSize());
	
	if (LOG.isDebugEnabled()) {
		LOG.debug("Iterator initialized using " + numPages + " memory buffers.");
	}
}
 

public RecordArea(int segmentSize) {
	int segmentSizeBits = MathUtils.log2strict(segmentSize);

	if ((segmentSize & (segmentSize - 1)) != 0) {
		throw new IllegalArgumentException("Segment size must be a power of 2!");
	}

	this.segmentSizeBits = segmentSizeBits;
	this.segmentSizeMask = segmentSize - 1;

	outView = new RecordAreaOutputView(segmentSize);
	try {
		addSegment();
	} catch (EOFException ex) {
		throw new RuntimeException("Bug in InPlaceMutableHashTable: we should have caught it earlier " +
			"that we don't have enough segments.");
	}
	inView = new RandomAccessInputView(segments, segmentSize);
}
 

public InPlaceMutableHashTable(TypeSerializer<T> serializer, TypeComparator<T> comparator, List<MemorySegment> memory) {
	super(serializer, comparator);
	this.numAllMemorySegments = memory.size();
	this.freeMemorySegments = new ArrayList<>(memory);

	// some sanity checks first
	if (freeMemorySegments.size() < MIN_NUM_MEMORY_SEGMENTS) {
		throw new IllegalArgumentException("Too few memory segments provided. InPlaceMutableHashTable needs at least " +
			MIN_NUM_MEMORY_SEGMENTS + " memory segments.");
	}

	// Get the size of the first memory segment and record it. All further buffers must have the same size.
	// the size must also be a power of 2
	segmentSize = freeMemorySegments.get(0).size();
	if ( (segmentSize & segmentSize - 1) != 0) {
		throw new IllegalArgumentException("Hash Table requires buffers whose size is a power of 2.");
	}

	this.numBucketsPerSegment = segmentSize / bucketSize;
	this.numBucketsPerSegmentBits = MathUtils.log2strict(this.numBucketsPerSegment);
	this.numBucketsPerSegmentMask = (1 << this.numBucketsPerSegmentBits) - 1;

	recordArea = new RecordArea(segmentSize);

	stagingSegments = new ArrayList<>();
	stagingSegments.add(forcedAllocateSegment());
	stagingSegmentsInView = new RandomAccessInputView(stagingSegments, segmentSize);
	stagingSegmentsOutView = new StagingOutputView(stagingSegments, segmentSize);

	prober = new HashTableProber<>(buildSideComparator, new SameTypePairComparator<>(buildSideComparator));

	enableResize = buildSideSerializer.getLength() == -1;
}
 

private void load(long numElements, RandomAccessInputView recordInputView) throws IOException {
	for (int index = 0; index < numElements; index++) {
		serializer.checkSkipReadForFixLengthPart(recordInputView);
		long pointer = recordInputView.getReadPosition();
		BinaryRow row = serializer1.mapFromPages(row1, recordInputView);
		valueSerializer.checkSkipReadForFixLengthPart(recordInputView);
		recordInputView.skipBytes(recordInputView.readInt());
		boolean success = checkNextIndexOffset();
		checkArgument(success);
		writeIndexAndNormalizedKey(row, pointer);
	}
}
 

public BinaryIndexedSortable(
		NormalizedKeyComputer normalizedKeyComputer,
		BinaryRowSerializer serializer,
		RecordComparator comparator,
		ArrayList<MemorySegment> recordBufferSegments,
		MemorySegmentPool memorySegmentPool) throws IOException {
	if (normalizedKeyComputer == null || serializer == null) {
		throw new NullPointerException();
	}
	this.normalizedKeyComputer = normalizedKeyComputer;
	this.serializer = serializer;
	this.comparator = comparator;
	this.memorySegmentPool = memorySegmentPool;
	this.useNormKeyUninverted = !normalizedKeyComputer.invertKey();

	this.numKeyBytes = normalizedKeyComputer.getNumKeyBytes();

	int segmentSize = memorySegmentPool.pageSize();
	this.recordBuffer = new RandomAccessInputView(recordBufferSegments, segmentSize);
	this.recordBufferForComparison = new RandomAccessInputView(recordBufferSegments, segmentSize);

	this.normalizedKeyFullyDetermines = normalizedKeyComputer.isKeyFullyDetermines();

	// compute the index entry size and limits
	this.indexEntrySize = numKeyBytes + OFFSET_LEN;
	this.indexEntriesPerSegment = segmentSize / this.indexEntrySize;
	this.lastIndexEntryOffset = (this.indexEntriesPerSegment - 1) * this.indexEntrySize;

	this.serializer1 = (BinaryRowSerializer) serializer.duplicate();
	this.serializer2 = (BinaryRowSerializer) serializer.duplicate();
	this.row1 = this.serializer1.createInstance();
	this.row2 = this.serializer2.createInstance();

	// set to initial state
	this.sortIndex = new ArrayList<>(16);
	this.currentSortIndexSegment = nextMemorySegment();
	sortIndex.add(currentSortIndexSegment);
}
 

private BuildSideBuffer(MemorySegment initialSegment, MemorySegmentSource memSource) {
	super(initialSegment, initialSegment.size(), 0);

	this.memSource = memSource;
	this.sizeBits = MathUtils.log2strict(initialSegment.size());
	this.targetList = new ArrayList<>();
	this.buildStageSegments = new ArrayList<>();
	this.buildStageSegments.add(initialSegment);
	this.buildStageInputView = new RandomAccessInputView(
			buildStageSegments, initialSegment.size());
}
 

private InMemoryBufferIterator newIterator(int beginRow, long offset) {
	checkArgument(offset >= 0, "`offset` can't be negative!");

	RandomAccessInputView recordBuffer = new RandomAccessInputView(
			this.recordBufferSegments, this.segmentSize, numBytesInLastBuffer);
	return new InMemoryBufferIterator(recordCount, beginRow, offset, recordBuffer);
}
 

@Test
public void testGenericObject() throws Exception {

	GenericTypeInfo<MyObj> info = new GenericTypeInfo<>(MyObj.class);
	TypeSerializer<MyObj> genericSerializer = info.createSerializer(new ExecutionConfig());

	BinaryRow row = new BinaryRow(4);
	BinaryRowWriter writer = new BinaryRowWriter(row);
	writer.writeInt(0, 0);

	BinaryGeneric<MyObj> myObj1 = new BinaryGeneric<>(new MyObj(0, 1), genericSerializer);
	writer.writeGeneric(1, myObj1);
	BinaryGeneric<MyObj> myObj2 = new BinaryGeneric<>(new MyObj(123, 5.0), genericSerializer);
	myObj2.ensureMaterialized();
	writer.writeGeneric(2, myObj2);
	BinaryGeneric<MyObj> myObj3 = new BinaryGeneric<>(new MyObj(1, 1), genericSerializer);
	writer.writeGeneric(3, myObj3);
	writer.complete();

	assertTestGenericObjectRow(row, genericSerializer);

	// getBytes from var-length memorySegments.
	BinaryRowSerializer serializer = new BinaryRowSerializer(4);
	MemorySegment[] memorySegments = new MemorySegment[3];
	ArrayList<MemorySegment> memorySegmentList = new ArrayList<>();
	for (int i = 0; i < 3; i++) {
		memorySegments[i] = MemorySegmentFactory.wrap(new byte[64]);
		memorySegmentList.add(memorySegments[i]);
	}
	RandomAccessOutputView out = new RandomAccessOutputView(memorySegments, 64);
	serializer.serializeToPages(row, out);

	BinaryRow mapRow = serializer.mapFromPages(new RandomAccessInputView(memorySegmentList, 64));
	assertTestGenericObjectRow(mapRow, genericSerializer);
}
 

@Test
public void testHashAndCopy() throws IOException {
	MemorySegment[] segments = new MemorySegment[3];
	for (int i = 0; i < 3; i++) {
		segments[i] = MemorySegmentFactory.wrap(new byte[64]);
	}
	RandomAccessOutputView out = new RandomAccessOutputView(segments, 64);
	BinaryRowSerializer serializer = new BinaryRowSerializer(2);

	BinaryRow row = new BinaryRow(2);
	BinaryRowWriter writer = new BinaryRowWriter(row);
	writer.writeString(0, BinaryString.fromString("hahahahahahahahahahahahahahahahahahahhahahahahahahahahah"));
	writer.writeString(1, BinaryString.fromString("hahahahahahahahahahahahahahahahahahahhahahahahahahahahaa"));
	writer.complete();
	serializer.serializeToPages(row, out);

	ArrayList<MemorySegment> segmentList = new ArrayList<>(Arrays.asList(segments));
	RandomAccessInputView input = new RandomAccessInputView(segmentList, 64, 64);

	BinaryRow mapRow = serializer.mapFromPages(input);
	assertEquals(row, mapRow);
	assertEquals(row.getString(0), mapRow.getString(0));
	assertEquals(row.getString(1), mapRow.getString(1));
	assertNotEquals(row.getString(0), mapRow.getString(1));

	// test if the hash code before and after serialization are the same
	assertEquals(row.hashCode(), mapRow.hashCode());
	assertEquals(row.getString(0).hashCode(), mapRow.getString(0).hashCode());
	assertEquals(row.getString(1).hashCode(), mapRow.getString(1).hashCode());

	// test if the copy method produce a row with the same contents
	assertEquals(row.copy(), mapRow.copy());
	assertEquals(row.getString(0).copy(), mapRow.getString(0).copy());
	assertEquals(row.getString(1).copy(), mapRow.getString(1).copy());
}
 

public InPlaceMutableHashTable(TypeSerializer<T> serializer, TypeComparator<T> comparator, List<MemorySegment> memory) {
	super(serializer, comparator);
	this.numAllMemorySegments = memory.size();
	this.freeMemorySegments = new ArrayList<>(memory);

	// some sanity checks first
	if (freeMemorySegments.size() < MIN_NUM_MEMORY_SEGMENTS) {
		throw new IllegalArgumentException("Too few memory segments provided. InPlaceMutableHashTable needs at least " +
			MIN_NUM_MEMORY_SEGMENTS + " memory segments.");
	}

	// Get the size of the first memory segment and record it. All further buffers must have the same size.
	// the size must also be a power of 2
	segmentSize = freeMemorySegments.get(0).size();
	if ( (segmentSize & segmentSize - 1) != 0) {
		throw new IllegalArgumentException("Hash Table requires buffers whose size is a power of 2.");
	}

	this.numBucketsPerSegment = segmentSize / bucketSize;
	this.numBucketsPerSegmentBits = MathUtils.log2strict(this.numBucketsPerSegment);
	this.numBucketsPerSegmentMask = (1 << this.numBucketsPerSegmentBits) - 1;

	recordArea = new RecordArea(segmentSize);

	stagingSegments = new ArrayList<>();
	stagingSegments.add(forcedAllocateSegment());
	stagingSegmentsInView = new RandomAccessInputView(stagingSegments, segmentSize);
	stagingSegmentsOutView = new StagingOutputView(stagingSegments, segmentSize);

	prober = new HashTableProber<>(buildSideComparator, new SameTypePairComparator<>(buildSideComparator));

	enableResize = buildSideSerializer.getLength() == -1;
}
 

private void load(long numElements, RandomAccessInputView recordInputView) throws IOException {
	for (int index = 0; index < numElements; index++) {
		serializer.checkSkipReadForFixLengthPart(recordInputView);
		long pointer = recordInputView.getReadPosition();
		BinaryRowData row = serializer1.mapFromPages(row1, recordInputView);
		valueSerializer.checkSkipReadForFixLengthPart(recordInputView);
		recordInputView.skipBytes(recordInputView.readInt());
		boolean success = checkNextIndexOffset();
		checkArgument(success);
		writeIndexAndNormalizedKey(row, pointer);
	}
}
 

public BinaryIndexedSortable(
		NormalizedKeyComputer normalizedKeyComputer,
		BinaryRowDataSerializer serializer,
		RecordComparator comparator,
		ArrayList<MemorySegment> recordBufferSegments,
		MemorySegmentPool memorySegmentPool) {
	if (normalizedKeyComputer == null || serializer == null) {
		throw new NullPointerException();
	}
	this.normalizedKeyComputer = normalizedKeyComputer;
	this.serializer = serializer;
	this.comparator = comparator;
	this.memorySegmentPool = memorySegmentPool;
	this.useNormKeyUninverted = !normalizedKeyComputer.invertKey();

	this.numKeyBytes = normalizedKeyComputer.getNumKeyBytes();

	int segmentSize = memorySegmentPool.pageSize();
	this.recordBuffer = new RandomAccessInputView(recordBufferSegments, segmentSize);
	this.recordBufferForComparison = new RandomAccessInputView(recordBufferSegments, segmentSize);

	this.normalizedKeyFullyDetermines = normalizedKeyComputer.isKeyFullyDetermines();

	// compute the index entry size and limits
	this.indexEntrySize = numKeyBytes + OFFSET_LEN;
	this.indexEntriesPerSegment = segmentSize / this.indexEntrySize;
	this.lastIndexEntryOffset = (this.indexEntriesPerSegment - 1) * this.indexEntrySize;

	this.serializer1 = (BinaryRowDataSerializer) serializer.duplicate();
	this.serializer2 = (BinaryRowDataSerializer) serializer.duplicate();
	this.row1 = this.serializer1.createInstance();
	this.row2 = this.serializer2.createInstance();

	// set to initial state
	this.sortIndex = new ArrayList<>(16);
	this.currentSortIndexSegment = nextMemorySegment();
	sortIndex.add(currentSortIndexSegment);
}
 

private BuildSideBuffer(MemorySegment initialSegment, MemorySegmentSource memSource) {
	super(initialSegment, initialSegment.size(), 0);

	this.memSource = memSource;
	this.sizeBits = MathUtils.log2strict(initialSegment.size());
	this.targetList = new ArrayList<>();
	this.buildStageSegments = new ArrayList<>();
	this.buildStageSegments.add(initialSegment);
	this.buildStageInputView = new RandomAccessInputView(
			buildStageSegments, initialSegment.size());
}
 

private InMemoryBufferIterator newIterator(int beginRow, long offset) {
	checkArgument(offset >= 0, "`offset` can't be negative!");

	RandomAccessInputView recordBuffer = new RandomAccessInputView(
			this.recordBufferSegments, segmentSize, numBytesInLastBuffer);
	return new InMemoryBufferIterator(recordCount, beginRow, offset, recordBuffer);
}
 

@Test
public void testGenericObject() throws Exception {

	GenericTypeInfo<MyObj> info = new GenericTypeInfo<>(MyObj.class);
	TypeSerializer<MyObj> genericSerializer = info.createSerializer(new ExecutionConfig());
	RawValueDataSerializer<MyObj> binarySerializer = new RawValueDataSerializer<>(genericSerializer);

	BinaryRowData row = new BinaryRowData(4);
	BinaryRowWriter writer = new BinaryRowWriter(row);
	writer.writeInt(0, 0);

	RawValueData<MyObj> myObj1 = RawValueData.fromObject(new MyObj(0, 1));
	writer.writeRawValue(1, myObj1, binarySerializer);
	RawValueData<MyObj> myObj2 = RawValueData.fromObject(new MyObj(123, 5.0));
	writer.writeRawValue(2, myObj2, binarySerializer);
	RawValueData<MyObj> myObj3 = RawValueData.fromObject(new MyObj(1, 1));
	writer.writeRawValue(3, myObj3, binarySerializer);
	writer.complete();

	assertTestGenericObjectRow(row, genericSerializer);

	// getBytes from var-length memorySegments.
	BinaryRowDataSerializer serializer = new BinaryRowDataSerializer(4);
	MemorySegment[] memorySegments = new MemorySegment[3];
	ArrayList<MemorySegment> memorySegmentList = new ArrayList<>();
	for (int i = 0; i < 3; i++) {
		memorySegments[i] = MemorySegmentFactory.wrap(new byte[64]);
		memorySegmentList.add(memorySegments[i]);
	}
	RandomAccessOutputView out = new RandomAccessOutputView(memorySegments, 64);
	serializer.serializeToPages(row, out);

	BinaryRowData mapRow = serializer.mapFromPages(new RandomAccessInputView(memorySegmentList, 64));
	assertTestGenericObjectRow(mapRow, genericSerializer);
}
 

@Test
public void testHashAndCopy() throws IOException {
	MemorySegment[] segments = new MemorySegment[3];
	for (int i = 0; i < 3; i++) {
		segments[i] = MemorySegmentFactory.wrap(new byte[64]);
	}
	RandomAccessOutputView out = new RandomAccessOutputView(segments, 64);
	BinaryRowDataSerializer serializer = new BinaryRowDataSerializer(2);

	BinaryRowData row = new BinaryRowData(2);
	BinaryRowWriter writer = new BinaryRowWriter(row);
	writer.writeString(0, fromString("hahahahahahahahahahahahahahahahahahahhahahahahahahahahah"));
	writer.writeString(1, fromString("hahahahahahahahahahahahahahahahahahahhahahahahahahahahaa"));
	writer.complete();
	serializer.serializeToPages(row, out);

	ArrayList<MemorySegment> segmentList = new ArrayList<>(Arrays.asList(segments));
	RandomAccessInputView input = new RandomAccessInputView(segmentList, 64, 64);

	BinaryRowData mapRow = serializer.mapFromPages(input);
	assertEquals(row, mapRow);
	assertEquals(row.getString(0), mapRow.getString(0));
	assertEquals(row.getString(1), mapRow.getString(1));
	assertNotEquals(row.getString(0), mapRow.getString(1));

	// test if the hash code before and after serialization are the same
	assertEquals(row.hashCode(), mapRow.hashCode());
	assertEquals(row.getString(0).hashCode(), mapRow.getString(0).hashCode());
	assertEquals(row.getString(1).hashCode(), mapRow.getString(1).hashCode());

	// test if the copy method produce a row with the same contents
	assertEquals(row.copy(), mapRow.copy());
	assertEquals(
		((BinaryStringData) row.getString(0)).copy(),
		((BinaryStringData) mapRow.getString(0)).copy());
	assertEquals(
		((BinaryStringData) row.getString(1)).copy(),
		((BinaryStringData) mapRow.getString(1)).copy());
}
 

public InPlaceMutableHashTable(TypeSerializer<T> serializer, TypeComparator<T> comparator, List<MemorySegment> memory) {
	super(serializer, comparator);
	this.numAllMemorySegments = memory.size();
	this.freeMemorySegments = new ArrayList<>(memory);

	// some sanity checks first
	if (freeMemorySegments.size() < MIN_NUM_MEMORY_SEGMENTS) {
		throw new IllegalArgumentException("Too few memory segments provided. InPlaceMutableHashTable needs at least " +
			MIN_NUM_MEMORY_SEGMENTS + " memory segments.");
	}

	// Get the size of the first memory segment and record it. All further buffers must have the same size.
	// the size must also be a power of 2
	segmentSize = freeMemorySegments.get(0).size();
	if ( (segmentSize & segmentSize - 1) != 0) {
		throw new IllegalArgumentException("Hash Table requires buffers whose size is a power of 2.");
	}

	this.numBucketsPerSegment = segmentSize / bucketSize;
	this.numBucketsPerSegmentBits = MathUtils.log2strict(this.numBucketsPerSegment);
	this.numBucketsPerSegmentMask = (1 << this.numBucketsPerSegmentBits) - 1;

	recordArea = new RecordArea(segmentSize);

	stagingSegments = new ArrayList<>();
	stagingSegments.add(forcedAllocateSegment());
	stagingSegmentsInView = new RandomAccessInputView(stagingSegments, segmentSize);
	stagingSegmentsOutView = new StagingOutputView(stagingSegments, segmentSize);

	prober = new HashTableProber<>(buildSideComparator, new SameTypePairComparator<>(buildSideComparator));

	enableResize = buildSideSerializer.getLength() == -1;
}
 

public NormalizedKeySorter(TypeSerializer<T> serializer, TypeComparator<T> comparator, 
		List<MemorySegment> memory, int maxNormalizedKeyBytes)
{
	if (serializer == null || comparator == null || memory == null) {
		throw new NullPointerException();
	}
	if (maxNormalizedKeyBytes < 0) {
		throw new IllegalArgumentException("Maximal number of normalized key bytes must not be negative.");
	}
	
	this.serializer = serializer;
	this.comparator = comparator;
	this.useNormKeyUninverted = !comparator.invertNormalizedKey();
	
	// check the size of the first buffer and record it. all further buffers must have the same size.
	// the size must also be a power of 2
	this.totalNumBuffers = memory.size();
	if (this.totalNumBuffers < MIN_REQUIRED_BUFFERS) {
		throw new IllegalArgumentException("Normalized-Key sorter requires at least " + MIN_REQUIRED_BUFFERS + " memory buffers.");
	}
	this.segmentSize = memory.get(0).size();
	this.freeMemory = new ArrayList<MemorySegment>(memory);
	
	// create the buffer collections
	this.sortIndex = new ArrayList<MemorySegment>(16);
	this.recordBufferSegments = new ArrayList<MemorySegment>(16);
	
	// the views for the record collections
	this.recordCollector = new SimpleCollectingOutputView(this.recordBufferSegments,
		new ListMemorySegmentSource(this.freeMemory), this.segmentSize);
	this.recordBuffer = new RandomAccessInputView(this.recordBufferSegments, this.segmentSize);
	this.recordBufferForComparison = new RandomAccessInputView(this.recordBufferSegments, this.segmentSize);
	
	// set up normalized key characteristics
	if (this.comparator.supportsNormalizedKey()) {
		// compute the max normalized key length
		int numPartialKeys;
		try {
			numPartialKeys = this.comparator.getFlatComparators().length;
		} catch (Throwable t) {
			numPartialKeys = 1;
		}
		
		int maxLen = Math.min(maxNormalizedKeyBytes, MAX_NORMALIZED_KEY_LEN_PER_ELEMENT * numPartialKeys);
		
		this.numKeyBytes = Math.min(this.comparator.getNormalizeKeyLen(), maxLen);
		this.normalizedKeyFullyDetermines = !this.comparator.isNormalizedKeyPrefixOnly(this.numKeyBytes);
	}
	else {
		this.numKeyBytes = 0;
		this.normalizedKeyFullyDetermines = false;
	}
	
	// compute the index entry size and limits
	this.indexEntrySize = this.numKeyBytes + OFFSET_LEN;
	this.indexEntriesPerSegment = this.segmentSize / this.indexEntrySize;
	this.lastIndexEntryOffset = (this.indexEntriesPerSegment - 1) * this.indexEntrySize;
	this.swapBuffer = new byte[this.indexEntrySize];
	
	// set to initial state
	this.currentSortIndexSegment = nextMemorySegment();
	this.sortIndex.add(this.currentSortIndexSegment);
}
 

RecordArea() {
	this.outView = new SimpleCollectingOutputView(segments, new RecordAreaMemorySource(), segmentSize);
	this.inView = new RandomAccessInputView(segments, segmentSize);
}
 

RandomAccessInputView getBuildStateInputView() {
	return this.buildSideWriteBuffer.getBuildStageInputView();
}
 

RandomAccessInputView getBuildStageInputView() {
	return buildStageInputView;
}
 

/**
 * For distinct build.
 */
private boolean findFirstSameBuildRow(
		MemorySegment bucket,
		int searchHashCode,
		int bucketInSegmentOffset,
		BinaryRow buildRowToInsert) {
	int posInSegment = bucketInSegmentOffset + BUCKET_HEADER_LENGTH;
	int countInBucket = bucket.getShort(bucketInSegmentOffset + HEADER_COUNT_OFFSET);
	int numInBucket = 0;
	RandomAccessInputView view = partition.getBuildStateInputView();
	while (countInBucket != 0) {
		while (numInBucket < countInBucket) {

			final int thisCode = bucket.getInt(posInSegment);
			posInSegment += HASH_CODE_LEN;

			if (thisCode == searchHashCode) {
				final int pointer = bucket.getInt(bucketInSegmentOffset +
						BUCKET_POINTER_START_OFFSET + (numInBucket * POINTER_LEN));
				numInBucket++;
				try {
					view.setReadPosition(pointer);
					BinaryRow row = table.binaryBuildSideSerializer.mapFromPages(table.reuseBuildRow, view);
					if (buildRowToInsert.equals(row)) {
						return true;
					}
				} catch (IOException e) {
					throw new RuntimeException("Error deserializing key or value from the hashtable: " +
							e.getMessage(), e);
				}
			} else {
				numInBucket++;
			}
		}

		// this segment is done. check if there is another chained bucket
		final int forwardPointer = bucket.getInt(bucketInSegmentOffset + HEADER_FORWARD_OFFSET);
		if (forwardPointer == BUCKET_FORWARD_POINTER_NOT_SET) {
			return false;
		}

		final int overflowSegIndex = forwardPointer >>> table.segmentSizeBits;
		bucket = overflowSegments[overflowSegIndex];
		bucketInSegmentOffset = forwardPointer & table.segmentSizeMask;
		countInBucket = bucket.getShort(bucketInSegmentOffset + HEADER_COUNT_OFFSET);
		posInSegment = bucketInSegmentOffset + BUCKET_HEADER_LENGTH;
		numInBucket = 0;
	}
	return false;
}