| // 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.common; |
| |
| import com.google.common.annotations.VisibleForTesting; |
| import com.google.common.base.Preconditions; |
| import com.google.common.collect.ImmutableMap; |
| import org.apache.commons.io.IOUtils; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.fs.FileStatus; |
| import org.apache.hadoop.fs.FileSystem; |
| import org.apache.hadoop.fs.FileUtil; |
| import org.apache.hadoop.fs.LocalFileSystem; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.hadoop.fs.RemoteIterator; |
| import org.apache.hadoop.fs.adl.AdlFileSystem; |
| import org.apache.hadoop.fs.azurebfs.AzureBlobFileSystem; |
| import org.apache.hadoop.fs.azurebfs.SecureAzureBlobFileSystem; |
| import org.apache.hadoop.fs.s3a.S3AFileSystem; |
| import org.apache.hadoop.hdfs.DistributedFileSystem; |
| import org.apache.hadoop.hdfs.client.HdfsAdmin; |
| import org.apache.hadoop.hdfs.protocol.EncryptionZone; |
| import org.apache.impala.catalog.HdfsCompression; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import java.io.FileNotFoundException; |
| import java.io.IOException; |
| import java.io.InputStream; |
| import java.net.URI; |
| import java.util.Map; |
| import java.util.NoSuchElementException; |
| import java.util.Stack; |
| import java.util.UUID; |
| |
| /** |
| * Common utility functions for operating on FileSystem objects. |
| */ |
| public class FileSystemUtil { |
| private static final Configuration CONF = new Configuration(); |
| private static final Logger LOG = LoggerFactory.getLogger(FileSystemUtil.class); |
| |
| /** |
| * Performs a non-recursive delete of all visible (non-hidden) files in a given |
| * directory. Returns the number of files deleted as part of this operation. |
| */ |
| public static int deleteAllVisibleFiles(Path directory) |
| throws IOException { |
| FileSystem fs = directory.getFileSystem(CONF); |
| Preconditions.checkState(fs.getFileStatus(directory).isDirectory()); |
| int numFilesDeleted = 0; |
| for (FileStatus fStatus: fs.listStatus(directory)) { |
| // Only delete files that are not hidden. |
| if (fStatus.isFile() && !isHiddenFile(fStatus.getPath().getName())) { |
| if (LOG.isTraceEnabled()) LOG.trace("Removing: " + fStatus.getPath()); |
| fs.delete(fStatus.getPath(), false); |
| ++numFilesDeleted; |
| } |
| } |
| return numFilesDeleted; |
| } |
| |
| /** |
| * Returns the total number of visible (non-hidden) files in a directory. |
| */ |
| public static int getTotalNumVisibleFiles(Path directory) throws IOException { |
| FileSystem fs = directory.getFileSystem(CONF); |
| Preconditions.checkState(fs.getFileStatus(directory).isDirectory()); |
| int numFiles = 0; |
| for (FileStatus fStatus: fs.listStatus(directory)) { |
| // Only delete files that are not hidden. |
| if (fStatus.isFile() && !isHiddenFile(fStatus.getPath().getName())) { |
| ++numFiles; |
| } |
| } |
| return numFiles; |
| } |
| |
| /** |
| * Returns true if path p1 and path p2 are in the same encryption zone in HDFS. |
| * Returns false if they are in different encryption zones or if either of the paths |
| * are not on HDFS. |
| */ |
| private static boolean arePathsInSameHdfsEncryptionZone(FileSystem fs, Path p1, |
| Path p2) throws IOException { |
| // Only distributed file systems have encryption zones. |
| if (!isDistributedFileSystem(p1) || !isDistributedFileSystem(p2)) return false; |
| HdfsAdmin hdfsAdmin = new HdfsAdmin(fs.getUri(), CONF); |
| EncryptionZone z1 = hdfsAdmin.getEncryptionZoneForPath(p1); |
| EncryptionZone z2 = hdfsAdmin.getEncryptionZoneForPath(p2); |
| if (z1 == null && z2 == null) return true; |
| if (z1 == null || z2 == null) return false; |
| return z1.equals(z2); |
| } |
| |
| /** |
| * Relocates all visible (non-hidden) files from a source directory to a destination |
| * directory. Files are moved (renamed) to the new location unless the source and |
| * destination directories are in different encryption zones, in which case the files |
| * are copied so that they are decrypted and/or encrypted. Naming conflicts are |
| * resolved by appending a UUID to the base file name. Any sub-directories within the |
| * source directory are skipped. Returns the number of files relocated as part of this |
| * operation. |
| */ |
| public static int relocateAllVisibleFiles(Path sourceDir, Path destDir) |
| throws IOException { |
| FileSystem destFs = destDir.getFileSystem(CONF); |
| FileSystem sourceFs = sourceDir.getFileSystem(CONF); |
| Preconditions.checkState(destFs.isDirectory(destDir)); |
| Preconditions.checkState(sourceFs.isDirectory(sourceDir)); |
| |
| // Use the same UUID to resolve all file name conflicts. This helps mitigate problems |
| // that might happen if there is a conflict moving a set of files that have |
| // dependent file names. For example, foo.lzo and foo.lzo_index. |
| UUID uuid = UUID.randomUUID(); |
| |
| // Enumerate all the files in the source |
| int numFilesMoved = 0; |
| for (FileStatus fStatus: sourceFs.listStatus(sourceDir)) { |
| if (fStatus.isDirectory()) { |
| if (LOG.isTraceEnabled()) { |
| LOG.trace("Skipping copy of directory: " + fStatus.getPath()); |
| } |
| continue; |
| } else if (isHiddenFile(fStatus.getPath().getName())) { |
| continue; |
| } |
| |
| Path destFile = new Path(destDir, fStatus.getPath().getName()); |
| if (destFs.exists(destFile)) { |
| destFile = new Path(destDir, |
| appendToBaseFileName(destFile.getName(), uuid.toString())); |
| } |
| FileSystemUtil.relocateFile(fStatus.getPath(), destFile, false); |
| ++numFilesMoved; |
| } |
| return numFilesMoved; |
| } |
| |
| /** |
| * Relocates the given file to a new location (either another directory or a |
| * file in the same or different filesystem). The file is generally moved (renamed) to |
| * the new location. However, the file is copied if the source and destination are in |
| * different encryption zones so that the file can be decrypted and/or encrypted, or if |
| * the source and destination are in different filesystems. If renameIfAlreadyExists is |
| * true, no error will be thrown if a file with the same name already exists in the |
| * destination location. Instead, a UUID will be appended to the base file name, |
| * preserving the existing file extension. If renameIfAlreadyExists is false, an |
| * IOException will be thrown if there is a file name conflict. |
| */ |
| public static void relocateFile(Path sourceFile, Path dest, |
| boolean renameIfAlreadyExists) throws IOException { |
| FileSystem destFs = dest.getFileSystem(CONF); |
| FileSystem sourceFs = sourceFile.getFileSystem(CONF); |
| |
| Path destFile = |
| destFs.isDirectory(dest) ? new Path(dest, sourceFile.getName()) : dest; |
| // If a file with the same name does not already exist in the destination location |
| // then use the same file name. Otherwise, generate a unique file name. |
| if (renameIfAlreadyExists && destFs.exists(destFile)) { |
| Path destDir = destFs.isDirectory(dest) ? dest : dest.getParent(); |
| destFile = new Path(destDir, |
| appendToBaseFileName(destFile.getName(), UUID.randomUUID().toString())); |
| } |
| boolean sameFileSystem = isPathOnFileSystem(sourceFile, destFs); |
| boolean destIsDfs = isDistributedFileSystem(destFs); |
| |
| // If the source and the destination are on different file systems, or in different |
| // encryption zones, files can't be moved from one location to the other and must be |
| // copied instead. |
| boolean sameEncryptionZone = |
| arePathsInSameHdfsEncryptionZone(destFs, sourceFile, destFile); |
| // We can do a rename if the src and dst are in the same encryption zone in the same |
| // distributed filesystem. |
| boolean doRename = destIsDfs && sameFileSystem && sameEncryptionZone; |
| // Alternatively, we can do a rename if the src and dst are on the same |
| // non-distributed filesystem. |
| if (!doRename) doRename = !destIsDfs && sameFileSystem; |
| if (doRename) { |
| if (LOG.isTraceEnabled()) { |
| LOG.trace(String.format( |
| "Moving '%s' to '%s'", sourceFile.toString(), destFile.toString())); |
| } |
| // Move (rename) the file. |
| destFs.rename(sourceFile, destFile); |
| return; |
| } |
| if (destIsDfs && sameFileSystem) { |
| Preconditions.checkState(!doRename); |
| // We must copy rather than move if the source and dest are in different |
| // encryption zones. A move would return an error from the NN because a move is a |
| // metadata-only operation and the files would not be encrypted/decrypted properly |
| // on the DNs. |
| if (LOG.isTraceEnabled()) { |
| LOG.trace(String.format( |
| "Copying source '%s' to '%s' because HDFS encryption zones are different.", |
| sourceFile, destFile)); |
| } |
| } else { |
| Preconditions.checkState(!sameFileSystem); |
| if (LOG.isTraceEnabled()) { |
| LOG.trace(String.format("Copying '%s' to '%s' between filesystems.", |
| sourceFile, destFile)); |
| } |
| } |
| FileUtil.copy(sourceFs, sourceFile, destFs, destFile, true, true, CONF); |
| } |
| |
| /** |
| * Reads the file at path and returns the contents. |
| */ |
| public static String readFile(Path file) throws IOException { |
| FileSystem fs = file.getFileSystem(CONF); |
| InputStream fileStream = fs.open(file); |
| try { |
| return IOUtils.toString(fileStream); |
| } finally { |
| IOUtils.closeQuietly(fileStream); |
| } |
| } |
| |
| /** |
| * Builds a new file name based on a base file name. This is done by inserting |
| * the given appendStr into the base file name, preserving the file extension (if |
| * one exists). |
| * For example, this could be passed a UUID string to uniquify files: |
| * file1.snap -> file1_<uuid>.snap |
| * file1 -> file1_<uuid> |
| */ |
| private static String appendToBaseFileName(String baseFileName, String appendStr) { |
| StringBuilder sb = new StringBuilder(baseFileName); |
| // Insert the string to append, preserving the file extension. |
| int extensionIdx = baseFileName.lastIndexOf('.'); |
| if (extensionIdx != -1) { |
| sb.replace(extensionIdx, extensionIdx + 1, "_" + appendStr + "."); |
| } else { |
| sb.append("_" + appendStr); |
| } |
| return sb.toString(); |
| } |
| |
| /** |
| * Returns true if the given Path contains any visible sub directories, otherwise false. |
| */ |
| public static boolean containsVisibleSubdirectory(Path directory) |
| throws FileNotFoundException, IOException { |
| FileSystem fs = directory.getFileSystem(CONF); |
| // Enumerate all the files in the source |
| for (FileStatus fStatus: fs.listStatus(directory)) { |
| String pathName = fStatus.getPath().getName(); |
| if (fStatus.isDirectory() && !isHiddenFile(pathName)) { |
| return true; |
| } |
| } |
| return false; |
| } |
| |
| /** |
| * Makes a temporary unique directory within the given directory. |
| */ |
| public static Path makeTmpSubdirectory(Path directory) throws IOException { |
| FileSystem fs = directory.getFileSystem(CONF); |
| Path tmpDir = new Path(directory, ".tmp_" + UUID.randomUUID().toString()); |
| fs.mkdirs(tmpDir); |
| return tmpDir; |
| } |
| |
| public static boolean isHiddenFile(String fileName) { |
| // Hidden files start with '.' or '_'. The '.copying' suffix is used by some |
| // filesystem utilities (e.g. hdfs put) as a temporary destination when copying |
| // files. The '.tmp' suffix is Flume's default for temporary files. |
| String lcFileName = fileName.toLowerCase(); |
| return lcFileName.startsWith(".") || lcFileName.startsWith("_") || |
| lcFileName.endsWith(".copying") || lcFileName.endsWith(".tmp"); |
| } |
| |
| /** |
| * Returns true if the file corresponding to 'fileStatus' is a valid data file as |
| * per Impala's partitioning rules. A fileStatus is considered invalid if its a |
| * directory/hidden file/LZO index file. LZO index files are skipped because they are |
| * read by the scanner directly. Currently Impala doesn't allow subdirectories in the |
| * partition paths. |
| */ |
| public static boolean isValidDataFile(FileStatus fileStatus) { |
| String fileName = fileStatus.getPath().getName(); |
| return !(fileStatus.isDirectory() || FileSystemUtil.isHiddenFile(fileName) || |
| HdfsCompression.fromFileName(fileName) == HdfsCompression.LZO_INDEX); |
| } |
| |
| /** |
| * Returns true if the filesystem supports storage UUIDs in BlockLocation calls. |
| */ |
| public static boolean supportsStorageIds(FileSystem fs) { |
| // Common case. |
| if (isDistributedFileSystem(fs)) return true; |
| // Blacklist FileSystems that are known to not to include storage UUIDs. |
| return !(fs instanceof S3AFileSystem || fs instanceof LocalFileSystem || |
| fs instanceof AzureBlobFileSystem || fs instanceof SecureAzureBlobFileSystem || |
| fs instanceof AdlFileSystem); |
| } |
| |
| /** |
| * Returns true iff the filesystem is a S3AFileSystem. |
| */ |
| public static boolean isS3AFileSystem(FileSystem fs) { |
| return fs instanceof S3AFileSystem; |
| } |
| |
| /** |
| * Returns true iff the path is on a S3AFileSystem. |
| */ |
| public static boolean isS3AFileSystem(Path path) throws IOException { |
| return isS3AFileSystem(path.getFileSystem(CONF)); |
| } |
| |
| /** |
| * Returns true iff the filesystem is AdlFileSystem. |
| */ |
| public static boolean isADLFileSystem(FileSystem fs) { |
| return fs instanceof AdlFileSystem; |
| } |
| |
| /** |
| * Returns true iff the path is on AdlFileSystem. |
| */ |
| public static boolean isADLFileSystem(Path path) throws IOException { |
| return isADLFileSystem(path.getFileSystem(CONF)); |
| } |
| |
| /** |
| * Returns true iff the filesystem is AzureBlobFileSystem or |
| * SecureAzureBlobFileSystem. This function is unique in that there are 2 |
| * distinct classes it checks for, but the ony functional difference is the |
| * use of wire encryption. Some features like OAuth authentication do require |
| * wire encryption but that does not matter in usages of this function. |
| */ |
| public static boolean isABFSFileSystem(FileSystem fs) { |
| return fs instanceof AzureBlobFileSystem |
| || fs instanceof SecureAzureBlobFileSystem; |
| } |
| |
| /** |
| * Returns true iff the path is on AzureBlobFileSystem or |
| * SecureAzureBlobFileSystem. |
| */ |
| public static boolean isABFSFileSystem(Path path) throws IOException { |
| return isABFSFileSystem(path.getFileSystem(CONF)); |
| } |
| |
| /** |
| * Returns true iff the filesystem is an instance of LocalFileSystem. |
| */ |
| public static boolean isLocalFileSystem(FileSystem fs) { |
| return fs instanceof LocalFileSystem; |
| } |
| |
| /** |
| * Return true iff path is on a local filesystem. |
| */ |
| public static boolean isLocalFileSystem(Path path) throws IOException { |
| return isLocalFileSystem(path.getFileSystem(CONF)); |
| } |
| |
| /** |
| * Returns true iff the filesystem is a DistributedFileSystem. |
| */ |
| public static boolean isDistributedFileSystem(FileSystem fs) { |
| return fs instanceof DistributedFileSystem; |
| } |
| |
| /** |
| * Return true iff path is on a DFS filesystem. |
| */ |
| public static boolean isDistributedFileSystem(Path path) throws IOException { |
| return isDistributedFileSystem(path.getFileSystem(CONF)); |
| } |
| |
| /** |
| * Represents the type of filesystem being used. Typically associated with a |
| * {@link org.apache.hadoop.fs.FileSystem} instance that is used to read data. |
| * |
| * <p> |
| * Unlike the {@code is*FileSystem} methods above. A FsType is more |
| * generic in that it is capable of grouping different filesystems to the |
| * same type. For example, the FsType {@link FsType#ADLS} maps to |
| * multiple filesystems: {@link AdlFileSystem}, |
| * {@link AzureBlobFileSystem}, and {@link SecureAzureBlobFileSystem}. |
| * </p> |
| */ |
| public enum FsType { |
| ADLS, |
| HDFS, |
| LOCAL, |
| S3; |
| |
| private static final Map<String, FsType> SCHEME_TO_FS_MAPPING = |
| ImmutableMap.<String, FsType>builder() |
| .put("abfs", ADLS) |
| .put("abfss", ADLS) |
| .put("adl", ADLS) |
| .put("file", LOCAL) |
| .put("hdfs", HDFS) |
| .put("s3a", S3) |
| .build(); |
| |
| /** |
| * Provides a mapping between filesystem schemes and filesystems types. This can be |
| * useful as there are often multiple filesystem connectors for a give fs, each |
| * with its own scheme (e.g. abfs, abfss, adl are all ADLS connectors). |
| * Returns the {@link FsType} associated with a given filesystem scheme (e.g. local, |
| * hdfs, s3a, etc.) |
| */ |
| public static FsType getFsType(String scheme) { |
| return SCHEME_TO_FS_MAPPING.get(scheme); |
| } |
| } |
| |
| public static FileSystem getDefaultFileSystem() throws IOException { |
| Path path = new Path(FileSystem.getDefaultUri(CONF)); |
| FileSystem fs = path.getFileSystem(CONF); |
| return fs; |
| } |
| |
| /** |
| * Returns the FileSystem object for a given path using the cached config. |
| */ |
| public static FileSystem getFileSystemForPath(Path p) throws IOException { |
| return p.getFileSystem(CONF); |
| } |
| |
| public static DistributedFileSystem getDistributedFileSystem() throws IOException { |
| FileSystem fs = getDefaultFileSystem(); |
| Preconditions.checkState(fs instanceof DistributedFileSystem); |
| return (DistributedFileSystem) fs; |
| } |
| |
| /** |
| * Fully-qualifies the given path based on the FileSystem configuration. |
| */ |
| public static Path createFullyQualifiedPath(Path location) { |
| URI defaultUri = FileSystem.getDefaultUri(CONF); |
| URI locationUri = location.toUri(); |
| // Use the default URI only if location has no scheme or it has the same scheme as |
| // the default URI. Otherwise, Path.makeQualified() will incorrectly use the |
| // authority from the default URI even though the schemes don't match. See HDFS-7031. |
| if (locationUri.getScheme() == null || |
| locationUri.getScheme().equalsIgnoreCase(defaultUri.getScheme())) { |
| return location.makeQualified(defaultUri, location); |
| } |
| // Already qualified (has scheme). |
| return location; |
| } |
| |
| /** |
| * Return true iff the path is on the given filesystem. |
| */ |
| public static boolean isPathOnFileSystem(Path path, FileSystem fs) { |
| try { |
| // Call makeQualified() for the side-effect of FileSystem.checkPath() which will |
| // throw an exception if path is not on fs. |
| fs.makeQualified(path); |
| return true; |
| } catch (IllegalArgumentException e) { |
| // Path is not on fs. |
| return false; |
| } |
| } |
| |
| /** |
| * Copies the source file to a destination path on the local filesystem. |
| * Throws IOException on failure. |
| */ |
| public static void copyToLocal(Path source, Path dest) throws IOException { |
| FileSystem fs = source.getFileSystem(CONF); |
| fs.copyToLocalFile(source, dest); |
| } |
| |
| /** |
| * Delete the file at 'path' if it exists. |
| */ |
| public static void deleteIfExists(Path path) { |
| try { |
| FileSystem fs = path.getFileSystem(CONF); |
| if (!fs.exists(path)) return; |
| fs.delete(path); |
| } catch (IOException e) { |
| LOG.warn("Encountered an exception deleting file at path " + path.toString(), e); |
| } |
| } |
| |
| /** |
| * Returns true if the given path is a location which supports caching (e.g. HDFS). |
| */ |
| public static boolean isPathCacheable(Path path) { |
| try { |
| return isDistributedFileSystem(path); |
| } catch (IOException e) { |
| return false; |
| } |
| } |
| |
| /** |
| * Returns the configuration. |
| */ |
| public static Configuration getConfiguration() { |
| return CONF; |
| } |
| |
| /** |
| * Returns true iff the given location is on a filesystem that Impala can write to. |
| */ |
| public static boolean isImpalaWritableFilesystem(String location) |
| throws IOException { |
| Path path = new Path(location); |
| return (FileSystemUtil.isDistributedFileSystem(path) || |
| FileSystemUtil.isLocalFileSystem(path) || |
| FileSystemUtil.isS3AFileSystem(path) || |
| FileSystemUtil.isABFSFileSystem(path) || |
| FileSystemUtil.isADLFileSystem(path)); |
| } |
| |
| /** |
| * Wrapper around FileSystem.listStatus() that specifically handles the case when |
| * the path does not exist. This helps simplify the caller code in cases where |
| * the file does not exist and also saves an RPC as the caller need not do a separate |
| * exists check for the path. Returns null if the path does not exist. |
| * |
| * If 'recursive' is true, all underlying files and directories will be yielded. |
| * Note that the order (breadth-first vs depth-first, sorted vs not) is undefined. |
| */ |
| public static RemoteIterator<? extends FileStatus> listStatus(FileSystem fs, Path p, |
| boolean recursive) throws IOException { |
| try { |
| if (recursive) { |
| // The Hadoop FileSystem API doesn't provide a recursive listStatus call that |
| // doesn't also fetch block locations, and fetching block locations is expensive. |
| // Here, our caller specifically doesn't need block locations, so we don't want to |
| // call the expensive 'listFiles' call on HDFS. Instead, we need to "manually" |
| // recursively call FileSystem.listStatusIterator(). |
| // |
| // Note that this "manual" recursion is not actually any slower than the recursion |
| // provided by the HDFS 'listFiles(recursive=true)' API, since the HDFS wire |
| // protocol doesn't provide any such recursive support anyway. In other words, |
| // the API that looks like a single recursive call is just as bad as what we're |
| // doing here. |
| // |
| // However, S3 actually implements 'listFiles(recursive=true)' with a faster path |
| // which natively recurses. In that case, it's quite preferable to use 'listFiles' |
| // even though it returns LocatedFileStatus objects with "fake" blocks which we |
| // will ignore. |
| if (isS3AFileSystem(fs)) { |
| return listFiles(fs, p, true); |
| } |
| |
| return new FilterIterator(p, new RecursingIterator(fs, p)); |
| } |
| |
| return new FilterIterator(p, fs.listStatusIterator(p)); |
| } catch (FileNotFoundException e) { |
| if (LOG.isWarnEnabled()) LOG.warn("Path does not exist: " + p.toString(), e); |
| return null; |
| } |
| } |
| |
| /** |
| * Wrapper around FileSystem.listFiles(), similar to the listStatus() wrapper above. |
| */ |
| public static RemoteIterator<? extends FileStatus> listFiles(FileSystem fs, Path p, |
| boolean recursive) throws IOException { |
| try { |
| return new FilterIterator(p, fs.listFiles(p, recursive)); |
| } catch (FileNotFoundException e) { |
| if (LOG.isWarnEnabled()) LOG.warn("Path does not exist: " + p.toString(), e); |
| return null; |
| } |
| } |
| |
| /** |
| * Returns true if the path 'p' is a directory, false otherwise. |
| */ |
| public static boolean isDir(Path p) throws IOException { |
| FileSystem fs = getFileSystemForPath(p); |
| return fs.isDirectory(p); |
| } |
| |
| /** |
| * Returns true if the path 'p' is a file, false if not. Throws if path does not exist. |
| */ |
| public static boolean isFile(Path p) throws IOException, FileNotFoundException { |
| FileSystem fs = getFileSystemForPath(p); |
| return fs.getFileStatus(p).isFile(); |
| } |
| |
| /** |
| * Return the path of 'path' relative to the startPath. This may |
| * differ from simply the file name in the case of recursive listings. |
| */ |
| public static String relativizePath(Path path, Path startPath) { |
| URI relUri = startPath.toUri().relativize(path.toUri()); |
| if (relUri.isAbsolute() || relUri.getPath().startsWith("/")) { |
| throw new RuntimeException("FileSystem returned an unexpected path " + |
| path + " for a file within " + startPath); |
| } |
| return relUri.getPath(); |
| } |
| |
| /** |
| * Util method to check if the given file status relative to its parent is contained |
| * in a ignored directory. This is useful to ignore the files which seemingly are valid |
| * just by themselves but should still be ignored if they are contained in a |
| * directory which needs to be ignored |
| * |
| * @return true if the fileStatus should be ignored, false otherwise |
| */ |
| @VisibleForTesting |
| static boolean isInIgnoredDirectory(Path parent, FileStatus fileStatus) { |
| Preconditions.checkNotNull(fileStatus); |
| Path currentPath = fileStatus.isDirectory() ? fileStatus.getPath() : |
| fileStatus.getPath().getParent(); |
| while (currentPath != null && !currentPath.equals(parent)) { |
| if (isIgnoredDir(currentPath)) { |
| LOG.debug("Ignoring {} since it is either in a hidden directory or a temporary " |
| + "staging directory {}", fileStatus.getPath(), currentPath); |
| return true; |
| } |
| currentPath = currentPath.getParent(); |
| } |
| return false; |
| } |
| |
| /** |
| * Prefix string used by hive to write certain temporary or "non-data" files in the |
| * table location |
| */ |
| public static final String HIVE_TEMP_FILE_PREFIX = "_tmp."; |
| |
| public static final String DOT = "."; |
| |
| /** |
| * Util method used to filter out hidden and temporary staging directories |
| * which tools like Hive create in the table/partition directories when a query is |
| * inserting data into them. |
| */ |
| @VisibleForTesting |
| static boolean isIgnoredDir(Path path) { |
| String filename = path.getName(); |
| return filename.startsWith(DOT) || filename.startsWith(HIVE_TEMP_FILE_PREFIX); |
| } |
| |
| /** |
| * A remote iterator which takes in another Remote Iterator and a start path and filters |
| * all the ignored directories |
| * (See {@link org.apache.impala.common.FileSystemUtil#isInIgnoredDirectory}) from the |
| * listing of the remote iterator |
| */ |
| static class FilterIterator implements RemoteIterator<FileStatus> { |
| private final RemoteIterator<? extends FileStatus> baseIterator_; |
| private FileStatus curFile_ = null; |
| private final Path startPath_; |
| |
| FilterIterator(Path startPath, RemoteIterator<? extends FileStatus> baseIterator) { |
| startPath_ = Preconditions.checkNotNull(startPath); |
| baseIterator_ = Preconditions.checkNotNull(baseIterator); |
| } |
| |
| @Override |
| public boolean hasNext() throws IOException { |
| // Pull the next file to be returned into 'curFile'. If we've already got one, |
| // we don't need to do anything (extra calls to hasNext() must not affect |
| // state) |
| while (curFile_ == null) { |
| FileStatus next; |
| try { |
| if (!baseIterator_.hasNext()) return false; |
| // if the next fileStatus is in ignored directory skip it |
| next = baseIterator_.next(); |
| } catch (FileNotFoundException ex) { |
| // in case of concurrent operations by multiple engines it is possible that |
| // some temporary files are deleted while Impala is loading the table. For |
| // instance, hive deletes the temporary files in the .hive-staging directory |
| // after an insert query from Hive completes. If we are loading the table at |
| // the same time, we may get a FileNotFoundException which is safe to ignore. |
| LOG.warn(ex.getMessage()); |
| continue; |
| } |
| if (!isInIgnoredDirectory(startPath_, next)) { |
| curFile_ = next; |
| return true; |
| } |
| } |
| return true; |
| } |
| |
| @Override |
| public FileStatus next() throws IOException { |
| if (hasNext()) { |
| FileStatus next = curFile_; |
| curFile_ = null; |
| return next; |
| } |
| throw new NoSuchElementException("No more entries"); |
| } |
| } |
| |
| /** |
| * Iterator which recursively visits directories on a FileSystem, yielding |
| * files in an unspecified order. |
| */ |
| private static class RecursingIterator implements RemoteIterator<FileStatus> { |
| private final FileSystem fs_; |
| private final Stack<RemoteIterator<FileStatus>> iters_ = new Stack<>(); |
| private RemoteIterator<FileStatus> curIter_; |
| private FileStatus curFile_; |
| |
| private RecursingIterator(FileSystem fs, Path startPath) throws IOException { |
| this.fs_ = Preconditions.checkNotNull(fs); |
| curIter_ = fs.listStatusIterator(Preconditions.checkNotNull(startPath)); |
| } |
| |
| @Override |
| public boolean hasNext() throws IOException { |
| // Pull the next file to be returned into 'curFile'. If we've already got one, |
| // we don't need to do anything (extra calls to hasNext() must not affect |
| // state) |
| while (curFile_ == null) { |
| if (curIter_.hasNext()) { |
| // Consume the next file or directory from the current iterator. |
| handleFileStat(curIter_.next()); |
| } else if (!iters_.empty()) { |
| // We ran out of entries in the current one, but we might still have |
| // entries at a higher level of recursion. |
| curIter_ = iters_.pop(); |
| } else { |
| // No iterators left to process, so we are entirely done. |
| return false; |
| } |
| } |
| return true; |
| } |
| |
| /** |
| * Process the input stat. |
| * If it is a file, return the file stat. |
| * If it is a directory, traverse the directory if recursive is true; |
| * ignore it if recursive is false. |
| * @param fileStatus input status |
| * @throws IOException if any IO error occurs |
| */ |
| private void handleFileStat(FileStatus fileStatus) throws IOException { |
| if (fileStatus.isFile()) { |
| curFile_ = fileStatus; |
| return; |
| } |
| iters_.push(curIter_); |
| curIter_ = fs_.listStatusIterator(fileStatus.getPath()); |
| curFile_ = fileStatus; |
| } |
| |
| @Override |
| public FileStatus next() throws IOException { |
| if (hasNext()) { |
| FileStatus result = curFile_; |
| // Reset back to 'null' so that hasNext() will pull a new entry on the next |
| // call. |
| curFile_ = null; |
| return result; |
| } |
| throw new NoSuchElementException("No more entries"); |
| } |
| } |
| } |
