apache / carbondata / 714

Build # 714

Build Type

push

jenkins

Committed by GitHub

Commit Message

<a href="https://github.com/apache/carbondata/commit/<a class=hub.com/apache/carbondata/commit/4ffb991a9e41e0de153b23bebcea76321dc8f663">4ffb991a9<a href="https://github.com/apache/carbondata/commit/4ffb991a9e41e0de153b23bebcea76321dc8f663">">Merge </a><a class="double-link" href="https://github.com/apache/carbondata/commit/<a class="double-link" href="https://github.com/apache/carbondata/commit/6c7d565403d3eacc73f87106be815d225aa81fa2">6c7d56540</a>">6c7d56540</a><a href="https://github.com/apache/carbondata/commit/4ffb991a9e41e0de153b23bebcea76321dc8f663"> into 3cd8b947c">3cd8b947c</a>

Run Details

62 of 62 new or added lines in 6 files covered. (100.0%)

60396 of 75014 relevant lines covered (80.51%)

1.06 hits per line

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

61.54

/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.LogService;
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;

public final class FileFactory {
  /**
   * LOGGER
   */
  private static final LogService 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
  }

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

apache / carbondata / 714

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