@Test public void testTruncate() throws Exception { long startTransactionId = rand.nextInt(1000); long segmentSizeThreshold = 1000; Segment.create(key, segmentPath, indexPath, 0, startTransactionId); Segment segment = new Segment(key, segmentPath, indexPath, controlFile.getPartitionInfo(0), segmentSizeThreshold); segment.setWritable(); assertEquals(startTransactionId, segment.firstTransactionId()); assertEquals(Segment.FILE_HEADER_SIZE, segment.size()); ArrayList<Long> transactionnId = new ArrayList<>(); ArrayList<Long> sizeAfterAppend = new ArrayList<>(); long count = 0; while (segment.size() < segmentSizeThreshold) { ArrayList<Record> records = new ArrayList<>(); byte[] data = new byte[100]; records.add(new Record(startTransactionId + count, reqId(), 0, data, Utils.checksum(data))); if (segment.append(records, 0) == 0) { break; } transactionnId.add(startTransactionId + count); sizeAfterAppend.add(segment.size()); count++; } segment.truncate(segment.maxTransactionId()); for (int i = sizeAfterAppend.size() - 1; i >= 0; i--) { segment.truncate(transactionnId.get(i)); assertEquals(sizeAfterAppend.get(i), (Long) segment.size()); } segment.truncate(startTransactionId - 1L); assertEquals(Segment.FILE_HEADER_SIZE, segment.size()); segment.close(); }
public void truncate(long transactionId) throws StorageException, IOException { synchronized (this) { if (closed) { throw new StorageException("segment closed"); } if (transactionId + 1 == nextTransactionId) { return; } if (transactionId + 1 < header.firstTransactionId || transactionId + 1 > nextTransactionId) { throw new StorageException("transaction id out of range: file=" + file.toString()); } long offset = index.get(transactionId + 1); if (offset < 0) { throw new StorageException("corrupted"); } index.truncate(transactionId); channel.truncate(offset); channel.force(true); nextOffset = offset; nextTransactionId = transactionId + 1; } }
Segment { public void truncate(long transactionId) throws StorageException, IOException { synchronized (this) { if (closed) { throw new StorageException("segment closed"); } if (transactionId + 1 == nextTransactionId) { return; } if (transactionId + 1 < header.firstTransactionId || transactionId + 1 > nextTransactionId) { throw new StorageException("transaction id out of range: file=" + file.toString()); } long offset = index.get(transactionId + 1); if (offset < 0) { throw new StorageException("corrupted"); } index.truncate(transactionId); channel.truncate(offset); channel.force(true); nextOffset = offset; nextTransactionId = transactionId + 1; } } }
Segment { public void truncate(long transactionId) throws StorageException, IOException { synchronized (this) { if (closed) { throw new StorageException("segment closed"); } if (transactionId + 1 == nextTransactionId) { return; } if (transactionId + 1 < header.firstTransactionId || transactionId + 1 > nextTransactionId) { throw new StorageException("transaction id out of range: file=" + file.toString()); } long offset = index.get(transactionId + 1); if (offset < 0) { throw new StorageException("corrupted"); } index.truncate(transactionId); channel.truncate(offset); channel.force(true); nextOffset = offset; nextTransactionId = transactionId + 1; } } Segment(UUID key, Path file, Path indexFile, PartitionInfo partitionInfo, long segmentSizeThreshold); }
Segment { public void truncate(long transactionId) throws StorageException, IOException { synchronized (this) { if (closed) { throw new StorageException("segment closed"); } if (transactionId + 1 == nextTransactionId) { return; } if (transactionId + 1 < header.firstTransactionId || transactionId + 1 > nextTransactionId) { throw new StorageException("transaction id out of range: file=" + file.toString()); } long offset = index.get(transactionId + 1); if (offset < 0) { throw new StorageException("corrupted"); } index.truncate(transactionId); channel.truncate(offset); channel.force(true); nextOffset = offset; nextTransactionId = transactionId + 1; } } Segment(UUID key, Path file, Path indexFile, PartitionInfo partitionInfo, long segmentSizeThreshold); void setReadOnly(); void setWritable(); boolean isWritable(); boolean isClosed(); boolean isChannelClosed(); void closeChannel(); long firstTransactionId(); long nextTransactionId(); long maxTransactionId(); long size(); void flush(); void checksum(CRC32 crc32); void close(); void delete(); int append(ArrayList<Record> records, int off); RecordHeader getRecordHeader(long transactionId); Record getRecord(long transactionId); int checkRecord(long offset, long transactionId); void truncate(long transactionId); String toString(); static void create(UUID key, Path segmentFile, Path indexFile, int partitionId, long firstTransaction); }
Segment { public void truncate(long transactionId) throws StorageException, IOException { synchronized (this) { if (closed) { throw new StorageException("segment closed"); } if (transactionId + 1 == nextTransactionId) { return; } if (transactionId + 1 < header.firstTransactionId || transactionId + 1 > nextTransactionId) { throw new StorageException("transaction id out of range: file=" + file.toString()); } long offset = index.get(transactionId + 1); if (offset < 0) { throw new StorageException("corrupted"); } index.truncate(transactionId); channel.truncate(offset); channel.force(true); nextOffset = offset; nextTransactionId = transactionId + 1; } } Segment(UUID key, Path file, Path indexFile, PartitionInfo partitionInfo, long segmentSizeThreshold); void setReadOnly(); void setWritable(); boolean isWritable(); boolean isClosed(); boolean isChannelClosed(); void closeChannel(); long firstTransactionId(); long nextTransactionId(); long maxTransactionId(); long size(); void flush(); void checksum(CRC32 crc32); void close(); void delete(); int append(ArrayList<Record> records, int off); RecordHeader getRecordHeader(long transactionId); Record getRecord(long transactionId); int checkRecord(long offset, long transactionId); void truncate(long transactionId); String toString(); static void create(UUID key, Path segmentFile, Path indexFile, int partitionId, long firstTransaction); final Index index; }