fe/src/main/java/org/apache/impala/catalog/FileMetadataLoader.java - impala - Git at Google

 // Licensed to the Apache Software Foundation (ASF) under one
 // or more contributor license agreements.  See the NOTICE file
 // distributed with this work for additional information
 // regarding copyright ownership.  The ASF licenses this file
 // to you under the Apache License, Version 2.0 (the
 // "License"); you may not use this file except in compliance
 // with the License.  You may obtain a copy of the License at
 //
 //   http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing,
 // software distributed under the License is distributed on an
 // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 // KIND, either express or implied.  See the License for the
 // specific language governing permissions and limitations
 // under the License.
 package org.apache.impala.catalog;

 import com.google.common.base.Objects;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Maps;

 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.BlockLocation;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.LocatedFileStatus;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.RemoteIterator;
 import org.apache.hadoop.hive.common.ValidTxnList;
 import org.apache.hadoop.hive.common.ValidWriteIdList;
 import org.apache.impala.catalog.HdfsPartition.FileDescriptor;
 import org.apache.impala.common.FileSystemUtil;
 import org.apache.impala.common.Reference;
 import org.apache.impala.compat.HdfsShim;
 import org.apache.impala.thrift.TNetworkAddress;
 import org.apache.impala.util.AcidUtils;
 import org.apache.impala.util.ListMap;
 import org.apache.impala.util.ThreadNameAnnotator;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;

 import javax.annotation.Nullable;

 /**
  * Utility for loading file metadata within a partition directory.
  */
 public class FileMetadataLoader {
   private final static Logger LOG = LoggerFactory.getLogger(FileMetadataLoader.class);
   private static final Configuration CONF = new Configuration();

   private final Path partDir_;
   private final boolean recursive_;
   private final ImmutableMap<String, FileDescriptor> oldFdsByRelPath_;
   private final ListMap<TNetworkAddress> hostIndex_;
   @Nullable
   private final ValidWriteIdList writeIds_;
   @Nullable
   private final ValidTxnList validTxnList_;

   private boolean forceRefreshLocations = false;

   private List<FileDescriptor> loadedFds_;
   private LoadStats loadStats_;

   /**
    * @param partDir the dir for which to fetch file metadata
    * @param recursive whether to recursively list files
    * @param oldFds any pre-existing file descriptors loaded for this table, used
    *   to optimize refresh if available.
    * @param hostIndex the host index with which to associate the file descriptors
    * @param validTxnList if non-null, it can tell whether a given transaction id is
    *   committed or not. We need it to ignore base directories of in-progress
    *   compactions.
    * @param writeIds if non-null, a write-id list which will filter the returned
    *   file descriptors to only include those indicated to be valid.
    */
   public FileMetadataLoader(Path partDir, boolean recursive, List<FileDescriptor> oldFds,
       ListMap<TNetworkAddress> hostIndex, @Nullable ValidTxnList validTxnList,
       @Nullable ValidWriteIdList writeIds) {
     // Either both validTxnList and writeIds are null, or none of them.
     Preconditions.checkState((validTxnList == null && writeIds == null) ||
                              (validTxnList != null && writeIds != null));
     partDir_ = Preconditions.checkNotNull(partDir);
     recursive_ = recursive;
     hostIndex_ = Preconditions.checkNotNull(hostIndex);
     oldFdsByRelPath_ = Maps.uniqueIndex(oldFds, FileDescriptor::getRelativePath);
     writeIds_ = writeIds;
     validTxnList_ = validTxnList;

     if (writeIds_ != null) {
       Preconditions.checkArgument(recursive_, "ACID tables must be listed recursively");
     }
   }

   /**
    * If 'refresh' is true, force re-fetching block locations even if a file does not
    * appear to have changed.
    */
   public void setForceRefreshBlockLocations(boolean refresh) {
     forceRefreshLocations = refresh;
   }

   /**
    * @return the file descriptors that were loaded after an invocation of load()
    */
   public List<FileDescriptor> getLoadedFds() {
     Preconditions.checkState(loadedFds_ != null,
         "Must have successfully loaded first");
     return loadedFds_;
   }

   /**
    * @return statistics about the descriptor loading process, after an invocation of
    * load()
    */
   public LoadStats getStats() {
     Preconditions.checkState(loadedFds_ != null,
         "Must have successfully loaded first");
     return loadStats_;
   }

   Path getPartDir() { return partDir_; }

   /**
    * Load the file descriptors, which may later be fetched using {@link #getLoadedFds()}.
    * After a successful load, stats may be fetched using {@link #getStats()}.
    *
    * If the directory does not exist, this succeeds and yields an empty list of
    * descriptors.
    *
    * @throws IOException if listing fails.
    */
   public void load() throws IOException {
     Preconditions.checkState(loadStats_ == null, "already loaded");
     loadStats_ = new LoadStats();
     FileSystem fs = partDir_.getFileSystem(CONF);

     // If we don't have any prior FDs from which we could re-use old block location info,
     // we'll need to fetch info for every returned file. In this case we can inline
     // that request with the 'list' call and save a round-trip per file.
     //
     // In the case that we _do_ have existing FDs which we can reuse, we'll optimistically
     // assume that most _can_ be reused, in which case it's faster to _not_ prefetch
     // the locations.
     boolean listWithLocations = FileSystemUtil.supportsStorageIds(fs) &&
         (oldFdsByRelPath_.isEmpty() || forceRefreshLocations);

     String msg = String.format("%s file metadata%s from path %s",
           oldFdsByRelPath_.isEmpty() ? "Loading" : "Refreshing",
           listWithLocations ? " with eager location-fetching" : "",
           partDir_);
     LOG.trace(msg);
     try (ThreadNameAnnotator tna = new ThreadNameAnnotator(msg)) {
       RemoteIterator<? extends FileStatus> fileStatuses;
       if (listWithLocations) {
         fileStatuses = FileSystemUtil.listFiles(fs, partDir_, recursive_);
       } else {
         fileStatuses = FileSystemUtil.listStatus(fs, partDir_, recursive_);

         // TODO(todd): we could look at the result of listing without locations, and if
         // we see that a substantial number of the files have changed, it may be better
         // to go back and re-list with locations vs doing an RPC per file.
       }
       loadedFds_ = new ArrayList<>();
       if (fileStatuses == null) return;

       Reference<Long> numUnknownDiskIds = new Reference<Long>(Long.valueOf(0));

       List<FileStatus> stats = new ArrayList<>();
       while (fileStatuses.hasNext()) {
         stats.add(fileStatuses.next());
       }

       if (writeIds_ != null) {
         stats = AcidUtils.filterFilesForAcidState(stats, partDir_, validTxnList_,
             writeIds_, loadStats_);
       }

       for (FileStatus fileStatus : stats) {
         if (fileStatus.isDirectory()) {
           continue;
         }

         if (!FileSystemUtil.isValidDataFile(fileStatus)) {
           ++loadStats_.hiddenFiles;
           continue;
         }
         String relPath = FileSystemUtil.relativizePath(fileStatus.getPath(), partDir_);
         FileDescriptor fd = oldFdsByRelPath_.get(relPath);
         if (listWithLocations || forceRefreshLocations ||
             hasFileChanged(fd, fileStatus)) {
           fd = createFd(fs, fileStatus, relPath, numUnknownDiskIds);
           ++loadStats_.loadedFiles;
         } else {
           ++loadStats_.skippedFiles;
         }
         loadedFds_.add(Preconditions.checkNotNull(fd));;
       }
       loadStats_.unknownDiskIds += numUnknownDiskIds.getRef();
       if (LOG.isTraceEnabled()) {
         LOG.trace(loadStats_.debugString());
       }
     }
   }

   /**
    * Create a FileDescriptor for the given FileStatus. If the FS supports block locations,
    * and FileStatus is a LocatedFileStatus (i.e. the location was prefetched) this uses
    * the already-loaded information; otherwise, this may have to remotely look up the
    * locations.
    */
   private FileDescriptor createFd(FileSystem fs, FileStatus fileStatus,
       String relPath, Reference<Long> numUnknownDiskIds) throws IOException {
     if (!FileSystemUtil.supportsStorageIds(fs)) {
       return FileDescriptor.createWithNoBlocks(fileStatus, relPath);
     }
     BlockLocation[] locations;
     if (fileStatus instanceof LocatedFileStatus) {
       locations = ((LocatedFileStatus)fileStatus).getBlockLocations();
     } else {
       locations = fs.getFileBlockLocations(fileStatus, 0, fileStatus.getLen());
     }
     return FileDescriptor.create(fileStatus, relPath, locations, hostIndex_,
         HdfsShim.isErasureCoded(fileStatus), numUnknownDiskIds);
   }

   /**
    * Compares the modification time and file size between the FileDescriptor and the
    * FileStatus to determine if the file has changed. Returns true if the file has changed
    * and false otherwise.
    */
   private static boolean hasFileChanged(FileDescriptor fd, FileStatus status) {
     return (fd == null) || (fd.getFileLength() != status.getLen()) ||
       (fd.getModificationTime() != status.getModificationTime());
   }

   // File/Block metadata loading stats for a single HDFS path.
   public class LoadStats {
     /** Number of files skipped because they pertain to an uncommitted ACID transaction */
     public int uncommittedAcidFilesSkipped = 0;

     /**
      * Number of files skipped because they pertain to ACID directories superceded
      * by later base data.
      */
     public int filesSupercededByNewerBase = 0;

     // Number of files for which the metadata was loaded.
     public int loadedFiles = 0;

     // Number of hidden files excluded from file metadata loading. More details at
     // isValidDataFile().
     public int hiddenFiles = 0;

     // Number of files skipped from file metadata loading because the files have not
     // changed since the last load. More details at hasFileChanged().
     //
     // TODO(todd) rename this to something indicating it was fast-pathed, not skipped
     public int skippedFiles = 0;

     // Number of unknown disk IDs encountered while loading block
     // metadata for this path.
     public int unknownDiskIds = 0;

     public String debugString() {
       return Objects.toStringHelper("")
         .add("path", partDir_)
         .add("loaded files", loadedFiles)
         .add("hidden files", nullIfZero(hiddenFiles))
         .add("skipped files", nullIfZero(skippedFiles))
         .add("uncommited files", nullIfZero(uncommittedAcidFilesSkipped))
         .add("superceded files", nullIfZero(filesSupercededByNewerBase))
         .add("unknown diskIds", nullIfZero(unknownDiskIds))
         .omitNullValues()
         .toString();
     }

     private Integer nullIfZero(int x) {
       return x > 0 ? x : null;
     }
   }
 }
	// Licensed to the Apache Software Foundation (ASF) under one
	// or more contributor license agreements. See the NOTICE file
	// distributed with this work for additional information
	// regarding copyright ownership. The ASF licenses this file
	// to you under the Apache License, Version 2.0 (the
	// "License"); you may not use this file except in compliance
	// with the License. You may obtain a copy of the License at
	//
	// http://www.apache.org/licenses/LICENSE-2.0
	//
	// Unless required by applicable law or agreed to in writing,
	// software distributed under the License is distributed on an
	// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	// KIND, either express or implied. See the License for the
	// specific language governing permissions and limitations
	// under the License.
	package org.apache.impala.catalog;

	import com.google.common.base.Objects;
	import com.google.common.base.Preconditions;
	import com.google.common.collect.ImmutableMap;
	import com.google.common.collect.Maps;

	import org.apache.hadoop.conf.Configuration;
	import org.apache.hadoop.fs.BlockLocation;
	import org.apache.hadoop.fs.FileStatus;
	import org.apache.hadoop.fs.FileSystem;
	import org.apache.hadoop.fs.LocatedFileStatus;
	import org.apache.hadoop.fs.Path;
	import org.apache.hadoop.fs.RemoteIterator;
	import org.apache.hadoop.hive.common.ValidTxnList;
	import org.apache.hadoop.hive.common.ValidWriteIdList;
	import org.apache.impala.catalog.HdfsPartition.FileDescriptor;
	import org.apache.impala.common.FileSystemUtil;
	import org.apache.impala.common.Reference;
	import org.apache.impala.compat.HdfsShim;
	import org.apache.impala.thrift.TNetworkAddress;
	import org.apache.impala.util.AcidUtils;
	import org.apache.impala.util.ListMap;
	import org.apache.impala.util.ThreadNameAnnotator;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import java.io.IOException;
	import java.util.ArrayList;
	import java.util.List;

	import javax.annotation.Nullable;

	/**
	* Utility for loading file metadata within a partition directory.
	*/
	public class FileMetadataLoader {
	private final static Logger LOG = LoggerFactory.getLogger(FileMetadataLoader.class);
	private static final Configuration CONF = new Configuration();

	private final Path partDir_;
	private final boolean recursive_;
	private final ImmutableMap<String, FileDescriptor> oldFdsByRelPath_;
	private final ListMap<TNetworkAddress> hostIndex_;
	@Nullable
	private final ValidWriteIdList writeIds_;
	@Nullable
	private final ValidTxnList validTxnList_;

	private boolean forceRefreshLocations = false;

	private List<FileDescriptor> loadedFds_;
	private LoadStats loadStats_;

	/**
	* @param partDir the dir for which to fetch file metadata
	* @param recursive whether to recursively list files
	* @param oldFds any pre-existing file descriptors loaded for this table, used
	* to optimize refresh if available.
	* @param hostIndex the host index with which to associate the file descriptors
	* @param validTxnList if non-null, it can tell whether a given transaction id is
	* committed or not. We need it to ignore base directories of in-progress
	* compactions.
	* @param writeIds if non-null, a write-id list which will filter the returned
	* file descriptors to only include those indicated to be valid.
	*/
	public FileMetadataLoader(Path partDir, boolean recursive, List<FileDescriptor> oldFds,
	ListMap<TNetworkAddress> hostIndex, @Nullable ValidTxnList validTxnList,
	@Nullable ValidWriteIdList writeIds) {
	// Either both validTxnList and writeIds are null, or none of them.
	Preconditions.checkState((validTxnList == null && writeIds == null) \|\|
	(validTxnList != null && writeIds != null));
	partDir_ = Preconditions.checkNotNull(partDir);
	recursive_ = recursive;
	hostIndex_ = Preconditions.checkNotNull(hostIndex);
	oldFdsByRelPath_ = Maps.uniqueIndex(oldFds, FileDescriptor::getRelativePath);
	writeIds_ = writeIds;
	validTxnList_ = validTxnList;

	if (writeIds_ != null) {
	Preconditions.checkArgument(recursive_, "ACID tables must be listed recursively");
	}
	}

	/**
	* If 'refresh' is true, force re-fetching block locations even if a file does not
	* appear to have changed.
	*/
	public void setForceRefreshBlockLocations(boolean refresh) {
	forceRefreshLocations = refresh;
	}

	/**
	* @return the file descriptors that were loaded after an invocation of load()
	*/
	public List<FileDescriptor> getLoadedFds() {
	Preconditions.checkState(loadedFds_ != null,
	"Must have successfully loaded first");
	return loadedFds_;
	}

	/**
	* @return statistics about the descriptor loading process, after an invocation of
	* load()
	*/
	public LoadStats getStats() {
	Preconditions.checkState(loadedFds_ != null,
	"Must have successfully loaded first");
	return loadStats_;
	}

	Path getPartDir() { return partDir_; }

	/**
	* Load the file descriptors, which may later be fetched using {@link #getLoadedFds()}.
	* After a successful load, stats may be fetched using {@link #getStats()}.
	*
	* If the directory does not exist, this succeeds and yields an empty list of
	* descriptors.
	*
	* @throws IOException if listing fails.
	*/
	public void load() throws IOException {
	Preconditions.checkState(loadStats_ == null, "already loaded");
	loadStats_ = new LoadStats();
	FileSystem fs = partDir_.getFileSystem(CONF);

	// If we don't have any prior FDs from which we could re-use old block location info,
	// we'll need to fetch info for every returned file. In this case we can inline
	// that request with the 'list' call and save a round-trip per file.
	//
	// In the case that we _do_ have existing FDs which we can reuse, we'll optimistically
	// assume that most _can_ be reused, in which case it's faster to _not_ prefetch
	// the locations.
	boolean listWithLocations = FileSystemUtil.supportsStorageIds(fs) &&
	(oldFdsByRelPath_.isEmpty() \|\| forceRefreshLocations);

	String msg = String.format("%s file metadata%s from path %s",
	oldFdsByRelPath_.isEmpty() ? "Loading" : "Refreshing",
	listWithLocations ? " with eager location-fetching" : "",
	partDir_);
	LOG.trace(msg);
	try (ThreadNameAnnotator tna = new ThreadNameAnnotator(msg)) {
	RemoteIterator<? extends FileStatus> fileStatuses;
	if (listWithLocations) {
	fileStatuses = FileSystemUtil.listFiles(fs, partDir_, recursive_);
	} else {
	fileStatuses = FileSystemUtil.listStatus(fs, partDir_, recursive_);

	// TODO(todd): we could look at the result of listing without locations, and if
	// we see that a substantial number of the files have changed, it may be better
	// to go back and re-list with locations vs doing an RPC per file.
	}
	loadedFds_ = new ArrayList<>();
	if (fileStatuses == null) return;

	Reference<Long> numUnknownDiskIds = new Reference<Long>(Long.valueOf(0));

	List<FileStatus> stats = new ArrayList<>();
	while (fileStatuses.hasNext()) {
	stats.add(fileStatuses.next());
	}

	if (writeIds_ != null) {
	stats = AcidUtils.filterFilesForAcidState(stats, partDir_, validTxnList_,
	writeIds_, loadStats_);
	}

	for (FileStatus fileStatus : stats) {
	if (fileStatus.isDirectory()) {
	continue;
	}

	if (!FileSystemUtil.isValidDataFile(fileStatus)) {
	++loadStats_.hiddenFiles;
	continue;
	}
	String relPath = FileSystemUtil.relativizePath(fileStatus.getPath(), partDir_);
	FileDescriptor fd = oldFdsByRelPath_.get(relPath);
	if (listWithLocations \|\| forceRefreshLocations \|\|
	hasFileChanged(fd, fileStatus)) {
	fd = createFd(fs, fileStatus, relPath, numUnknownDiskIds);
	++loadStats_.loadedFiles;
	} else {
	++loadStats_.skippedFiles;
	}
	loadedFds_.add(Preconditions.checkNotNull(fd));;
	}
	loadStats_.unknownDiskIds += numUnknownDiskIds.getRef();
	if (LOG.isTraceEnabled()) {
	LOG.trace(loadStats_.debugString());
	}
	}
	}

	/**
	* Create a FileDescriptor for the given FileStatus. If the FS supports block locations,
	* and FileStatus is a LocatedFileStatus (i.e. the location was prefetched) this uses
	* the already-loaded information; otherwise, this may have to remotely look up the
	* locations.
	*/
	private FileDescriptor createFd(FileSystem fs, FileStatus fileStatus,
	String relPath, Reference<Long> numUnknownDiskIds) throws IOException {
	if (!FileSystemUtil.supportsStorageIds(fs)) {
	return FileDescriptor.createWithNoBlocks(fileStatus, relPath);
	}
	BlockLocation[] locations;
	if (fileStatus instanceof LocatedFileStatus) {
	locations = ((LocatedFileStatus)fileStatus).getBlockLocations();
	} else {
	locations = fs.getFileBlockLocations(fileStatus, 0, fileStatus.getLen());
	}
	return FileDescriptor.create(fileStatus, relPath, locations, hostIndex_,
	HdfsShim.isErasureCoded(fileStatus), numUnknownDiskIds);
	}

	/**
	* Compares the modification time and file size between the FileDescriptor and the
	* FileStatus to determine if the file has changed. Returns true if the file has changed
	* and false otherwise.
	*/
	private static boolean hasFileChanged(FileDescriptor fd, FileStatus status) {
	return (fd == null) \|\| (fd.getFileLength() != status.getLen()) \|\|
	(fd.getModificationTime() != status.getModificationTime());
	}

	// File/Block metadata loading stats for a single HDFS path.
	public class LoadStats {
	/** Number of files skipped because they pertain to an uncommitted ACID transaction */
	public int uncommittedAcidFilesSkipped = 0;

	/**
	* Number of files skipped because they pertain to ACID directories superceded
	* by later base data.
	*/
	public int filesSupercededByNewerBase = 0;

	// Number of files for which the metadata was loaded.
	public int loadedFiles = 0;

	// Number of hidden files excluded from file metadata loading. More details at
	// isValidDataFile().
	public int hiddenFiles = 0;

	// Number of files skipped from file metadata loading because the files have not
	// changed since the last load. More details at hasFileChanged().
	//
	// TODO(todd) rename this to something indicating it was fast-pathed, not skipped
	public int skippedFiles = 0;

	// Number of unknown disk IDs encountered while loading block
	// metadata for this path.
	public int unknownDiskIds = 0;

	public String debugString() {
	return Objects.toStringHelper("")
	.add("path", partDir_)
	.add("loaded files", loadedFiles)
	.add("hidden files", nullIfZero(hiddenFiles))
	.add("skipped files", nullIfZero(skippedFiles))
	.add("uncommited files", nullIfZero(uncommittedAcidFilesSkipped))
	.add("superceded files", nullIfZero(filesSupercededByNewerBase))
	.add("unknown diskIds", nullIfZero(unknownDiskIds))
	.omitNullValues()
	.toString();
	}

	private Integer nullIfZero(int x) {
	return x > 0 ? x : null;
	}
	}
	}