<commit_before>/*
 * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
 * (the "License"). You may not use this work except in compliance with the License, which is
 * available at www.apache.org/licenses/LICENSE-2.0
 *
 * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
 * either express or implied, as more fully set forth in the License.
 *
 * See the NOTICE file distributed with this work for information regarding copyright ownership.
 */

package alluxio.underfs;

import alluxio.AlluxioURI;
import alluxio.Configuration;
import alluxio.PropertyKey;
import alluxio.exception.ExceptionMessage;
import alluxio.retry.ExponentialBackoffRetry;
import alluxio.retry.RetryPolicy;
import alluxio.underfs.options.CreateOptions;
import alluxio.underfs.options.DeleteOptions;
import alluxio.underfs.options.FileLocationOptions;
import alluxio.underfs.options.ListOptions;
import alluxio.underfs.options.MkdirsOptions;
import alluxio.underfs.options.OpenOptions;
import alluxio.util.CommonUtils;
import alluxio.util.executor.ExecutorServiceFactories;
import alluxio.util.io.PathUtils;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;

import javax.annotation.Nullable;
import javax.annotation.concurrent.NotThreadSafe;
import javax.annotation.concurrent.ThreadSafe;

/**
 * An object based abstract {@link UnderFileSystem}. Object Stores implementing the
 * {@link UnderFileSystem} interface should derive from this class.
 */
@ThreadSafe
public abstract class ObjectUnderFileSystem extends BaseUnderFileSystem {
  private static final Logger LOG = LoggerFactory.getLogger(ObjectUnderFileSystem.class);

  /** Default maximum length for a single listing query. */
  private static final int DEFAULT_MAX_LISTING_CHUNK_LENGTH = 1000;

  /** Value used to indicate nested structure. */
  protected static final char PATH_SEPARATOR_CHAR = '/';

  /**
   * Value used to indicate nested structure. This is a string representation of
   * {@link ObjectUnderFileSystem#PATH_SEPARATOR_CHAR}.
   */
  protected static final String PATH_SEPARATOR = String.valueOf(PATH_SEPARATOR_CHAR);

  /** Executor service used for parallel UFS operations such as bulk deletes. */
  protected ExecutorService mExecutorService;

  /** Retry policy parameters for {@link UnderFileSystem#open(String, OpenOptions)}. */
  protected int mOpenRetryBaseSleepMs;
  protected int mOpenRetryMaxSleepMs;
  protected int mOpenRetryMaxNum;

  /**
   * Constructs an {@link ObjectUnderFileSystem}.
   *
   * @param uri the {@link AlluxioURI} used to create this ufs
   * @param ufsConf UFS configuration
   */
  protected ObjectUnderFileSystem(AlluxioURI uri, UnderFileSystemConfiguration ufsConf) {
    super(uri, ufsConf);

    int numThreads = Configuration.getInt(PropertyKey.UNDERFS_OBJECT_STORE_SERVICE_THREADS);
    mExecutorService = ExecutorServiceFactories.fixedThreadPoolExecutorServiceFactory(
        "alluxio-underfs-object-service-worker", numThreads).create();

    mOpenRetryBaseSleepMs =
        (int) Configuration.getMs(PropertyKey.UNDERFS_OBJECT_STORE_READ_RETRY_BASE_SLEEP_MS);
    mOpenRetryMaxSleepMs =
        (int) Configuration.getMs(PropertyKey.UNDERFS_OBJECT_STORE_READ_RETRY_MAX_SLEEP_MS);
    mOpenRetryMaxNum =
        Configuration.getInt(PropertyKey.UNDERFS_OBJECT_STORE_READ_RETRY_MAX_NUM);
  }

  /**
   * Information about a single object in object UFS.
   */
  protected class ObjectStatus {
    private static final long INVALID_CONTENT_LENGTH = -1L;
    private static final long INVALID_MODIFIED_TIME = -1L;

    private final long mContentLength;
    private final long mLastModifiedTimeMs;
    private final String mName;

