9874

Build # 9874

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/68300ecd0729c03382240e52127a47d3854a05db">68300ecd0<a href="https://github.com/apache/carbondata/commit/68300ecd0729c03382240e52127a47d3854a05db">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/c2fc6d5964b56d4a111a13b6f9dad7ed6748934f">c2fc6d596</a>">c2fc6d596</a><a href="https://github.com/apache/carbondata/commit/68300ecd0729c03382240e52127a47d3854a05db"> into 382ce430a">382ce430a</a>

Run Details

21 of 21 new or added lines in 2 files covered. (100.0%)

61602 of 77425 relevant lines covered (79.56%)

1.05 hits per line

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

62.58

/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.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, configuration)
        .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
  }

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

apache / carbondata / 9874

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