core/src/main/java/org/apache/carbondata/core/datastore/filesystem/LocalCarbonFile.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.File;
 import java.io.FileInputStream;
 import java.io.FileNotFoundException;
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.nio.channels.FileChannel;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
 import java.util.Objects;
 import java.util.zip.GZIPInputStream;
 import java.util.zip.GZIPOutputStream;

 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.util.CarbonUtil;

 import com.github.luben.zstd.ZstdInputStream;
 import com.github.luben.zstd.ZstdOutputStream;
 import net.jpountz.lz4.LZ4BlockInputStream;
 import net.jpountz.lz4.LZ4BlockOutputStream;
 import org.apache.commons.compress.compressors.bzip2.BZip2CompressorInputStream;
 import org.apache.commons.compress.compressors.bzip2.BZip2CompressorOutputStream;
 import org.apache.commons.io.FileUtils;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.PathFilter;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.log4j.Logger;
 import org.xerial.snappy.SnappyInputStream;
 import org.xerial.snappy.SnappyOutputStream;

 public class LocalCarbonFile implements CarbonFile {
   private static final Logger LOGGER =
       LogServiceFactory.getLogService(LocalCarbonFile.class.getName());

   private File file;

   private String absoluteFilePath;

   public LocalCarbonFile(String filePath) {
     Path pathWithoutSchemeAndAuthority = Path.getPathWithoutSchemeAndAuthority(new Path(filePath));
     filePath = pathWithoutSchemeAndAuthority.toString().replace("\\", "/");
     absoluteFilePath = FileFactory.getUpdatedFilePath(filePath);
     file = new File(absoluteFilePath);
   }

   LocalCarbonFile(File file) {
     this.file = file;
     String filePath = file.getAbsolutePath().replace("\\", "/");
     absoluteFilePath = FileFactory.getUpdatedFilePath(filePath);
   }

   @Override
   public String getAbsolutePath() {
     return file.getAbsolutePath();
   }

   @Override
   public CarbonFile[] listFiles(final CarbonFileFilter fileFilter) {
     if (!isDirectory()) {
       return new CarbonFile[0];
     }
     File[] files = file.listFiles(pathname -> fileFilter.accept(new LocalCarbonFile(pathname)));
     if (files == null) {
       return new CarbonFile[0];
     }
     CarbonFile[] carbonFiles = new CarbonFile[files.length];
     for (int i = 0; i < carbonFiles.length; i++) {
       carbonFiles[i] = new LocalCarbonFile(files[i]);
     }
     return carbonFiles;
   }

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

   @Override
   public boolean isDirectory() {
     return file.isDirectory();
   }

   @Override
   public boolean exists() {
     return file.exists();
   }

   @Override
   public String getCanonicalPath() {
     try {
       return file.getCanonicalPath();
     } catch (IOException e) {
       LOGGER.error("Exception occurred" + e.getMessage(), e);
     }
     return null;
   }

   @Override
   public CarbonFile getParentFile() {
     return new LocalCarbonFile(file.getParentFile());
   }

   @Override
   public String getPath() {
     return file.getPath();
   }

   @Override
   public long getSize() {
     return file.length();
   }

   public boolean renameTo(String changeToName) {
     changeToName = FileFactory.getUpdatedFilePath(changeToName);
     return file.renameTo(new File(changeToName));
   }

   public boolean delete() {
     return deleteFile();
   }

   @Override
   public CarbonFile[] listFiles() {
     if (!file.isDirectory()) {
       return new CarbonFile[0];
     }
     File[] files = file.listFiles();
     if (files == null) {
       return new CarbonFile[0];
     }
     CarbonFile[] carbonFiles = new CarbonFile[files.length];
     for (int i = 0; i < carbonFiles.length; i++) {
       carbonFiles[i] = new LocalCarbonFile(files[i]);
     }
     return carbonFiles;
   }

   @Override
   public CarbonFile[] listFiles(boolean recursive, int maxCount)
       throws IOException {
     // ignore the maxCount for local filesystem
     return listFiles();
   }

   @Override
   public List<CarbonFile> listFiles(Boolean recursive) {
     if (!isDirectory()) {
       return new ArrayList<>();
     }
     Collection<File> fileCollection = FileUtils.listFiles(file, null, true);
     List<CarbonFile> carbonFiles = new ArrayList<>();
     for (File file : fileCollection) {
       carbonFiles.add(new LocalCarbonFile(file));
     }
     return carbonFiles;
   }

   @Override
   public List<CarbonFile> listFiles(boolean recursive, CarbonFileFilter fileFilter) {
     if (!file.isDirectory()) {
       return new ArrayList<>();
     }
     Collection<File> fileCollection = FileUtils.listFiles(file, null, recursive);
     if (fileCollection.isEmpty()) {
       return new ArrayList<>();
     }
     List<CarbonFile> carbonFiles = new ArrayList<>();
     for (File file : fileCollection) {
       CarbonFile carbonFile = new LocalCarbonFile(file);
       if (fileFilter.accept(carbonFile)) {
         carbonFiles.add(carbonFile);
       }
     }
     return carbonFiles;
   }

   @Override
   public boolean createNewFile() {
     try {
       return file.createNewFile();
     } catch (IOException e) {
       return false;
     }
   }

   @Override
   public long getLastModifiedTime() {
     return file.lastModified();
   }

   @Override
   public boolean setLastModifiedTime(long timestamp) {
     return file.setLastModified(timestamp);
   }

   /**
    * This method will delete the data in file data from a given offset
    */
   @Override
   public boolean truncate(String fileName, long validDataEndOffset) {
     FileChannel source = null;
     FileChannel destination = null;
     boolean fileTruncatedSuccessfully = false;
     // temporary file name
     String tempWriteFilePath = fileName + CarbonCommonConstants.TEMPWRITEFILEEXTENSION;
     try {
       CarbonFile tempFile;
       // delete temporary file if it already exists at a given path
       if (isFileExist()) {
         tempFile = FileFactory.getCarbonFile(tempWriteFilePath);
         tempFile.delete();
       }
       // create new temporary file
       FileFactory.createNewFile(tempWriteFilePath);
       tempFile = FileFactory.getCarbonFile(tempWriteFilePath);
       source = new FileInputStream(fileName).getChannel();
       destination = new FileOutputStream(tempWriteFilePath).getChannel();
       long read = destination.transferFrom(source, 0, validDataEndOffset);
       long totalBytesRead = read;
       long remaining = validDataEndOffset - totalBytesRead;
       // read till required data offset is not reached
       while (remaining > 0) {
         read = destination.transferFrom(source, totalBytesRead, remaining);
         totalBytesRead = totalBytesRead + read;
         remaining = remaining - totalBytesRead;
       }
       CarbonUtil.closeStreams(source, destination);
       // 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(source, destination);
     }
     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 true if the file is modified otherwise return false.
    */
   @Override
   public boolean isFileModified(long fileTimeStamp, long endOffset) {
     boolean isFileModified = false;
     if (getLastModifiedTime() > fileTimeStamp || getSize() > endOffset) {
       isFileModified = true;
     }
     return isFileModified;
   }

   @Override
   public boolean renameForce(String changeToName) {
     File destFile = new File(changeToName);
     if (destFile.exists() && !file.getAbsolutePath().equals(destFile.getAbsolutePath())) {
       if (destFile.delete()) {
         return file.renameTo(new File(changeToName));
       }
     }
     return file.renameTo(new File(changeToName));
   }

   @Override
   public DataOutputStream getDataOutputStream(int bufferSize, boolean append) throws
       FileNotFoundException {
     return new DataOutputStream(
         new BufferedOutputStream(new FileOutputStream(absoluteFilePath, append), bufferSize));
   }

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

   @Override
   public DataInputStream getDataInputStream(int bufferSize, String compressor)
       throws IOException {
     InputStream inputStream;
     if (compressor.isEmpty()) {
       inputStream = new FileInputStream(absoluteFilePath);
     } else if ("GZIP".equalsIgnoreCase(compressor)) {
       inputStream = new GZIPInputStream(new FileInputStream(absoluteFilePath));
     } else if ("BZIP2".equalsIgnoreCase(compressor)) {
       inputStream = new BZip2CompressorInputStream(new FileInputStream(absoluteFilePath));
     } else if ("SNAPPY".equalsIgnoreCase(compressor)) {
       inputStream = new SnappyInputStream(new FileInputStream(absoluteFilePath));
     } else if ("LZ4".equalsIgnoreCase(compressor)) {
       inputStream = new LZ4BlockInputStream(new FileInputStream(absoluteFilePath));
     } else if ("ZSTD".equalsIgnoreCase(compressor)) {
       inputStream = new ZstdInputStream(new FileInputStream(absoluteFilePath));
     } else {
       throw new IOException("Unsupported compressor: " + compressor);
     }

     if (bufferSize <= 0) {
       return new DataInputStream(new BufferedInputStream(inputStream));
     } else {
       return new DataInputStream(new BufferedInputStream(inputStream, bufferSize));
     }
   }

   /**
    * return the DataInputStream which is seek to the offset of file
    *
    * @param bufferSize
    * @param offset
    * @return DataInputStream
    * @throws IOException
    */
   @Override
   public DataInputStream getDataInputStream(int bufferSize, long offset) throws
       IOException {
     FileInputStream fis = new FileInputStream(absoluteFilePath);
     long actualSkipSize = 0;
     long skipSize = offset;
     try {
       while (actualSkipSize != offset) {
         actualSkipSize += fis.skip(skipSize);
         skipSize = skipSize - actualSkipSize;
       }
     } catch (IOException ioe) {
       CarbonUtil.closeStream(fis);
       throw ioe;
     }
     return new DataInputStream(new BufferedInputStream(fis));
   }

   @Override
   public DataOutputStream getDataOutputStream()
       throws IOException {
     return new DataOutputStream(new BufferedOutputStream(new FileOutputStream(absoluteFilePath)));
   }

   @Override
   public DataOutputStream getDataOutputStream(int bufferSize, long blockSize) throws IOException {
     return getDataOutputStream(bufferSize, blockSize, (short) 1);
   }

   @Override
   public DataOutputStream getDataOutputStream(int bufferSize, long blockSize, short replication)
       throws IOException {
     return new DataOutputStream(
         new BufferedOutputStream(new FileOutputStream(absoluteFilePath), bufferSize));
   }

   @Override
   public DataOutputStream getDataOutputStream(int bufferSize, String compressor)
       throws IOException {
     OutputStream outputStream;
     if (compressor.isEmpty()) {
       outputStream = new FileOutputStream(absoluteFilePath);
     } else if ("GZIP".equalsIgnoreCase(compressor)) {
       outputStream = new GZIPOutputStream(new FileOutputStream(absoluteFilePath));
     } else if ("BZIP2".equalsIgnoreCase(compressor)) {
       outputStream = new BZip2CompressorOutputStream(new FileOutputStream(absoluteFilePath));
     } else if ("SNAPPY".equalsIgnoreCase(compressor)) {
       outputStream = new SnappyOutputStream(new FileOutputStream(absoluteFilePath));
     } else if ("LZ4".equalsIgnoreCase(compressor)) {
       outputStream = new LZ4BlockOutputStream(new FileOutputStream(absoluteFilePath));
     } else if ("ZSTD".equalsIgnoreCase(compressor)) {
       // compression level 1 is cost-effective for sort temp file
       // which is not used for storage
       outputStream = new ZstdOutputStream(new FileOutputStream(absoluteFilePath), 1);
     } else {
       throw new IOException("Unsupported compressor: " + compressor);
     }

     if (bufferSize <= 0) {
       return new DataOutputStream(new BufferedOutputStream(outputStream));
     } else {
       return new DataOutputStream(new BufferedOutputStream(outputStream, bufferSize));
     }
   }

   @Override
   public boolean isFileExist(boolean performFileCheck) {
     if (performFileCheck) {
       return file.exists() && file.isFile();
     } else {
       return file.exists();
     }
   }

   @Override
   public boolean isFileExist() {
     return file.exists();
   }

   @Override
   public boolean createNewFile(final FsPermission permission) throws IOException {
     return file.createNewFile();
   }

   @Override
   public boolean deleteFile() {
     return FileFactory.deleteAllFilesOfDir(file);
   }

   @Override
   public boolean mkdirs() {
     return file.mkdirs();
   }

   @Override
   public DataOutputStream getDataOutputStreamUsingAppend() throws IOException {
     return new DataOutputStream(
         new BufferedOutputStream(new FileOutputStream(absoluteFilePath, true)));
   }

   @Override
   public boolean createNewLockFile() throws IOException {
     return file.createNewFile();
   }

   @Override
   public CarbonFile[] locationAwareListFiles(PathFilter pathFilter) {
     return listFiles();
   }

   @Override
   public String[] getLocations() {
     return new String[]{"localhost"};
   }

   @Override
   public boolean setReplication(short replication) {
     // local carbon file does not need replication
     return true;
   }

   @Override
   public short getDefaultReplication() {
     return 1;
   }

   @Override
   public long getLength() {
     return file.length();
   }

   @Override
   public boolean equals(Object o) {
     if (this == o) {
       return true;
     }
     if (o == null || getClass() != o.getClass()) {
       return false;
     }
     LocalCarbonFile that = (LocalCarbonFile) o;
     return Objects.equals(file.getAbsolutePath(), that.file.getAbsolutePath());
   }

   @Override
   public int hashCode() {
     return Objects.hash(file.getAbsolutePath());
   }
 }
	/*
	* 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.File;
	import java.io.FileInputStream;
	import java.io.FileNotFoundException;
	import java.io.FileOutputStream;
	import java.io.IOException;
	import java.io.InputStream;
	import java.io.OutputStream;
	import java.nio.channels.FileChannel;
	import java.util.ArrayList;
	import java.util.Collection;
	import java.util.List;
	import java.util.Objects;
	import java.util.zip.GZIPInputStream;
	import java.util.zip.GZIPOutputStream;

	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.util.CarbonUtil;

	import com.github.luben.zstd.ZstdInputStream;
	import com.github.luben.zstd.ZstdOutputStream;
	import net.jpountz.lz4.LZ4BlockInputStream;
	import net.jpountz.lz4.LZ4BlockOutputStream;
	import org.apache.commons.compress.compressors.bzip2.BZip2CompressorInputStream;
	import org.apache.commons.compress.compressors.bzip2.BZip2CompressorOutputStream;
	import org.apache.commons.io.FileUtils;
	import org.apache.hadoop.fs.Path;
	import org.apache.hadoop.fs.PathFilter;
	import org.apache.hadoop.fs.permission.FsPermission;
	import org.apache.log4j.Logger;
	import org.xerial.snappy.SnappyInputStream;
	import org.xerial.snappy.SnappyOutputStream;

	public class LocalCarbonFile implements CarbonFile {
	private static final Logger LOGGER =
	LogServiceFactory.getLogService(LocalCarbonFile.class.getName());

	private File file;

	private String absoluteFilePath;

	public LocalCarbonFile(String filePath) {
	Path pathWithoutSchemeAndAuthority = Path.getPathWithoutSchemeAndAuthority(new Path(filePath));
	filePath = pathWithoutSchemeAndAuthority.toString().replace("\\", "/");
	absoluteFilePath = FileFactory.getUpdatedFilePath(filePath);
	file = new File(absoluteFilePath);
	}

	LocalCarbonFile(File file) {
	this.file = file;
	String filePath = file.getAbsolutePath().replace("\\", "/");
	absoluteFilePath = FileFactory.getUpdatedFilePath(filePath);
	}

	@Override
	public String getAbsolutePath() {
	return file.getAbsolutePath();
	}

	@Override
	public CarbonFile[] listFiles(final CarbonFileFilter fileFilter) {
	if (!isDirectory()) {
	return new CarbonFile[0];
	}
	File[] files = file.listFiles(pathname -> fileFilter.accept(new LocalCarbonFile(pathname)));
	if (files == null) {
	return new CarbonFile[0];
	}
	CarbonFile[] carbonFiles = new CarbonFile[files.length];
	for (int i = 0; i < carbonFiles.length; i++) {
	carbonFiles[i] = new LocalCarbonFile(files[i]);
	}
	return carbonFiles;
	}

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

	@Override
	public boolean isDirectory() {
	return file.isDirectory();
	}

	@Override
	public boolean exists() {
	return file.exists();
	}

	@Override
	public String getCanonicalPath() {
	try {
	return file.getCanonicalPath();
	} catch (IOException e) {
	LOGGER.error("Exception occurred" + e.getMessage(), e);
	}
	return null;
	}

	@Override
	public CarbonFile getParentFile() {
	return new LocalCarbonFile(file.getParentFile());
	}

	@Override
	public String getPath() {
	return file.getPath();
	}

	@Override
	public long getSize() {
	return file.length();
	}

	public boolean renameTo(String changeToName) {
	changeToName = FileFactory.getUpdatedFilePath(changeToName);
	return file.renameTo(new File(changeToName));
	}

	public boolean delete() {
	return deleteFile();
	}

	@Override
	public CarbonFile[] listFiles() {
	if (!file.isDirectory()) {
	return new CarbonFile[0];
	}
	File[] files = file.listFiles();
	if (files == null) {
	return new CarbonFile[0];
	}
	CarbonFile[] carbonFiles = new CarbonFile[files.length];
	for (int i = 0; i < carbonFiles.length; i++) {
	carbonFiles[i] = new LocalCarbonFile(files[i]);
	}
	return carbonFiles;
	}

	@Override
	public CarbonFile[] listFiles(boolean recursive, int maxCount)
	throws IOException {
	// ignore the maxCount for local filesystem
	return listFiles();
	}

	@Override
	public List<CarbonFile> listFiles(Boolean recursive) {
	if (!isDirectory()) {
	return new ArrayList<>();
	}
	Collection<File> fileCollection = FileUtils.listFiles(file, null, true);
	List<CarbonFile> carbonFiles = new ArrayList<>();
	for (File file : fileCollection) {
	carbonFiles.add(new LocalCarbonFile(file));
	}
	return carbonFiles;
	}

	@Override
	public List<CarbonFile> listFiles(boolean recursive, CarbonFileFilter fileFilter) {
	if (!file.isDirectory()) {
	return new ArrayList<>();
	}
	Collection<File> fileCollection = FileUtils.listFiles(file, null, recursive);
	if (fileCollection.isEmpty()) {
	return new ArrayList<>();
	}
	List<CarbonFile> carbonFiles = new ArrayList<>();
	for (File file : fileCollection) {
	CarbonFile carbonFile = new LocalCarbonFile(file);
	if (fileFilter.accept(carbonFile)) {
	carbonFiles.add(carbonFile);
	}
	}
	return carbonFiles;
	}

	@Override
	public boolean createNewFile() {
	try {
	return file.createNewFile();
	} catch (IOException e) {
	return false;
	}
	}

	@Override
	public long getLastModifiedTime() {
	return file.lastModified();
	}

	@Override
	public boolean setLastModifiedTime(long timestamp) {
	return file.setLastModified(timestamp);
	}

	/**
	* This method will delete the data in file data from a given offset
	*/
	@Override
	public boolean truncate(String fileName, long validDataEndOffset) {
	FileChannel source = null;
	FileChannel destination = null;
	boolean fileTruncatedSuccessfully = false;
	// temporary file name
	String tempWriteFilePath = fileName + CarbonCommonConstants.TEMPWRITEFILEEXTENSION;
	try {
	CarbonFile tempFile;
	// delete temporary file if it already exists at a given path
	if (isFileExist()) {
	tempFile = FileFactory.getCarbonFile(tempWriteFilePath);
	tempFile.delete();
	}
	// create new temporary file
	FileFactory.createNewFile(tempWriteFilePath);
	tempFile = FileFactory.getCarbonFile(tempWriteFilePath);
	source = new FileInputStream(fileName).getChannel();
	destination = new FileOutputStream(tempWriteFilePath).getChannel();
	long read = destination.transferFrom(source, 0, validDataEndOffset);
	long totalBytesRead = read;
	long remaining = validDataEndOffset - totalBytesRead;
	// read till required data offset is not reached
	while (remaining > 0) {
	read = destination.transferFrom(source, totalBytesRead, remaining);
	totalBytesRead = totalBytesRead + read;
	remaining = remaining - totalBytesRead;
	}
	CarbonUtil.closeStreams(source, destination);
	// 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(source, destination);
	}
	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 true if the file is modified otherwise return false.
	*/
	@Override
	public boolean isFileModified(long fileTimeStamp, long endOffset) {
	boolean isFileModified = false;
	if (getLastModifiedTime() > fileTimeStamp \|\| getSize() > endOffset) {
	isFileModified = true;
	}
	return isFileModified;
	}

	@Override
	public boolean renameForce(String changeToName) {
	File destFile = new File(changeToName);
	if (destFile.exists() && !file.getAbsolutePath().equals(destFile.getAbsolutePath())) {
	if (destFile.delete()) {
	return file.renameTo(new File(changeToName));
	}
	}
	return file.renameTo(new File(changeToName));
	}

	@Override
	public DataOutputStream getDataOutputStream(int bufferSize, boolean append) throws
	FileNotFoundException {
	return new DataOutputStream(
	new BufferedOutputStream(new FileOutputStream(absoluteFilePath, append), bufferSize));
	}

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

	@Override
	public DataInputStream getDataInputStream(int bufferSize, String compressor)
	throws IOException {
	InputStream inputStream;
	if (compressor.isEmpty()) {
	inputStream = new FileInputStream(absoluteFilePath);
	} else if ("GZIP".equalsIgnoreCase(compressor)) {
	inputStream = new GZIPInputStream(new FileInputStream(absoluteFilePath));
	} else if ("BZIP2".equalsIgnoreCase(compressor)) {
	inputStream = new BZip2CompressorInputStream(new FileInputStream(absoluteFilePath));
	} else if ("SNAPPY".equalsIgnoreCase(compressor)) {
	inputStream = new SnappyInputStream(new FileInputStream(absoluteFilePath));
	} else if ("LZ4".equalsIgnoreCase(compressor)) {
	inputStream = new LZ4BlockInputStream(new FileInputStream(absoluteFilePath));
	} else if ("ZSTD".equalsIgnoreCase(compressor)) {
	inputStream = new ZstdInputStream(new FileInputStream(absoluteFilePath));
	} else {
	throw new IOException("Unsupported compressor: " + compressor);
	}

	if (bufferSize <= 0) {
	return new DataInputStream(new BufferedInputStream(inputStream));
	} else {
	return new DataInputStream(new BufferedInputStream(inputStream, bufferSize));
	}
	}

	/**
	* return the DataInputStream which is seek to the offset of file
	*
	* @param bufferSize
	* @param offset
	* @return DataInputStream
	* @throws IOException
	*/
	@Override
	public DataInputStream getDataInputStream(int bufferSize, long offset) throws
	IOException {
	FileInputStream fis = new FileInputStream(absoluteFilePath);
	long actualSkipSize = 0;
	long skipSize = offset;
	try {
	while (actualSkipSize != offset) {
	actualSkipSize += fis.skip(skipSize);
	skipSize = skipSize - actualSkipSize;
	}
	} catch (IOException ioe) {
	CarbonUtil.closeStream(fis);
	throw ioe;
	}
	return new DataInputStream(new BufferedInputStream(fis));
	}

	@Override
	public DataOutputStream getDataOutputStream()
	throws IOException {
	return new DataOutputStream(new BufferedOutputStream(new FileOutputStream(absoluteFilePath)));
	}

	@Override
	public DataOutputStream getDataOutputStream(int bufferSize, long blockSize) throws IOException {
	return getDataOutputStream(bufferSize, blockSize, (short) 1);
	}

	@Override
	public DataOutputStream getDataOutputStream(int bufferSize, long blockSize, short replication)
	throws IOException {
	return new DataOutputStream(
	new BufferedOutputStream(new FileOutputStream(absoluteFilePath), bufferSize));
	}

	@Override
	public DataOutputStream getDataOutputStream(int bufferSize, String compressor)
	throws IOException {
	OutputStream outputStream;
	if (compressor.isEmpty()) {
	outputStream = new FileOutputStream(absoluteFilePath);
	} else if ("GZIP".equalsIgnoreCase(compressor)) {
	outputStream = new GZIPOutputStream(new FileOutputStream(absoluteFilePath));
	} else if ("BZIP2".equalsIgnoreCase(compressor)) {
	outputStream = new BZip2CompressorOutputStream(new FileOutputStream(absoluteFilePath));
	} else if ("SNAPPY".equalsIgnoreCase(compressor)) {
	outputStream = new SnappyOutputStream(new FileOutputStream(absoluteFilePath));
	} else if ("LZ4".equalsIgnoreCase(compressor)) {
	outputStream = new LZ4BlockOutputStream(new FileOutputStream(absoluteFilePath));
	} else if ("ZSTD".equalsIgnoreCase(compressor)) {
	// compression level 1 is cost-effective for sort temp file
	// which is not used for storage
	outputStream = new ZstdOutputStream(new FileOutputStream(absoluteFilePath), 1);
	} else {
	throw new IOException("Unsupported compressor: " + compressor);
	}

	if (bufferSize <= 0) {
	return new DataOutputStream(new BufferedOutputStream(outputStream));
	} else {
	return new DataOutputStream(new BufferedOutputStream(outputStream, bufferSize));
	}
	}

	@Override
	public boolean isFileExist(boolean performFileCheck) {
	if (performFileCheck) {
	return file.exists() && file.isFile();
	} else {
	return file.exists();
	}
	}

	@Override
	public boolean isFileExist() {
	return file.exists();
	}

	@Override
	public boolean createNewFile(final FsPermission permission) throws IOException {
	return file.createNewFile();
	}

	@Override
	public boolean deleteFile() {
	return FileFactory.deleteAllFilesOfDir(file);
	}

	@Override
	public boolean mkdirs() {
	return file.mkdirs();
	}

	@Override
	public DataOutputStream getDataOutputStreamUsingAppend() throws IOException {
	return new DataOutputStream(
	new BufferedOutputStream(new FileOutputStream(absoluteFilePath, true)));
	}

	@Override
	public boolean createNewLockFile() throws IOException {
	return file.createNewFile();
	}

	@Override
	public CarbonFile[] locationAwareListFiles(PathFilter pathFilter) {
	return listFiles();
	}

	@Override
	public String[] getLocations() {
	return new String[]{"localhost"};
	}

	@Override
	public boolean setReplication(short replication) {
	// local carbon file does not need replication
	return true;
	}

	@Override
	public short getDefaultReplication() {
	return 1;
	}

	@Override
	public long getLength() {
	return file.length();
	}

	@Override
	public boolean equals(Object o) {
	if (this == o) {
	return true;
	}
	if (o == null \|\| getClass() != o.getClass()) {
	return false;
	}
	LocalCarbonFile that = (LocalCarbonFile) o;
	return Objects.equals(file.getAbsolutePath(), that.file.getAbsolutePath());
	}

	@Override
	public int hashCode() {
	return Objects.hash(file.getAbsolutePath());
	}
	}