    public ObjectStatus(String name, long contentLength, long lastModifiedTimeMs) {
      mContentLength = contentLength;
      mLastModifiedTimeMs = lastModifiedTimeMs;
      mName = name;
    }

    public ObjectStatus(String name) {
      mContentLength = INVALID_CONTENT_LENGTH;
      mLastModifiedTimeMs = INVALID_MODIFIED_TIME;
      mName = name;
    }

    /**
     * Gets the size of object contents in bytes.
     *
     * @return the content length in bytes
     */
    public long getContentLength() {
      return mContentLength;
    }

    /**
     * Gets the last modified epoch time in ms.
     *
     * @return modification time in milliseconds
     */
    public long getLastModifiedTimeMs() {
      return mLastModifiedTimeMs;
    }

    /**
     * Gets the name of the object.
     *
     * @return object name
     */
    public String getName() {
      return mName;
    }
  }

  /**
   * A chunk of listing results.
   */
  public interface ObjectListingChunk {
    /**
     * Returns objects in a pseudo-directory which may be a file or a directory.
     *
     * @return a list of object statuses
     */
    ObjectStatus[] getObjectStatuses();

    /**
     * Use common prefixes to infer pseudo-directories in object store.
     *
     * @return a list of common prefixes
     */
    String[] getCommonPrefixes();

    /**
     * Gets next chunk of object listings.
     *
     * @return null if listing did not find anything or is done, otherwise return new
     * {@link ObjectListingChunk} for the next chunk
     */
    @Nullable
    ObjectListingChunk getNextChunk() throws IOException;
  }

  /**
   * Permissions in object UFS.
   */
  public class ObjectPermissions {
    final String mOwner;
    final String mGroup;
    final short mMode;

    /**
     * Creates a new ObjectPermissions.
     *
     * @param owner the owner of the object
     * @param group the group of the object
     * @param mode the representation of the permission bits
     */
    public ObjectPermissions(String owner, String group, short mode) {
      mOwner = owner;
      mGroup = group;
      mMode = mode;
    }

    /**
     * @return the name of the owner
     */
    public String getOwner() {
      return mOwner;
    }

    /**
     * @return the name of the group
     */
    public String getGroup() {
      return mGroup;
    }

    /**
     * @return the short representation of the permission bits
     */
    public short getMode() {
      return mMode;
    }
  }

  @Override
  public void close() throws IOException {
  }

  @Override
  public void connectFromMaster(String hostname) {
  }

  @Override
  public void connectFromWorker(String hostname) {
  }

  @Override
  public OutputStream create(String path, CreateOptions options) throws IOException {
    if (options.getCreateParent() && !mkdirs(getParentPath(path))) {
      throw new IOException(ExceptionMessage.PARENT_CREATION_FAILED.getMessage(path));
    }
    return createObject(stripPrefixIfPresent(path));
  }

  @Override
  public boolean deleteFile(String path) throws IOException {
    return deleteObject(stripPrefixIfPresent(path));
  }

  @Override
  public boolean deleteDirectory(String path, DeleteOptions options) throws IOException {
    if (!options.isRecursive()) {
      UfsStatus[] children = listInternal(path, ListOptions.defaults());
      if (children == null) {
        LOG.error("Unable to delete path because {} is not a directory ", path);
        return false;
      }
      if (children.length != 0) {
        LOG.error("Unable to delete {} because it is a non empty directory. Specify "
                + "recursive as true in order to delete non empty directories.", path);
        return false;
      }
      // Delete the directory itself
      return deleteObject(stripPrefixIfPresent(convertToFolderName(path)));
    }

    // Delete children
    DeleteBuffer deleteBuffer = new DeleteBuffer();
    UfsStatus[] pathsToDelete = listInternal(path, ListOptions.defaults().setRecursive(true));
    if (pathsToDelete == null) {
      LOG.warn("Unable to delete {} because listInternal returns null", path);
      return false;
    }
    for (UfsStatus pathToDelete : pathsToDelete) {
      String pathKey = stripPrefixIfPresent(PathUtils.concatPath(path, pathToDelete.getName()));
      if (pathToDelete.isDirectory()) {
        deleteBuffer.add(convertToFolderName(pathKey));
      } else {
        deleteBuffer.add(pathKey);
      }
    }
    deleteBuffer.add(stripPrefixIfPresent(convertToFolderName(path)));
    int filesDeleted = deleteBuffer.getResult().size();
    if (filesDeleted != deleteBuffer.mEntriesAdded) {
      LOG.warn("Failed to delete directory, successfully deleted {} files out of {}.",
          filesDeleted, deleteBuffer.mEntriesAdded);
      return false;
    }
    return true;
  }

