SeekableFileChannelInputView extends AbstractPagedInputView { public void seek(long position) throws IOException { final int block = MathUtils.checkedDownCast(position / segmentSize); final int positionInBlock = (int) (position % segmentSize); if (position < 0 || block >= numBlocksTotal || (block == numBlocksTotal - 1 && positionInBlock > sizeOfLastBlock)) { throw new IllegalArgumentException("Position is out of range"); } clear(); if (reader != null) { reader.close(); } reader = ioManager.createBlockChannelReader(channelId); if (block > 0) { reader.seekToPosition(((long) block) * segmentSize); } this.numBlocksRemaining = this.numBlocksTotal - block; this.numRequestsRemaining = numBlocksRemaining; for (int i = 0; i < memory.size(); i++) { sendReadRequest(memory.get(i)); } numBlocksRemaining--; seekInput(reader.getNextReturnedBlock(), positionInBlock, numBlocksRemaining == 0 ? sizeOfLastBlock : segmentSize); } SeekableFileChannelInputView(IOManager ioManager, FileIOChannel.ID channelId, MemoryManager memManager, List<MemorySegment> memory, int sizeOfLastBlock); void seek(long position); void close(); void closeAndDelete();  }
@Test public void testSeek() { final IOManager ioManager = new IOManagerAsync(); final int PAGE_SIZE = 16 * 1024; final int NUM_RECORDS = 120000; try { MemoryManager memMan = new MemoryManager(4 * PAGE_SIZE, 1, PAGE_SIZE, MemoryType.HEAP, true); List<MemorySegment> memory = new ArrayList<MemorySegment>(); memMan.allocatePages(new DummyInvokable(), memory, 4); FileIOChannel.ID channel = ioManager.createChannel(); BlockChannelWriter<MemorySegment> writer = ioManager.createBlockChannelWriter(channel); FileChannelOutputView out = new FileChannelOutputView(writer, memMan, memory, memMan.getPageSize()); for (int i = 0; i < NUM_RECORDS; i += 4) { out.writeInt(i); } out.close(); assertTrue(memMan.verifyEmpty()); memMan.allocatePages(new DummyInvokable(), memory, 4); SeekableFileChannelInputView in = new SeekableFileChannelInputView(ioManager, channel, memMan, memory, out.getBytesInLatestSegment()); for (int i = 0; i < NUM_RECORDS; i += 4) { assertEquals(i, in.readInt()); } try { in.readInt(); fail("should throw EOF exception"); } catch (EOFException ignored) {} int i = 2 * PAGE_SIZE + PAGE_SIZE / 4; in.seek(i); for (; i < NUM_RECORDS; i += 4) { assertEquals(i, in.readInt()); } try { in.readInt(); fail("should throw EOF exception"); } catch (EOFException ignored) {} i = 120000 - 4; in.seek(i); for (; i < NUM_RECORDS; i += 4) { assertEquals(i, in.readInt()); } try { in.readInt(); fail("should throw EOF exception"); } catch (EOFException ignored) {} i = 0; in.seek(i); for (; i < NUM_RECORDS; i += 4) { assertEquals(i, in.readInt()); } try { in.readInt(); fail("should throw EOF exception"); } catch (EOFException ignored) {} i = PAGE_SIZE; in.seek(i); for (; i < NUM_RECORDS; i += 4) { assertEquals(i, in.readInt()); } try { in.readInt(); fail("should throw EOF exception"); } catch (EOFException ignored) {} i = 3 * PAGE_SIZE; in.seek(i); for (; i < NUM_RECORDS; i += 4) { assertEquals(i, in.readInt()); } try { in.readInt(); fail("should throw EOF exception"); } catch (EOFException ignored) {} i = NUM_RECORDS; in.seek(i); try { in.readInt(); fail("should throw EOF exception"); } catch (EOFException ignored) {} try { in.seek(-10); fail("should throw an exception"); } catch (IllegalArgumentException ignored) {} try { in.seek(NUM_RECORDS + 1); fail("should throw an exception"); } catch (IllegalArgumentException ignored) {} } catch (Exception e) { e.printStackTrace(); fail(e.getMessage()); } finally { ioManager.shutdown(); } }