@Test public void testSpill() throws Exception { ResettableExternalBuffer buffer = newBuffer(DEFAULT_PAGE_SIZE * 2); int number = 5000; List<Long> expected = insertMulti(buffer, number); assertEquals(buffer.size(), number); assertBuffer(expected, buffer); assertTrue(buffer.getSpillChannels().size() > 0); assertBuffer(expected, buffer); buffer.newIterator(); assertBuffer(expected, buffer); buffer.close(); }
private void spill() throws IOException { FileIOChannel.ID channel = ioManager.createChannel(); final BlockChannelWriter<MemorySegment> writer = this.ioManager.createBlockChannelWriter(channel); int numRecordBuffers = inMemoryBuffer.getNumRecordBuffers(); ArrayList<MemorySegment> segments = inMemoryBuffer.getRecordBufferSegments(); try { for (int i = 0; i < numRecordBuffers; i++) { writer.writeBlock(segments.get(i)); } LOG.info("here spill the reset buffer data with {} bytes", writer.getSize()); writer.close(); } catch (IOException e) { writer.closeAndDelete(); throw e; } spillSize += numRecordBuffers * segmentSize; spilledChannelIDs.add(new ChannelWithMeta( channel, inMemoryBuffer.getNumRecordBuffers(), inMemoryBuffer.getNumBytesInLastBuffer())); this.spilledChannelRowOffsets.add(numRows); inMemoryBuffer.reset(); }
ResettableExternalBuffer implements ResettableRowBuffer { private void spill() throws IOException { FileIOChannel.ID channel = ioManager.createChannel(); final BlockChannelWriter<MemorySegment> writer = this.ioManager.createBlockChannelWriter(channel); int numRecordBuffers = inMemoryBuffer.getNumRecordBuffers(); ArrayList<MemorySegment> segments = inMemoryBuffer.getRecordBufferSegments(); try { for (int i = 0; i < numRecordBuffers; i++) { writer.writeBlock(segments.get(i)); } LOG.info("here spill the reset buffer data with {} bytes", writer.getSize()); writer.close(); } catch (IOException e) { writer.closeAndDelete(); throw e; } spillSize += numRecordBuffers * segmentSize; spilledChannelIDs.add(new ChannelWithMeta( channel, inMemoryBuffer.getNumRecordBuffers(), inMemoryBuffer.getNumBytesInLastBuffer())); this.spilledChannelRowOffsets.add(numRows); inMemoryBuffer.reset(); } }
ResettableExternalBuffer implements ResettableRowBuffer { private void spill() throws IOException { FileIOChannel.ID channel = ioManager.createChannel(); final BlockChannelWriter<MemorySegment> writer = this.ioManager.createBlockChannelWriter(channel); int numRecordBuffers = inMemoryBuffer.getNumRecordBuffers(); ArrayList<MemorySegment> segments = inMemoryBuffer.getRecordBufferSegments(); try { for (int i = 0; i < numRecordBuffers; i++) { writer.writeBlock(segments.get(i)); } LOG.info("here spill the reset buffer data with {} bytes", writer.getSize()); writer.close(); } catch (IOException e) { writer.closeAndDelete(); throw e; } spillSize += numRecordBuffers * segmentSize; spilledChannelIDs.add(new ChannelWithMeta( channel, inMemoryBuffer.getNumRecordBuffers(), inMemoryBuffer.getNumBytesInLastBuffer())); this.spilledChannelRowOffsets.add(numRows); inMemoryBuffer.reset(); } ResettableExternalBuffer(
		MemoryManager memoryManager,
		IOManager ioManager,
		List<MemorySegment> memory,
		AbstractRowSerializer serializer,
		boolean isRowAllInFixedPart); }
ResettableExternalBuffer implements ResettableRowBuffer { private void spill() throws IOException { FileIOChannel.ID channel = ioManager.createChannel(); final BlockChannelWriter<MemorySegment> writer = this.ioManager.createBlockChannelWriter(channel); int numRecordBuffers = inMemoryBuffer.getNumRecordBuffers(); ArrayList<MemorySegment> segments = inMemoryBuffer.getRecordBufferSegments(); try { for (int i = 0; i < numRecordBuffers; i++) { writer.writeBlock(segments.get(i)); } LOG.info("here spill the reset buffer data with {} bytes", writer.getSize()); writer.close(); } catch (IOException e) { writer.closeAndDelete(); throw e; } spillSize += numRecordBuffers * segmentSize; spilledChannelIDs.add(new ChannelWithMeta( channel, inMemoryBuffer.getNumRecordBuffers(), inMemoryBuffer.getNumBytesInLastBuffer())); this.spilledChannelRowOffsets.add(numRows); inMemoryBuffer.reset(); } ResettableExternalBuffer(
		MemoryManager memoryManager,
		IOManager ioManager,
		List<MemorySegment> memory,
		AbstractRowSerializer serializer,
		boolean isRowAllInFixedPart); @Override void reset(); @Override void add(BaseRow row); @Override void complete(); @Override BufferIterator newIterator(); @Override BufferIterator newIterator(int beginRow); @Override void close(); int size(); long getUsedMemoryInBytes(); int getNumSpillFiles(); long getSpillInBytes(); }
ResettableExternalBuffer implements ResettableRowBuffer { private void spill() throws IOException { FileIOChannel.ID channel = ioManager.createChannel(); final BlockChannelWriter<MemorySegment> writer = this.ioManager.createBlockChannelWriter(channel); int numRecordBuffers = inMemoryBuffer.getNumRecordBuffers(); ArrayList<MemorySegment> segments = inMemoryBuffer.getRecordBufferSegments(); try { for (int i = 0; i < numRecordBuffers; i++) { writer.writeBlock(segments.get(i)); } LOG.info("here spill the reset buffer data with {} bytes", writer.getSize()); writer.close(); } catch (IOException e) { writer.closeAndDelete(); throw e; } spillSize += numRecordBuffers * segmentSize; spilledChannelIDs.add(new ChannelWithMeta( channel, inMemoryBuffer.getNumRecordBuffers(), inMemoryBuffer.getNumBytesInLastBuffer())); this.spilledChannelRowOffsets.add(numRows); inMemoryBuffer.reset(); } ResettableExternalBuffer(
		MemoryManager memoryManager,
		IOManager ioManager,
		List<MemorySegment> memory,
		AbstractRowSerializer serializer,
		boolean isRowAllInFixedPart); @Override void reset(); @Override void add(BaseRow row); @Override void complete(); @Override BufferIterator newIterator(); @Override BufferIterator newIterator(int beginRow); @Override void close(); int size(); long getUsedMemoryInBytes(); int getNumSpillFiles(); long getSpillInBytes();  }