  /**
   * Objects added to a {@link DeleteBuffer} will be deleted in batches. Multiple batches are
   * processed in parallel.
   */
  @NotThreadSafe
  protected class DeleteBuffer {
    /** A list of objects in batches to be deleted in parallel. */
    private ArrayList<List<String>> mBatches;
    /** A list of the successfully deleted objects for each batch delete. */
    private ArrayList<Future<List<String>>> mBatchesResult;
    /** Buffer for a batch of objects to be deleted. */
    private List<String> mCurrentBatchBuffer;
    /** Total number of objects to be deleted across batches. */
    private int mEntriesAdded;

    /**
     * Construct a new {@link DeleteBuffer} instance.
     */
    public DeleteBuffer() {
      mBatches = new ArrayList<>();
      mBatchesResult = new ArrayList<>();
      mCurrentBatchBuffer = new ArrayList<>();
      mEntriesAdded = 0;
    }

    /**
     * Add a new object to be deleted.
     *
     * @param path of object
     * @throws IOException if a non-Alluxio error occurs
     */
    public void add(String path) throws IOException {
      // Delete batch size is same as listing length
      if (mCurrentBatchBuffer.size() == getListingChunkLength()) {
        // Batch is full
        submitBatch();
      }
      mCurrentBatchBuffer.add(path);
      mEntriesAdded++;
    }

    /**
     * Get the combined result from all batches.
     *
     * @return a list of successfully deleted objects
     * @throws IOException if a non-Alluxio error occurs
     */
    public List<String> getResult() throws IOException {
      submitBatch();
      List<String> result = new ArrayList<>();
      for (Future<List<String>> list : mBatchesResult) {
        try {
          result.addAll(list.get());
        } catch (InterruptedException e) {
          // If operation was interrupted do not add to successfully deleted list
          LOG.warn("Interrupted while waiting for the result of batch delete. UFS and Alluxio "
              + "state may be inconsistent. Error: {}", e.getMessage());
          Thread.currentThread().interrupt();
        } catch (ExecutionException e) {
          // If operation failed to execute do not add to successfully deleted list
          LOG.warn("A batch delete failed. UFS and Alluxio state may be inconsistent. Error: {}",
              e.getMessage());
        }
      }
      return result;
    }

    /**
     * Process the current batch asynchronously.
     */
    private void submitBatch() throws IOException {
      if (mCurrentBatchBuffer.size() != 0) {
        int batchNumber = mBatches.size();
        mBatches.add(new ArrayList<>(mCurrentBatchBuffer));
        mCurrentBatchBuffer.clear();
        mBatchesResult.add(batchNumber,
            mExecutorService.submit(new DeleteThread(mBatches.get(batchNumber))));
      }
    }

    /**
     * Thread class to delete a batch of objects.
     */
    @NotThreadSafe
    protected class DeleteThread implements Callable<List<String>> {

      List<String> mBatch;

      /**
       * Delete a batch of objects.
       * @param batch a list of objects to delete
       */
      public DeleteThread(List<String> batch) {
        mBatch = batch;
      }

      @Override
      public List<String> call() {
        try {
          return deleteObjects(mBatch);
        } catch (IOException e) {
          // Do not append to success list
          return Collections.emptyList();
        }
      }
    }
  }

  /**
   * Gets the block size in bytes. This method defaults to the default user block size in Alluxio.
   *
   * @param path the file name
   * @return the default Alluxio user block size
   */
  @Override
  public long getBlockSizeByte(String path) throws IOException {
    return Configuration.getBytes(PropertyKey.USER_BLOCK_SIZE_BYTES_DEFAULT);
  }

