10334

Build # 10334

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/9fbcf60589cb611c1c903830a12ab3b5182950c0">9fbcf6058<a href="https://github.com/apache/carbondata/commit/9fbcf60589cb611c1c903830a12ab3b5182950c0">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/3aec67c5f525e90604a30c9db504b35896d0ed3c">3aec67c5f</a>">3aec67c5f</a><a href="https://github.com/apache/carbondata/commit/9fbcf60589cb611c1c903830a12ab3b5182950c0"> into e193df0a1">e193df0a1</a>

Run Details

61652 of 77972 relevant lines covered (79.07%)

1.05 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

62.11

/core/src/main/java/org/apache/carbondata/core/datastore/impl/FileFactory.java

/*
 * 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.carbondata.core.datastore.impl;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.lang.reflect.Method;
import java.nio.channels.FileChannel;

import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.constants.CarbonCommonConstants;
import org.apache.carbondata.core.datastore.FileReader;
import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
import org.apache.carbondata.core.util.ThreadLocalSessionInfo;

import org.apache.commons.io.FileUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.log4j.Logger;

public final class FileFactory {
  /**
   * LOGGER
   */
  private static final Logger LOGGER =
      LogServiceFactory.getLogService(FileFactory.class.getName());
  private static Configuration configuration = null;

  static {
    configuration = new Configuration();
    configuration.addResource(new Path("../core-default.xml"));
  }

  private static FileTypeInterface fileFileTypeInterface = new DefaultFileTypeProvider();
  public static void setFileTypeInterface(FileTypeInterface fileTypeInterface) {
    fileFileTypeInterface = fileTypeInterface;
  }
  private FileFactory() {

  }

  public static Configuration getConfiguration() {
    Configuration conf;
    Object confObject = ThreadLocalSessionInfo.getOrCreateCarbonSessionInfo()
        .getNonSerializableExtraInfo().get("carbonConf");
    if (confObject == null) {
      conf = configuration;
    } else {
      conf = (Configuration) confObject;
    }
    return conf;
  }

  public static FileReader getFileHolder(FileType fileType) {
    return fileFileTypeInterface.getFileHolder(fileType, getConfiguration());
  }

  public static FileReader getFileHolder(FileType fileType, Configuration configuration) {
    return fileFileTypeInterface.getFileHolder(fileType, configuration);
  }

  public static FileType getFileType(String path) {
    String lowerPath = path.toLowerCase();
    if (lowerPath.startsWith(CarbonCommonConstants.HDFSURL_PREFIX)) {
      return FileType.HDFS;
    } else if (lowerPath.startsWith(CarbonCommonConstants.ALLUXIOURL_PREFIX)) {
      return FileType.ALLUXIO;
    } else if (lowerPath.startsWith(CarbonCommonConstants.VIEWFSURL_PREFIX)) {
      return FileType.VIEWFS;
    } else if (lowerPath.startsWith(CarbonCommonConstants.S3N_PREFIX) ||
        lowerPath.startsWith(CarbonCommonConstants.S3A_PREFIX) ||
        lowerPath.startsWith(CarbonCommonConstants.S3_PREFIX)) {
      return FileType.S3;
    }
    return FileType.LOCAL;
  }

  public static CarbonFile getCarbonFile(String path) {
    return fileFileTypeInterface.getCarbonFile(path, getFileType(path));
  }
  public static CarbonFile getCarbonFile(String path, FileType fileType) {
    return fileFileTypeInterface.getCarbonFile(path, fileType);
  }
  public static CarbonFile getCarbonFile(String path,
      Configuration hadoopConf) {
    return fileFileTypeInterface.getCarbonFile(path, getFileType(path), hadoopConf);
  }

  public static DataInputStream getDataInputStream(String path, FileType fileType)
      throws IOException {
    return getDataInputStream(path, fileType, -1);
  }

  public static DataInputStream getDataInputStream(String path, FileType fileType,
      Configuration configuration) throws IOException {
    return getDataInputStream(path, fileType, -1, configuration);
  }

  public static DataInputStream getDataInputStream(String path, FileType fileType, int bufferSize)
      throws IOException {
    return getDataInputStream(path, fileType, bufferSize, getConfiguration());
  }
  public static DataInputStream getDataInputStream(String path, FileType fileType, int bufferSize,
      Configuration configuration) throws IOException {
    return getCarbonFile(path).getDataInputStream(path, fileType, bufferSize, configuration);
  }

  /**
   * get data input stream
   * @param path
   * @param fileType
   * @param bufferSize
   * @param compressorName name of compressor to read this file
   * @return data input stream
   * @throws IOException
   */
  public static DataInputStream getDataInputStream(String path, FileType fileType, int bufferSize,
      String compressorName) throws IOException {
    return getCarbonFile(path).getDataInputStream(path, fileType, bufferSize, compressorName);
  }
  /**
   * return the datainputStream which is seek to the offset of file
   *
   * @param path
   * @param fileType
   * @param bufferSize
   * @param offset
   * @return DataInputStream
   * @throws IOException
   */
  public static DataInputStream getDataInputStream(String path, FileType fileType, int bufferSize,
      long offset) throws IOException {
    return getCarbonFile(path).getDataInputStream(path, fileType, bufferSize, offset);
  }

  public static DataOutputStream getDataOutputStream(String path, FileType fileType)
      throws IOException {
    return getCarbonFile(path).getDataOutputStream(path, fileType);
  }

  public static DataOutputStream getDataOutputStream(String path, FileType fileType, int bufferSize,
      boolean append) throws IOException {
    return getCarbonFile(path).getDataOutputStream(path, fileType, bufferSize, append);
  }

  public static DataOutputStream getDataOutputStream(String path, FileType fileType, int bufferSize,
      long blockSize) throws IOException {
    return getCarbonFile(path).getDataOutputStream(path, fileType, bufferSize, blockSize);
  }

  /**
   * get data output stream
   * @param path file path
   * @param fileType file type
   * @param bufferSize write buffer size
   * @param blockSize block size
   * @param replication replication
   * @return data output stream
   * @throws IOException if error occurs
   */
  public static DataOutputStream getDataOutputStream(String path, FileType fileType, int bufferSize,
      long blockSize, short replication) throws IOException {
    return getCarbonFile(path).getDataOutputStream(path, fileType, bufferSize, blockSize,
        replication);
  }
  /**
   * get data out put stream
   * @param path
   * @param fileType
   * @param bufferSize
   * @param compressorName name of compressor to write this file
   * @return data out put stram
   * @throws IOException
   */
  public static DataOutputStream getDataOutputStream(String path, FileType fileType, int bufferSize,
      String compressorName) throws IOException {
    return getCarbonFile(path).getDataOutputStream(path, fileType, bufferSize, compressorName);
  }
  /**
   * This method checks the given path exists or not and also is it file or
   * not if the performFileCheck is true
   *
   * @param filePath         - Path
   * @param fileType         - FileType Local/HDFS
   * @param performFileCheck - Provide false for folders, true for files and
   */
  public static boolean isFileExist(String filePath, FileType fileType, boolean performFileCheck)
      throws IOException {
    return getCarbonFile(filePath).isFileExist(filePath, fileType, performFileCheck);
  }

  /**
   * This method checks the given path exists or not.
   *
   * @param filePath - Path
   * @param fileType - FileType Local/HDFS
   */
  public static boolean isFileExist(String filePath, FileType fileType) throws IOException {
    return getCarbonFile(filePath).isFileExist(filePath, fileType);
  }

  /**
   * This method checks the given path exists or not.
   *
   * @param filePath - Path
   */
  public static boolean isFileExist(String filePath) throws IOException {
    return isFileExist(filePath, getFileType(filePath));
  }

  public static boolean createNewFile(String filePath, FileType fileType) throws IOException {
    return createNewFile(filePath, fileType, true, null);
  }
  public static boolean createNewFile(
      String filePath,
      FileType fileType,
      boolean doAs,
      final FsPermission permission) throws IOException {
    return getCarbonFile(filePath).createNewFile(filePath, fileType, doAs, permission);
  }
  public static boolean deleteFile(String filePath, FileType fileType) throws IOException {
    return getCarbonFile(filePath).deleteFile(filePath, fileType);
  }

  public static boolean deleteAllFilesOfDir(File path) {
    if (!path.exists()) {
      return true;
    }
    if (path.isFile()) {
      return path.delete();
    }
    File[] files = path.listFiles();
    if (null == files) {
      return true;
    }
    for (int i = 0; i < files.length; i++) {
      deleteAllFilesOfDir(files[i]);
    }
    return path.delete();
  }

  public static boolean deleteAllCarbonFilesOfDir(CarbonFile path) {
    if (!path.exists()) {
      return true;
    }
    if (!path.isDirectory()) {
      return path.delete();
    }
    CarbonFile[] files = path.listFiles();
    for (int i = 0; i < files.length; i++) {
      deleteAllCarbonFilesOfDir(files[i]);
    }
    return path.delete();
  }

  public static boolean mkdirs(String filePath, FileType fileType) throws IOException {
    return getCarbonFile(filePath).mkdirs(filePath);
  }

  public static boolean mkdirs(String filePath, Configuration configuration) throws IOException {
    return getCarbonFile(filePath, configuration).mkdirs(filePath);
  }

  /**
   * for getting the dataoutput stream using the hdfs filesystem append API.
   *
   * @param path
   * @param fileType
   * @return
   * @throws IOException
   */
  public static DataOutputStream getDataOutputStreamUsingAppend(String path, FileType fileType)
      throws IOException {
    return getCarbonFile(path).getDataOutputStreamUsingAppend(path, fileType);
  }

  /**
   * this method will truncate the file to the new size.
   * @param path
   * @param fileType
   * @param newSize
   * @throws IOException
   */
  public static void truncateFile(String path, FileType fileType, long newSize) throws IOException {
    path = path.replace("\\", "/");
    FileChannel fileChannel = null;
    switch (fileType) {
      case LOCAL:
        path = getUpdatedFilePath(path, fileType);
        fileChannel = new FileOutputStream(path, true).getChannel();
        try {
          fileChannel.truncate(newSize);
        } finally {
          if (fileChannel != null) {
            fileChannel.close();
          }
        }
        return;
      case HDFS:
      case ALLUXIO:
      case VIEWFS:
      case S3:
        // if hadoop version >= 2.7, it can call method 'FileSystem.truncate' to truncate file,
        // this method was new in hadoop 2.7, otherwise use CarbonFile.truncate to do this.
        try {
          Path pt = new Path(path);
          FileSystem fs = pt.getFileSystem(getConfiguration());
          Method truncateMethod = fs.getClass().getDeclaredMethod("truncate",
              new Class[]{Path.class, long.class});
          truncateMethod.invoke(fs, new Object[]{pt, newSize});
        } catch (NoSuchMethodException e) {
          LOGGER.error("the version of hadoop is below 2.7, there is no 'truncate'"
              + " method in FileSystem, It needs to use 'CarbonFile.truncate'.");
          CarbonFile carbonFile = FileFactory.getCarbonFile(path, fileType);
          carbonFile.truncate(path, newSize);
        } catch (Exception e) {
          LOGGER.error("Other exception occurred while truncating the file " + e.getMessage());
        }
        return;
      default:
        fileChannel = new FileOutputStream(path, true).getChannel();
        try {
          fileChannel.truncate(newSize);
        } finally {
          if (fileChannel != null) {
            fileChannel.close();
          }
        }
        return;
    }
  }

  /**
   * for creating a new Lock file and if it is successfully created
   * then in case of abrupt shutdown then the stream to that file will be closed.
   *
   * @param filePath
   * @param fileType
   * @return
   * @throws IOException
   */
  public static boolean createNewLockFile(String filePath, FileType fileType) throws IOException {
    return getCarbonFile(filePath).createNewLockFile(filePath, fileType);
  }

  public enum FileType {
    LOCAL, HDFS, ALLUXIO, VIEWFS, S3
  }

  /**
   * Adds the schema to file path if not exists to the file path.
   * @param filePath path of file
   * @return Updated filepath
   */
  public static String addSchemeIfNotExists(String filePath) {
    FileType fileType = getFileType(filePath);
    switch (fileType) {
      case LOCAL:
        if (filePath.startsWith("file:")) {
          return filePath;
        } else {
          return new Path("file://" + filePath).toString();
        }
      case HDFS:
      case ALLUXIO:
      case VIEWFS:
      case S3:
      default:
        return filePath;
    }
  }

  /**
   * below method will be used to update the file path
   * for local type
   * it removes the file:/ from the path
   *
   * @param filePath
   * @param fileType
   * @return updated file path without url for local
   */
  public static String getUpdatedFilePath(String filePath, FileType fileType) {
    switch (fileType) {
      case HDFS:
      case VIEWFS:
      case S3:
        return filePath;
      case ALLUXIO:
        return StringUtils.startsWith(filePath, "alluxio") ? filePath : "alluxio:///" + filePath;
      case LOCAL:
      default:
        if (filePath != null && !filePath.isEmpty()) {
          // If the store path is relative then convert to absolute path.
          if (filePath.startsWith("./")) {
            try {
              return new File(filePath).getCanonicalPath();
            } catch (IOException e) {
              throw new RuntimeException(e);
            }
          } else {
            Path pathWithoutSchemeAndAuthority =
                Path.getPathWithoutSchemeAndAuthority(new Path(filePath));
            return pathWithoutSchemeAndAuthority.toString();
          }
        } else {
          return filePath;
        }
    }
  }

  /**
   * below method will be used to update the file path
   * for local type
   * it removes the file:/ from the path
   *
   * @param filePath
   * @return updated file path without url for local
   */
  public static String getUpdatedFilePath(String filePath) {
    FileType fileType = getFileType(filePath);
    return getUpdatedFilePath(filePath, fileType);
  }

  /**
   * It computes size of directory
   *
   * @param filePath
   * @return size in bytes
   * @throws IOException
   */
  public static long getDirectorySize(String filePath) throws IOException {
    FileType fileType = getFileType(filePath);
    switch (fileType) {
      case HDFS:
      case ALLUXIO:
      case VIEWFS:
      case S3:
        Path path = new Path(filePath);
        FileSystem fs = path.getFileSystem(getConfiguration());
        return fs.getContentSummary(path).getLength();
      case LOCAL:
      default:
        filePath = getUpdatedFilePath(filePath, fileType);
        File file = new File(filePath);
        return FileUtils.sizeOfDirectory(file);
    }
  }

  /**
   * This method will create the path object for a given file
   *
   * @param filePath
   * @return
   */
  public static Path getPath(String filePath) {
    return new Path(filePath);
  }

  /**
   * This method will return the filesystem instance
   *
   * @param path
   * @return
   * @throws IOException
   */
  public static FileSystem getFileSystem(Path path) throws IOException {
    return path.getFileSystem(getConfiguration());
  }


  public static void createDirectoryAndSetPermission(String directoryPath, FsPermission permission)
      throws IOException {
    FileFactory.FileType fileType = FileFactory.getFileType(directoryPath);
    switch (fileType) {
      case S3:
      case HDFS:
      case ALLUXIO:
      case VIEWFS:
        try {
          Path path = new Path(directoryPath);
          FileSystem fs = path.getFileSystem(getConfiguration());
          if (!fs.exists(path)) {
            fs.mkdirs(path);
            fs.setPermission(path, permission);
          }
        } catch (IOException e) {
          LOGGER.error("Exception occurred : " + e.getMessage());
          throw e;
        }
        return;
      case LOCAL:
      default:
        directoryPath = FileFactory.getUpdatedFilePath(directoryPath, fileType);
        File file = new File(directoryPath);
        if (!file.mkdirs()) {
          LOGGER.error(" Failed to create directory path " + directoryPath);
        }

    }
  }

  /**
   * Check and append the hadoop's defaultFS to the path
   */
  public static String checkAndAppendDefaultFs(String path, Configuration conf) {
    String defaultFs = conf.get(CarbonCommonConstants.FS_DEFAULT_FS);
    String lowerPath = path.toLowerCase();
    if (lowerPath.startsWith(CarbonCommonConstants.HDFSURL_PREFIX) || lowerPath
        .startsWith(CarbonCommonConstants.ALLUXIOURL_PREFIX) || lowerPath
        .startsWith(CarbonCommonConstants.VIEWFSURL_PREFIX) || lowerPath
        .startsWith(CarbonCommonConstants.S3N_PREFIX) || lowerPath
        .startsWith(CarbonCommonConstants.S3A_PREFIX) || lowerPath
        .startsWith(CarbonCommonConstants.S3_PREFIX)) {
      return path;
    } else if (defaultFs != null) {
      return defaultFs + CarbonCommonConstants.FILE_SEPARATOR + path;
    } else {
      return path;
    }
  }

  /**
   * set the file replication
   *
   * @param path file path
   * @param fileType file type
   * @param replication replication
   * @return true, if success; false, if failed
   * @throws IOException if error occurs
   */
  public static boolean setReplication(String path, FileFactory.FileType fileType,
      short replication) throws IOException {
    return getCarbonFile(path, fileType).setReplication(path, replication);
  }

  /**
   * get the default replication
   *
   * @param path file path
   * @param fileType file type
   * @return replication
   * @throws IOException if error occurs
   */
  public static short getDefaultReplication(String path, FileFactory.FileType fileType)
      throws IOException {
    return getCarbonFile(path, fileType).getDefaultReplication(path);
  }
}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one or more
3	* contributor license agreements. See the NOTICE file distributed with
4	* this work for additional information regarding copyright ownership.
5	* The ASF licenses this file to You under the Apache License, Version 2.0
6	* (the "License"); you may not use this file except in compliance with
7	* the License. You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*/
17
18	package org.apache.carbondata.core.datastore.impl;
19
20	import java.io.DataInputStream;
21	import java.io.DataOutputStream;
22	import java.io.File;
23	import java.io.FileOutputStream;
24	import java.io.IOException;
25	import java.lang.reflect.Method;
26	import java.nio.channels.FileChannel;
27
28	import org.apache.carbondata.common.logging.LogServiceFactory;
29	import org.apache.carbondata.core.constants.CarbonCommonConstants;
30	import org.apache.carbondata.core.datastore.FileReader;
31	import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
32	import org.apache.carbondata.core.util.ThreadLocalSessionInfo;
33
34	import org.apache.commons.io.FileUtils;
35	import org.apache.commons.lang.StringUtils;
36	import org.apache.hadoop.conf.Configuration;
37	import org.apache.hadoop.fs.FileSystem;
38	import org.apache.hadoop.fs.Path;
39	import org.apache.hadoop.fs.permission.FsPermission;
40	import org.apache.log4j.Logger;
41
42	public final class FileFactory {
43	/**
44	* LOGGER
45	*/
46	private static final Logger LOGGER =	2✔
47	LogServiceFactory.getLogService(FileFactory.class.getName());	2✔
48	private static Configuration configuration = null;	2✔
49
50	static {
51	configuration = new Configuration();	2✔
52	configuration.addResource(new Path("../core-default.xml"));	2✔
53	}
54
55	private static FileTypeInterface fileFileTypeInterface = new DefaultFileTypeProvider();	2✔
56	public static void setFileTypeInterface(FileTypeInterface fileTypeInterface) {
57	fileFileTypeInterface = fileTypeInterface;	×
58	}	×
59	private FileFactory() {	×
60
61	}	×
62
63	public static Configuration getConfiguration() {
64	Configuration conf;
65	Object confObject = ThreadLocalSessionInfo.getOrCreateCarbonSessionInfo()	2✔
66	.getNonSerializableExtraInfo().get("carbonConf");	2✔
67	if (confObject == null) {	2✔
68	conf = configuration;	2✔
69	} else {
70	conf = (Configuration) confObject;	1✔
71	}
72	return conf;	2✔
73	}
74
75	public static FileReader getFileHolder(FileType fileType) {
76	return fileFileTypeInterface.getFileHolder(fileType, getConfiguration());	2✔
77	}
78
79	public static FileReader getFileHolder(FileType fileType, Configuration configuration) {
80	return fileFileTypeInterface.getFileHolder(fileType, configuration);	1✔
81	}
82
83	public static FileType getFileType(String path) {
84	String lowerPath = path.toLowerCase();	2✔
85	if (lowerPath.startsWith(CarbonCommonConstants.HDFSURL_PREFIX)) {	2✔
86	return FileType.HDFS;	×
87	} else if (lowerPath.startsWith(CarbonCommonConstants.ALLUXIOURL_PREFIX)) {	2✔
88	return FileType.ALLUXIO;	×
89	} else if (lowerPath.startsWith(CarbonCommonConstants.VIEWFSURL_PREFIX)) {	2✔
90	return FileType.VIEWFS;	×
91	} else if (lowerPath.startsWith(CarbonCommonConstants.S3N_PREFIX) \|\|	2✔
92	lowerPath.startsWith(CarbonCommonConstants.S3A_PREFIX) \|\|	2✔
93	lowerPath.startsWith(CarbonCommonConstants.S3_PREFIX)) {	2✔
94	return FileType.S3;	×
95	}
96	return FileType.LOCAL;	2✔
97	}
98
99	public static CarbonFile getCarbonFile(String path) {
100	return fileFileTypeInterface.getCarbonFile(path, getFileType(path));	2✔
101	}
102	public static CarbonFile getCarbonFile(String path, FileType fileType) {
103	return fileFileTypeInterface.getCarbonFile(path, fileType);	2✔
104	}
105	public static CarbonFile getCarbonFile(String path,
106	Configuration hadoopConf) {
107	return fileFileTypeInterface.getCarbonFile(path, getFileType(path), hadoopConf);	1✔
108	}
109
110	public static DataInputStream getDataInputStream(String path, FileType fileType)
111	throws IOException {
112	return getDataInputStream(path, fileType, -1);	2✔
113	}
114
115	public static DataInputStream getDataInputStream(String path, FileType fileType,
116	Configuration configuration) throws IOException {
117	return getDataInputStream(path, fileType, -1, configuration);	2✔
118	}
119
120	public static DataInputStream getDataInputStream(String path, FileType fileType, int bufferSize)
121	throws IOException {
122	return getDataInputStream(path, fileType, bufferSize, getConfiguration());	2✔
123	}
124	public static DataInputStream getDataInputStream(String path, FileType fileType, int bufferSize,
125	Configuration configuration) throws IOException {
126	return getCarbonFile(path).getDataInputStream(path, fileType, bufferSize, configuration);	2✔
127	}
128
129	/**
130	* get data input stream
131	* @param path
132	* @param fileType
133	* @param bufferSize
134	* @param compressorName name of compressor to read this file
135	* @return data input stream
136	* @throws IOException
137	*/
138	public static DataInputStream getDataInputStream(String path, FileType fileType, int bufferSize,
139	String compressorName) throws IOException {
140	return getCarbonFile(path).getDataInputStream(path, fileType, bufferSize, compressorName);	1✔
141	}
142	/**
143	* return the datainputStream which is seek to the offset of file
144	*
145	* @param path
146	* @param fileType
147	* @param bufferSize
148	* @param offset
149	* @return DataInputStream
150	* @throws IOException
151	*/
152	public static DataInputStream getDataInputStream(String path, FileType fileType, int bufferSize,
153	long offset) throws IOException {
154	return getCarbonFile(path).getDataInputStream(path, fileType, bufferSize, offset);	×
155	}
156
157	public static DataOutputStream getDataOutputStream(String path, FileType fileType)
158	throws IOException {
159	return getCarbonFile(path).getDataOutputStream(path, fileType);	2✔
160	}
161
162	public static DataOutputStream getDataOutputStream(String path, FileType fileType, int bufferSize,
163	boolean append) throws IOException {
164	return getCarbonFile(path).getDataOutputStream(path, fileType, bufferSize, append);	2✔
165	}
166
167	public static DataOutputStream getDataOutputStream(String path, FileType fileType, int bufferSize,
168	long blockSize) throws IOException {
169	return getCarbonFile(path).getDataOutputStream(path, fileType, bufferSize, blockSize);	1✔
170	}
171
172	/**
173	* get data output stream
174	* @param path file path
175	* @param fileType file type
176	* @param bufferSize write buffer size
177	* @param blockSize block size
178	* @param replication replication
179	* @return data output stream
180	* @throws IOException if error occurs
181	*/
182	public static DataOutputStream getDataOutputStream(String path, FileType fileType, int bufferSize,
183	long blockSize, short replication) throws IOException {
184	return getCarbonFile(path).getDataOutputStream(path, fileType, bufferSize, blockSize,	×
185	replication);
186	}
187	/**
188	* get data out put stream
189	* @param path
190	* @param fileType
191	* @param bufferSize
192	* @param compressorName name of compressor to write this file
193	* @return data out put stram
194	* @throws IOException
195	*/
196	public static DataOutputStream getDataOutputStream(String path, FileType fileType, int bufferSize,
197	String compressorName) throws IOException {
198	return getCarbonFile(path).getDataOutputStream(path, fileType, bufferSize, compressorName);	1✔
199	}
200	/**
201	* This method checks the given path exists or not and also is it file or
202	* not if the performFileCheck is true
203	*
204	* @param filePath - Path
205	* @param fileType - FileType Local/HDFS
206	* @param performFileCheck - Provide false for folders, true for files and
207	*/
208	public static boolean isFileExist(String filePath, FileType fileType, boolean performFileCheck)
209	throws IOException {
210	return getCarbonFile(filePath).isFileExist(filePath, fileType, performFileCheck);	2✔
211	}
212
213	/**
214	* This method checks the given path exists or not.
215	*
216	* @param filePath - Path
217	* @param fileType - FileType Local/HDFS
218	*/
219	public static boolean isFileExist(String filePath, FileType fileType) throws IOException {
220	return getCarbonFile(filePath).isFileExist(filePath, fileType);	2✔
221	}
222
223	/**
224	* This method checks the given path exists or not.
225	*
226	* @param filePath - Path
227	*/
228	public static boolean isFileExist(String filePath) throws IOException {
229	return isFileExist(filePath, getFileType(filePath));	1✔
230	}
231
232	public static boolean createNewFile(String filePath, FileType fileType) throws IOException {
233	return createNewFile(filePath, fileType, true, null);	2✔
234	}
235	public static boolean createNewFile(
236	String filePath,
237	FileType fileType,
238	boolean doAs,
239	final FsPermission permission) throws IOException {
240	return getCarbonFile(filePath).createNewFile(filePath, fileType, doAs, permission);	2✔
241	}
242	public static boolean deleteFile(String filePath, FileType fileType) throws IOException {
243	return getCarbonFile(filePath).deleteFile(filePath, fileType);	2✔
244	}
245
246	public static boolean deleteAllFilesOfDir(File path) {
247	if (!path.exists()) {	2✔
248	return true;	2✔
249	}
250	if (path.isFile()) {	2✔
251	return path.delete();	2✔
252	}
253	File[] files = path.listFiles();	2✔
254	if (null == files) {	2✔
255	return true;	×
256	}
257	for (int i = 0; i < files.length; i++) {	2✔
258	deleteAllFilesOfDir(files[i]);	1✔
259	}
260	return path.delete();	2✔
261	}
262
263	public static boolean deleteAllCarbonFilesOfDir(CarbonFile path) {
264	if (!path.exists()) {	1✔
265	return true;	1✔
266	}
267	if (!path.isDirectory()) {	1✔
268	return path.delete();	1✔
269	}
270	CarbonFile[] files = path.listFiles();	1✔
271	for (int i = 0; i < files.length; i++) {	1✔
272	deleteAllCarbonFilesOfDir(files[i]);	1✔
273	}
274	return path.delete();	1✔
275	}
276
277	public static boolean mkdirs(String filePath, FileType fileType) throws IOException {
278	return getCarbonFile(filePath).mkdirs(filePath);	2✔
279	}
280
281	public static boolean mkdirs(String filePath, Configuration configuration) throws IOException {
282	return getCarbonFile(filePath, configuration).mkdirs(filePath);	1✔
283	}
284
285	/**
286	* for getting the dataoutput stream using the hdfs filesystem append API.
287	*
288	* @param path
289	* @param fileType
290	* @return
291	* @throws IOException
292	*/
293	public static DataOutputStream getDataOutputStreamUsingAppend(String path, FileType fileType)
294	throws IOException {
295	return getCarbonFile(path).getDataOutputStreamUsingAppend(path, fileType);	2✔
296	}
297
298	/**
299	* this method will truncate the file to the new size.
300	* @param path
301	* @param fileType
302	* @param newSize
303	* @throws IOException
304	*/
305	public static void truncateFile(String path, FileType fileType, long newSize) throws IOException {
306	path = path.replace("\\", "/");	2✔
307	FileChannel fileChannel = null;	2✔
308	switch (fileType) {	2✔
309	case LOCAL:
310	path = getUpdatedFilePath(path, fileType);	2✔
311	fileChannel = new FileOutputStream(path, true).getChannel();	2✔
312	try {
313	fileChannel.truncate(newSize);	2✔
314	} finally {
315	if (fileChannel != null) {	2✔
316	fileChannel.close();	2✔
317	}
318	}
319	return;	2✔
320	case HDFS:
321	case ALLUXIO:
322	case VIEWFS:
323	case S3:
324	// if hadoop version >= 2.7, it can call method 'FileSystem.truncate' to truncate file,
325	// this method was new in hadoop 2.7, otherwise use CarbonFile.truncate to do this.
326	try {
327	Path pt = new Path(path);	×
328	FileSystem fs = pt.getFileSystem(getConfiguration());	×
329	Method truncateMethod = fs.getClass().getDeclaredMethod("truncate",	×
330	new Class[]{Path.class, long.class});
331	truncateMethod.invoke(fs, new Object[]{pt, newSize});	×
332	} catch (NoSuchMethodException e) {	×
333	LOGGER.error("the version of hadoop is below 2.7, there is no 'truncate'"	×
334	+ " method in FileSystem, It needs to use 'CarbonFile.truncate'.");
335	CarbonFile carbonFile = FileFactory.getCarbonFile(path, fileType);	×
336	carbonFile.truncate(path, newSize);	×
337	} catch (Exception e) {	×
338	LOGGER.error("Other exception occurred while truncating the file " + e.getMessage());	×
339	}	×
340	return;	×
341	default:
342	fileChannel = new FileOutputStream(path, true).getChannel();	×
343	try {
344	fileChannel.truncate(newSize);	×
345	} finally {
346	if (fileChannel != null) {	×
347	fileChannel.close();	×
348	}
349	}
350	return;	×
351	}
352	}
353
354	/**
355	* for creating a new Lock file and if it is successfully created
356	* then in case of abrupt shutdown then the stream to that file will be closed.
357	*
358	* @param filePath
359	* @param fileType
360	* @return
361	* @throws IOException
362	*/
363	public static boolean createNewLockFile(String filePath, FileType fileType) throws IOException {
364	return getCarbonFile(filePath).createNewLockFile(filePath, fileType);	2✔
365	}
366
367	public enum FileType {	2✔
368	LOCAL, HDFS, ALLUXIO, VIEWFS, S3	2✔
369	}
370
371	/**
372	* Adds the schema to file path if not exists to the file path.
373	* @param filePath path of file
374	* @return Updated filepath
375	*/
376	public static String addSchemeIfNotExists(String filePath) {
377	FileType fileType = getFileType(filePath);	1✔
378	switch (fileType) {	1✔
379	case LOCAL:
380	if (filePath.startsWith("file:")) {	1✔
381	return filePath;	×
382	} else {
383	return new Path("file://" + filePath).toString();	1✔
384	}
385	case HDFS:
386	case ALLUXIO:
387	case VIEWFS:
388	case S3:
389	default:
390	return filePath;	×
391	}
392	}
393
394	/**
395	* below method will be used to update the file path
396	* for local type
397	* it removes the file:/ from the path
398	*
399	* @param filePath
400	* @param fileType
401	* @return updated file path without url for local
402	*/
403	public static String getUpdatedFilePath(String filePath, FileType fileType) {
404	switch (fileType) {	2✔
405	case HDFS:
406	case VIEWFS:
407	case S3:
408	return filePath;	2✔
409	case ALLUXIO:
410	return StringUtils.startsWith(filePath, "alluxio") ? filePath : "alluxio:///" + filePath;	×
411	case LOCAL:
412	default:
413	if (filePath != null && !filePath.isEmpty()) {	2✔
414	// If the store path is relative then convert to absolute path.
415	if (filePath.startsWith("./")) {	2✔
416	try {
417	return new File(filePath).getCanonicalPath();	1✔
418	} catch (IOException e) {	×
419	throw new RuntimeException(e);	×
420	}
421	} else {
422	Path pathWithoutSchemeAndAuthority =	2✔
423	Path.getPathWithoutSchemeAndAuthority(new Path(filePath));	2✔
424	return pathWithoutSchemeAndAuthority.toString();	2✔
425	}
426	} else {
427	return filePath;	×
428	}
429	}
430	}
431
432	/**
433	* below method will be used to update the file path
434	* for local type
435	* it removes the file:/ from the path
436	*
437	* @param filePath
438	* @return updated file path without url for local
439	*/
440	public static String getUpdatedFilePath(String filePath) {
441	FileType fileType = getFileType(filePath);	2✔
442	return getUpdatedFilePath(filePath, fileType);	2✔
443	}
444
445	/**
446	* It computes size of directory
447	*
448	* @param filePath
449	* @return size in bytes
450	* @throws IOException
451	*/
452	public static long getDirectorySize(String filePath) throws IOException {
453	FileType fileType = getFileType(filePath);	1✔
454	switch (fileType) {	1✔
455	case HDFS:
456	case ALLUXIO:
457	case VIEWFS:
458	case S3:
459	Path path = new Path(filePath);	×
460	FileSystem fs = path.getFileSystem(getConfiguration());	×
461	return fs.getContentSummary(path).getLength();	×
462	case LOCAL:
463	default:
464	filePath = getUpdatedFilePath(filePath, fileType);	1✔
465	File file = new File(filePath);	1✔
466	return FileUtils.sizeOfDirectory(file);	1✔
467	}
468	}
469
470	/**
471	* This method will create the path object for a given file
472	*
473	* @param filePath
474	* @return
475	*/
476	public static Path getPath(String filePath) {
477	return new Path(filePath);	1✔
478	}
479
480	/**
481	* This method will return the filesystem instance
482	*
483	* @param path
484	* @return
485	* @throws IOException
486	*/
487	public static FileSystem getFileSystem(Path path) throws IOException {
488	return path.getFileSystem(getConfiguration());	1✔
489	}
490
491
492	public static void createDirectoryAndSetPermission(String directoryPath, FsPermission permission)
493	throws IOException {
494	FileFactory.FileType fileType = FileFactory.getFileType(directoryPath);	1✔
495	switch (fileType) {	1✔
496	case S3:
497	case HDFS:
498	case ALLUXIO:
499	case VIEWFS:
500	try {
501	Path path = new Path(directoryPath);	×
502	FileSystem fs = path.getFileSystem(getConfiguration());	×
503	if (!fs.exists(path)) {	×
504	fs.mkdirs(path);	×
505	fs.setPermission(path, permission);	×
506	}
507	} catch (IOException e) {	×
508	LOGGER.error("Exception occurred : " + e.getMessage());	×
509	throw e;	×
510	}	×
511	return;	×
512	case LOCAL:
513	default:
514	directoryPath = FileFactory.getUpdatedFilePath(directoryPath, fileType);	1✔
515	File file = new File(directoryPath);	1✔
516	if (!file.mkdirs()) {	1✔
517	LOGGER.error(" Failed to create directory path " + directoryPath);	1✔
518	}
519
520	}
521	}	1✔
522
523	/**
524	* Check and append the hadoop's defaultFS to the path
525	*/
526	public static String checkAndAppendDefaultFs(String path, Configuration conf) {
527	String defaultFs = conf.get(CarbonCommonConstants.FS_DEFAULT_FS);	×
528	String lowerPath = path.toLowerCase();	×
529	if (lowerPath.startsWith(CarbonCommonConstants.HDFSURL_PREFIX) \|\| lowerPath	×
530	.startsWith(CarbonCommonConstants.ALLUXIOURL_PREFIX) \|\| lowerPath	×
531	.startsWith(CarbonCommonConstants.VIEWFSURL_PREFIX) \|\| lowerPath	×
532	.startsWith(CarbonCommonConstants.S3N_PREFIX) \|\| lowerPath	×
533	.startsWith(CarbonCommonConstants.S3A_PREFIX) \|\| lowerPath	×
534	.startsWith(CarbonCommonConstants.S3_PREFIX)) {	×
535	return path;	×
536	} else if (defaultFs != null) {	×
537	return defaultFs + CarbonCommonConstants.FILE_SEPARATOR + path;	×
538	} else {
539	return path;	×
540	}
541	}
542
543	/**
544	* set the file replication
545	*
546	* @param path file path
547	* @param fileType file type
548	* @param replication replication
549	* @return true, if success; false, if failed
550	* @throws IOException if error occurs
551	*/
552	public static boolean setReplication(String path, FileFactory.FileType fileType,
553	short replication) throws IOException {
554	return getCarbonFile(path, fileType).setReplication(path, replication);	×
555	}
556
557	/**
558	* get the default replication
559	*
560	* @param path file path
561	* @param fileType file type
562	* @return replication
563	* @throws IOException if error occurs
564	*/
565	public static short getDefaultReplication(String path, FileFactory.FileType fileType)
566	throws IOException {
567	return getCarbonFile(path, fileType).getDefaultReplication(path);	×
568	}
569	}

apache / carbondata / 10334

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous