/*
 * 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.hadoop.fs.s3a;

import java.nio.charset.StandardCharsets;

import org.assertj.core.api.Assertions;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.PathFilter;
import org.apache.hadoop.fs.s3a.auth.ITestRestrictedReadAccess;
import org.apache.hadoop.fs.statistics.IOStatistics;
import org.apache.hadoop.mapred.LocatedFileStatusFetcher;

import static org.apache.hadoop.fs.contract.ContractTestUtils.createFile;
import static org.apache.hadoop.fs.contract.ContractTestUtils.touch;
import static org.apache.hadoop.fs.statistics.IOStatisticAssertions.assertThatStatisticCounter;
import static org.apache.hadoop.fs.statistics.IOStatisticAssertions.extractStatistics;
import static org.apache.hadoop.fs.statistics.StoreStatisticNames.OBJECT_LIST_REQUEST;
import static org.apache.hadoop.mapreduce.lib.input.FileInputFormat.LIST_STATUS_NUM_THREADS;

/**
 * Test the LocatedFileStatusFetcher can do.
 * This is related to HADOOP-16458.
 * There's basic tests in ITestS3AFSMainOperations; this
 * is see if we can create better corner cases.
 * <p></p>
 * Much of the class is based on tests in {@link ITestRestrictedReadAccess},
 * but whereas that tests failure paths, this looks at the performance
 * of successful invocations.
 */
public class ITestLocatedFileStatusFetcher extends AbstractS3ATestBase {

  private static final Logger LOG =
      LoggerFactory.getLogger(ITestLocatedFileStatusFetcher.class);

  /** Filter to select everything. */
  private static final PathFilter EVERYTHING = t -> true;

  /** Filter to select .txt files. */
  private static final PathFilter TEXT_FILE =
      path -> path.toUri().toString().endsWith(".txt");

  /** The same path filter used in FileInputFormat. */
  private static final PathFilter HIDDEN_FILE_FILTER =
      (p) -> {
        String n = p.getName();
        return !n.startsWith("_") && !n.startsWith(".");
      };

  /**
   * Text found in LocatedFileStatusFetcher exception when the glob
   * returned "null".
   */
  private static final String DOES_NOT_EXIST = "does not exist";

  /**
   * Text found in LocatedFileStatusFetcher exception when
   * the glob returned an empty list.
   */
  private static final String MATCHES_0_FILES = "matches 0 files";

  /**
   * Text used in files.
   */
  public static final byte[] HELLO = "hello".getBytes(StandardCharsets.UTF_8);

  /**
   * How many list calls are expected in a run which collects them: {@value}.
   */
  private static final int EXPECTED_LIST_COUNT = 4;

  private Path basePath;

  private Path emptyDir;

  private Path emptyFile;

  private Path subDir;

  private Path subdirFile;

  private Path subDir2;

  private Path subdir2File1;

  private Path subdir2File2;

  private Configuration listConfig;

  @Override
  public void setup() throws Exception {
    super.setup();
    S3AFileSystem fs
        = getFileSystem();

    basePath = methodPath();

    // define the paths and create them.
    describe("Creating test directories and files");

    // an empty directory directory under the noReadDir
    emptyDir = new Path(basePath, "emptyDir");
    fs.mkdirs(emptyDir);

    // an empty file directory under the noReadDir
    emptyFile = new Path(basePath, "emptyFile.txt");
    touch(fs, emptyFile);

    // a subdirectory
    subDir = new Path(basePath, "subDir");

    // and a file in that subdirectory
    subdirFile = new Path(subDir, "subdirFile.txt");
    createFile(fs, subdirFile, true, HELLO);
    subDir2 = new Path(subDir, "subDir2");
    subdir2File1 = new Path(subDir2, "subdir2File1.txt");
    subdir2File2 = new Path(subDir2, "subdir2File2.txt");
    createFile(fs, subdir2File1, true, HELLO);
    createFile(fs, subdir2File2, true, HELLO);
    listConfig = new Configuration(getConfiguration());
  }

  /**
   * Assert that the fetcher stats logs the expected number of calls.
   * @param fetcher fetcher
   * @param expectedListCount expected number of list calls
   */
  private void assertListCount(final LocatedFileStatusFetcher fetcher,
      final int expectedListCount) {
    IOStatistics iostats = extractStatistics(fetcher);
    LOG.info("Statistics of fetcher: {}", iostats);
    assertThatStatisticCounter(iostats,
        OBJECT_LIST_REQUEST)
        .describedAs("stats of %s", iostats)
        .isEqualTo(expectedListCount);
  }

  /**
   * Run a located file status fetcher against the directory tree.
   */
  @Test
  public void testSingleThreadedLocatedFileStatus() throws Throwable {

    describe("LocatedFileStatusFetcher operations");
    // use the same filter as FileInputFormat; single thread.

    listConfig.setInt(LIST_STATUS_NUM_THREADS, 1);
    LocatedFileStatusFetcher fetcher =
        new LocatedFileStatusFetcher(
            listConfig,
            new Path[]{basePath},
            true,
            HIDDEN_FILE_FILTER,
            true);
    Iterable<FileStatus> stats = fetcher.getFileStatuses();
    Assertions.assertThat(stats)
        .describedAs("result of located scan")
        .flatExtracting(FileStatus::getPath)
        .containsExactlyInAnyOrder(
            emptyFile,
            subdirFile,
            subdir2File1,
            subdir2File2);
    assertListCount(fetcher, EXPECTED_LIST_COUNT);
  }

  /**
   * Run a located file status fetcher against the directory tree.
   */
  @Test
  public void testLocatedFileStatusFourThreads() throws Throwable {

    // four threads and the text filter.
    int threads = 4;
    describe("LocatedFileStatusFetcher with %d", threads);
    listConfig.setInt(LIST_STATUS_NUM_THREADS, threads);
    LocatedFileStatusFetcher fetcher =
        new LocatedFileStatusFetcher(
            listConfig,
            new Path[]{basePath},
            true,
            EVERYTHING,
            true);
    Iterable<FileStatus> stats = fetcher.getFileStatuses();
    IOStatistics iostats = extractStatistics(fetcher);
    LOG.info("Statistics of fetcher: {}", iostats);
    Assertions.assertThat(stats)
        .describedAs("result of located scan")
        .isNotNull()
        .flatExtracting(FileStatus::getPath)
        .containsExactlyInAnyOrder(
            emptyFile,
            subdirFile,
            subdir2File1,
            subdir2File2);
    assertListCount(fetcher, EXPECTED_LIST_COUNT);
  }

  /**
   * Run a located file status fetcher against a file.
   */
  @Test
  public void testLocatedFileStatusScanFile() throws Throwable {
    // pass in a file as the base of the scan.
    describe("LocatedFileStatusFetcher with file %s", subdirFile);
    listConfig.setInt(LIST_STATUS_NUM_THREADS, 16);
    LocatedFileStatusFetcher fetcher
        = new LocatedFileStatusFetcher(
        listConfig,
        new Path[]{subdirFile},
        true,
        TEXT_FILE,
        true);
    Iterable<FileStatus> stats = fetcher.getFileStatuses();
    Assertions.assertThat(stats)
        .describedAs("result of located scan")
        .isNotNull()
        .flatExtracting(FileStatus::getPath)
        .containsExactly(subdirFile);
    IOStatistics ioStatistics = fetcher.getIOStatistics();
    Assertions.assertThat(ioStatistics)
        .describedAs("IO statistics of %s", fetcher)
        .isNull();
  }

}