  @Override
  public UfsDirectoryStatus getDirectoryStatus(String path) throws IOException {
    if (isDirectory(path)) {
      ObjectPermissions permissions = getPermissions();
      return new UfsDirectoryStatus(path, permissions.getOwner(), permissions.getGroup(),
          permissions.getMode());
    }
    LOG.warn("Error fetching directory status, assuming directory {} does not exist", path);
    throw new FileNotFoundException(path);
  }

  // Not supported
  @Override
  @Nullable
  public List<String> getFileLocations(String path) throws IOException {
    LOG.debug("getFileLocations is not supported when using default ObjectUnderFileSystem.");
    return null;
  }

  // Not supported
  @Override
  @Nullable
  public List<String> getFileLocations(String path, FileLocationOptions options)
      throws IOException {
    LOG.debug("getFileLocations is not supported when using default ObjectUnderFileSystem.");
    return null;
  }

  // This call is currently only used for the web ui, where a negative value implies unknown.
  @Override
  public long getSpace(String path, SpaceType type) throws IOException {
    return -1;
  }

  @Override
  public UfsFileStatus getFileStatus(String path) throws IOException {
    ObjectStatus details = getObjectStatus(stripPrefixIfPresent(path));
    if (details != null) {
      ObjectPermissions permissions = getPermissions();
      return new UfsFileStatus(path, details.getContentLength(), details.getLastModifiedTimeMs(),
          permissions.getOwner(), permissions.getGroup(), permissions.getMode());
    } else {
      LOG.warn("Error fetching file status, assuming file {} does not exist", path);
      throw new FileNotFoundException(path);
    }
  }

  @Override
  public boolean isDirectory(String path) throws IOException {
    // Root is always a folder
    if (isRoot(path)) {
      return true;
    }
    String keyAsFolder = convertToFolderName(stripPrefixIfPresent(path));
    if (getObjectStatus(keyAsFolder) != null) {
      return true;
    }
    return getObjectListingChunkForPath(path, true) != null;
  }

  @Override
  public boolean isFile(String path) throws IOException {
    // Directly try to get the file metadata, if we fail it either is a folder or does not exist
    return !isRoot(path) && (getObjectStatus(stripPrefixIfPresent(path)) != null);
  }

  @Override
  public boolean isObjectStorage() {
    return true;
  }

  @Override
  public UfsStatus[] listStatus(String path) throws IOException {
    return listInternal(path, ListOptions.defaults());
  }

  @Override
  public UfsStatus[] listStatus(String path, ListOptions options)
      throws IOException {
    return listInternal(path, options);
  }

  @Override
  public boolean mkdirs(String path, MkdirsOptions options) throws IOException {
    if (path == null) {
      return false;
    }
    if (isDirectory(path)) {
      return true;
    }
    if (isFile(path)) {
      LOG.error("Cannot create directory {} because it is already a file.", path);
      return false;
    }
    if (!options.getCreateParent()) {
      if (parentExists(path)) {
        // Parent directory exists
        return mkdirsInternal(path);
      } else {
        LOG.error("Cannot create directory {} because parent does not exist", path);
        return false;
      }
    }
    // Parent directories should be created
    if (parentExists(path)) {
      // Parent directory exists
      return mkdirsInternal(path);
    } else {
      String parentKey = getParentPath(path);
      // Recursively make the parent folders
      return mkdirs(parentKey) && mkdirsInternal(path);
    }
  }

