fe/src/main/java/com/cloudera/impala/catalog/HdfsTable.java
/**
   * Loads the file block metadata for the given collection of FileDescriptors.
   * The FileDescriptors are passed as a Map of partition location to list of
   * files that exist under that directory.
   */
Loads the file block metadata for the given collection of FileDescriptors.
The FileDescriptors are passed as a Map of partition location to list of
files that exist under that directory.
Loads the file block metadata for the given collection of FileDescriptors . The FileDescriptors are passed as a Map of partition location to list of files that exist under that directory .
private void loadBlockMd(Map<String, List<FileDescriptor>> fileDescriptors)
      throws RuntimeException {
    LOG.debug("load block md for " + name_);
    List<BlockLocation> blockLocations = Lists.newArrayList();
    for (String parentPath: fileDescriptors.keySet()) {
      for (FileDescriptor fileDescriptor: fileDescriptors.get(parentPath)) {
        Path p = new Path(parentPath, fileDescriptor.getFileName());
        BlockLocation[] locations = null;
        try {
          FileStatus fileStatus = DFS.getFileStatus(p);
          Preconditions.checkArgument(!fileStatus.isDirectory());
          locations = DFS.getFileBlockLocations(fileStatus, 0, fileStatus.getLen());
          Preconditions.checkNotNull(locations);
          blockLocations.addAll(Arrays.asList(locations));
          for (BlockLocation block: locations) {
            String[] blockHostPorts = block.getNames();
            try {
              blockHostPorts = block.getNames();
            } catch (IOException e) {
              String errorMsg = "BlockLocation.getNames() failed:\n" + e.getMessage();
              LOG.error(errorMsg);
              throw new IllegalStateException(errorMsg);
            }
            List<Integer> replicaHostIdxs = new ArrayList<Integer>(blockHostPorts.length);
            for (int i = 0; i < blockHostPorts.length; ++i) {
              String[] ip_port = blockHostPorts[i].split(":");
              Preconditions.checkState(ip_port.length == 2);
              TNetworkAddress network_address = new TNetworkAddress(ip_port[0],
                  Integer.parseInt(ip_port[1]));
              Integer hostIdx = hostMap_.get(network_address);
              if (hostIdx == null) {
                hostList_.add(network_address);
                hostMap_.put(network_address, hostList_.size() - 1);
                hostIdx = hostList_.size() - 1;
              }
              replicaHostIdxs.add(hostIdx);
            }
            fileDescriptor.addFileBlock(
                new FileBlock(block.getOffset(), block.getLength(), replicaHostIdxs));
          }
        } catch (IOException e) {
          throw new RuntimeException("couldn't determine block locations for path '"
              + p + "':\n" + e.getMessage(), e);
        }
      }
    }
    if (SUPPORTS_VOLUME_ID) {
      LOG.trace("loading disk ids for: " + getFullName() +
          ". nodes: " + getNumNodes());
      loadDiskIds(blockLocations, fileDescriptors);
      LOG.trace("completed load of disk ids for: " + getFullName());
    }
  }
private void loadBlockMd ( Map < String , List < FileDescriptor > > fileDescriptors ) throws RuntimeException { LOG . debug ( "load block md for " + name_ ) ; List < BlockLocation > blockLocations = Lists . newArrayList ( ) ; for ( String parentPath : fileDescriptors . keySet ( ) ) { for ( FileDescriptor fileDescriptor : fileDescriptors . get ( parentPath ) ) { Path p = new Path ( parentPath , fileDescriptor . getFileName ( ) ) ; BlockLocation [ ] locations = null ; try { FileStatus fileStatus = DFS . getFileStatus ( p ) ; Preconditions . checkArgument ( ! fileStatus . isDirectory ( ) ) ; locations = DFS . getFileBlockLocations ( fileStatus , 0 , fileStatus . getLen ( ) ) ; Preconditions . checkNotNull ( locations ) ; blockLocations . addAll ( Arrays . asList ( locations ) ) ; for ( BlockLocation block : locations ) { String [ ] blockHostPorts = block . getNames ( ) ; try { blockHostPorts = block . getNames ( ) ; } catch ( IOException e ) { String errorMsg = "BlockLocation.getNames() failed:\n" + e . getMessage ( ) ; LOG . error ( errorMsg ) ; throw new IllegalStateException ( errorMsg ) ; } List < Integer > replicaHostIdxs = new ArrayList < Integer > ( blockHostPorts . length ) ; for ( int i = 0 ; i < blockHostPorts . length ; ++ i ) { String [ ] ip_port = blockHostPorts [ i ] . split ( ":" ) ; Preconditions . checkState ( ip_port . length == 2 ) ; TNetworkAddress network_address = new TNetworkAddress ( ip_port [ 0 ] , Integer . parseInt ( ip_port [ 1 ] ) ) ; Integer hostIdx = hostMap_ . get ( network_address ) ; if ( hostIdx == null ) { hostList_ . add ( network_address ) ; hostMap_ . put ( network_address , hostList_ . size ( ) - 1 ) ; hostIdx = hostList_ . size ( ) - 1 ; } replicaHostIdxs . add ( hostIdx ) ; } fileDescriptor . addFileBlock ( new FileBlock ( block . getOffset ( ) , block . getLength ( ) , replicaHostIdxs ) ) ; } } catch ( IOException e ) { throw new RuntimeException ( "couldn't determine block locations for path '" + p + "':\n" + e . getMessage ( ) , e ) ; } } } if ( SUPPORTS_VOLUME_ID ) { LOG . trace ( "loading disk ids for: " + getFullName ( ) + ". nodes: " + getNumNodes ( ) ) ; loadDiskIds ( blockLocations , fileDescriptors ) ; LOG . trace ( "completed load of disk ids for: " + getFullName ( ) ) ; } }
Loads the file block metadata for the given collection of FileDescriptors.
Loads the file block metadata for the given collection of FileDescriptors .
// Store all BlockLocations so they can be reused when loading the disk IDs. // loop over all files and record their block metadata, minus volume ids // fileDescriptors should not contain directories. // Loop over all blocks in the file. // this shouldn't happen, getNames() doesn't throw anything // Now enumerate all replicas of the block, adding any unknown hosts // to hostMap_/hostList_. The host ID (index in to the hostList_) for each // replica is stored in replicaHostIdxs. // No match was found, add a new entry for this host to the hostMap_.