core/src/main/java/org/apache/carbondata/core/datastore/filesystem/AbstractDFSCarbonFile.java - carbondata - Git at Google

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

 package org.apache.carbondata.core.datastore.filesystem;

 import java.io.BufferedInputStream;
 import java.io.BufferedOutputStream;
 import java.io.DataInputStream;
 import java.io.DataOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;

 import org.apache.carbondata.common.logging.LogServiceFactory;
 import org.apache.carbondata.core.constants.CarbonCommonConstants;
 import org.apache.carbondata.core.datastore.impl.FileFactory;
 import org.apache.carbondata.core.exception.CarbonFileException;
 import org.apache.carbondata.core.util.CarbonUtil;

 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.BlockLocation;
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.LocatedFileStatus;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.PathFilter;
 import org.apache.hadoop.fs.RemoteIterator;
 import org.apache.hadoop.fs.permission.FsAction;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.io.compress.BZip2Codec;
 import org.apache.hadoop.io.compress.CompressionCodec;
 import org.apache.hadoop.io.compress.CompressionCodecFactory;
 import org.apache.hadoop.io.compress.GzipCodec;
 import org.apache.hadoop.io.compress.Lz4Codec;
 import org.apache.hadoop.io.compress.SnappyCodec;
 import org.apache.log4j.Logger;

 public abstract class AbstractDFSCarbonFile implements CarbonFile {
   /**
    * LOGGER
    */
   private static final Logger LOGGER =
       LogServiceFactory.getLogService(AbstractDFSCarbonFile.class.getName());
   protected FileSystem fileSystem;
   protected Configuration hadoopConf;
   protected Path path;
   protected FileStatus fileStatus;

   AbstractDFSCarbonFile(String filePath) {
     this(filePath, FileFactory.getConfiguration());
   }

   AbstractDFSCarbonFile(String filePath, Configuration hadoopConf) {
     this(new Path(filePath), hadoopConf);
   }

   AbstractDFSCarbonFile(Path path) {
     this(path, FileFactory.getConfiguration());
   }

   AbstractDFSCarbonFile(Path path, Configuration hadoopConf) {
     this.hadoopConf = hadoopConf;
     try {
       Path newPath = new Path(path.toString().replace("\\", "/"));
       fileSystem = newPath.getFileSystem(this.hadoopConf);
       this.path = newPath;
     } catch (IOException e) {
       throw new CarbonFileException("Error while getting File System: ", e);
     }
   }

   AbstractDFSCarbonFile(FileStatus fileStatus) {
     this(fileStatus.getPath());
     this.fileStatus = fileStatus;
   }

   @Override
   public boolean createNewFile() {
     try {
       return fileSystem.createNewFile(path);
     } catch (IOException e) {
       throw new CarbonFileException("Unable to create file: " + path.toString(), e);
     }
   }

   @Override
   public String getAbsolutePath() {
     try {
       return fileSystem.getFileStatus(path).getPath().toString();
     } catch (IOException e) {
       throw new CarbonFileException("Unable to get file status: ", e);
     }
   }

   @Override
   public String getName() {
     return path.getName();
   }

   @Override
   public boolean isDirectory() {
     try {
       return fileSystem.getFileStatus(path).isDirectory();
     } catch (IOException e) {
       throw new CarbonFileException("Unable to get file status: ", e);
     }
   }

   @Override
   public CarbonFile getParentFile() {
     Path parentPath = path.getParent();
     return parentPath == null ? null : FileFactory.getCarbonFile(parentPath.toString());
   }

   @Override
   public boolean exists() {
     try {
       return fileSystem.exists(path);
     } catch (IOException e) {
       throw new CarbonFileException(e);
     }
   }

   @Override
   public String getCanonicalPath() {
     return getAbsolutePath();
   }

   @Override
   public String getPath() {
     return getAbsolutePath();
   }

   @Override
   public long getSize() {
     try {
       return fileSystem.getFileStatus(path).getLen();
     } catch (IOException e) {
       throw new CarbonFileException("Unable to get file status for " + path.toString(), e);
     }
   }

   @Override
   public boolean renameTo(String changeToName) {
     try {
       return fileSystem.rename(path, new Path(changeToName));
     } catch (IOException e) {
       throw new CarbonFileException("Failed to rename file: ", e);
     }
   }

   public boolean delete() {
     try {
       return fileSystem.delete(path, true);
     } catch (IOException e) {
       throw new CarbonFileException("Failed to delete file:", e);
     }
   }

   @Override
   public long getLastModifiedTime() {
     try {
       if (this.fileStatus != null) {
         return this.fileStatus.getModificationTime();
       }
       return fileSystem.getFileStatus(path).getModificationTime();
     } catch (IOException e) {
       throw new CarbonFileException("Unable to get file status: ", e);
     }
   }

   @Override
   public boolean setLastModifiedTime(long timestamp) {
     try {
       fileSystem.setTimes(path, timestamp, timestamp);
       return true;
     } catch (IOException e) {
       throw new CarbonFileException("Error while setting modified time: ", e);
     }
   }

   /**
    * This method will delete the data in file data from a given offset
    */
   @Override
   public boolean truncate(String fileName, long validDataEndOffset) {
     DataOutputStream dataOutputStream = null;
     DataInputStream dataInputStream = null;
     boolean fileTruncatedSuccessfully = false;
     // if bytes to read less than 1024 then buffer size should be equal to the given offset
     int bufferSize = validDataEndOffset > CarbonCommonConstants.BYTE_TO_KB_CONVERSION_FACTOR ?
         CarbonCommonConstants.BYTE_TO_KB_CONVERSION_FACTOR :
         (int) validDataEndOffset;
     // temporary file name
     String tempWriteFilePath = fileName + CarbonCommonConstants.TEMPWRITEFILEEXTENSION;
     try {
       CarbonFile tempFile;
       // delete temporary file if it already exists at a given path
       if (FileFactory.isFileExist(tempWriteFilePath)) {
         tempFile = FileFactory.getCarbonFile(tempWriteFilePath);
         tempFile.delete();
       }
       // create new temporary file
       FileFactory.createNewFile(tempWriteFilePath);
       tempFile = FileFactory.getCarbonFile(tempWriteFilePath);
       byte[] buff = new byte[bufferSize];
       dataInputStream = FileFactory.getDataInputStream(fileName);
       // read the data
       int read = dataInputStream.read(buff, 0, buff.length);
       dataOutputStream = FileFactory.getDataOutputStream(tempWriteFilePath);
       dataOutputStream.write(buff, 0, read);
       long remaining = validDataEndOffset - read;
       // anytime we should not cross the offset to be read
       while (remaining > 0) {
         if (remaining > bufferSize) {
           buff = new byte[bufferSize];
         } else {
           buff = new byte[(int) remaining];
         }
         read = dataInputStream.read(buff, 0, buff.length);
         dataOutputStream.write(buff, 0, read);
         remaining = remaining - read;
       }
       CarbonUtil.closeStreams(dataInputStream, dataOutputStream);
       // rename the temp file to original file
       tempFile.renameForce(fileName);
       fileTruncatedSuccessfully = true;
     } catch (IOException e) {
       LOGGER.error("Exception occurred while truncating the file " + e.getMessage(), e);
     } finally {
       CarbonUtil.closeStreams(dataOutputStream, dataInputStream);
     }
     return fileTruncatedSuccessfully;
   }

   /**
    * This method will be used to check whether a file has been modified or not
    *
    * @param fileTimeStamp time to be compared with latest timestamp of file
    * @param endOffset     file length to be compared with current length of file
    * @return whether a file has been modified or not
    */
   @Override
   public boolean isFileModified(long fileTimeStamp, long endOffset) {
     boolean isFileModified = false;
     if (getLastModifiedTime() > fileTimeStamp || getSize() > endOffset) {
       isFileModified = true;
     }
     return isFileModified;
   }

   @Override
   public DataOutputStream getDataOutputStream(int bufferSize, boolean append)
       throws IOException {
     FSDataOutputStream stream = null;
     if (append) {
       // append to a file only if file already exists else file not found
       // exception will be thrown by hdfs
       if (CarbonUtil.isFileExists(path.toString())) {
         stream = fileSystem.append(path, bufferSize);
       } else {
         stream = fileSystem.create(path, true, bufferSize);
       }
     } else {
       stream = fileSystem.create(path, true, bufferSize);
     }
     return stream;
   }

   @Override
   public DataInputStream getDataInputStream(int bufferSize) throws IOException {
     return getDataInputStream(bufferSize,
         CarbonUtil.inferCompressorFromFileName(path.toString()));
   }

   /**
    * return the DataInputStream which is seek to the offset of file
    *
    * @return DataInputStream
    * @throws IOException
    */
   @Override
   public DataInputStream getDataInputStream(int bufferSize, long offset)
       throws IOException {
     FSDataInputStream stream = fileSystem.open(path, bufferSize);
     stream.seek(offset);
     return new DataInputStream(new BufferedInputStream(stream));
   }

   @Override
   public DataInputStream getDataInputStream(int bufferSize, String compressor)
       throws IOException {
     InputStream inputStream;
     if (bufferSize <= 0) {
       inputStream = fileSystem.open(path);
     } else {
       inputStream = fileSystem.open(path, bufferSize);
     }
     String codecName = getCodecNameFromCompressor(compressor);
     if (!codecName.isEmpty()) {
       CompressionCodec codec = new CompressionCodecFactory(hadoopConf).getCodecByName(codecName);
       inputStream = codec.createInputStream(inputStream);
     }
     if (bufferSize <= 0 && inputStream instanceof FSDataInputStream) {
       return (DataInputStream) inputStream;
     } else {
       return new DataInputStream(new BufferedInputStream(inputStream));
     }
   }

   /**
    * get codec name from user specified compressor name
    *
    * @param compressorName user specified compressor name
    * @return name of codec
    * @throws IOException
    */
   private String getCodecNameFromCompressor(String compressorName) throws IOException {
     if (compressorName.isEmpty()) {
       return "";
     } else if ("GZIP".equalsIgnoreCase(compressorName)) {
       return GzipCodec.class.getName();
     } else if ("BZIP2".equalsIgnoreCase(compressorName)) {
       return BZip2Codec.class.getName();
     } else if ("SNAPPY".equalsIgnoreCase(compressorName)) {
       return SnappyCodec.class.getName();
     } else if ("LZ4".equalsIgnoreCase(compressorName)) {
       return Lz4Codec.class.getName();
     } else {
       throw new IOException("Unsupported compressor: " + compressorName);
     }
   }

   @Override
   public DataOutputStream getDataOutputStream() throws IOException {
     return fileSystem.create(path, true);
   }

   @Override
   public DataOutputStream getDataOutputStream(int bufferSize, long blockSize)
       throws IOException {
     short replication = fileSystem.getDefaultReplication(path);
     return getDataOutputStream(bufferSize, blockSize, replication);
   }

   @Override
   public DataOutputStream getDataOutputStream(int bufferSize, long blockSize, short replication)
       throws IOException {
     return fileSystem.create(path, true, bufferSize, replication, blockSize);
   }

   @Override
   public DataOutputStream getDataOutputStream(int bufferSize, String compressor)
       throws IOException {
     OutputStream outputStream;
     if (bufferSize <= 0) {
       outputStream = fileSystem.create(path);
     } else {
       outputStream = fileSystem.create(path, true, bufferSize);
     }
     String codecName = getCodecNameFromCompressor(compressor);
     if (!codecName.isEmpty()) {
       CompressionCodec codec = new CompressionCodecFactory(hadoopConf).getCodecByName(codecName);
       outputStream = codec.createOutputStream(outputStream);
     }
     return new DataOutputStream(new BufferedOutputStream(outputStream));
   }

   @Override
   public boolean isFileExist(boolean performFileCheck) throws IOException {
     if (performFileCheck) {
       return fileSystem.exists(path) && fileSystem.isFile(path);
     } else {
       return fileSystem.exists(path);
     }
   }

   @Override
   public boolean isFileExist() throws IOException {
     return isFileExist(false);
   }

   @Override
   public boolean createNewFile(FsPermission permission) throws IOException {
     if (fileSystem.exists(path)) {
       return false;
     } else {
       if (permission == null) {
         permission =
             FsPermission.getFileDefault().applyUMask(FsPermission.getUMask(fileSystem.getConf()));
       }
       // Pass the permissions during file creation itself
       fileSystem
           .create(path, permission, false, fileSystem.getConf().getInt("io.file.buffer.size", 4096),
               fileSystem.getDefaultReplication(path), fileSystem.getDefaultBlockSize(path), null)
           .close();
       return true;
     }
   }

   @Override
   public boolean deleteFile() throws IOException {
     return fileSystem.delete(path, true);
   }

   @Override
   public boolean mkdirs() throws IOException {
     return fileSystem.mkdirs(path);
   }

   @Override
   public DataOutputStream getDataOutputStreamUsingAppend() throws IOException {
     return fileSystem.append(path);
   }

   @Override
   public boolean createNewLockFile() throws IOException {
     if (fileSystem.exists(path)) {
       return false;
     } else {
       // Pass the permissions during file creation itself
       fileSystem.create(path, new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL), false,
           fileSystem.getConf().getInt("io.file.buffer.size", 4096),
           fileSystem.getDefaultReplication(path), fileSystem.getDefaultBlockSize(path), null)
           .close();
       // hadoop masks the permission according to configured permission, so need to set permission
       // forcefully
       fileSystem.setPermission(path, new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL));
       return true;
     }
   }

   @Override
   public CarbonFile[] listFiles() {
     FileStatus[] listStatus;
     try {
       listStatus = fileSystem.listStatus(path);
     } catch (IOException e) {
       LOGGER.warn("Exception occurred: " + e.getMessage(), e);
       return new CarbonFile[0];
     }
     return getFiles(listStatus);
   }

   /**
    * Get the CarbonFiles from FileStatus array
    */
   protected abstract CarbonFile[] getFiles(FileStatus[] listStatus);

   @Override
   public List<CarbonFile> listFiles(Boolean recursive) throws IOException {
     RemoteIterator<LocatedFileStatus> listStatus = fileSystem.listFiles(path, recursive);
     return getFiles(listStatus);
   }

   @Override
   public CarbonFile[] locationAwareListFiles(PathFilter pathFilter) throws IOException {
     List<FileStatus> listStatus = new ArrayList<>();
     RemoteIterator<LocatedFileStatus> iter = fileSystem.listLocatedStatus(path);
     while (iter.hasNext()) {
       LocatedFileStatus fileStatus = iter.next();
       if (pathFilter.accept(fileStatus.getPath()) && fileStatus.getLen() > 0) {
         listStatus.add(fileStatus);
       }
     }
     return getFiles(listStatus.toArray(new FileStatus[listStatus.size()]));
   }

   protected List<CarbonFile> getFiles(RemoteIterator<LocatedFileStatus> listStatus)
       throws IOException {
     List<CarbonFile> carbonFiles = new ArrayList<>();
     while (listStatus.hasNext()) {
       Path filePath = listStatus.next().getPath();
       carbonFiles.add(FileFactory.getCarbonFile(filePath.toString()));
     }
     return carbonFiles;
   }

   /**
    * Method used to list files recursively and apply file filter on the result.
    *
    */
   @Override
   public List<CarbonFile> listFiles(boolean recursive, CarbonFileFilter fileFilter)
       throws IOException {
     List<CarbonFile> carbonFiles = new ArrayList<>();
     FileStatus fileStatus = fileSystem.getFileStatus(path);
     if (null != fileStatus && fileStatus.isDirectory()) {
       RemoteIterator<LocatedFileStatus> listStatus = fileSystem.listFiles(fileStatus.getPath(),
           recursive);
       while (listStatus.hasNext()) {
         LocatedFileStatus locatedFileStatus = listStatus.next();
         CarbonFile carbonFile = FileFactory.getCarbonFile(locatedFileStatus.getPath().toString());
         if (fileFilter.accept(carbonFile)) {
           carbonFiles.add(carbonFile);
         }
       }
     }
     return carbonFiles;
   }

   @Override
   public CarbonFile[] listFiles(boolean recursive, int maxCount)
       throws IOException {
     List<CarbonFile> carbonFiles = new ArrayList<>();
     FileStatus fileStatus = fileSystem.getFileStatus(path);
     if (null != fileStatus && fileStatus.isDirectory()) {
       RemoteIterator<LocatedFileStatus> listStatus = fileSystem.listFiles(path, recursive);
       int counter = 0;
       while (counter < maxCount && listStatus.hasNext()) {
         LocatedFileStatus locatedFileStatus = listStatus.next();
         CarbonFile carbonFile = FileFactory.getCarbonFile(locatedFileStatus.getPath().toString());
         carbonFiles.add(carbonFile);
         counter++;
       }
     }
     return carbonFiles.toArray(new CarbonFile[0]);
   }

   @Override
   public String[] getLocations() throws IOException {
     BlockLocation[] blkLocations;
     FileStatus fileStatus = fileSystem.getFileStatus(path);
     if (fileStatus instanceof LocatedFileStatus) {
       blkLocations = ((LocatedFileStatus) fileStatus).getBlockLocations();
     } else {
       blkLocations =
           fileSystem.getFileBlockLocations(fileStatus.getPath(), 0L, fileStatus.getLen());
     }
     return blkLocations[0].getHosts();
   }

   @Override
   public boolean setReplication(short replication) throws IOException {
     return fileSystem.setReplication(path, replication);
   }

   @Override
   public CarbonFile[] listFiles(final CarbonFileFilter fileFilter) {
     CarbonFile[] files = listFiles();
     if (files != null && files.length >= 1) {
       List<CarbonFile> fileList = new ArrayList<CarbonFile>(files.length);
       for (int i = 0; i < files.length; i++) {
         if (fileFilter.accept(files[i])) {
           fileList.add(files[i]);
         }
       }
       if (fileList.size() >= 1) {
         return fileList.toArray(new CarbonFile[fileList.size()]);
       } else {
         return new CarbonFile[0];
       }
     }
     return files;
   }

   @Override
   public short getDefaultReplication() {
     return fileSystem.getDefaultReplication(path);
   }

   @Override
   public boolean equals(Object o) {
     if (this == o) {
       return true;
     }
     if (o == null || getClass() != o.getClass()) {
       return false;
     }
     AbstractDFSCarbonFile that = (AbstractDFSCarbonFile) o;
     if (path == null || that.path == null) {
       return false;
     }
     return path.equals(that.path);
   }

   @Override
   public int hashCode() {
     if (path == null) {
       return 0;
     }
     return Objects.hash(path);
   }

   @Override
   public long getLength() throws IOException {
     return fileSystem.getFileStatus(path).getLen();
   }
 }
	/*
	* 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.filesystem;

	import java.io.BufferedInputStream;
	import java.io.BufferedOutputStream;
	import java.io.DataInputStream;
	import java.io.DataOutputStream;
	import java.io.IOException;
	import java.io.InputStream;
	import java.io.OutputStream;
	import java.util.ArrayList;
	import java.util.List;
	import java.util.Objects;

	import org.apache.carbondata.common.logging.LogServiceFactory;
	import org.apache.carbondata.core.constants.CarbonCommonConstants;
	import org.apache.carbondata.core.datastore.impl.FileFactory;
	import org.apache.carbondata.core.exception.CarbonFileException;
	import org.apache.carbondata.core.util.CarbonUtil;

	import org.apache.hadoop.conf.Configuration;
	import org.apache.hadoop.fs.BlockLocation;
	import org.apache.hadoop.fs.FSDataInputStream;
	import org.apache.hadoop.fs.FSDataOutputStream;
	import org.apache.hadoop.fs.FileStatus;
	import org.apache.hadoop.fs.FileSystem;
	import org.apache.hadoop.fs.LocatedFileStatus;
	import org.apache.hadoop.fs.Path;
	import org.apache.hadoop.fs.PathFilter;
	import org.apache.hadoop.fs.RemoteIterator;
	import org.apache.hadoop.fs.permission.FsAction;
	import org.apache.hadoop.fs.permission.FsPermission;
	import org.apache.hadoop.io.compress.BZip2Codec;
	import org.apache.hadoop.io.compress.CompressionCodec;
	import org.apache.hadoop.io.compress.CompressionCodecFactory;
	import org.apache.hadoop.io.compress.GzipCodec;
	import org.apache.hadoop.io.compress.Lz4Codec;
	import org.apache.hadoop.io.compress.SnappyCodec;
	import org.apache.log4j.Logger;

	public abstract class AbstractDFSCarbonFile implements CarbonFile {
	/**
	* LOGGER
	*/
	private static final Logger LOGGER =
	LogServiceFactory.getLogService(AbstractDFSCarbonFile.class.getName());
	protected FileSystem fileSystem;
	protected Configuration hadoopConf;
	protected Path path;
	protected FileStatus fileStatus;

	AbstractDFSCarbonFile(String filePath) {
	this(filePath, FileFactory.getConfiguration());
	}

	AbstractDFSCarbonFile(String filePath, Configuration hadoopConf) {
	this(new Path(filePath), hadoopConf);
	}

	AbstractDFSCarbonFile(Path path) {
	this(path, FileFactory.getConfiguration());
	}

	AbstractDFSCarbonFile(Path path, Configuration hadoopConf) {
	this.hadoopConf = hadoopConf;
	try {
	Path newPath = new Path(path.toString().replace("\\", "/"));
	fileSystem = newPath.getFileSystem(this.hadoopConf);
	this.path = newPath;
	} catch (IOException e) {
	throw new CarbonFileException("Error while getting File System: ", e);
	}
	}

	AbstractDFSCarbonFile(FileStatus fileStatus) {
	this(fileStatus.getPath());
	this.fileStatus = fileStatus;
	}

	@Override
	public boolean createNewFile() {
	try {
	return fileSystem.createNewFile(path);
	} catch (IOException e) {
	throw new CarbonFileException("Unable to create file: " + path.toString(), e);
	}
	}

	@Override
	public String getAbsolutePath() {
	try {
	return fileSystem.getFileStatus(path).getPath().toString();
	} catch (IOException e) {
	throw new CarbonFileException("Unable to get file status: ", e);
	}
	}

	@Override
	public String getName() {
	return path.getName();
	}

	@Override
	public boolean isDirectory() {
	try {
	return fileSystem.getFileStatus(path).isDirectory();
	} catch (IOException e) {
	throw new CarbonFileException("Unable to get file status: ", e);
	}
	}

	@Override
	public CarbonFile getParentFile() {
	Path parentPath = path.getParent();
	return parentPath == null ? null : FileFactory.getCarbonFile(parentPath.toString());
	}

	@Override
	public boolean exists() {
	try {
	return fileSystem.exists(path);
	} catch (IOException e) {
	throw new CarbonFileException(e);
	}
	}

	@Override
	public String getCanonicalPath() {
	return getAbsolutePath();
	}

	@Override
	public String getPath() {
	return getAbsolutePath();
	}

	@Override
	public long getSize() {
	try {
	return fileSystem.getFileStatus(path).getLen();
	} catch (IOException e) {
	throw new CarbonFileException("Unable to get file status for " + path.toString(), e);
	}
	}

	@Override
	public boolean renameTo(String changeToName) {
	try {
	return fileSystem.rename(path, new Path(changeToName));
	} catch (IOException e) {
	throw new CarbonFileException("Failed to rename file: ", e);
	}
	}

	public boolean delete() {
	try {
	return fileSystem.delete(path, true);
	} catch (IOException e) {
	throw new CarbonFileException("Failed to delete file:", e);
	}
	}

	@Override
	public long getLastModifiedTime() {
	try {
	if (this.fileStatus != null) {
	return this.fileStatus.getModificationTime();
	}
	return fileSystem.getFileStatus(path).getModificationTime();
	} catch (IOException e) {
	throw new CarbonFileException("Unable to get file status: ", e);
	}
	}

	@Override
	public boolean setLastModifiedTime(long timestamp) {
	try {
	fileSystem.setTimes(path, timestamp, timestamp);
	return true;
	} catch (IOException e) {
	throw new CarbonFileException("Error while setting modified time: ", e);
	}
	}

	/**
	* This method will delete the data in file data from a given offset
	*/
	@Override
	public boolean truncate(String fileName, long validDataEndOffset) {
	DataOutputStream dataOutputStream = null;
	DataInputStream dataInputStream = null;
	boolean fileTruncatedSuccessfully = false;
	// if bytes to read less than 1024 then buffer size should be equal to the given offset
	int bufferSize = validDataEndOffset > CarbonCommonConstants.BYTE_TO_KB_CONVERSION_FACTOR ?
	CarbonCommonConstants.BYTE_TO_KB_CONVERSION_FACTOR :
	(int) validDataEndOffset;
	// temporary file name
	String tempWriteFilePath = fileName + CarbonCommonConstants.TEMPWRITEFILEEXTENSION;
	try {
	CarbonFile tempFile;
	// delete temporary file if it already exists at a given path
	if (FileFactory.isFileExist(tempWriteFilePath)) {
	tempFile = FileFactory.getCarbonFile(tempWriteFilePath);
	tempFile.delete();
	}
	// create new temporary file
	FileFactory.createNewFile(tempWriteFilePath);
	tempFile = FileFactory.getCarbonFile(tempWriteFilePath);
	byte[] buff = new byte[bufferSize];
	dataInputStream = FileFactory.getDataInputStream(fileName);
	// read the data
	int read = dataInputStream.read(buff, 0, buff.length);
	dataOutputStream = FileFactory.getDataOutputStream(tempWriteFilePath);
	dataOutputStream.write(buff, 0, read);
	long remaining = validDataEndOffset - read;
	// anytime we should not cross the offset to be read
	while (remaining > 0) {
	if (remaining > bufferSize) {
	buff = new byte[bufferSize];
	} else {
	buff = new byte[(int) remaining];
	}
	read = dataInputStream.read(buff, 0, buff.length);
	dataOutputStream.write(buff, 0, read);
	remaining = remaining - read;
	}
	CarbonUtil.closeStreams(dataInputStream, dataOutputStream);
	// rename the temp file to original file
	tempFile.renameForce(fileName);
	fileTruncatedSuccessfully = true;
	} catch (IOException e) {
	LOGGER.error("Exception occurred while truncating the file " + e.getMessage(), e);
	} finally {
	CarbonUtil.closeStreams(dataOutputStream, dataInputStream);
	}
	return fileTruncatedSuccessfully;
	}

	/**
	* This method will be used to check whether a file has been modified or not
	*
	* @param fileTimeStamp time to be compared with latest timestamp of file
	* @param endOffset file length to be compared with current length of file
	* @return whether a file has been modified or not
	*/
	@Override
	public boolean isFileModified(long fileTimeStamp, long endOffset) {
	boolean isFileModified = false;
	if (getLastModifiedTime() > fileTimeStamp \|\| getSize() > endOffset) {
	isFileModified = true;
	}
	return isFileModified;
	}

	@Override
	public DataOutputStream getDataOutputStream(int bufferSize, boolean append)
	throws IOException {
	FSDataOutputStream stream = null;
	if (append) {
	// append to a file only if file already exists else file not found
	// exception will be thrown by hdfs
	if (CarbonUtil.isFileExists(path.toString())) {
	stream = fileSystem.append(path, bufferSize);
	} else {
	stream = fileSystem.create(path, true, bufferSize);
	}
	} else {
	stream = fileSystem.create(path, true, bufferSize);
	}
	return stream;
	}

	@Override
	public DataInputStream getDataInputStream(int bufferSize) throws IOException {
	return getDataInputStream(bufferSize,
	CarbonUtil.inferCompressorFromFileName(path.toString()));
	}

	/**
	* return the DataInputStream which is seek to the offset of file
	*
	* @return DataInputStream
	* @throws IOException
	*/
	@Override
	public DataInputStream getDataInputStream(int bufferSize, long offset)
	throws IOException {
	FSDataInputStream stream = fileSystem.open(path, bufferSize);
	stream.seek(offset);
	return new DataInputStream(new BufferedInputStream(stream));
	}

	@Override
	public DataInputStream getDataInputStream(int bufferSize, String compressor)
	throws IOException {
	InputStream inputStream;
	if (bufferSize <= 0) {
	inputStream = fileSystem.open(path);
	} else {
	inputStream = fileSystem.open(path, bufferSize);
	}
	String codecName = getCodecNameFromCompressor(compressor);
	if (!codecName.isEmpty()) {
	CompressionCodec codec = new CompressionCodecFactory(hadoopConf).getCodecByName(codecName);
	inputStream = codec.createInputStream(inputStream);
	}
	if (bufferSize <= 0 && inputStream instanceof FSDataInputStream) {
	return (DataInputStream) inputStream;
	} else {
	return new DataInputStream(new BufferedInputStream(inputStream));
	}
	}

	/**
	* get codec name from user specified compressor name
	*
	* @param compressorName user specified compressor name
	* @return name of codec
	* @throws IOException
	*/
	private String getCodecNameFromCompressor(String compressorName) throws IOException {
	if (compressorName.isEmpty()) {
	return "";
	} else if ("GZIP".equalsIgnoreCase(compressorName)) {
	return GzipCodec.class.getName();
	} else if ("BZIP2".equalsIgnoreCase(compressorName)) {
	return BZip2Codec.class.getName();
	} else if ("SNAPPY".equalsIgnoreCase(compressorName)) {
	return SnappyCodec.class.getName();
	} else if ("LZ4".equalsIgnoreCase(compressorName)) {
	return Lz4Codec.class.getName();
	} else {
	throw new IOException("Unsupported compressor: " + compressorName);
	}
	}

	@Override
	public DataOutputStream getDataOutputStream() throws IOException {
	return fileSystem.create(path, true);
	}

	@Override
	public DataOutputStream getDataOutputStream(int bufferSize, long blockSize)
	throws IOException {
	short replication = fileSystem.getDefaultReplication(path);
	return getDataOutputStream(bufferSize, blockSize, replication);
	}

	@Override
	public DataOutputStream getDataOutputStream(int bufferSize, long blockSize, short replication)
	throws IOException {
	return fileSystem.create(path, true, bufferSize, replication, blockSize);
	}

	@Override
	public DataOutputStream getDataOutputStream(int bufferSize, String compressor)
	throws IOException {
	OutputStream outputStream;
	if (bufferSize <= 0) {
	outputStream = fileSystem.create(path);
	} else {
	outputStream = fileSystem.create(path, true, bufferSize);
	}
	String codecName = getCodecNameFromCompressor(compressor);
	if (!codecName.isEmpty()) {
	CompressionCodec codec = new CompressionCodecFactory(hadoopConf).getCodecByName(codecName);
	outputStream = codec.createOutputStream(outputStream);
	}
	return new DataOutputStream(new BufferedOutputStream(outputStream));
	}

	@Override
	public boolean isFileExist(boolean performFileCheck) throws IOException {
	if (performFileCheck) {
	return fileSystem.exists(path) && fileSystem.isFile(path);
	} else {
	return fileSystem.exists(path);
	}
	}

	@Override
	public boolean isFileExist() throws IOException {
	return isFileExist(false);
	}

	@Override
	public boolean createNewFile(FsPermission permission) throws IOException {
	if (fileSystem.exists(path)) {
	return false;
	} else {
	if (permission == null) {
	permission =
	FsPermission.getFileDefault().applyUMask(FsPermission.getUMask(fileSystem.getConf()));
	}
	// Pass the permissions during file creation itself
	fileSystem
	.create(path, permission, false, fileSystem.getConf().getInt("io.file.buffer.size", 4096),
	fileSystem.getDefaultReplication(path), fileSystem.getDefaultBlockSize(path), null)
	.close();
	return true;
	}
	}

	@Override
	public boolean deleteFile() throws IOException {
	return fileSystem.delete(path, true);
	}

	@Override
	public boolean mkdirs() throws IOException {
	return fileSystem.mkdirs(path);
	}

	@Override
	public DataOutputStream getDataOutputStreamUsingAppend() throws IOException {
	return fileSystem.append(path);
	}

	@Override
	public boolean createNewLockFile() throws IOException {
	if (fileSystem.exists(path)) {
	return false;
	} else {
	// Pass the permissions during file creation itself
	fileSystem.create(path, new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL), false,
	fileSystem.getConf().getInt("io.file.buffer.size", 4096),
	fileSystem.getDefaultReplication(path), fileSystem.getDefaultBlockSize(path), null)
	.close();
	// hadoop masks the permission according to configured permission, so need to set permission
	// forcefully
	fileSystem.setPermission(path, new FsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL));
	return true;
	}
	}

	@Override
	public CarbonFile[] listFiles() {
	FileStatus[] listStatus;
	try {
	listStatus = fileSystem.listStatus(path);
	} catch (IOException e) {
	LOGGER.warn("Exception occurred: " + e.getMessage(), e);
	return new CarbonFile[0];
	}
	return getFiles(listStatus);
	}

	/**
	* Get the CarbonFiles from FileStatus array
	*/
	protected abstract CarbonFile[] getFiles(FileStatus[] listStatus);

	@Override
	public List<CarbonFile> listFiles(Boolean recursive) throws IOException {
	RemoteIterator<LocatedFileStatus> listStatus = fileSystem.listFiles(path, recursive);
	return getFiles(listStatus);
	}

	@Override
	public CarbonFile[] locationAwareListFiles(PathFilter pathFilter) throws IOException {
	List<FileStatus> listStatus = new ArrayList<>();
	RemoteIterator<LocatedFileStatus> iter = fileSystem.listLocatedStatus(path);
	while (iter.hasNext()) {
	LocatedFileStatus fileStatus = iter.next();
	if (pathFilter.accept(fileStatus.getPath()) && fileStatus.getLen() > 0) {
	listStatus.add(fileStatus);
	}
	}
	return getFiles(listStatus.toArray(new FileStatus[listStatus.size()]));
	}

	protected List<CarbonFile> getFiles(RemoteIterator<LocatedFileStatus> listStatus)
	throws IOException {
	List<CarbonFile> carbonFiles = new ArrayList<>();
	while (listStatus.hasNext()) {
	Path filePath = listStatus.next().getPath();
	carbonFiles.add(FileFactory.getCarbonFile(filePath.toString()));
	}
	return carbonFiles;
	}

	/**
	* Method used to list files recursively and apply file filter on the result.
	*
	*/
	@Override
	public List<CarbonFile> listFiles(boolean recursive, CarbonFileFilter fileFilter)
	throws IOException {
	List<CarbonFile> carbonFiles = new ArrayList<>();
	FileStatus fileStatus = fileSystem.getFileStatus(path);
	if (null != fileStatus && fileStatus.isDirectory()) {
	RemoteIterator<LocatedFileStatus> listStatus = fileSystem.listFiles(fileStatus.getPath(),
	recursive);
	while (listStatus.hasNext()) {
	LocatedFileStatus locatedFileStatus = listStatus.next();
	CarbonFile carbonFile = FileFactory.getCarbonFile(locatedFileStatus.getPath().toString());
	if (fileFilter.accept(carbonFile)) {
	carbonFiles.add(carbonFile);
	}
	}
	}
	return carbonFiles;
	}

	@Override
	public CarbonFile[] listFiles(boolean recursive, int maxCount)
	throws IOException {
	List<CarbonFile> carbonFiles = new ArrayList<>();
	FileStatus fileStatus = fileSystem.getFileStatus(path);
	if (null != fileStatus && fileStatus.isDirectory()) {
	RemoteIterator<LocatedFileStatus> listStatus = fileSystem.listFiles(path, recursive);
	int counter = 0;
	while (counter < maxCount && listStatus.hasNext()) {
	LocatedFileStatus locatedFileStatus = listStatus.next();
	CarbonFile carbonFile = FileFactory.getCarbonFile(locatedFileStatus.getPath().toString());
	carbonFiles.add(carbonFile);
	counter++;
	}
	}
	return carbonFiles.toArray(new CarbonFile[0]);
	}

	@Override
	public String[] getLocations() throws IOException {
	BlockLocation[] blkLocations;
	FileStatus fileStatus = fileSystem.getFileStatus(path);
	if (fileStatus instanceof LocatedFileStatus) {
	blkLocations = ((LocatedFileStatus) fileStatus).getBlockLocations();
	} else {
	blkLocations =
	fileSystem.getFileBlockLocations(fileStatus.getPath(), 0L, fileStatus.getLen());
	}
	return blkLocations[0].getHosts();
	}

	@Override
	public boolean setReplication(short replication) throws IOException {
	return fileSystem.setReplication(path, replication);
	}

	@Override
	public CarbonFile[] listFiles(final CarbonFileFilter fileFilter) {
	CarbonFile[] files = listFiles();
	if (files != null && files.length >= 1) {
	List<CarbonFile> fileList = new ArrayList<CarbonFile>(files.length);
	for (int i = 0; i < files.length; i++) {
	if (fileFilter.accept(files[i])) {
	fileList.add(files[i]);
	}
	}
	if (fileList.size() >= 1) {
	return fileList.toArray(new CarbonFile[fileList.size()]);
	} else {
	return new CarbonFile[0];
	}
	}
	return files;
	}

	@Override
	public short getDefaultReplication() {
	return fileSystem.getDefaultReplication(path);
	}

	@Override
	public boolean equals(Object o) {
	if (this == o) {
	return true;
	}
	if (o == null \|\| getClass() != o.getClass()) {
	return false;
	}
	AbstractDFSCarbonFile that = (AbstractDFSCarbonFile) o;
	if (path == null \|\| that.path == null) {
	return false;
	}
	return path.equals(that.path);
	}

	@Override
	public int hashCode() {
	if (path == null) {
	return 0;
	}
	return Objects.hash(path);
	}

	@Override
	public long getLength() throws IOException {
	return fileSystem.getFileStatus(path).getLen();
	}
	}