  @Override
  public InputStream open(String path, OpenOptions options) throws IOException {
    IOException thrownException = null;
    RetryPolicy retryPolicy =
        new ExponentialBackoffRetry(mOpenRetryBaseSleepMs, mOpenRetryMaxSleepMs, mOpenRetryMaxNum);
    while (retryPolicy.attemptRetry()) {
      try {
        return openObject(stripPrefixIfPresent(path), options);
      } catch (IOException e) {
        LOG.warn("{} attempt to open {} failed with exception : {}", retryPolicy.getRetryCount(),
            path, e.getMessage());
        thrownException = e;
      }
    }
    throw thrownException;
  }<commit_msg>Address comment<commit_after>/*
 * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
 * (the "License"). You may not use this work except in compliance with the License, which is
 * available at www.apache.org/licenses/LICENSE-2.0
 *
 * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
 * either express or implied, as more fully set forth in the License.
 *
 * See the NOTICE file distributed with this work for information regarding copyright ownership.
 */

package alluxio.underfs;

import alluxio.AlluxioURI;
import alluxio.Configuration;
import alluxio.PropertyKey;
import alluxio.exception.ExceptionMessage;
import alluxio.retry.ExponentialBackoffRetry;
import alluxio.retry.RetryPolicy;
import alluxio.underfs.options.CreateOptions;
import alluxio.underfs.options.DeleteOptions;
import alluxio.underfs.options.FileLocationOptions;
import alluxio.underfs.options.ListOptions;
import alluxio.underfs.options.MkdirsOptions;
import alluxio.underfs.options.OpenOptions;
import alluxio.util.CommonUtils;
import alluxio.util.executor.ExecutorServiceFactories;
import alluxio.util.io.PathUtils;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;

import javax.annotation.Nullable;
import javax.annotation.concurrent.NotThreadSafe;
import javax.annotation.concurrent.ThreadSafe;

/**
 * An object based abstract {@link UnderFileSystem}. Object Stores implementing the
 * {@link UnderFileSystem} interface should derive from this class.
 */
@ThreadSafe
public abstract class ObjectUnderFileSystem extends BaseUnderFileSystem {
  private static final Logger LOG = LoggerFactory.getLogger(ObjectUnderFileSystem.class);

  /** Default maximum length for a single listing query. */
  private static final int DEFAULT_MAX_LISTING_CHUNK_LENGTH = 1000;

  /** Value used to indicate nested structure. */
  protected static final char PATH_SEPARATOR_CHAR = '/';

  /**
   * Value used to indicate nested structure. This is a string representation of
   * {@link ObjectUnderFileSystem#PATH_SEPARATOR_CHAR}.
   */
  protected static final String PATH_SEPARATOR = String.valueOf(PATH_SEPARATOR_CHAR);

  /** Executor service used for parallel UFS operations such as bulk deletes. */
  protected ExecutorService mExecutorService;

  /**
   * Constructs an {@link ObjectUnderFileSystem}.
   *
   * @param uri the {@link AlluxioURI} used to create this ufs
   * @param ufsConf UFS configuration
   */
  protected ObjectUnderFileSystem(AlluxioURI uri, UnderFileSystemConfiguration ufsConf) {
    super(uri, ufsConf);

    int numThreads = Configuration.getInt(PropertyKey.UNDERFS_OBJECT_STORE_SERVICE_THREADS);
    mExecutorService = ExecutorServiceFactories.fixedThreadPoolExecutorServiceFactory(
        "alluxio-underfs-object-service-worker", numThreads).create();
  }

  /**
   * Information about a single object in object UFS.
   */
  protected class ObjectStatus {
    private static final long INVALID_CONTENT_LENGTH = -1L;
    private static final long INVALID_MODIFIED_TIME = -1L;

    private final long mContentLength;
    private final long mLastModifiedTimeMs;
    private final String mName;

    public ObjectStatus(String name, long contentLength, long lastModifiedTimeMs) {
      mContentLength = contentLength;
      mLastModifiedTimeMs = lastModifiedTimeMs;
      mName = name;
    }

    public ObjectStatus(String name) {
      mContentLength = INVALID_CONTENT_LENGTH;
      mLastModifiedTimeMs = INVALID_MODIFIED_TIME;
      mName = name;
    }

    /**
     * Gets the size of object contents in bytes.
     *
     * @return the content length in bytes
     */
    public long getContentLength() {
      return mContentLength;
    }

    /**
     * Gets the last modified epoch time in ms.
     *
     * @return modification time in milliseconds
     */
    public long getLastModifiedTimeMs() {
      return mLastModifiedTimeMs;
    }

