content: /*
   * 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.Arrays;
  import java.util.List;
  import java.util.Objects;
  import java.util.stream.Collectors;
  
  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 {
        if (fileStatus != null) {
          return fileStatus.getPath().toString();
        }
        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 {
        if (fileStatus != null) {
          return fileStatus.getLen();
        }
        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());
        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 = this.fileStatus;
      if (fileStatus == null) {
        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();
    }
  
    @Override
    public List<CarbonFile> listDirs() throws IOException {
      FileStatus[] listStatus = null;
      if (null != fileStatus && fileStatus.isDirectory()) {
        Path path = fileStatus.getPath();
        listStatus = fileSystem.listStatus(path);
        CarbonFile[] dirs = getFiles(listStatus);
        List<CarbonFile> result = new ArrayList<CarbonFile>(Arrays.asList(dirs));
        return result.stream().filter(x -> x.isDirectory()).collect(Collectors.toList());
      } else {
        return new ArrayList<CarbonFile>();
      }
    }
  }
  