format/src/main/java/com/github/sadikovi/riff/FileReader.java
/**
   * Prepare row buffer based on file path.
   * Method 'close()' should be called on row buffer after reading all data to close stream.
   * @param filter optional filter, if null then no filter applied
   * @return row buffer
   * @throws FileNotFoundException if either data or header file is not found
   * @throws IOException
   */
Prepare row buffer based on file path.
Method 'close()' should be called on row buffer after reading all data to close stream.
@param filter optional filter, if null then no filter applied
@return row buffer
@throws FileNotFoundException if either data or header file is not found
@throws IOException
Prepare row buffer based on file path . Method ' close () ' should be called on row buffer after reading all data to close stream . @param filter optional filter if null then no filter applied @return row buffer @throws FileNotFoundException if either data or header file is not found @throws IOException
public RowBuffer prepareRead(Tree filter) throws FileNotFoundException, IOException {
    if (readPrepared) throw new IOException("Reader reuse");
    FSDataInputStream in = null;
    try {
      in = fs.open(fileStatus.getPath(), hdfsBufferSize);
      fileHeader = FileHeader.readFrom(in);
      fileFooter = FileFooter.readFrom(in, fileStatus.getLen());
      LOG.debug("Found type description {}", fileHeader.getTypeDescription());
      CompressionCodec codec = Riff.decodeCompressionCodec(fileHeader.state(0));
      if (codec == null) {
        LOG.debug("Found no compression codec, using uncompressed");
      } else {
        LOG.debug("Found compression codec {}", codec);
      }
      PredicateState state = null;
      if (filter != null) {
        state = new PredicateState(filter, fileHeader.getTypeDescription());
      }
      boolean skipFile = false;
      if (state != null) {
        if (state.hasIndexedTreeOnly()) {
          skipFile = !state.indexTree().evaluateState(fileFooter.getFileStatistics());
        } else {
          skipFile = !state.tree().evaluateState(fileFooter.getFileStatistics());
        }
      }
      if (skipFile) {
        LOG.debug("Skip file {}", fileStatus.getPath());
        return Buffers.emptyRowBuffer(in);
      }
      StripeInformation[] stripes = fileFooter.getStripeInformation();
      stripes = evaluateStripes(stripes, state);
      LOG.debug("Prepare iterator to read data from {} stripes", stripes.length);
      readPrepared = true;
      return Buffers.prepareRowBuffer(in, stripes, fileHeader.getTypeDescription(), codec,
        bufferSize, state);
    } catch (IOException ioe) {
      if (in != null) {
        in.close();
      }
      throw ioe;
    }
  }
public RowBuffer prepareRead ( Tree filter ) throws FileNotFoundException , IOException { if ( readPrepared ) throw new IOException ( "Reader reuse" ) ; FSDataInputStream in = null ; try { in = fs . open ( fileStatus . getPath ( ) , hdfsBufferSize ) ; fileHeader = FileHeader . readFrom ( in ) ; fileFooter = FileFooter . readFrom ( in , fileStatus . getLen ( ) ) ; LOG . debug ( "Found type description {}" , fileHeader . getTypeDescription ( ) ) ; CompressionCodec codec = Riff . decodeCompressionCodec ( fileHeader . state ( 0 ) ) ; if ( codec == null ) { LOG . debug ( "Found no compression codec, using uncompressed" ) ; } else { LOG . debug ( "Found compression codec {}" , codec ) ; } PredicateState state = null ; if ( filter != null ) { state = new PredicateState ( filter , fileHeader . getTypeDescription ( ) ) ; } boolean skipFile = false ; if ( state != null ) { if ( state . hasIndexedTreeOnly ( ) ) { skipFile = ! state . indexTree ( ) . evaluateState ( fileFooter . getFileStatistics ( ) ) ; } else { skipFile = ! state . tree ( ) . evaluateState ( fileFooter . getFileStatistics ( ) ) ; } } if ( skipFile ) { LOG . debug ( "Skip file {}" , fileStatus . getPath ( ) ) ; return Buffers . emptyRowBuffer ( in ) ; } StripeInformation [ ] stripes = fileFooter . getStripeInformation ( ) ; stripes = evaluateStripes ( stripes , state ) ; LOG . debug ( "Prepare iterator to read data from {} stripes" , stripes . length ) ; readPrepared = true ; return Buffers . prepareRowBuffer ( in , stripes , fileHeader . getTypeDescription ( ) , codec , bufferSize , state ) ; } catch ( IOException ioe ) { if ( in != null ) { in . close ( ) ; } throw ioe ; } }
Prepare row buffer based on file path.
Prepare row buffer based on file path .
// we start with reading file header and extracting all information that is required to // validate file and/or resolve statistics // read input stream and return file state // initialize valid predicate state if necessary // if predicate state is available - evaluate tree and decide on whether or not to read the // file any further. // reevaluate stripes based on predicate tree