    /**
     * Gets the name of the object.
     *
     * @return object name
     */
    public String getName() {
      return mName;
    }
  }

  /**
   * A chunk of listing results.
   */
  public interface ObjectListingChunk {
    /**
     * Returns objects in a pseudo-directory which may be a file or a directory.
     *
     * @return a list of object statuses
     */
    ObjectStatus[] getObjectStatuses();

    /**
     * Use common prefixes to infer pseudo-directories in object store.
     *
     * @return a list of common prefixes
     */
    String[] getCommonPrefixes();

    /**
     * Gets next chunk of object listings.
     *
     * @return null if listing did not find anything or is done, otherwise return new
     * {@link ObjectListingChunk} for the next chunk
     */
    @Nullable
    ObjectListingChunk getNextChunk() throws IOException;
  }

  /**
   * Permissions in object UFS.
   */
  public class ObjectPermissions {
    final String mOwner;
    final String mGroup;
    final short mMode;

    /**
     * Creates a new ObjectPermissions.
     *
     * @param owner the owner of the object
     * @param group the group of the object
     * @param mode the representation of the permission bits
     */
    public ObjectPermissions(String owner, String group, short mode) {
      mOwner = owner;
      mGroup = group;
      mMode = mode;
    }

    /**
     * @return the name of the owner
     */
    public String getOwner() {
      return mOwner;
    }

    /**
     * @return the name of the group
     */
    public String getGroup() {
      return mGroup;
    }

    /**
     * @return the short representation of the permission bits
     */
    public short getMode() {
      return mMode;
    }
  }

  @Override
  public void close() throws IOException {
  }

  @Override
  public void connectFromMaster(String hostname) {
  }

  @Override
  public void connectFromWorker(String hostname) {
  }

  @Override
  public OutputStream create(String path, CreateOptions options) throws IOException {
    if (options.getCreateParent() && !mkdirs(getParentPath(path))) {
      throw new IOException(ExceptionMessage.PARENT_CREATION_FAILED.getMessage(path));
    }
    return createObject(stripPrefixIfPresent(path));
  }

  @Override
  public boolean deleteFile(String path) throws IOException {
    return deleteObject(stripPrefixIfPresent(path));
  }

  @Override
  public boolean deleteDirectory(String path, DeleteOptions options) throws IOException {
    if (!options.isRecursive()) {
      UfsStatus[] children = listInternal(path, ListOptions.defaults());
      if (children == null) {
        LOG.error("Unable to delete path because {} is not a directory ", path);
        return false;
      }
      if (children.length != 0) {
        LOG.error("Unable to delete {} because it is a non empty directory. Specify "
                + "recursive as true in order to delete non empty directories.", path);
        return false;
      }
      // Delete the directory itself
      return deleteObject(stripPrefixIfPresent(convertToFolderName(path)));
    }

    // Delete children
    DeleteBuffer deleteBuffer = new DeleteBuffer();
    UfsStatus[] pathsToDelete = listInternal(path, ListOptions.defaults().setRecursive(true));
    if (pathsToDelete == null) {
      LOG.warn("Unable to delete {} because listInternal returns null", path);
      return false;
    }
    for (UfsStatus pathToDelete : pathsToDelete) {
      String pathKey = stripPrefixIfPresent(PathUtils.concatPath(path, pathToDelete.getName()));
      if (pathToDelete.isDirectory()) {
        deleteBuffer.add(convertToFolderName(pathKey));
      } else {
        deleteBuffer.add(pathKey);
      }
    }
    deleteBuffer.add(stripPrefixIfPresent(convertToFolderName(path)));
    int filesDeleted = deleteBuffer.getResult().size();
    if (filesDeleted != deleteBuffer.mEntriesAdded) {
      LOG.warn("Failed to delete directory, successfully deleted {} files out of {}.",
          filesDeleted, deleteBuffer.mEntriesAdded);
      return false;
    }
    return true;
  }

