/** * Method returns the reader given the specified arguments. * TODO This is a bad abstraction. See HBASE-6635. * * @param path hfile's path * @param fsdis stream of path's file * @param size max size of the trailer. * @param cacheConf Cache configuation values, cannot be null. * @param hfs * @return an appropriate instance of HFileReader * @throws IOException If file is invalid, will throw CorruptHFileException flavored IOException */ @edu.umd.cs.findbugs.annotations.SuppressWarnings(value = "SF_SWITCH_FALLTHROUGH", justification = "Intentional") private static Reader pickReaderVersion(Path path, FSDataInputStreamWrapper fsdis, long size, CacheConfig cacheConf, HFileSystem hfs, Configuration conf) throws IOException { FixedFileTrailer trailer = null; try { boolean isHBaseChecksum = fsdis.shouldUseHBaseChecksum(); assert !isHBaseChecksum; // Initially we must read with FS checksum. trailer = FixedFileTrailer.readFromStream(fsdis.getStream(isHBaseChecksum), size); switch (trailer.getMajorVersion()) { case 2: return new HFileReaderV2(path, trailer, fsdis, size, cacheConf, hfs, conf); case 3: return new HFileReaderV3(path, trailer, fsdis, size, cacheConf, hfs, conf); default: throw new IllegalArgumentException("Invalid HFile version " + trailer.getMajorVersion()); } } catch (Throwable t) { try { fsdis.close(); } catch (Throwable t2) { LOG.warn("Error closing fsdis FSDataInputStreamWrapper", t2); } throw new CorruptHFileException("Problem reading HFile Trailer from file " + path, t); } }
@Test public void testNewBlocksHaveDefaultChecksum() throws IOException { Path path = new Path(TEST_UTIL.getDataTestDir(), "default_checksum"); FSDataOutputStream os = fs.create(path); HFileContext meta = new HFileContextBuilder().build(); HFileBlock.Writer hbw = new HFileBlock.Writer(null, meta); DataOutputStream dos = hbw.startWriting(BlockType.DATA); for (int i = 0; i < 1000; ++i) dos.writeInt(i); hbw.writeHeaderAndData(os); int totalSize = hbw.getOnDiskSizeWithHeader(); os.close(); // Use hbase checksums. assertEquals(true, hfs.useHBaseChecksum()); FSDataInputStreamWrapper is = new FSDataInputStreamWrapper(fs, path); meta = new HFileContextBuilder().withHBaseCheckSum(true).build(); HFileBlock.FSReader hbr = new HFileBlock.FSReaderImpl( is, totalSize, (HFileSystem) fs, path, meta); HFileBlock b = hbr.readBlockData(0, -1, -1, false); assertEquals(b.getChecksumType(), ChecksumType.getDefaultChecksumType().getCode()); }
/** * Read all blocks from {@code path} to populate {@code blockCache}. */ private static void cacheBlocks(Configuration conf, CacheConfig cacheConfig, FileSystem fs, Path path, HFileContext cxt) throws IOException { FSDataInputStreamWrapper fsdis = new FSDataInputStreamWrapper(fs, path); long fileSize = fs.getFileStatus(path).getLen(); FixedFileTrailer trailer = FixedFileTrailer.readFromStream(fsdis.getStream(false), fileSize); HFileReaderV2 reader = new HFileReaderV2(path, trailer, fsdis, fileSize, cacheConfig, fsdis.getHfs(), conf); reader.loadFileInfo(); long offset = trailer.getFirstDataBlockOffset(), max = trailer.getLastDataBlockOffset(); List<HFileBlock> blocks = new ArrayList<HFileBlock>(4); HFileBlock block; while (offset <= max) { block = reader.readBlock(offset, -1, /* cacheBlock */ true, /* pread */ false, /* isCompaction */ false, /* updateCacheMetrics */ true, null, null); offset += block.getOnDiskSizeWithHeader(); blocks.add(block); } LOG.info("read " + Iterables.toString(blocks)); }
/** * Method returns the reader given the specified arguments. * TODO This is a bad abstraction. See HBASE-6635. * * @param path hfile's path * @param fsdis stream of path's file * @param size max size of the trailer. * @param cacheConf Cache configuation values, cannot be null. * @param hfs * @return an appropriate instance of HFileReader * @throws IOException If file is invalid, will throw CorruptHFileException flavored IOException */ private static Reader pickReaderVersion(Path path, FSDataInputStreamWrapper fsdis, long size, CacheConfig cacheConf, HFileSystem hfs, Configuration conf) throws IOException { FixedFileTrailer trailer = null; try { boolean isHBaseChecksum = fsdis.shouldUseHBaseChecksum(); assert !isHBaseChecksum; // Initially we must read with FS checksum. trailer = FixedFileTrailer.readFromStream(fsdis.getStream(isHBaseChecksum), size); switch (trailer.getMajorVersion()) { case 2: return new HFileReaderV2(path, trailer, fsdis, size, cacheConf, hfs, conf); case 3 : return new HFileReaderV3(path, trailer, fsdis, size, cacheConf, hfs, conf); default: throw new IllegalArgumentException("Invalid HFile version " + trailer.getMajorVersion()); } } catch (Throwable t) { try { fsdis.close(); } catch (Throwable t2) { LOG.warn("Error closing fsdis FSDataInputStreamWrapper", t2); } throw new CorruptHFileException("Problem reading HFile Trailer from file " + path, t); } }
/** * @param fs fileystem to read from * @param p path to the file * @param in {@link FSDataInputStreamWrapper} * @param size Full size of the file * @param cacheConf * @param r original reference file. This will be not null only when reading a split file. * @param reader the base reader instance * @return The reader to use * @throws IOException */ public StoreFile.Reader postStoreFileReaderOpen(final FileSystem fs, final Path p, final FSDataInputStreamWrapper in, final long size, final CacheConfig cacheConf, final Reference r, StoreFile.Reader reader) throws IOException { ObserverContext<RegionCoprocessorEnvironment> ctx = null; for (RegionEnvironment env : coprocessors) { if (env.getInstance() instanceof RegionObserver) { ctx = ObserverContext.createAndPrepare(env, ctx); Thread currentThread = Thread.currentThread(); ClassLoader cl = currentThread.getContextClassLoader(); try { currentThread.setContextClassLoader(env.getClassLoader()); reader = ((RegionObserver) env.getInstance()).postStoreFileReaderOpen(ctx, fs, p, in, size, cacheConf, r, reader); } catch (Throwable e) { handleCoprocessorThrowable(env, e); } finally { currentThread.setContextClassLoader(cl); } if (ctx.shouldComplete()) { break; } } } return reader; }
@Override public StoreFile.Reader preStoreFileReaderOpen( ObserverContext<RegionCoprocessorEnvironment> ctx, FileSystem fs, Path p, FSDataInputStreamWrapper in, long size, CacheConfig cacheConf, Reference r, StoreFile.Reader reader) throws IOException { try { long delay = rnd.nextInt(3); LOG.info("@@@ Delaying region " + ctx.getEnvironment().getRegion().getRegionInfo(). getRegionNameAsString() + " for " + delay + " seconds..."); Thread.sleep(delay * 1000); } catch (InterruptedException ie) { LOG.error(ie); } return reader; }
/** * @param fs fileystem to read from * @param p path to the file * @param in {@link FSDataInputStreamWrapper} * @param size Full size of the file * @param cacheConf * @param r original reference file. This will be not null only when reading a split file. * @return a Reader instance to use instead of the base reader if overriding * default behavior, null otherwise * @throws IOException */ public StoreFileReader preStoreFileReaderOpen(final FileSystem fs, final Path p, final FSDataInputStreamWrapper in, final long size, final CacheConfig cacheConf, final Reference r) throws IOException { if (coprocEnvironments.isEmpty()) { return null; } return execOperationWithResult( new ObserverOperationWithResult<RegionObserver, StoreFileReader>(regionObserverGetter, null) { @Override public StoreFileReader call(RegionObserver observer) throws IOException { return observer.preStoreFileReaderOpen(this, fs, p, in, size, cacheConf, r, getResult()); } }); }
/** * @param fs fileystem to read from * @param p path to the file * @param in {@link FSDataInputStreamWrapper} * @param size Full size of the file * @param cacheConf * @param r original reference file. This will be not null only when reading a split file. * @param reader the base reader instance * @return The reader to use * @throws IOException */ public StoreFileReader postStoreFileReaderOpen(final FileSystem fs, final Path p, final FSDataInputStreamWrapper in, final long size, final CacheConfig cacheConf, final Reference r, final StoreFileReader reader) throws IOException { if (this.coprocEnvironments.isEmpty()) { return reader; } return execOperationWithResult( new ObserverOperationWithResult<RegionObserver, StoreFileReader>(regionObserverGetter, reader) { @Override public StoreFileReader call(RegionObserver observer) throws IOException { return observer.postStoreFileReaderOpen(this, fs, p, in, size, cacheConf, r, getResult()); } }); }
FSReaderImpl(FSDataInputStreamWrapper stream, long fileSize, HFileSystem hfs, Path path, HFileContext fileContext) throws IOException { this.fileSize = fileSize; this.hfs = hfs; if (path != null) { this.pathName = path.toString(); } this.fileContext = fileContext; this.hdrSize = headerSize(fileContext.isUseHBaseChecksum()); this.streamWrapper = stream; // Older versions of HBase didn't support checksum. this.streamWrapper.prepareForBlockReader(!fileContext.isUseHBaseChecksum()); defaultDecodingCtx = new HFileBlockDefaultDecodingContext(fileContext); encodedBlockDecodingCtx = defaultDecodingCtx; }
@Test public void testNewBlocksHaveDefaultChecksum() throws IOException { Path path = new Path(TEST_UTIL.getDataTestDir(), "default_checksum"); FSDataOutputStream os = fs.create(path); HFileContext meta = new HFileContextBuilder().build(); HFileBlock.Writer hbw = new HFileBlock.Writer(null, meta); DataOutputStream dos = hbw.startWriting(BlockType.DATA); for (int i = 0; i < 1000; ++i) dos.writeInt(i); hbw.writeHeaderAndData(os); int totalSize = hbw.getOnDiskSizeWithHeader(); os.close(); // Use hbase checksums. assertEquals(true, hfs.useHBaseChecksum()); FSDataInputStreamWrapper is = new FSDataInputStreamWrapper(fs, path); meta = new HFileContextBuilder().withHBaseCheckSum(true).build(); HFileBlock.FSReader hbr = new HFileBlock.FSReaderImpl( is, totalSize, (HFileSystem) fs, path, meta); HFileBlock b = hbr.readBlockData(0, -1, false, false); assertEquals(b.getChecksumType(), ChecksumType.getDefaultChecksumType().getCode()); }
/** * Read all blocks from {@code path} to populate {@code blockCache}. */ private static void cacheBlocks(Configuration conf, CacheConfig cacheConfig, FileSystem fs, Path path, HFileContext cxt) throws IOException { FSDataInputStreamWrapper fsdis = new FSDataInputStreamWrapper(fs, path); long fileSize = fs.getFileStatus(path).getLen(); FixedFileTrailer trailer = FixedFileTrailer.readFromStream(fsdis.getStream(false), fileSize); HFile.Reader reader = new HFileReaderImpl(path, trailer, fsdis, fileSize, cacheConfig, fsdis.getHfs(), conf); reader.loadFileInfo(); long offset = trailer.getFirstDataBlockOffset(), max = trailer.getLastDataBlockOffset(); List<HFileBlock> blocks = new ArrayList<>(4); HFileBlock block; while (offset <= max) { block = reader.readBlock(offset, -1, /* cacheBlock */ true, /* pread */ false, /* isCompaction */ false, /* updateCacheMetrics */ true, null, null); offset += block.getOnDiskSizeWithHeader(); blocks.add(block); } LOG.info("read " + Iterables.toString(blocks)); }
/** * Method returns the reader given the specified arguments. * TODO This is a bad abstraction. See HBASE-6635. * * @param path hfile's path * @param fsdis stream of path's file * @param size max size of the trailer. * @param cacheConf Cache configuation values, cannot be null. * @param hfs * @return an appropriate instance of HFileReader * @throws IOException If file is invalid, will throw CorruptHFileException flavored IOException */ private static Reader pickReaderVersion(Path path, FSDataInputStreamWrapper fsdis, long size, CacheConfig cacheConf, HFileSystem hfs) throws IOException { FixedFileTrailer trailer = null; try { boolean isHBaseChecksum = fsdis.shouldUseHBaseChecksum(); assert !isHBaseChecksum; // Initially we must read with FS checksum. trailer = FixedFileTrailer.readFromStream(fsdis.getStream(isHBaseChecksum), size); switch (trailer.getMajorVersion()) { case 2: return new HFileReaderV2( path, trailer, fsdis, size, cacheConf, hfs); default: throw new CorruptHFileException("Invalid HFile version " + trailer.getMajorVersion()); } } catch (Throwable t) { try { fsdis.close(); } catch (Throwable t2) { LOG.warn("Error closing fsdis FSDataInputStreamWrapper", t2); } throw new CorruptHFileException("Problem reading HFile Trailer from file " + path, t); } }
/** * @param fs fileystem to read from * @param p path to the file * @param in {@link FSDataInputStreamWrapper} * @param size Full size of the file * @param cacheConf * @param r original reference file. This will be not null only when reading a split file. * @return a Reader instance to use instead of the base reader if overriding * default behavior, null otherwise * @throws IOException */ public StoreFile.Reader preStoreFileReaderOpen(final FileSystem fs, final Path p, final FSDataInputStreamWrapper in, final long size, final CacheConfig cacheConf, final Reference r) throws IOException { return execOperationWithResult(null, coprocessors.isEmpty() ? null : new RegionOperationWithResult<StoreFile.Reader>() { @Override public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx) throws IOException { setResult(oserver.preStoreFileReaderOpen(ctx, fs, p, in, size, cacheConf, r, getResult())); } }); }
/** * @param fs fileystem to read from * @param p path to the file * @param in {@link FSDataInputStreamWrapper} * @param size Full size of the file * @param cacheConf * @param r original reference file. This will be not null only when reading a split file. * @param reader the base reader instance * @return The reader to use * @throws IOException */ public StoreFile.Reader postStoreFileReaderOpen(final FileSystem fs, final Path p, final FSDataInputStreamWrapper in, final long size, final CacheConfig cacheConf, final Reference r, final StoreFile.Reader reader) throws IOException { return execOperationWithResult(reader, coprocessors.isEmpty() ? null : new RegionOperationWithResult<StoreFile.Reader>() { @Override public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx) throws IOException { setResult(oserver.postStoreFileReaderOpen(ctx, fs, p, in, size, cacheConf, r, getResult())); } }); }
public FSReaderImpl(FSDataInputStreamWrapper stream, long fileSize, HFileSystem hfs, Path path, HFileContext fileContext) throws IOException { super(fileSize, hfs, path, fileContext); this.streamWrapper = stream; // Older versions of HBase didn't support checksum. this.streamWrapper.prepareForBlockReader(!fileContext.isUseHBaseChecksum()); defaultDecodingCtx = new HFileBlockDefaultDecodingContext(fileContext); encodedBlockDecodingCtx = defaultDecodingCtx; }
protected HFileContext createHFileContext(FSDataInputStreamWrapper fsdis, long fileSize, HFileSystem hfs, Path path, FixedFileTrailer trailer) throws IOException { return new HFileContextBuilder() .withIncludesMvcc(this.includesMemstoreTS) .withCompression(this.compressAlgo) .withHBaseCheckSum(trailer.getMinorVersion() >= MINOR_VERSION_WITH_CHECKSUM) .build(); }
@Override public Reader preStoreFileReaderOpen(ObserverContext<RegionCoprocessorEnvironment> ctx, FileSystem fs, Path p, FSDataInputStreamWrapper in, long size, CacheConfig cacheConf, Reference r, Reader reader) throws IOException { ctPreStoreFileReaderOpen.incrementAndGet(); return null; }
@Override public Reader postStoreFileReaderOpen(ObserverContext<RegionCoprocessorEnvironment> ctx, FileSystem fs, Path p, FSDataInputStreamWrapper in, long size, CacheConfig cacheConf, Reference r, Reader reader) throws IOException { ctPostStoreFileReaderOpen.incrementAndGet(); return reader; }
/** * Test all checksum types by writing and reading back blocks. */ @Test public void testAllChecksumTypes() throws IOException { List<ChecksumType> cktypes = new ArrayList<>(Arrays.asList(ChecksumType.values())); for (Iterator<ChecksumType> itr = cktypes.iterator(); itr.hasNext(); ) { ChecksumType cktype = itr.next(); Path path = new Path(TEST_UTIL.getDataTestDir(), "checksum" + cktype.getName()); FSDataOutputStream os = fs.create(path); HFileContext meta = new HFileContextBuilder() .withChecksumType(cktype).build(); HFileBlock.Writer hbw = new HFileBlock.Writer(null, meta); DataOutputStream dos = hbw.startWriting(BlockType.DATA); for (int i = 0; i < 1000; ++i) dos.writeInt(i); hbw.writeHeaderAndData(os); int totalSize = hbw.getOnDiskSizeWithHeader(); os.close(); // Use hbase checksums. assertEquals(true, hfs.useHBaseChecksum()); FSDataInputStreamWrapper is = new FSDataInputStreamWrapper(fs, path); meta = new HFileContextBuilder().withHBaseCheckSum(true).build(); HFileBlock.FSReader hbr = new HFileBlock.FSReaderImpl( is, totalSize, (HFileSystem) fs, path, meta); HFileBlock b = hbr.readBlockData(0, -1, -1, false); ByteBuffer data = b.getBufferWithoutHeader(); for (int i = 0; i < 1000; i++) { assertEquals(i, data.getInt()); } boolean exception_thrown = false; try { data.getInt(); } catch (BufferUnderflowException e) { exception_thrown = true; } assertTrue(exception_thrown); assertEquals(0, HFile.getChecksumFailuresCount()); } }
/** * This factory method is used only by unit tests */ static Reader createReaderFromStream(Path path, FSDataInputStream fsdis, long size, CacheConfig cacheConf, Configuration conf) throws IOException { FSDataInputStreamWrapper wrapper = new FSDataInputStreamWrapper(fsdis); return pickReaderVersion(path, wrapper, size, cacheConf, null, conf); }
/** * @param fs fileystem to read from * @param p path to the file * @param in {@link FSDataInputStreamWrapper} * @param size Full size of the file * @param cacheConf * @param r original reference file. This will be not null only when reading a split file. * @return a Reader instance to use instead of the base reader if overriding * default behavior, null otherwise * @throws IOException */ public StoreFile.Reader preStoreFileReaderOpen(final FileSystem fs, final Path p, final FSDataInputStreamWrapper in, final long size, final CacheConfig cacheConf, final Reference r) throws IOException { StoreFile.Reader reader = null; ObserverContext<RegionCoprocessorEnvironment> ctx = null; for (RegionEnvironment env : coprocessors) { if (env.getInstance() instanceof RegionObserver) { ctx = ObserverContext.createAndPrepare(env, ctx); Thread currentThread = Thread.currentThread(); ClassLoader cl = currentThread.getContextClassLoader(); try { currentThread.setContextClassLoader(env.getClassLoader()); reader = ((RegionObserver) env.getInstance()).preStoreFileReaderOpen(ctx, fs, p, in, size, cacheConf, r, reader); } catch (Throwable e) { handleCoprocessorThrowable(env, e); } finally { currentThread.setContextClassLoader(cl); } if (ctx.shouldComplete()) { break; } } } return reader; }
public FSReaderV2(FSDataInputStreamWrapper stream, long fileSize, HFileSystem hfs, Path path, HFileContext fileContext) throws IOException { super(fileSize, hfs, path, fileContext); this.streamWrapper = stream; // Older versions of HBase didn't support checksum. this.streamWrapper.prepareForBlockReader(!fileContext.isUseHBaseChecksum()); defaultDecodingCtx = new HFileBlockDefaultDecodingContext(fileContext); encodedBlockDecodingCtx = new HFileBlockDefaultDecodingContext(fileContext); }
@Override public Reader preStoreFileReaderOpen(ObserverContext<RegionCoprocessorEnvironment> ctx, FileSystem fs, Path path, FSDataInputStreamWrapper in, long size, CacheConfig cacheConf, Reference r, Reader reader) throws IOException { Configuration conf = ctx.getEnvironment().getConfiguration(); if (reader == null && r != null && isIndexRegionReference(path)) { return new IndexHalfStoreFileReader(fs, path, cacheConf, in, size, r, conf); } return reader; }
/** * @param p * @param cacheConf * @param r * @throws IOException */ public IndexHalfStoreFileReader(final FileSystem fs, final Path p, final CacheConfig cacheConf, final FSDataInputStreamWrapper in, long size, final Reference r, final Configuration conf) throws IOException { super(fs, p, in, size, cacheConf, conf); this.splitkey = r.getSplitKey(); // Is it top or bottom half? this.top = Reference.isTopFileRegion(r.getFileRegion()); this.splitRow = KeyValue.createKeyValueFromKey(splitkey).getRow(); }
@Override public StoreFile.Reader preStoreFileReaderOpen( ObserverContext<RegionCoprocessorEnvironment> ctx, FileSystem fs, Path p, FSDataInputStreamWrapper in, long size, CacheConfig cacheConf, Reference r, StoreFile.Reader reader) throws IOException { return null; }
@Override public StoreFile.Reader postStoreFileReaderOpen( ObserverContext<RegionCoprocessorEnvironment> ctx, FileSystem fs, Path p, FSDataInputStreamWrapper in, long size, CacheConfig cacheConf, Reference r, StoreFile.Reader reader) throws IOException { return null; }
/** * Method returns the reader given the specified arguments. * TODO This is a bad abstraction. See HBASE-6635. * * @param path hfile's path * @param fsdis stream of path's file * @param size max size of the trailer. * @param cacheConf Cache configuation values, cannot be null. * @param hfs * @param primaryReplicaReader true if this is a reader for primary replica * @return an appropriate instance of HFileReader * @throws IOException If file is invalid, will throw CorruptHFileException flavored IOException */ @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="SF_SWITCH_FALLTHROUGH", justification="Intentional") private static Reader openReader(Path path, FSDataInputStreamWrapper fsdis, long size, CacheConfig cacheConf, HFileSystem hfs, boolean primaryReplicaReader, Configuration conf) throws IOException { FixedFileTrailer trailer = null; try { boolean isHBaseChecksum = fsdis.shouldUseHBaseChecksum(); assert !isHBaseChecksum; // Initially we must read with FS checksum. trailer = FixedFileTrailer.readFromStream(fsdis.getStream(isHBaseChecksum), size); switch (trailer.getMajorVersion()) { case 2: LOG.debug("Opening HFile v2 with v3 reader"); // Fall through. FindBugs: SF_SWITCH_FALLTHROUGH case 3: return new HFileReaderImpl(path, trailer, fsdis, size, cacheConf, hfs, primaryReplicaReader, conf); default: throw new IllegalArgumentException("Invalid HFile version " + trailer.getMajorVersion()); } } catch (Throwable t) { IOUtils.closeQuietly(fsdis); throw new CorruptHFileException("Problem reading HFile Trailer from file " + path, t); } finally { fsdis.unbuffer(); } }