  /**
   * Objects added to a {@link DeleteBuffer} will be deleted in batches. Multiple batches are
   * processed in parallel.
   */
  @NotThreadSafe
  protected class DeleteBuffer {
    /** A list of objects in batches to be deleted in parallel. */
    private ArrayList<List<String>> mBatches;
    /** A list of the successfully deleted objects for each batch delete. */
    private ArrayList<Future<List<String>>> mBatchesResult;
    /** Buffer for a batch of objects to be deleted. */
    private List<String> mCurrentBatchBuffer;
    /** Total number of objects to be deleted across batches. */
    private int mEntriesAdded;

    /**
     * Construct a new {@link DeleteBuffer} instance.
     */
    public DeleteBuffer() {
      mBatches = new ArrayList<>();
      mBatchesResult = new ArrayList<>();
      mCurrentBatchBuffer = new ArrayList<>();
      mEntriesAdded = 0;
    }

    /**
     * Add a new object to be deleted.
     *
     * @param path of object
     * @throws IOException if a non-Alluxio error occurs
     */
    public void add(String path) throws IOException {
      // Delete batch size is same as listing length
      if (mCurrentBatchBuffer.size() == getListingChunkLength()) {
        // Batch is full
        submitBatch();
      }
      mCurrentBatchBuffer.add(path);
      mEntriesAdded++;
    }

    /**
     * Get the combined result from all batches.
     *
     * @return a list of successfully deleted objects
     * @throws IOException if a non-Alluxio error occurs
     */
    public List<String> getResult() throws IOException {
      submitBatch();
      List<String> result = new ArrayList<>();
      for (Future<List<String>> list : mBatchesResult) {
        try {
          result.addAll(list.get());
        } catch (InterruptedException e) {
          // If operation was interrupted do not add to successfully deleted list
          LOG.warn("Interrupted while waiting for the result of batch delete. UFS and Alluxio "
              + "state may be inconsistent. Error: {}", e.getMessage());
          Thread.currentThread().interrupt();
        } catch (ExecutionException e) {
          // If operation failed to execute do not add to successfully deleted list
          LOG.warn("A batch delete failed. UFS and Alluxio state may be inconsistent. Error: {}",
              e.getMessage());
        }
      }
      return result;
    }

    /**
     * Process the current batch asynchronously.
     */
    private void submitBatch() throws IOException {
      if (mCurrentBatchBuffer.size() != 0) {
        int batchNumber = mBatches.size();
        mBatches.add(new ArrayList<>(mCurrentBatchBuffer));
        mCurrentBatchBuffer.clear();
        mBatchesResult.add(batchNumber,
            mExecutorService.submit(new DeleteThread(mBatches.get(batchNumber))));
      }
    }

    /**
     * Thread class to delete a batch of objects.
     */
    @NotThreadSafe
    protected class DeleteThread implements Callable<List<String>> {

      List<String> mBatch;

      /**
       * Delete a batch of objects.
       * @param batch a list of objects to delete
       */
      public DeleteThread(List<String> batch) {
        mBatch = batch;
      }

      @Override
      public List<String> call() {
        try {
          return deleteObjects(mBatch);
        } catch (IOException e) {
          // Do not append to success list
          return Collections.emptyList();
        }
      }
    }
  }

  /**
   * Gets the block size in bytes. This method defaults to the default user block size in Alluxio.
   *
   * @param path the file name
   * @return the default Alluxio user block size
   */
  @Override
  public long getBlockSizeByte(String path) throws IOException {
    return Configuration.getBytes(PropertyKey.USER_BLOCK_SIZE_BYTES_DEFAULT);
  }

  @Override
  public UfsDirectoryStatus getDirectoryStatus(String path) throws IOException {
    if (isDirectory(path)) {
      ObjectPermissions permissions = getPermissions();
      return new UfsDirectoryStatus(path, permissions.getOwner(), permissions.getGroup(),
          permissions.getMode());
    }
    LOG.warn("Error fetching directory status, assuming directory {} does not exist", path);
    throw new FileNotFoundException(path);
  }

  // Not supported
  @Override
  @Nullable
  public List<String> getFileLocations(String path) throws IOException {
    LOG.debug("getFileLocations is not supported when using default ObjectUnderFileSystem.");
    return null;
  }

  // Not supported
  @Override
  @Nullable
  public List<String> getFileLocations(String path, FileLocationOptions options)
      throws IOException {
    LOG.debug("getFileLocations is not supported when using default ObjectUnderFileSystem.");
    return null;
  }

  // This call is currently only used for the web ui, where a negative value implies unknown.
  @Override
  public long getSpace(String path, SpaceType type) throws IOException {
    return -1;
  }

  @Override
  public UfsFileStatus getFileStatus(String path) throws IOException {
    ObjectStatus details = getObjectStatus(stripPrefixIfPresent(path));
    if (details != null) {
      ObjectPermissions permissions = getPermissions();
      return new UfsFileStatus(path, details.getContentLength(), details.getLastModifiedTimeMs(),
          permissions.getOwner(), permissions.getGroup(), permissions.getMode());
    } else {
      LOG.warn("Error fetching file status, assuming file {} does not exist", path);
      throw new FileNotFoundException(path);
    }
  }

  @Override
  public boolean isDirectory(String path) throws IOException {
    // Root is always a folder
    if (isRoot(path)) {
      return true;
    }
    String keyAsFolder = convertToFolderName(stripPrefixIfPresent(path));
    if (getObjectStatus(keyAsFolder) != null) {
      return true;
    }
    return getObjectListingChunkForPath(path, true) != null;
  }

  @Override
  public boolean isFile(String path) throws IOException {
    // Directly try to get the file metadata, if we fail it either is a folder or does not exist
    return !isRoot(path) && (getObjectStatus(stripPrefixIfPresent(path)) != null);
  }

  @Override
  public boolean isObjectStorage() {
    return true;
  }

  @Override
  public UfsStatus[] listStatus(String path) throws IOException {
    return listInternal(path, ListOptions.defaults());
  }

  @Override
  public UfsStatus[] listStatus(String path, ListOptions options)
      throws IOException {
    return listInternal(path, options);
  }

  @Override
  public boolean mkdirs(String path, MkdirsOptions options) throws IOException {
    if (path == null) {
      return false;
    }
    if (isDirectory(path)) {
      return true;
    }
    if (isFile(path)) {
      LOG.error("Cannot create directory {} because it is already a file.", path);
      return false;
    }
    if (!options.getCreateParent()) {
      if (parentExists(path)) {
        // Parent directory exists
        return mkdirsInternal(path);
      } else {
        LOG.error("Cannot create directory {} because parent does not exist", path);
        return false;
      }
    }
    // Parent directories should be created
    if (parentExists(path)) {
      // Parent directory exists
      return mkdirsInternal(path);
    } else {
      String parentKey = getParentPath(path);
      // Recursively make the parent folders
      return mkdirs(parentKey) && mkdirsInternal(path);
    }
  }

  @Override
  public InputStream open(String path, OpenOptions options) throws IOException {
    IOException thrownException = null;
    RetryPolicy retryPolicy = new ExponentialBackoffRetry(
        (int) Configuration.getMs(PropertyKey.UNDERFS_OBJECT_STORE_READ_RETRY_BASE_SLEEP_MS),
        (int) Configuration.getMs(PropertyKey.UNDERFS_OBJECT_STORE_READ_RETRY_MAX_SLEEP_MS),
        Configuration.getInt(PropertyKey.UNDERFS_OBJECT_STORE_READ_RETRY_MAX_NUM));
    while (retryPolicy.attemptRetry()) {
      try {
        return openObject(stripPrefixIfPresent(path), options);
      } catch (IOException e) {
        LOG.warn("{} attempt to open {} failed with exception : {}", retryPolicy.getRetryCount(),
            path, e.getMessage());
        thrownException = e;
      }
    }
    throw thrownException;
  }
{'fraction_non_alphanumeric': 0.08915084560021325, 'fraction_numerical': 0.0009477830761484465, 'mean_word_length': 3.096081523717093, 'pattern_counts': {'":': 0, '<': 47, '<?xml version=': 0, '>': 47, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3110051', 'n_tokens_mistral': 9171, 'n_tokens_neox': 9062, 'n_words': 2810}