fs.s3a.bucket.gcs-container.path.style.access
true
diff --git a/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/troubleshooting_s3a.md b/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/troubleshooting_s3a.md
index be359336463ad..c1b499e3da95d 100644
--- a/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/troubleshooting_s3a.md
+++ b/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/troubleshooting_s3a.md
@@ -132,9 +132,7 @@ was built.
This can also be caused by having more than one version of an AWS SDK
JAR on the classpath. If the full `bundle.jar` JAR is on the
-classpath, do not add any of the `aws-sdk-` JARs *except* for
-`aws-crt.jar` (which is required) and
-`eventstream.jar` which is required when using S3 Select.
+classpath, do not add any of the `aws-sdk-` JARs.
### `java.lang.NoSuchMethodError` referencing an `org.apache.hadoop` class
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/TestRequestFactory.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/TestRequestFactory.java
index 1fb576a55514c..f5e91fae2a33e 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/TestRequestFactory.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/impl/TestRequestFactory.java
@@ -178,7 +178,6 @@ private void createFactoryObjects(RequestFactory factory) throws IOException {
PutObjectOptions.keepingDirs(), -1, true));
a(factory.newPutObjectRequestBuilder(path,
PutObjectOptions.deletingDirs(), 1024, false));
- a(factory.newSelectRequestBuilder(path));
}
/**
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/AbstractS3SelectTest.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/AbstractS3SelectTest.java
deleted file mode 100644
index a3d41116182e5..0000000000000
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/AbstractS3SelectTest.java
+++ /dev/null
@@ -1,756 +0,0 @@
-/*
- * 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.select;
-
-import java.io.BufferedReader;
-import java.io.EOFException;
-import java.io.IOException;
-import java.io.InputStreamReader;
-import java.time.Duration;
-import java.time.ZonedDateTime;
-import java.time.format.DateTimeFormatter;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.NoSuchElementException;
-import java.util.Optional;
-import java.util.Scanner;
-import java.util.function.Consumer;
-
-import org.junit.Assume;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.CommonConfigurationKeys;
-import org.apache.hadoop.fs.FSDataInputStream;
-import org.apache.hadoop.fs.FileContext;
-import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.FutureDataInputStreamBuilder;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.StreamCapabilities;
-import org.apache.hadoop.fs.s3a.AWSServiceIOException;
-import org.apache.hadoop.fs.s3a.AbstractS3ATestBase;
-import org.apache.hadoop.fs.s3a.S3AFileSystem;
-import org.apache.hadoop.fs.s3a.commit.AbstractCommitITest;
-import org.apache.hadoop.io.LongWritable;
-import org.apache.hadoop.io.Text;
-import org.apache.hadoop.io.compress.PassthroughCodec;
-import org.apache.hadoop.mapred.JobConf;
-import org.apache.hadoop.mapreduce.MRJobConfig;
-import org.apache.hadoop.mapreduce.RecordReader;
-import org.apache.hadoop.mapreduce.TaskAttemptContext;
-import org.apache.hadoop.mapreduce.TaskAttemptID;
-import org.apache.hadoop.mapreduce.lib.input.FileSplit;
-import org.apache.hadoop.mapreduce.lib.input.LineRecordReader;
-import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;
-import org.apache.hadoop.util.DurationInfo;
-
-import static org.apache.hadoop.fs.s3a.Constants.STORAGE_CLASS;
-import static org.apache.hadoop.fs.s3a.S3ATestUtils.getLandsatCSVPath;
-import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
-import static org.apache.hadoop.fs.s3a.select.CsvFile.ALL_QUOTES;
-import static org.apache.hadoop.fs.s3a.select.SelectConstants.*;
-import static org.apache.hadoop.test.LambdaTestUtils.intercept;
-import static org.apache.hadoop.util.functional.FutureIO.awaitFuture;
-
-/**
- * Superclass for S3 Select tests.
- * A lot of the work here goes into creating and querying a simple CSV test
- * format, with various datatypes which can be used in type-casting queries.
- *
- * 1 "ID": index of the row
- * 2 "date": date as ISO 8601
- * 3 "timestamp": timestamp in seconds of epoch
- * 4 "name", entry-$row
- * 5 "odd", odd/even as boolean. True means odd,
- * 6 "oddint", odd/even as int : 1 for odd, 0 for even
- * 7 "oddrange": odd/even as 1 for odd, -1 for even
- *
- */
-public abstract class AbstractS3SelectTest extends AbstractS3ATestBase {
-
- /**
- * Number of columns in the CSV file: {@value}.
- */
- public static final int CSV_COLUMN_COUNT = 7;
-
- protected static final String TRUE = q("TRUE");
-
- protected static final String FALSE = q("FALSE");
-
- public static final String SELECT_EVERYTHING = "SELECT * FROM S3OBJECT s";
-
- public static final String SELECT_EVEN_ROWS_NO_HEADER =
- "SELECT * FROM S3OBJECT s WHERE s._5 = " + TRUE;
- public static final String SELECT_ODD_ROWS
- = "SELECT s.name FROM S3OBJECT s WHERE s.odd = " + TRUE;
-
- public static final String SELECT_ODD_ENTRIES
- = "SELECT * FROM S3OBJECT s WHERE s.odd = `TRUE`";
-
- public static final String SELECT_ODD_ENTRIES_BOOL
- = "SELECT * FROM S3OBJECT s WHERE CAST(s.odd AS BOOL) = TRUE";
-
- public static final String SELECT_ODD_ENTRIES_INT
- = "SELECT * FROM S3OBJECT s WHERE CAST(s.\"oddint\" AS INT) = 1";
-
- public static final String SELECT_ODD_ENTRIES_DECIMAL
- = "SELECT * FROM S3OBJECT s WHERE CAST(s.\"oddint\" AS DECIMAL) = 1";
-
- /**
- * Playing with timestamps: {@value}.
- */
- public static final String SELECT_TO_DATE
- = "SELECT\n"
- + "CAST(s.\"date\" AS TIMESTAMP)\n"
- + "FROM S3OBJECT s";
-
-
- /**
- * How many rows are being generated.
- */
- protected static final int ALL_ROWS_COUNT = 10;
-
- /**
- * Row count of all rows + header.
- */
- protected static final int ALL_ROWS_COUNT_WITH_HEADER = ALL_ROWS_COUNT + 1;
-
- /**
- * Number of odd rows expected: {@value}.
- */
- protected static final int ODD_ROWS_COUNT = ALL_ROWS_COUNT / 2;
-
- /**
- * Number of even rows expected: {@value}.
- * This is the same as the odd row count; it's separate just to
- * be consistent on tests which select even results.
- */
- protected static final int EVEN_ROWS_COUNT = ODD_ROWS_COUNT;
-
- protected static final String ENTRY_0001 = "\"entry-0001\"";
-
- protected static final String ENTRY_0002 = "\"entry-0002\"";
-
- /**
- * Path to the landsat csv.gz file.
- */
- private Path landsatGZ;
-
- /**
- * The filesystem with the landsat data.
- */
- private S3AFileSystem landsatFS;
-
-
- // A random task attempt id for testing.
- private String attempt0;
-
- private TaskAttemptID taskAttempt0;
-
- private String jobId;
-
- /**
- * Base CSV file is headers.
- *
- * 1 "ID": index of the row
- * 2 "date": date as Date.toString
- * 3 "timestamp": timestamp in seconds of epoch
- * 4 "name", entry-$row
- * 5 "odd", odd/even as boolean
- * 6 "oddint", odd/even as int : 1 for odd, 0 for even
- * 7 "oddrange": odd/even as 1 for odd, -1 for even
- *
- * @param fs filesystem
- * @param path path to write
- * @param header should the standard header be printed?
- * @param quoteHeaderPolicy what the header quote policy is.
- * @param quoteRowPolicy what the row quote policy is.
- * @param rows number of rows
- * @param separator column separator
- * @param eol end of line characters
- * @param quote quote char
- * @param footer callback to run after the main CSV file is written
- * @throws IOException IO failure.
- */
- public static void createStandardCsvFile(
- final FileSystem fs,
- final Path path,
- final boolean header,
- final long quoteHeaderPolicy,
- final long quoteRowPolicy,
- final int rows,
- final String separator,
- final String eol,
- final String quote,
- final Consumer footer) throws IOException {
- try (CsvFile csv = new CsvFile(fs,
- path,
- true,
- separator,
- eol,
- quote)) {
-
- if (header) {
- writeStandardHeader(csv, quoteHeaderPolicy);
- }
- DateTimeFormatter formatter
- = DateTimeFormatter.ISO_OFFSET_DATE_TIME;
- ZonedDateTime timestamp = ZonedDateTime.now();
- Duration duration = Duration.ofHours(20);
- // loop is at 1 for use in counters and flags
- for (int i = 1; i <= rows; i++) {
- // flip the odd flags
- boolean odd = (i & 1) == 1;
- // and move the timestamp back
- timestamp = timestamp.minus(duration);
- csv.row(quoteRowPolicy,
- i,
- timestamp.format(formatter),
- timestamp.toEpochSecond(),
- String.format("entry-%04d", i),
- odd ? "TRUE" : "FALSE",
- odd ? 1 : 0,
- odd ? 1 : -1
- );
- }
- // write the footer
- footer.accept(csv);
- }
- }
-
- /**
- * Write out the standard header to a CSV file.
- * @param csv CSV file to use.
- * @param quoteHeaderPolicy quote policy.
- * @return the input file.
- * @throws IOException failure to write.
- */
- private static CsvFile writeStandardHeader(final CsvFile csv,
- final long quoteHeaderPolicy) throws IOException {
- return csv.row(quoteHeaderPolicy,
- "id",
- "date",
- "timestamp",
- "name",
- "odd",
- "oddint",
- "oddrange");
- }
-
- /**
- * Verify that an exception has a specific error code.
- * if not: an assertion is raised containing the original value.
- * @param code expected code.
- * @param ex exception caught
- * @throws AssertionError on a mismatch
- */
- protected static AWSServiceIOException verifyErrorCode(final String code,
- final AWSServiceIOException ex) {
- logIntercepted(ex);
- if (!code.equals(ex.awsErrorDetails().errorCode())) {
- throw new AssertionError("Expected Error code" + code
- + " actual " + ex.awsErrorDetails().errorCode(),
- ex);
- }
- return ex;
- }
-
- /**
- * Probe for a filesystem instance supporting S3 Select.
- * @param filesystem filesystem
- * @return true iff the filesystem supports S3 Select.
- */
- boolean isSelectAvailable(final FileSystem filesystem) {
- return filesystem instanceof StreamCapabilities
- && ((StreamCapabilities) filesystem)
- .hasCapability(S3_SELECT_CAPABILITY);
- }
-
- @Override
- protected Configuration createConfiguration() {
- Configuration conf = super.createConfiguration();
- removeBaseAndBucketOverrides(conf, STORAGE_CLASS);
-
- return conf;
- }
-
- /**
- * Setup: requires select to be available.
- */
- @Override
- public void setup() throws Exception {
- super.setup();
- Assume.assumeTrue("S3 Select is not enabled on "
- + getFileSystem().getUri(),
- isSelectAvailable(getFileSystem()));
- Configuration conf = getConfiguration();
- landsatGZ = getLandsatCSVPath(conf);
- landsatFS = (S3AFileSystem) landsatGZ.getFileSystem(conf);
- Assume.assumeTrue("S3 Select is not enabled on " + landsatFS.getUri(),
- isSelectAvailable(landsatFS));
- // create some job info
- jobId = AbstractCommitITest.randomJobId();
- attempt0 = "attempt_" + jobId + "_m_000000_0";
- taskAttempt0 = TaskAttemptID.forName(attempt0);
- }
-
- /**
- * Build the SQL statement, using String.Format rules.
- * @param template template
- * @param args arguments for the template
- * @return the template to use
- */
- protected static String sql(
- final String template,
- final Object... args) {
- return args.length > 0 ? String.format(template, args) : template;
- }
-
- /**
- * Quote a constant with the SQL quote logic.
- * @param c constant
- * @return quoted constant
- */
- protected static String q(String c) {
- return '\'' + c + '\'';
- }
-
- /**
- * Select from a source file.
- * @param fileSystem FS.
- * @param source source file.
- * @param conf config for the select call.
- * @param sql template for a formatted SQL request.
- * @param args arguments for the formatted request.
- * @return the input stream.
- * @throws IOException failure
- */
- protected FSDataInputStream select(
- final FileSystem fileSystem,
- final Path source,
- final Configuration conf,
- final String sql,
- final Object... args)
- throws IOException {
- String expression = sql(sql, args);
- describe("Execution Select call: %s", expression);
- FutureDataInputStreamBuilder builder =
- fileSystem.openFile(source)
- .must(SELECT_SQL, expression);
- // propagate all known options
- for (String key : InternalSelectConstants.SELECT_OPTIONS) {
- String value = conf.get(key);
- if (value != null) {
- builder.must(key, value);
- }
- }
- return awaitFuture(builder.build());
- }
-
- /**
- * Select from a source file via the file context API.
- * @param fc file context
- * @param source source file.
- * @param conf config for the select call.
- * @param sql template for a formatted SQL request.
- * @param args arguments for the formatted request.
- * @return the input stream.
- * @throws IOException failure
- */
- protected FSDataInputStream select(
- final FileContext fc,
- final Path source,
- final Configuration conf,
- final String sql,
- final Object... args)
- throws IOException {
- String expression = sql(sql, args);
- describe("Execution Select call: %s", expression);
- FutureDataInputStreamBuilder builder = fc.openFile(source)
- .must(SELECT_SQL, expression);
- // propagate all known options
- InternalSelectConstants.SELECT_OPTIONS.forEach((key) ->
- Optional.ofNullable(conf.get(key))
- .map((v) -> builder.must(key, v)));
- return awaitFuture(builder.build());
- }
-
- /**
- * Parse a selection to lines; log at info.
- * @param selection selection input
- * @return a list of lines.
- * @throws IOException if raised during the read.
- */
- protected List parseToLines(final FSDataInputStream selection)
- throws IOException {
- return parseToLines(selection, getMaxLines());
- }
-
- /**
- * Enable the passthrough codec for a job, with the given extension.
- * @param conf configuration to update
- * @param extension extension to use
- */
- protected void enablePassthroughCodec(final Configuration conf,
- final String extension) {
- conf.set(CommonConfigurationKeys.IO_COMPRESSION_CODECS_KEY,
- PassthroughCodec.CLASSNAME);
- conf.set(PassthroughCodec.OPT_EXTENSION, extension);
- }
-
- /**
- * Override if a test suite is likely to ever return more lines.
- * @return the max number for parseToLines/1
- */
- protected int getMaxLines() {
- return 100;
- }
-
- /**
- * Parse a selection to lines; log at info.
- * @param selection selection input
- * @param maxLines maximum number of lines.
- * @return a list of lines.
- * @throws IOException if raised during the read.
- */
- protected List parseToLines(final FSDataInputStream selection,
- int maxLines)
- throws IOException {
- List result = new ArrayList<>();
- String stats;
- // the scanner assumes that any IOE => EOF; we don't want
- // that and so will check afterwards.
- try (Scanner scanner = new Scanner(
- new BufferedReader(new InputStreamReader(selection)))) {
- scanner.useDelimiter(CSV_INPUT_RECORD_DELIMITER_DEFAULT);
- while (maxLines > 0) {
- try {
- String l = scanner.nextLine();
- LOG.info("{}", l);
- result.add(l);
- maxLines--;
- } catch (NoSuchElementException e) {
- // EOL or an error
- break;
- }
- }
- stats = selection.toString();
- describe("Result line count: %s\nStatistics\n%s",
- result.size(), stats);
- // look for any raised error.
- IOException ioe = scanner.ioException();
- if (ioe != null && !(ioe instanceof EOFException)) {
- throw ioe;
- }
- }
- return result;
- }
-
- /**
- * Verify the selection count; return the original list.
- * If there's a mismatch, the whole list is logged at error, then
- * an assertion raised.
- * @param expected expected value.
- * @param expression expression -for error messages.
- * @param selection selected result.
- * @return the input list.
- */
- protected List verifySelectionCount(
- final int expected,
- final String expression,
- final List selection) {
- return verifySelectionCount(expected, expected, expression, selection);
- }
-
- /**
- * Verify the selection count is within a given range;
- * return the original list.
- * If there's a mismatch, the whole list is logged at error, then
- * an assertion raised.
- * @param min min value (exclusive).
- * @param max max value (exclusive). If -1: no maximum.
- * @param expression expression -for error messages.
- * @param selection selected result.
- * @return the input list.
- */
- protected List verifySelectionCount(
- final int min,
- final int max,
- final String expression,
- final List selection) {
- int size = selection.size();
- if (size < min || (max > -1 && size > max)) {
- // mismatch: log and then fail
- String listing = prepareToPrint(selection);
- LOG.error("\n{} => \n{}", expression, listing);
- fail("row count from select call " + expression
- + " is out of range " + min + " to " + max
- + ": " + size
- + " \n" + listing);
- }
- return selection;
- }
-
- /**
- * Do whatever is needed to prepare a string for logging.
- * @param selection selection
- * @return something printable.
- */
- protected String prepareToPrint(final List selection) {
- return String.join("\n", selection);
- }
-
- /**
- * Create "the standard" CSV file with the default row count.
- * @param fs filesystem
- * @param path path to write
- * @param quoteRowPolicy what the row quote policy is.
- * @throws IOException IO failure.
- */
- protected void createStandardCsvFile(
- final FileSystem fs,
- final Path path,
- final long quoteRowPolicy)
- throws IOException {
- createStandardCsvFile(
- fs, path,
- true,
- ALL_QUOTES,
- quoteRowPolicy,
- ALL_ROWS_COUNT,
- ",",
- "\n",
- "\"",
- c -> {});
- }
-
- /**
- * Set an MR Job input option.
- * @param conf configuration
- * @param key key to set
- * @param val value
- */
- void inputOpt(Configuration conf, String key, String val) {
- conf.set(MRJobConfig.INPUT_FILE_OPTION_PREFIX + key, val);
- }
-
- /**
- * Set a mandatory MR Job input option.
- * @param conf configuration
- * @param key key to set
- * @param val value
- */
- void inputMust(Configuration conf, String key, String val) {
- conf.set(MRJobConfig.INPUT_FILE_MANDATORY_PREFIX + key,
- val);
- }
-
- /**
- * Reads lines through a v2 RecordReader, as if it were part of a
- * MRv2 job.
- * @param conf job conf
- * @param path path to query
- * @param sql sql to add to the configuration.
- * @param initialCapacity capacity of the read
- * @param reader reader: this is closed after the read
- * @return the selected lines.
- * @throws Exception failure
- */
- protected List readRecords(JobConf conf,
- Path path,
- String sql,
- RecordReader reader,
- int initialCapacity) throws Exception {
-
- inputMust(conf, SELECT_SQL, sql);
- List lines = new ArrayList<>(initialCapacity);
- try {
- reader.initialize(
- createSplit(conf, path),
- createTaskAttemptContext(conf));
- while (reader.nextKeyValue()) {
- lines.add(reader.getCurrentValue().toString());
- }
- } finally {
- reader.close();
- }
- return lines;
- }
- /**
- * Reads lines through a v1 RecordReader, as if it were part of a
- * MRv1 job.
- * @param conf job conf
- * @param reader reader: this is closed after the read
- * @param initialCapacity capacity of the read
- * @return the selected lines.
- * @throws Exception failure
- */
- protected List readRecordsV1(JobConf conf,
- org.apache.hadoop.mapred.RecordReader reader,
- K key,
- V value,
- int initialCapacity) throws Exception {
- List lines = new ArrayList<>(initialCapacity);
- try {
- while (reader.next(key, value)) {
- lines.add(value.toString());
- }
- } finally {
- reader.close();
- }
- return lines;
- }
-
- /**
- * Create a task attempt context for a job, creating a random JobID to
- * do this.
- * @param conf job configuration.
- * @return a new task attempt context containing the job conf
- * @throws Exception failure.
- */
- protected TaskAttemptContext createTaskAttemptContext(final JobConf conf)
- throws Exception {
- String id = AbstractCommitITest.randomJobId();
- return new TaskAttemptContextImpl(conf,
- TaskAttemptID.forName("attempt_" + id + "_m_000000_0"));
- }
-
- /**
- * Create an MRv2 file input split.
- * @param conf job configuration
- * @param path path to file
- * @return the split
- * @throws IOException problems reading the file.
- */
- protected FileSplit createSplit(final JobConf conf, final Path path)
- throws IOException {
- FileSystem fs = path.getFileSystem(conf);
- FileStatus status = fs.getFileStatus(path);
- return new FileSplit(path, 0, status.getLen(),
- new String[]{"localhost"});
- }
-
- /**
- * Create an MRv1 file input split.
- * @param conf job configuration
- * @param path path to file
- * @return the split
- * @throws IOException problems reading the file.
- */
- protected org.apache.hadoop.mapred.FileSplit
- createSplitV1(final JobConf conf, final Path path)
- throws IOException {
- FileSystem fs = path.getFileSystem(conf);
- FileStatus status = fs.getFileStatus(path);
- return new org.apache.hadoop.mapred.FileSplit(path, 0, status.getLen(),
- new String[]{"localhost"});
- }
-
- /**
- * Create a v2 line record reader expecting newlines as the EOL marker.
- * @return a reader
- */
- protected RecordReader createLineRecordReader() {
- return new LineRecordReader(new byte[]{'\n'});
- }
-
- /**
- * Create a v1 line record reader.
- * @return a reader
- */
- protected org.apache.hadoop.mapred.RecordReader
- createLineRecordReaderV1(
- final JobConf conf,
- final Path path) throws IOException {
- return new org.apache.hadoop.mapred.LineRecordReader(
- conf, createSplitV1(conf, path));
- }
-
- /**
- * Get the path to the landsat file.
- * @return the landsat CSV.GZ path.
- */
- protected Path getLandsatGZ() {
- return landsatGZ;
- }
-
- /**
- * Get the filesystem for the landsat file.
- * @return the landsat FS.
- */
- protected S3AFileSystem getLandsatFS() {
- return landsatFS;
- }
-
- /**
- * Perform a seek: log duration of the operation.
- * @param stream stream to seek.
- * @param target target position.
- * @throws IOException on an error
- */
- protected void seek(final FSDataInputStream stream, final long target)
- throws IOException {
- try(DurationInfo ignored =
- new DurationInfo(LOG, "Seek to %d", target)) {
- stream.seek(target);
- }
- }
-
- /**
- * Execute a seek so far past the EOF that it will be rejected.
- * If the seek did not fail, the exception raised includes the toString()
- * value of the stream.
- * @param seekStream stream to seek in.
- * @param newpos new position
- * @return the EOF Exception raised.
- * @throws Exception any other exception.
- */
- protected EOFException expectSeekEOF(final FSDataInputStream seekStream,
- final int newpos) throws Exception {
- return intercept(EOFException.class,
- () -> {
- seek(seekStream, newpos);
- // return this for the test failure reports.
- return "Stream after seek to " + newpos + ": " + seekStream;
- });
- }
-
- public String getAttempt0() {
- return attempt0;
- }
-
- public TaskAttemptID getTaskAttempt0() {
- return taskAttempt0;
- }
-
- public String getJobId() {
- return jobId;
- }
-
- /**
- * Logs intercepted exceptions.
- * This generates the stack traces for the documentation.
- * @param ex exception
- * @return the exception passed in (for chaining)
- */
- protected static T logIntercepted(T ex) {
- LOG.info("Intercepted Exception is ", ex);
- return ex;
- }
-}
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/ITestS3Select.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/ITestS3Select.java
deleted file mode 100644
index cdb28c81b17d2..0000000000000
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/ITestS3Select.java
+++ /dev/null
@@ -1,981 +0,0 @@
-/*
- * 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.select;
-
-import java.io.BufferedReader;
-import java.io.FileNotFoundException;
-import java.io.IOException;
-import java.io.InputStreamReader;
-import java.util.List;
-import java.util.concurrent.CompletableFuture;
-
-import org.junit.Assume;
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.commons.io.IOUtils;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FSDataInputStream;
-import org.apache.hadoop.fs.FSExceptionMessages;
-import org.apache.hadoop.fs.FileContext;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.FutureDataInputStreamBuilder;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.PathIOException;
-import org.apache.hadoop.fs.Seekable;
-import org.apache.hadoop.fs.contract.ContractTestUtils;
-import org.apache.hadoop.fs.impl.AbstractFSBuilderImpl;
-import org.apache.hadoop.fs.s3a.AWSBadRequestException;
-import org.apache.hadoop.fs.s3a.AWSServiceIOException;
-import org.apache.hadoop.fs.s3a.Constants;
-import org.apache.hadoop.fs.s3a.S3AFileSystem;
-import org.apache.hadoop.fs.s3a.S3AInputStream;
-import org.apache.hadoop.fs.s3a.S3ATestUtils;
-import org.apache.hadoop.fs.s3a.Statistic;
-import org.apache.hadoop.fs.s3a.statistics.S3AInputStreamStatistics;
-import org.apache.hadoop.io.LongWritable;
-import org.apache.hadoop.io.Text;
-import org.apache.hadoop.mapred.JobConf;
-import org.apache.hadoop.mapreduce.InputSplit;
-import org.apache.hadoop.mapreduce.Job;
-import org.apache.hadoop.mapreduce.JobContext;
-import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
-import org.apache.hadoop.mapreduce.lib.input.LineRecordReader;
-import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
-import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
-import org.apache.hadoop.mapreduce.task.JobContextImpl;
-import org.apache.hadoop.util.DurationInfo;
-
-import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY;
-import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY_DEFAULT;
-import static org.apache.hadoop.fs.s3a.Constants.READAHEAD_RANGE;
-import static org.apache.hadoop.fs.s3a.S3ATestUtils.skipIfS3ExpressBucket;
-import static org.apache.hadoop.fs.s3a.select.CsvFile.ALL_QUOTES;
-import static org.apache.hadoop.fs.s3a.select.SelectBinding.expandBackslashChars;
-import static org.apache.hadoop.fs.s3a.select.SelectConstants.*;
-import static org.apache.hadoop.test.LambdaTestUtils.intercept;
-import static org.apache.hadoop.test.LambdaTestUtils.interceptFuture;
-import static org.hamcrest.CoreMatchers.hasItem;
-import static org.hamcrest.CoreMatchers.not;
-import static org.hamcrest.collection.IsCollectionWithSize.hasSize;
-
-/**
- * Test the S3 Select feature with some basic SQL Commands.
- * Executed if the destination store declares its support for the feature.
- */
-public class ITestS3Select extends AbstractS3SelectTest {
-
- private static final Logger LOG =
- LoggerFactory.getLogger(ITestS3Select.class);
-
- public static final String E_CAST_FAILED = "CastFailed";
-
- public static final String E_PARSE_INVALID_PATH_COMPONENT
- = "ParseInvalidPathComponent";
-
- public static final String E_INVALID_TABLE_ALIAS = "InvalidTableAlias";
-
- private Configuration selectConf;
-
- /** well formed CSV. */
- private Path csvPath;
-
- /** CSV file with fewer columns than expected, all fields parse badly. */
- private Path brokenCSV;
-
- @Override
- protected Configuration createConfiguration() {
- final Configuration conf = super.createConfiguration();
- skipIfS3ExpressBucket(conf);
- return conf;
- }
-
- @Override
- public void setup() throws Exception {
- super.setup();
- csvPath = path(getMethodName() + ".csv");
- Assume.assumeTrue("S3 Select is not enabled",
- getFileSystem().hasPathCapability(csvPath, S3_SELECT_CAPABILITY));
- selectConf = new Configuration(false);
- selectConf.setBoolean(SELECT_ERRORS_INCLUDE_SQL, true);
- createStandardCsvFile(getFileSystem(), csvPath, ALL_QUOTES);
- // create the broken CSV file.
- brokenCSV = path("testParseBrokenCSVFile");
- createStandardCsvFile(
- getFileSystem(), brokenCSV,
- true,
- ALL_QUOTES,
- ALL_ROWS_COUNT,
- ALL_ROWS_COUNT,
- ",",
- "\n",
- "\"",
- csv -> csv
- .line("# comment")
- .row(ALL_QUOTES, "bad", "Tuesday", 0, "entry-bad", "yes", false));
- }
-
- @Override
- public void teardown() throws Exception {
- describe("teardown");
- try {
- if (csvPath != null) {
- getFileSystem().delete(csvPath, false);
- }
- if (brokenCSV != null) {
- getFileSystem().delete(brokenCSV, false);
- }
- } finally {
- super.teardown();
- }
- }
-
- @Test
- public void testCapabilityProbe() throws Throwable {
-
- // this should always hold true if we get past test setup
- assertTrue("Select is not available on " + getFileSystem(),
- isSelectAvailable(getFileSystem()));
- }
-
- @SuppressWarnings("NestedAssignment")
- @Test
- public void testReadWholeFileClassicAPI() throws Throwable {
- describe("create and read the whole file. Verifies setup working");
- int lines;
- try (BufferedReader reader = new BufferedReader(
- new InputStreamReader(
- getFileSystem().open(csvPath)))) {
- lines = 0;
- // seek to 0, which is what some input formats do
- String line;
- while ((line = reader.readLine()) != null) {
- lines++;
- LOG.info("{}", line);
- }
- }
- assertEquals("line count", ALL_ROWS_COUNT_WITH_HEADER, lines);
- }
-
- @Test
- public void testSelectWholeFileNoHeader() throws Throwable {
- describe("Select the entire file, expect all rows but the header");
- expectSelected(
- ALL_ROWS_COUNT,
- selectConf,
- CSV_HEADER_OPT_USE,
- "SELECT * FROM S3OBJECT");
- }
-
- @Test
- public void testSelectFirstColumnNoHeader() throws Throwable {
- describe("Select the entire file, expect all rows but the header");
- expectSelected(
- ALL_ROWS_COUNT_WITH_HEADER,
- selectConf,
- CSV_HEADER_OPT_NONE,
- "SELECT s._1 FROM S3OBJECT s");
- }
-
- @Test
- public void testSelectSelfNoHeader() throws Throwable {
- describe("Select the entire file, expect all rows but the header");
- expectSelected(
- ALL_ROWS_COUNT_WITH_HEADER,
- selectConf,
- CSV_HEADER_OPT_NONE,
- "SELECT s._1 FROM S3OBJECT s WHERE s._1 = s._1");
- }
-
- @Test
- public void testSelectSelfUseHeader() throws Throwable {
- describe("Select the entire file, expect all rows including the header");
- expectSelected(
- ALL_ROWS_COUNT,
- selectConf,
- CSV_HEADER_OPT_USE,
- "SELECT s.id FROM S3OBJECT s WHERE s.id = s.id");
- }
-
- @Test
- public void testSelectID2UseHeader() throws Throwable {
- describe("Select where ID=2; use the header");
- expectSelected(
- 1,
- selectConf,
- CSV_HEADER_OPT_USE,
- "SELECT s.id FROM S3OBJECT s WHERE s.id = '2'");
- }
-
- @Test
- public void testSelectNoMatchingID() throws Throwable {
- describe("Select where there is no match; expect nothing back");
- expectSelected(
- 0,
- selectConf,
- CSV_HEADER_OPT_USE,
- "SELECT s.id FROM S3OBJECT s WHERE s.id = '0x8000'");
- }
-
- @Test
- public void testSelectId1() throws Throwable {
- describe("Select the first element in the file");
- expectSelected(
- 1,
- selectConf,
- CSV_HEADER_OPT_NONE,
- "SELECT * FROM S3OBJECT s WHERE s._1 = '1'",
- TRUE);
- }
-
- @Test
- public void testSelectEmptySQL() throws Throwable {
- describe("An empty SQL statement fails fast");
- FutureDataInputStreamBuilder builder = getFileSystem().openFile(
- csvPath)
- .must(SELECT_SQL, "");
- interceptFuture(IllegalArgumentException.class,
- SELECT_SQL,
- builder.build());
- }
-
- @Test
- public void testSelectEmptyFile() throws Throwable {
- describe("Select everything from an empty file");
- Path path = path("testSelectEmptyFile");
- S3AFileSystem fs = getFileSystem();
- ContractTestUtils.touch(fs, path);
- parseToLines(fs.openFile(path)
- .must(SELECT_SQL, SELECT_EVERYTHING)
- .withFileStatus(fs.getFileStatus(path))
- .build()
- .get(),
- 0);
- }
-
- @Test
- public void testSelectEmptyFileWithConditions() throws Throwable {
- describe("Select everything from an empty file with a more complex SQL");
- Path path = path("testSelectEmptyFileWithConditions");
- S3AFileSystem fs = getFileSystem();
- ContractTestUtils.touch(fs, path);
- String sql = "SELECT * FROM S3OBJECT s WHERE s._1 = `TRUE`";
- CompletableFuture future = fs.openFile(path)
- .must(SELECT_SQL, sql).build();
- assertEquals("Not at the end of the file", -1, future.get().read());
- }
-
- @Test
- public void testSelectSeek() throws Throwable {
- describe("Verify forward seeks work, not others");
-
- // start: read in the full data through the initial select
- // this makes asserting that contents match possible
- Path path = csvPath;
- S3AFileSystem fs = getFileSystem();
- int len = (int) fs.getFileStatus(path).getLen();
- byte[] fullData = new byte[len];
- int actualLen;
- try (DurationInfo ignored =
- new DurationInfo(LOG, "Initial read of %s", path);
- FSDataInputStream sourceStream =
- select(fs, path,
- selectConf,
- SELECT_EVERYTHING)) {
- // read it in
- actualLen = IOUtils.read(sourceStream, fullData);
- }
- int seekRange = 20;
-
- try (FSDataInputStream seekStream =
- select(fs, path,
- selectConf,
- SELECT_EVERYTHING)) {
- SelectInputStream sis
- = (SelectInputStream) seekStream.getWrappedStream();
- S3AInputStreamStatistics streamStats =
- sis.getS3AStreamStatistics();
- // lazy seek doesn't raise a problem here
- seekStream.seek(0);
- assertEquals("first byte read", fullData[0], seekStream.read());
-
- // and now the pos has moved, again, seek will be OK
- seekStream.seek(1);
- seekStream.seek(1);
- // but trying to seek elsewhere now fails
- PathIOException ex = intercept(PathIOException.class,
- SelectInputStream.SEEK_UNSUPPORTED,
- () -> seekStream.seek(0));
- LOG.info("Seek error is as expected", ex);
- // positioned reads from the current location work.
- byte[] buffer = new byte[1];
- long pos = seekStream.getPos();
- seekStream.readFully(pos, buffer);
- // but positioned backwards fail.
- intercept(PathIOException.class,
- SelectInputStream.SEEK_UNSUPPORTED,
- () -> seekStream.readFully(0, buffer));
- // the position has now moved on.
- assertPosition(seekStream, pos + 1);
- // so a seek to the old pos will fail
- intercept(PathIOException.class,
- SelectInputStream.SEEK_UNSUPPORTED,
- () -> seekStream.readFully(pos, buffer));
-
- // set the readahead to the default.
- // This verifies it reverts to the default.
- seekStream.setReadahead(null);
- assertEquals("Readahead in ",
- Constants.DEFAULT_READAHEAD_RANGE, sis.getReadahead());
- // forward seeks are implemented as 1+ skip
- long target = seekStream.getPos() + seekRange;
- seek(seekStream, target);
- assertPosition(seekStream, target);
- // now do a read and compare values
- assertEquals("byte at seek position",
- fullData[(int)seekStream.getPos()], seekStream.read());
- assertEquals("Seek bytes skipped in " + streamStats,
- seekRange, streamStats.getBytesSkippedOnSeek());
-
- // try an invalid readahead range
- intercept(IllegalArgumentException.class,
- S3AInputStream.E_NEGATIVE_READAHEAD_VALUE,
- () -> seekStream.setReadahead(-1L));
-
- // do a slightly forward offset read
- int read = seekStream.read(seekStream.getPos() + 2, buffer, 0, 1);
- assertEquals(1, read);
-
- // final fun: seek way past the EOF
- logIntercepted(expectSeekEOF(seekStream, actualLen * 2));
- assertPosition(seekStream, actualLen);
- assertEquals(-1, seekStream.read());
- LOG.info("Seek statistics {}", streamStats);
- // this will return no, but not fail
- assertFalse("Failed to seek to new source in " + seekStream,
- seekStream.seekToNewSource(0));
- // and set the readahead to 0 to see that close path works
- seekStream.setReadahead(0L);
- // then do a manual close even though there's one in the try resource.
- // which will verify that a double close is harmless
- seekStream.close();
- LOG.info("Final stream state {}", sis);
- }
- }
-
- /**
- * Assert that a stream is in a specific position.
- * @param stream stream or other seekable.
- * @param pos expected position.
- * @throws IOException failure of the getPos() call.
- * @throws AssertionError mismatch between expected and actual.
- */
- private void assertPosition(Seekable stream, long pos)
- throws IOException {
- assertEquals("Wrong stream position in " + stream,
- pos, stream.getPos());
- }
-
- @Test
- public void testSelectOddLinesNoHeader() throws Throwable {
- describe("Select odd lines, ignoring the header");
- expectSelected(
- ODD_ROWS_COUNT,
- selectConf,
- CSV_HEADER_OPT_IGNORE,
- "SELECT * FROM S3OBJECT s WHERE s._5 = `TRUE`");
- // and do a quick check on the instrumentation
- long bytesRead = getFileSystem().getInstrumentation()
- .getCounterValue(Statistic.STREAM_READ_BYTES);
- assertNotEquals("No bytes read count in filesystem instrumentation counter",
- 0, bytesRead);
- }
-
- @Test
- public void testSelectOddLinesHeader() throws Throwable {
- describe("Select the odd values");
- List selected = expectSelected(
- ODD_ROWS_COUNT,
- selectConf,
- CSV_HEADER_OPT_USE,
- SELECT_ODD_ROWS);
- // the list includes odd values
- assertThat(selected, hasItem(ENTRY_0001));
- // but not the evens
- assertThat(selected, not(hasItem(ENTRY_0002)));
- }
-
- @Test
- public void testSelectOddLinesHeaderTSVOutput() throws Throwable {
- describe("Select the odd values with tab spaced output");
- selectConf.set(CSV_OUTPUT_FIELD_DELIMITER, "\t");
- selectConf.set(CSV_OUTPUT_QUOTE_CHARACTER, "'");
- selectConf.set(CSV_OUTPUT_QUOTE_FIELDS,
- CSV_OUTPUT_QUOTE_FIELDS_AS_NEEEDED);
- selectConf.set(CSV_OUTPUT_RECORD_DELIMITER, "\r");
- List selected = expectSelected(
- ODD_ROWS_COUNT,
- selectConf,
- CSV_HEADER_OPT_USE,
- SELECT_ODD_ENTRIES_BOOL);
- // the list includes odd values
- String row1 = selected.get(0);
-
- // split that first line into columns: This is why TSV is better for code
- // to work with than CSV
- String[] columns = row1.split("\t", -1);
- assertEquals("Wrong column count from tab split line <" + row1 + ">",
- CSV_COLUMN_COUNT, columns.length);
- assertEquals("Wrong column value from tab split line <" + row1 + ">",
- "entry-0001", columns[3]);
- }
-
- @Test
- public void testSelectNotOperationHeader() throws Throwable {
- describe("Select the even values with a NOT call; quote the header name");
- List selected = expectSelected(
- EVEN_ROWS_COUNT,
- selectConf,
- CSV_HEADER_OPT_USE,
- "SELECT s.name FROM S3OBJECT s WHERE NOT s.\"odd\" = %s",
- TRUE);
- // the list includes no odd values
- assertThat(selected, not(hasItem(ENTRY_0001)));
- // but has the evens
- assertThat(selected, hasItem(ENTRY_0002));
- }
-
- @Test
- public void testBackslashExpansion() throws Throwable {
- assertEquals("\t\r\n", expandBackslashChars("\t\r\n"));
- assertEquals("\t", expandBackslashChars("\\t"));
- assertEquals("\r", expandBackslashChars("\\r"));
- assertEquals("\r \n", expandBackslashChars("\\r \\n"));
- assertEquals("\\", expandBackslashChars("\\\\"));
- }
-
- /**
- * This is an expanded example for the documentation.
- * Also helps catch out unplanned changes to the configuration strings.
- */
- @Test
- public void testSelectFileExample() throws Throwable {
- describe("Select the entire file, expect all rows but the header");
- int len = (int) getFileSystem().getFileStatus(csvPath).getLen();
- FutureDataInputStreamBuilder builder =
- getFileSystem().openFile(csvPath)
- .must("fs.s3a.select.sql",
- SELECT_ODD_ENTRIES)
- .must("fs.s3a.select.input.format", "CSV")
- .must("fs.s3a.select.input.compression", "NONE")
- .must("fs.s3a.select.input.csv.header", "use")
- .must("fs.s3a.select.output.format", "CSV");
-
- CompletableFuture future = builder.build();
- try (FSDataInputStream select = future.get()) {
- // process the output
- byte[] bytes = new byte[len];
- int actual = select.read(bytes);
- LOG.info("file length is {}; length of selected data is {}",
- len, actual);
- }
- }
-
- /**
- * This is an expanded example for the documentation.
- * Also helps catch out unplanned changes to the configuration strings.
- */
- @Test
- public void testSelectUnsupportedInputFormat() throws Throwable {
- describe("Request an Unsupported input format");
- FutureDataInputStreamBuilder builder = getFileSystem().openFile(csvPath)
- .must(SELECT_SQL, SELECT_ODD_ENTRIES)
- .must(SELECT_INPUT_FORMAT, "pptx");
- interceptFuture(IllegalArgumentException.class,
- "pptx",
- builder.build());
- }
-
- /**
- * Ask for an invalid output format.
- */
- @Test
- public void testSelectUnsupportedOutputFormat() throws Throwable {
- describe("Request a (currently) Unsupported output format");
- FutureDataInputStreamBuilder builder = getFileSystem().openFile(csvPath)
- .must(SELECT_SQL, SELECT_ODD_ENTRIES)
- .must(SELECT_INPUT_FORMAT, "csv")
- .must(SELECT_OUTPUT_FORMAT, "json");
- interceptFuture(IllegalArgumentException.class,
- "json",
- builder.build());
- }
-
- /**
- * Missing files fail lazy.
- */
- @Test
- public void testSelectMissingFile() throws Throwable {
-
- describe("Select a missing file, expect it to surface in the future");
-
- Path missing = path("missing");
-
- FutureDataInputStreamBuilder builder =
- getFileSystem().openFile(missing)
- .must(SELECT_SQL, SELECT_ODD_ENTRIES);
-
- interceptFuture(FileNotFoundException.class,
- "", builder.build());
- }
-
- @Test
- public void testSelectDirectoryFails() throws Throwable {
- describe("Verify that secondary select options are only valid on select"
- + " queries");
- S3AFileSystem fs = getFileSystem();
- Path dir = path("dir");
- // this will be an empty dir marker
- fs.mkdirs(dir);
-
- FutureDataInputStreamBuilder builder =
- getFileSystem().openFile(dir)
- .must(SELECT_SQL, SELECT_ODD_ENTRIES);
- interceptFuture(FileNotFoundException.class,
- "", builder.build());
-
- // try the parent
- builder = getFileSystem().openFile(dir.getParent())
- .must(SELECT_SQL,
- SELECT_ODD_ENTRIES);
- interceptFuture(FileNotFoundException.class,
- "", builder.build());
- }
-
- @Test
- public void testSelectRootFails() throws Throwable {
- describe("verify root dir selection is rejected");
- FutureDataInputStreamBuilder builder =
- getFileSystem().openFile(path("/"))
- .must(SELECT_SQL, SELECT_ODD_ENTRIES);
- interceptFuture(IOException.class,
- "", builder.build());
- }
-
- /**
- * Validate the abort logic.
- */
- @Test
- public void testCloseWithAbort() throws Throwable {
- describe("Close the stream with the readahead outstanding");
- S3ATestUtils.MetricDiff readOps = new S3ATestUtils.MetricDiff(
- getFileSystem(),
- Statistic.STREAM_READ_OPERATIONS_INCOMPLETE);
- selectConf.setInt(READAHEAD_RANGE, 2);
-
- FSDataInputStream stream = select(getFileSystem(), csvPath, selectConf,
- "SELECT * FROM S3OBJECT s");
- SelectInputStream sis = (SelectInputStream) stream.getWrappedStream();
- assertEquals("Readahead on " + sis, 2, sis.getReadahead());
- stream.setReadahead(1L);
- assertEquals("Readahead on " + sis, 1, sis.getReadahead());
- stream.read();
- S3AInputStreamStatistics stats
- = (S3AInputStreamStatistics)
- sis.getS3AStreamStatistics();
- assertEquals("Read count in " + sis,
- 1, stats.getBytesRead());
- stream.close();
- assertEquals("Abort count in " + sis,
- 1, stats.getAborted());
- readOps.assertDiffEquals("Read operations are still considered active",
- 0);
- intercept(PathIOException.class, FSExceptionMessages.STREAM_IS_CLOSED,
- () -> stream.read());
- }
-
- @Test
- public void testCloseWithNoAbort() throws Throwable {
- describe("Close the stream with the readahead outstandingV");
- FSDataInputStream stream = select(getFileSystem(), csvPath, selectConf,
- "SELECT * FROM S3OBJECT s");
- stream.setReadahead(0x1000L);
- SelectInputStream sis = (SelectInputStream) stream.getWrappedStream();
- S3AInputStreamStatistics stats
- = (S3AInputStreamStatistics)
- sis.getS3AStreamStatistics();
- stream.close();
- assertEquals("Close count in " + sis, 1, stats.getClosed());
- assertEquals("Abort count in " + sis, 0, stats.getAborted());
- assertTrue("No bytes read in close of " + sis,
- stats.getBytesReadInClose() > 0);
- }
-
- @Test
- public void testFileContextIntegration() throws Throwable {
- describe("Test that select works through FileContext");
- FileContext fc = S3ATestUtils.createTestFileContext(getConfiguration());
- selectConf.set(CSV_INPUT_HEADER, CSV_HEADER_OPT_USE);
-
- List selected =
- verifySelectionCount(ODD_ROWS_COUNT, SELECT_ODD_ENTRIES_INT,
- parseToLines(
- select(fc, csvPath, selectConf, SELECT_ODD_ROWS)));
- // the list includes odd values
- assertThat(selected, hasItem(ENTRY_0001));
- // but not the evens
- assertThat(selected, not(hasItem(ENTRY_0002)));
- }
-
- @Test
- public void testSelectOptionsOnlyOnSelectCalls() throws Throwable {
- describe("Secondary select options are only valid on select"
- + " queries");
- String key = CSV_INPUT_HEADER;
- intercept(IllegalArgumentException.class, key,
- () -> getFileSystem().openFile(csvPath)
- .must(key, CSV_HEADER_OPT_USE).build());
- }
-
- @Test
- public void testSelectMustBeEnabled() throws Throwable {
- describe("Verify that the FS must have S3 select enabled.");
- Configuration conf = new Configuration(getFileSystem().getConf());
- conf.setBoolean(FS_S3A_SELECT_ENABLED, false);
- try (FileSystem fs2 = FileSystem.newInstance(csvPath.toUri(), conf)) {
- intercept(UnsupportedOperationException.class,
- SELECT_UNSUPPORTED,
- () -> {
- assertFalse("S3 Select Capability must be disabled on " + fs2,
- isSelectAvailable(fs2));
- return fs2.openFile(csvPath)
- .must(SELECT_SQL, SELECT_ODD_ROWS)
- .build();
- });
- }
- }
-
- @Test
- public void testSelectOptionsRejectedOnNormalOpen() throws Throwable {
- describe("Verify that a normal open fails on select must() options");
- intercept(IllegalArgumentException.class,
- AbstractFSBuilderImpl.UNKNOWN_MANDATORY_KEY,
- () -> getFileSystem().openFile(csvPath)
- .must(CSV_INPUT_HEADER, CSV_HEADER_OPT_USE)
- .build());
- }
-
- @Test
- public void testSelectOddRecordsWithHeader()
- throws Throwable {
- describe("work through a record reader");
- JobConf conf = createJobConf();
- inputMust(conf, CSV_INPUT_HEADER, CSV_HEADER_OPT_USE);
- expectRecordsRead(ODD_ROWS_COUNT, conf, SELECT_ODD_ENTRIES_DECIMAL);
- }
-
- @Test
- public void testSelectDatestampsConverted()
- throws Throwable {
- describe("timestamp conversion in record IIO");
- JobConf conf = createJobConf();
- inputMust(conf, CSV_INPUT_HEADER, CSV_HEADER_OPT_USE);
- inputMust(conf, CSV_OUTPUT_QUOTE_FIELDS,
- CSV_OUTPUT_QUOTE_FIELDS_AS_NEEEDED);
- String sql = SELECT_TO_DATE;
- List records = expectRecordsRead(ALL_ROWS_COUNT, conf, sql);
- LOG.info("Result of {}\n{}", sql, prepareToPrint(records));
- }
-
- @Test
- public void testSelectNoMatch()
- throws Throwable {
- describe("when there's no match to a query, 0 records are returned,");
- JobConf conf = createJobConf();
- inputMust(conf, CSV_INPUT_HEADER, CSV_HEADER_OPT_USE);
- expectRecordsRead(0, conf,
- "SELECT * FROM S3OBJECT s WHERE s.odd = " + q("maybe"));
- }
-
- @Test
- public void testSelectOddRecordsIgnoreHeader()
- throws Throwable {
- describe("work through a record reader");
- JobConf conf = createJobConf();
- inputOpt(conf, CSV_INPUT_HEADER, CSV_HEADER_OPT_NONE);
- inputMust(conf, CSV_INPUT_HEADER, CSV_HEADER_OPT_IGNORE);
- expectRecordsRead(EVEN_ROWS_COUNT, conf,
- SELECT_EVEN_ROWS_NO_HEADER);
- }
-
- @Test
- public void testSelectRecordsUnknownMustOpt()
- throws Throwable {
- describe("verify reader key validation is remapped");
- JobConf conf = createJobConf();
- inputOpt(conf, CSV_INPUT_HEADER, CSV_HEADER_OPT_NONE);
- inputMust(conf, CSV_INPUT_HEADER + ".something", CSV_HEADER_OPT_IGNORE);
- intercept(IllegalArgumentException.class,
- AbstractFSBuilderImpl.UNKNOWN_MANDATORY_KEY,
- () -> readRecords(conf, SELECT_EVEN_ROWS_NO_HEADER));
- }
-
- @Test
- public void testSelectOddRecordsWithHeaderV1()
- throws Throwable {
- describe("work through a V1 record reader");
- JobConf conf = createJobConf();
- inputMust(conf, CSV_INPUT_HEADER, CSV_HEADER_OPT_USE);
- // using a double backslash here makes the string "\t" which will then
- // be parsed in the SelectBinding code as it if had come in on from an XML
- // entry
- inputMust(conf, CSV_OUTPUT_FIELD_DELIMITER, "\\t");
- inputMust(conf, CSV_OUTPUT_QUOTE_CHARACTER, "'");
- inputMust(conf, CSV_OUTPUT_QUOTE_FIELDS,
- CSV_OUTPUT_QUOTE_FIELDS_AS_NEEEDED);
- inputMust(conf, CSV_OUTPUT_RECORD_DELIMITER, "\n");
- verifySelectionCount(ODD_ROWS_COUNT,
- SELECT_ODD_ROWS,
- readRecordsV1(conf, SELECT_ODD_ROWS));
- }
-
- /**
- * Create a job conf for line reader tests.
- * This patches the job with the passthrough codec for
- * CSV files.
- * @return a job configuration
- */
- private JobConf createJobConf() {
- JobConf conf = new JobConf(getConfiguration());
- enablePassthroughCodec(conf, ".csv");
- return conf;
- }
-
- @Test
- public void testSelectOddRecordsIgnoreHeaderV1()
- throws Throwable {
- describe("work through a V1 record reader");
- JobConf conf = createJobConf();
- inputOpt(conf, CSV_INPUT_HEADER, CSV_HEADER_OPT_NONE);
- inputMust(conf, CSV_INPUT_HEADER, CSV_HEADER_OPT_IGNORE);
- inputMust(conf, FS_OPTION_OPENFILE_READ_POLICY,
- FS_OPTION_OPENFILE_READ_POLICY_DEFAULT);
- inputMust(conf, SELECT_ERRORS_INCLUDE_SQL, "true");
- verifySelectionCount(EVEN_ROWS_COUNT,
- SELECT_EVEN_ROWS_NO_HEADER,
- readRecordsV1(conf, SELECT_EVEN_ROWS_NO_HEADER));
- }
-
- protected List expectRecordsRead(final int expected,
- final JobConf conf,
- final String sql) throws Exception {
- return verifySelectionCount(expected, sql, readRecords(conf, sql));
- }
-
- /**
- * Reads lines through {@link LineRecordReader}, as if it were an MR
- * job.
- * @param conf jpb conf
- * @param sql sql to add to the configuration.
- * @return the selected lines.
- * @throws Exception failure
- */
- private List readRecords(JobConf conf, String sql) throws Exception {
- return readRecords(conf,
- csvPath,
- sql,
- createLineRecordReader(),
- ALL_ROWS_COUNT_WITH_HEADER);
- }
-
- /**
- * Reads lines through a v1 LineRecordReader}.
- * @param conf jpb conf
- * @param sql sql to add to the configuration.
- * @return the selected lines.
- * @throws Exception failure
- */
- private List readRecordsV1(JobConf conf, String sql)
- throws Exception {
- inputMust(conf, SELECT_SQL, sql);
- return super.readRecordsV1(conf,
- createLineRecordReaderV1(conf, csvPath),
- new LongWritable(),
- new Text(),
- ALL_ROWS_COUNT_WITH_HEADER);
- }
-
- /**
- * Issue a select call, expect the specific number of rows back.
- * Error text will include the SQL.
- * @param expected expected row count.
- * @param conf config for the select call.
- * @param header header option
- * @param sql template for a formatted SQL request.
- * @param args arguments for the formatted request.
- * @return the lines selected
- * @throws IOException failure
- */
- private List expectSelected(
- final int expected,
- final Configuration conf,
- final String header,
- final String sql,
- final Object...args) throws Exception {
- conf.set(CSV_INPUT_HEADER, header);
- return verifySelectionCount(expected, sql(sql, args),
- selectCsvFile(conf, sql, args));
- }
-
- /**
- * Select from the CSV file.
- * @param conf config for the select call.
- * @param sql template for a formatted SQL request.
- * @param args arguments for the formatted request.
- * @return the lines selected
- * @throws IOException failure
- */
- private List selectCsvFile(
- final Configuration conf,
- final String sql,
- final Object...args)
- throws Exception {
-
- return parseToLines(
- select(getFileSystem(), csvPath, conf, sql, args));
- }
-
- @Test
- public void testCommentsSkipped() throws Throwable {
- describe("Verify that comments are skipped");
- selectConf.set(CSV_INPUT_HEADER, CSV_HEADER_OPT_USE);
-
- List lines = verifySelectionCount(
- ALL_ROWS_COUNT_WITH_HEADER,
- "select s.id",
- parseToLines(
- select(getFileSystem(), brokenCSV, selectConf,
- "SELECT * FROM S3OBJECT s")));
- LOG.info("\n{}", prepareToPrint(lines));
- }
-
- @Test
- public void testEmptyColumnsRegenerated() throws Throwable {
- describe("if you ask for a column but your row doesn't have it,"
- + " an empty column is inserted");
- selectConf.set(CSV_INPUT_HEADER, CSV_HEADER_OPT_USE);
-
- List lines = verifySelectionCount(
- ALL_ROWS_COUNT_WITH_HEADER, "select s.oddrange",
- parseToLines(
- select(getFileSystem(), brokenCSV, selectConf,
- "SELECT s.oddrange FROM S3OBJECT s")));
- LOG.info("\n{}", prepareToPrint(lines));
- assertEquals("Final oddrange column is not regenerated empty",
- "\"\"", lines.get(lines.size() - 1));
- }
-
- @Test
- public void testIntCastFailure() throws Throwable {
- describe("Verify that int casts fail");
- expectSelectFailure(E_CAST_FAILED, SELECT_ODD_ENTRIES_INT);
-
- }
-
- @Test
- public void testSelectToDateParseFailure() throws Throwable {
- describe("Verify date parsing failure");
- expectSelectFailure(E_CAST_FAILED, SELECT_TO_DATE);
- }
-
- @Test
- public void testParseInvalidPathComponent() throws Throwable {
- describe("Verify bad SQL parseing");
- expectSelectFailure(E_PARSE_INVALID_PATH_COMPONENT,
- "SELECT * FROM S3OBJECT WHERE s.'oddf' = true");
- }
-
- @Test
- public void testSelectInvalidTableAlias() throws Throwable {
- describe("select with unknown column name");
- expectSelectFailure(E_INVALID_TABLE_ALIAS,
- "SELECT * FROM S3OBJECT WHERE s.\"oddf\" = 'true'");
- }
-
- @Test
- public void testSelectGeneratedAliases() throws Throwable {
- describe("select with a ._2 column when headers are enabled");
- expectSelectFailure(E_INVALID_TABLE_ALIAS,
- "SELECT * FROM S3OBJECT WHERE s._2 = 'true'");
- }
-
- /**
- * Expect select against the broken CSV file to fail with a specific
- * AWS exception error code.
- * If the is no failure, the results are included in the assertion raised.
- * @param expectedErrorCode error code in getErrorCode()
- * @param sql SQL to invoke
- * @return the exception, if it is as expected.
- * @throws Exception any other failure
- * @throws AssertionError when an exception is raised, but its error code
- * is different, or when no exception was raised.
- */
- protected AWSServiceIOException expectSelectFailure(
- String expectedErrorCode,
- String sql)
- throws Exception {
- selectConf.set(CSV_INPUT_HEADER, CSV_HEADER_OPT_USE);
- return verifyErrorCode(expectedErrorCode,
- intercept(AWSBadRequestException.class,
- () ->
- prepareToPrint(
- parseToLines(
- select(getFileSystem(), brokenCSV, selectConf, sql)
- ))));
-
- }
-
-
- @Test
- public void testInputSplit()
- throws Throwable {
- describe("Verify that only a single file is used for splits");
- JobConf conf = new JobConf(getConfiguration());
-
-
- inputMust(conf, CSV_INPUT_HEADER, CSV_HEADER_OPT_USE);
- final Path input = csvPath;
- S3AFileSystem fs = getFileSystem();
- final Path output = path("testLandsatSelect")
- .makeQualified(fs.getUri(), fs.getWorkingDirectory());
- conf.set(FileInputFormat.INPUT_DIR, input.toString());
- conf.set(FileOutputFormat.OUTDIR, output.toString());
-
- final Job job = Job.getInstance(conf, "testInputSplit");
- JobContext jobCtx = new JobContextImpl(job.getConfiguration(),
- getTaskAttempt0().getJobID());
-
- TextInputFormat tif = new TextInputFormat();
- List splits = tif.getSplits(jobCtx);
- assertThat("split count wrong", splits, hasSize(1));
-
- }
-
-}
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/ITestS3SelectCLI.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/ITestS3SelectCLI.java
deleted file mode 100644
index 2dd813bf40a91..0000000000000
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/ITestS3SelectCLI.java
+++ /dev/null
@@ -1,357 +0,0 @@
-/*
- * 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.select;
-
-import java.io.ByteArrayOutputStream;
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.nio.charset.StandardCharsets;
-import java.util.List;
-
-import org.junit.Assume;
-import org.junit.Test;
-
-import org.apache.commons.io.IOUtils;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.s3a.S3AFileSystem;
-import org.apache.hadoop.fs.s3a.S3ATestUtils;
-import org.apache.hadoop.fs.s3a.Statistic;
-import org.apache.hadoop.fs.s3a.impl.ChangeDetectionPolicy;
-import org.apache.hadoop.fs.s3a.impl.ChangeDetectionPolicy.Source;
-import org.apache.hadoop.fs.s3a.s3guard.S3GuardTool;
-import org.apache.hadoop.util.ExitUtil;
-import org.apache.hadoop.util.OperationDuration;
-import org.apache.hadoop.util.ToolRunner;
-
-import static org.apache.hadoop.util.Preconditions.checkNotNull;
-import static org.apache.hadoop.fs.s3a.S3ATestUtils.disableFilesystemCaching;
-import static org.apache.hadoop.fs.s3a.s3guard.S3GuardToolTestHelper.exec;
-import static org.apache.hadoop.fs.s3a.select.ITestS3SelectLandsat.SELECT_NOTHING;
-import static org.apache.hadoop.fs.s3a.select.ITestS3SelectLandsat.SELECT_SUNNY_ROWS_NO_LIMIT;
-import static org.apache.hadoop.fs.s3a.select.SelectConstants.*;
-import static org.apache.hadoop.fs.s3a.select.SelectTool.*;
-import static org.apache.hadoop.service.launcher.LauncherExitCodes.EXIT_COMMAND_ARGUMENT_ERROR;
-import static org.apache.hadoop.service.launcher.LauncherExitCodes.EXIT_NOT_FOUND;
-import static org.apache.hadoop.service.launcher.LauncherExitCodes.EXIT_SERVICE_UNAVAILABLE;
-import static org.apache.hadoop.service.launcher.LauncherExitCodes.EXIT_SUCCESS;
-import static org.apache.hadoop.service.launcher.LauncherExitCodes.EXIT_USAGE;
-import static org.apache.hadoop.test.LambdaTestUtils.intercept;
-
-/**
- * Test the S3 Select CLI through some operations against landsat
- * and files generated from it.
- */
-public class ITestS3SelectCLI extends AbstractS3SelectTest {
-
- public static final int LINE_COUNT = 100;
-
- public static final String SELECT_EVERYTHING = "SELECT * FROM S3OBJECT s";
-
- private SelectTool selectTool;
-
- private Configuration selectConf;
-
- public static final String D = "-D";
-
- private File localFile;
-
- private String landsatSrc;
-
- @Override
- public void setup() throws Exception {
- super.setup();
- selectTool = new SelectTool(getConfiguration());
- selectConf = new Configuration(getConfiguration());
- localFile = getTempFilename();
- landsatSrc = getLandsatGZ().toString();
- final S3AFileSystem landsatFS = getLandsatFS();
- ChangeDetectionPolicy changeDetectionPolicy =
- landsatFS.getChangeDetectionPolicy();
- Assume.assumeFalse("the standard landsat bucket doesn't have versioning",
- changeDetectionPolicy.getSource() == Source.VersionId
- && changeDetectionPolicy.isRequireVersion());
- Assume.assumeTrue("S3 Select is not enabled",
- landsatFS.hasPathCapability(new Path("/"), S3_SELECT_CAPABILITY));
- }
-
- @Override
- public void teardown() throws Exception {
- super.teardown();
- if (localFile != null) {
- localFile.delete();
- }
- }
-
- /**
- * Expect a command to succeed.
- * @param message any extra text to include in the assertion error message
- * @param tool tool to run
- * @param args arguments to the command
- * @return the output of any successful run
- * @throws Exception failure
- */
- protected static String expectSuccess(
- String message,
- S3GuardTool tool,
- String... args) throws Exception {
- ByteArrayOutputStream buf = new ByteArrayOutputStream();
- exec(EXIT_SUCCESS, message, tool, buf, args);
- return buf.toString();
- }
-
- /**
- * Run a S3GuardTool command from a varags list and the
- * configuration returned by {@code getConfiguration()}.
- * @param conf config to use
- * @param args argument list
- * @return the return code
- * @throws Exception any exception
- */
- protected int run(Configuration conf, S3GuardTool tool,
- String... args) throws Exception {
- return ToolRunner.run(conf, tool, args);
- }
-
- /**
- * Run a S3GuardTool command from a varags list, catch any raised
- * ExitException and verify the status code matches that expected.
- * @param status expected status code of the exception
- * @param conf config to use
- * @param args argument list
- * @throws Exception any exception
- */
- protected void runToFailure(int status, Configuration conf,
- String message,
- S3GuardTool tool, String... args)
- throws Exception {
- final ExitUtil.ExitException ex =
- intercept(ExitUtil.ExitException.class, message,
- () -> ToolRunner.run(conf, tool, args));
- if (ex.status != status) {
- throw ex;
- }
-
- }
-
- @Test
- public void testLandsatToFile() throws Throwable {
- describe("select part of the landsat to a file");
- int lineCount = LINE_COUNT;
- S3AFileSystem landsatFS =
- (S3AFileSystem) getLandsatGZ().getFileSystem(getConfiguration());
- S3ATestUtils.MetricDiff selectCount = new S3ATestUtils.MetricDiff(landsatFS,
- Statistic.OBJECT_SELECT_REQUESTS);
-
- run(selectConf, selectTool,
- D, v(CSV_OUTPUT_QUOTE_CHARACTER, "'"),
- D, v(CSV_OUTPUT_QUOTE_FIELDS, CSV_OUTPUT_QUOTE_FIELDS_AS_NEEEDED),
- "select",
- o(OPT_HEADER), CSV_HEADER_OPT_USE,
- o(OPT_COMPRESSION), COMPRESSION_OPT_GZIP,
- o(OPT_LIMIT), Integer.toString(lineCount),
- o(OPT_OUTPUT), localFile.toString(),
- landsatSrc,
- SELECT_SUNNY_ROWS_NO_LIMIT);
- List lines = IOUtils.readLines(new FileInputStream(localFile), StandardCharsets.UTF_8);
- LOG.info("Result from select:\n{}", lines.get(0));
- assertEquals(lineCount, lines.size());
- selectCount.assertDiffEquals("select count", 1);
- OperationDuration duration = selectTool.getSelectDuration();
- assertTrue("Select duration was not measured",
- duration.value() > 0);
- }
-
- private File getTempFilename() throws IOException {
- File dest = File.createTempFile("landat", ".csv");
- dest.delete();
- return dest;
- }
-
- @Test
- public void testLandsatToConsole() throws Throwable {
- describe("select part of the landsat to the console");
- // this verifies the input stream was actually closed
- S3ATestUtils.MetricDiff readOps = new S3ATestUtils.MetricDiff(
- getFileSystem(),
- Statistic.STREAM_READ_OPERATIONS_INCOMPLETE);
- run(selectConf, selectTool,
- D, v(CSV_OUTPUT_QUOTE_CHARACTER, "'"),
- D, v(CSV_OUTPUT_QUOTE_FIELDS, CSV_OUTPUT_QUOTE_FIELDS_ALWAYS),
- "select",
- o(OPT_HEADER), CSV_HEADER_OPT_USE,
- o(OPT_COMPRESSION), COMPRESSION_OPT_GZIP,
- o(OPT_LIMIT), Integer.toString(LINE_COUNT),
- landsatSrc,
- SELECT_SUNNY_ROWS_NO_LIMIT);
- assertEquals("Lines read and printed to console",
- LINE_COUNT, selectTool.getLinesRead());
- readOps.assertDiffEquals("Read operations are still considered active",
- 0); }
-
- @Test
- public void testSelectNothing() throws Throwable {
- describe("an empty select is not an error");
- run(selectConf, selectTool,
- "select",
- o(OPT_HEADER), CSV_HEADER_OPT_USE,
- o(OPT_COMPRESSION), COMPRESSION_OPT_GZIP,
- o(OPT_INPUTFORMAT), "csv",
- o(OPT_OUTPUTFORMAT), "csv",
- o(OPT_EXPECTED), "0",
- o(OPT_LIMIT), Integer.toString(LINE_COUNT),
- landsatSrc,
- SELECT_NOTHING);
- assertEquals("Lines read and printed to console",
- 0, selectTool.getLinesRead());
- }
-
- @Test
- public void testLandsatToRemoteFile() throws Throwable {
- describe("select part of the landsat to a file");
- Path dest = path("testLandsatToRemoteFile.csv");
- run(selectConf, selectTool,
- D, v(CSV_OUTPUT_QUOTE_CHARACTER, "'"),
- D, v(CSV_OUTPUT_QUOTE_FIELDS, CSV_OUTPUT_QUOTE_FIELDS_ALWAYS),
- "select",
- o(OPT_HEADER), CSV_HEADER_OPT_USE,
- o(OPT_COMPRESSION), COMPRESSION_OPT_GZIP,
- o(OPT_LIMIT), Integer.toString(LINE_COUNT),
- o(OPT_OUTPUT), dest.toString(),
- landsatSrc,
- SELECT_SUNNY_ROWS_NO_LIMIT);
- FileStatus status = getFileSystem().getFileStatus(dest);
- assertEquals(
- "Mismatch between bytes selected and file len in " + status,
- selectTool.getBytesRead(), status.getLen());
- assertIsFile(dest);
-
- // now select on that
- Configuration conf = getConfiguration();
- SelectTool tool2 = new SelectTool(conf);
- run(conf, tool2,
- "select",
- o(OPT_HEADER), CSV_HEADER_OPT_NONE,
- dest.toString(),
- SELECT_EVERYTHING);
- }
-
- @Test
- public void testUsage() throws Throwable {
- runToFailure(EXIT_USAGE, getConfiguration(), TOO_FEW_ARGUMENTS,
- selectTool, "select");
- }
-
- @Test
- public void testRejectionOfNonS3FS() throws Throwable {
- File dest = getTempFilename();
- runToFailure(EXIT_SERVICE_UNAVAILABLE,
- getConfiguration(),
- WRONG_FILESYSTEM,
- selectTool, "select", dest.toString(),
- SELECT_EVERYTHING);
- }
-
- @Test
- public void testFailMissingFile() throws Throwable {
- Path dest = path("testFailMissingFile.csv");
- runToFailure(EXIT_NOT_FOUND,
- getConfiguration(),
- "",
- selectTool, "select", dest.toString(),
- SELECT_EVERYTHING);
- }
-
- @Test
- public void testS3SelectDisabled() throws Throwable {
- Configuration conf = getConfiguration();
- conf.setBoolean(FS_S3A_SELECT_ENABLED, false);
- disableFilesystemCaching(conf);
- runToFailure(EXIT_SERVICE_UNAVAILABLE,
- conf,
- SELECT_IS_DISABLED,
- selectTool, "select",
- o(OPT_HEADER), CSV_HEADER_OPT_USE,
- o(OPT_COMPRESSION), COMPRESSION_OPT_GZIP,
- o(OPT_LIMIT), Integer.toString(LINE_COUNT),
- landsatSrc,
- SELECT_SUNNY_ROWS_NO_LIMIT);
- }
-
- @Test
- public void testSelectBadLimit() throws Throwable {
- runToFailure(EXIT_USAGE,
- getConfiguration(),
- "",
- selectTool, "select",
- o(OPT_HEADER), CSV_HEADER_OPT_USE,
- o(OPT_COMPRESSION), COMPRESSION_OPT_GZIP,
- o(OPT_LIMIT), "-1",
- landsatSrc,
- SELECT_NOTHING);
- }
-
- @Test
- public void testSelectBadInputFormat() throws Throwable {
- runToFailure(EXIT_COMMAND_ARGUMENT_ERROR,
- getConfiguration(),
- "",
- selectTool, "select",
- o(OPT_HEADER), CSV_HEADER_OPT_USE,
- o(OPT_INPUTFORMAT), "pptx",
- o(OPT_COMPRESSION), COMPRESSION_OPT_GZIP,
- landsatSrc,
- SELECT_NOTHING);
- }
-
- @Test
- public void testSelectBadOutputFormat() throws Throwable {
- runToFailure(EXIT_COMMAND_ARGUMENT_ERROR,
- getConfiguration(),
- "",
- selectTool, "select",
- o(OPT_HEADER), CSV_HEADER_OPT_USE,
- o(OPT_OUTPUTFORMAT), "pptx",
- o(OPT_COMPRESSION), COMPRESSION_OPT_GZIP,
- landsatSrc,
- SELECT_NOTHING);
- }
-
- /**
- * Take an option and add the "-" prefix.
- * @param in input option
- * @return value for the tool args list.
- */
- private static String o(String in) {
- return "-" + in;
- }
-
- /**
- * Create the key=value bit of the -D key=value pair.
- * @param key key to set
- * @param value value to use
- * @return a string for the tool args list.
- */
- private static String v(String key, String value) {
- return checkNotNull(key) + "=" + checkNotNull(value);
- }
-
-}
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/ITestS3SelectLandsat.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/ITestS3SelectLandsat.java
deleted file mode 100644
index 4d4af822ee50b..0000000000000
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/ITestS3SelectLandsat.java
+++ /dev/null
@@ -1,435 +0,0 @@
-/*
- * 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.select;
-
-import java.io.IOException;
-import java.util.List;
-
-import org.junit.Assume;
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.commons.io.IOUtils;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.CommonConfigurationKeys;
-import org.apache.hadoop.fs.FSDataInputStream;
-import org.apache.hadoop.fs.FileContext;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.PathIOException;
-import org.apache.hadoop.fs.s3a.impl.ChangeDetectionPolicy;
-import org.apache.hadoop.fs.s3a.impl.ChangeDetectionPolicy.Source;
-import org.apache.hadoop.fs.s3a.statistics.S3AInputStreamStatistics;
-import org.apache.hadoop.fs.s3a.S3AFileSystem;
-import org.apache.hadoop.fs.s3a.S3ATestUtils;
-import org.apache.hadoop.fs.s3a.Statistic;
-import org.apache.hadoop.mapred.JobConf;
-import org.apache.hadoop.util.DurationInfo;
-
-import static org.apache.hadoop.fs.s3a.S3ATestUtils.assume;
-import static org.apache.hadoop.fs.s3a.S3ATestUtils.getTestPropertyBool;
-import static org.apache.hadoop.fs.s3a.scale.S3AScaleTestBase._1KB;
-import static org.apache.hadoop.fs.s3a.scale.S3AScaleTestBase._1MB;
-import static org.apache.hadoop.fs.s3a.select.SelectConstants.*;
-import static org.apache.hadoop.test.LambdaTestUtils.intercept;
-import static org.hamcrest.CoreMatchers.containsString;
-import static org.hamcrest.CoreMatchers.not;
-
-/**
- * Test the S3 Select feature with the Landsat dataset.
- *
- * This helps explore larger datasets, compression and the like.
- *
- * This suite is only executed if the destination store declares its support for
- * the feature and the test CSV file configuration option points to the
- * standard landsat GZip file. That's because these tests require the specific
- * format of the landsat file.
- *
- * Normally working with the landsat file is a scale test.
- * Here, because of the select operations, there's a lot less data
- * to download.
- * For this to work: write aggressive select calls: filtering, using LIMIT
- * and projecting down to a few columns.
- *
- * For the structure, see
- * Landsat on AWS
- *
- *
- * entityId: String LC80101172015002LGN00
- * acquisitionDate: String 2015-01-02 15:49:05.571384
- * cloudCover: Float (possibly -ve) 80.81
- * processingLevel: String L1GT
- * path: Int 10
- * row: Int 117
- * min_lat: Float -79.09923
- * min_lon: Float -139.66082
- * max_lat: Float -77.7544
- * max_lon: Float 125.09297
- * download_url: HTTPS URL https://s3-us-west-2.amazonaws.com/landsat-pds/L8/010/117/LC80101172015002LGN00/index.html
- *
- * Ranges
- *
- * - Latitude should range in -180 <= lat <= 180
- * - Longitude in 0 <= lon <= 360
- * - Standard Greenwich Meridian (not the french one which still surfaces)
- * - Cloud cover Should be 0-100, but there are some negative ones.
- *
- *
- * Head of the file:
- *
- entityId,acquisitionDate,cloudCover,processingLevel,path,row,min_lat,min_lon,max_lat,max_lon,download_url
- * LC80101172015002LGN00,2015-01-02 15:49:05.571384,80.81,L1GT,10,117,-79.09923,-139.66082,-77.7544,-125.09297,https://s3-us-west-2.amazonaws.com/landsat-pds/L8/010/117/LC80101172015002LGN00/index.html
- * LC80260392015002LGN00,2015-01-02 16:56:51.399666,90.84,L1GT,26,39,29.23106,-97.48576,31.36421,-95.16029,https://s3-us-west-2.amazonaws.com/landsat-pds/L8/026/039/LC80260392015002LGN00/index.html
- * LC82270742015002LGN00,2015-01-02 13:53:02.047000,83.44,L1GT,227,74,-21.28598,-59.27736,-19.17398,-57.07423,https://s3-us-west-2.amazonaws.com/landsat-pds/L8/227/074/LC82270742015002LGN00/index.html
- * LC82270732015002LGN00,2015-01-02 13:52:38.110317,52.29,L1T,227,73,-19.84365,-58.93258,-17.73324,-56.74692,https://s3-us-west-2.amazonaws.com/landsat-pds/L8/227/073/LC82270732015002LGN00/index.html
- *
- *
- * For the Curious this is the Scala/Spark declaration of the schema.
- *
- * def addLandsatColumns(csv: DataFrame): DataFrame = {
- * csv
- * .withColumnRenamed("entityId", "id")
- * .withColumn("acquisitionDate",
- * csv.col("acquisitionDate").cast(TimestampType))
- * .withColumn("cloudCover", csv.col("cloudCover").cast(DoubleType))
- * .withColumn("path", csv.col("path").cast(IntegerType))
- * .withColumn("row", csv.col("row").cast(IntegerType))
- * .withColumn("min_lat", csv.col("min_lat").cast(DoubleType))
- * .withColumn("min_lon", csv.col("min_lon").cast(DoubleType))
- * .withColumn("max_lat", csv.col("max_lat").cast(DoubleType))
- * .withColumn("max_lon", csv.col("max_lon").cast(DoubleType))
- * .withColumn("year",
- * year(col("acquisitionDate")))
- * .withColumn("month",
- * month(col("acquisitionDate")))
- * .withColumn("day",
- * month(col("acquisitionDate")))
- * }
- *
- */
-public class ITestS3SelectLandsat extends AbstractS3SelectTest {
-
- private static final Logger LOG =
- LoggerFactory.getLogger(ITestS3SelectLandsat.class);
-
- private JobConf selectConf;
-
- /**
- * Normal limit for select operations.
- * Value: {@value}.
- */
- public static final int SELECT_LIMIT = 250;
-
- /**
- * And that select limit as a limit string.
- */
- public static final String LIMITED = " LIMIT " + SELECT_LIMIT;
-
- /**
- * Select days with 100% cloud cover, limited to {@link #SELECT_LIMIT}.
- * Value: {@value}.
- */
- public static final String SELECT_ENTITY_ID_ALL_CLOUDS =
- "SELECT\n"
- + "s.entityId from\n"
- + "S3OBJECT s WHERE\n"
- + "s.\"cloudCover\" = '100.0'\n"
- + LIMITED;
-
- /**
- * Select sunny days. There's no limit on the returned values, so
- * set one except for a scale test.
- * Value: {@value}.
- */
- public static final String SELECT_SUNNY_ROWS_NO_LIMIT
- = "SELECT * FROM S3OBJECT s WHERE s.cloudCover = '0.0'";
-
- /**
- * A Select call which returns nothing, always.
- * Value: {@value}.
- */
- public static final String SELECT_NOTHING
- = "SELECT * FROM S3OBJECT s WHERE s.cloudCover = 'sunny'";
-
- /**
- * Select the processing level; no limit.
- * Value: {@value}.
- */
- public static final String SELECT_PROCESSING_LEVEL_NO_LIMIT =
- "SELECT\n"
- + "s.processingLevel from\n"
- + "S3OBJECT s";
-
- @Override
- public void setup() throws Exception {
- super.setup();
-
- selectConf = new JobConf(false);
- // file is compressed.
- selectConf.set(SELECT_INPUT_COMPRESSION, COMPRESSION_OPT_GZIP);
- // and has a header
- selectConf.set(CSV_INPUT_HEADER, CSV_HEADER_OPT_USE);
- selectConf.setBoolean(SELECT_ERRORS_INCLUDE_SQL, true);
- inputMust(selectConf, CSV_INPUT_HEADER, CSV_HEADER_OPT_USE);
- inputMust(selectConf, SELECT_INPUT_FORMAT, SELECT_FORMAT_CSV);
- inputMust(selectConf, SELECT_OUTPUT_FORMAT, SELECT_FORMAT_CSV);
- inputMust(selectConf, SELECT_INPUT_COMPRESSION, COMPRESSION_OPT_GZIP);
- // disable the gzip codec, so that the record readers do not
- // get confused
- enablePassthroughCodec(selectConf, ".gz");
- ChangeDetectionPolicy changeDetectionPolicy =
- getLandsatFS().getChangeDetectionPolicy();
- Assume.assumeFalse("the standard landsat bucket doesn't have versioning",
- changeDetectionPolicy.getSource() == Source.VersionId
- && changeDetectionPolicy.isRequireVersion());
- }
-
- protected int getMaxLines() {
- return SELECT_LIMIT * 2;
- }
-
- @Test
- public void testSelectCloudcoverIgnoreHeader() throws Throwable {
- describe("select ignoring the header");
- selectConf.set(CSV_INPUT_HEADER, CSV_HEADER_OPT_IGNORE);
- String sql = "SELECT\n"
- + "* from\n"
- + "S3OBJECT s WHERE\n"
- + "s._3 = '0.0'\n"
- + LIMITED;
- List list = selectLandsatFile(selectConf, sql);
- LOG.info("Line count: {}", list.size());
- verifySelectionCount(1, SELECT_LIMIT, sql, list);
- }
-
- @Test
- public void testSelectCloudcoverUseHeader() throws Throwable {
- describe("select 100% cover using the header, "
- + "+ verify projection and incrementing select statistics");
- S3ATestUtils.MetricDiff selectCount = new S3ATestUtils.MetricDiff(
- getLandsatFS(),
- Statistic.OBJECT_SELECT_REQUESTS);
-
- List list = selectLandsatFile(selectConf,
- SELECT_ENTITY_ID_ALL_CLOUDS);
- LOG.info("Line count: {}", list.size());
- verifySelectionCount(1, SELECT_LIMIT, SELECT_ENTITY_ID_ALL_CLOUDS, list);
- String line1 = list.get(0);
- assertThat("no column filtering from " + SELECT_ENTITY_ID_ALL_CLOUDS,
- line1, not(containsString("100.0")));
- selectCount.assertDiffEquals("select count", 1);
- }
-
- @Test
- public void testFileContextIntegration() throws Throwable {
- describe("Test that select works through FileContext");
- FileContext fc = S3ATestUtils.createTestFileContext(getConfiguration());
-
- // there's a limit on the number of rows to read; this is larger
- // than the SELECT_LIMIT call to catch any failure where more than
- // that is returned, newline parsing fails, etc etc.
- List list = parseToLines(
- select(fc, getLandsatGZ(), selectConf, SELECT_ENTITY_ID_ALL_CLOUDS),
- SELECT_LIMIT * 2);
- LOG.info("Line count: {}", list.size());
- verifySelectionCount(1, SELECT_LIMIT, SELECT_ENTITY_ID_ALL_CLOUDS, list);
- }
-
- @Test
- public void testReadLandsatRecords() throws Throwable {
- describe("Use a record reader to read the records");
- inputMust(selectConf, CSV_OUTPUT_FIELD_DELIMITER, "\\t");
- inputMust(selectConf, CSV_OUTPUT_QUOTE_CHARACTER, "'");
- inputMust(selectConf, CSV_OUTPUT_QUOTE_FIELDS,
- CSV_OUTPUT_QUOTE_FIELDS_AS_NEEEDED);
- inputMust(selectConf, CSV_OUTPUT_RECORD_DELIMITER, "\n");
- List records = readRecords(
- selectConf,
- getLandsatGZ(),
- SELECT_ENTITY_ID_ALL_CLOUDS,
- createLineRecordReader(),
- SELECT_LIMIT);
- verifySelectionCount(1, SELECT_LIMIT, SELECT_ENTITY_ID_ALL_CLOUDS, records);
- }
-
- @Test
- public void testReadLandsatRecordsNoMatch() throws Throwable {
- describe("Verify the v2 record reader does not fail"
- + " when there are no results");
- verifySelectionCount(0, 0, SELECT_NOTHING,
- readRecords(
- selectConf,
- getLandsatGZ(),
- SELECT_NOTHING,
- createLineRecordReader(),
- SELECT_LIMIT));
- }
-
- @Test
- public void testReadLandsatRecordsGZipEnabled() throws Throwable {
- describe("Verify that by default, the gzip codec is connected to .gz"
- + " files, and so fails");
- // implicitly re-enable the gzip codec.
- selectConf.unset(CommonConfigurationKeys.IO_COMPRESSION_CODECS_KEY);
- intercept(IOException.class, "gzip",
- () -> readRecords(
- selectConf,
- getLandsatGZ(),
- SELECT_ENTITY_ID_ALL_CLOUDS,
- createLineRecordReader(),
- SELECT_LIMIT));
- }
-
- @Test
- public void testReadLandsatRecordsV1() throws Throwable {
- describe("Use a record reader to read the records");
-
- verifySelectionCount(1, SELECT_LIMIT, SELECT_ENTITY_ID_ALL_CLOUDS,
- readRecords(
- selectConf,
- getLandsatGZ(),
- SELECT_ENTITY_ID_ALL_CLOUDS,
- createLineRecordReader(),
- SELECT_LIMIT));
- }
-
- @Test
- public void testReadLandsatRecordsV1NoResults() throws Throwable {
- describe("verify that a select with no results is not an error");
-
- verifySelectionCount(0, 0, SELECT_NOTHING,
- readRecords(
- selectConf,
- getLandsatGZ(),
- SELECT_NOTHING,
- createLineRecordReader(),
- SELECT_LIMIT));
- }
-
- /**
- * Select from the landsat file.
- * @param conf config for the select call.
- * @param sql template for a formatted SQL request.
- * @param args arguments for the formatted request.
- * @return the lines selected
- * @throws IOException failure
- */
- private List selectLandsatFile(
- final Configuration conf,
- final String sql,
- final Object... args)
- throws Exception {
-
- // there's a limit on the number of rows to read; this is larger
- // than the SELECT_LIMIT call to catch any failure where more than
- // that is returned, newline parsing fails, etc etc.
- return parseToLines(
- select(getLandsatFS(), getLandsatGZ(), conf, sql, args));
- }
-
- /**
- * This is a larger-scale version of {@link ITestS3Select#testSelectSeek()}.
- */
- @Test
- public void testSelectSeekFullLandsat() throws Throwable {
- describe("Verify forward seeks work, not others");
-
- boolean enabled = getTestPropertyBool(
- getConfiguration(),
- KEY_SCALE_TESTS_ENABLED,
- DEFAULT_SCALE_TESTS_ENABLED);
- assume("Scale test disabled", enabled);
-
- // start: read in the full data through the initial select
- // this makes asserting that contents match possible
- final Path path = getLandsatGZ();
- S3AFileSystem fs = getLandsatFS();
-
- int len = (int) fs.getFileStatus(path).getLen();
- byte[] dataset = new byte[4 * _1MB];
- int actualLen;
- try (DurationInfo ignored =
- new DurationInfo(LOG, "Initial read of %s", path);
- FSDataInputStream sourceStream =
- select(fs, path,
- selectConf,
- SELECT_EVERYTHING)) {
- // read it in
- actualLen = IOUtils.read(sourceStream, dataset);
- }
- int seekRange = 16 * _1KB;
-
- try (FSDataInputStream seekStream =
- select(fs, path,
- selectConf,
- SELECT_EVERYTHING)) {
- SelectInputStream sis
- = (SelectInputStream) seekStream.getWrappedStream();
- S3AInputStreamStatistics streamStats
- = sis.getS3AStreamStatistics();
- // lazy seek doesn't raise a problem here
- seekStream.seek(0);
- assertEquals("first byte read", dataset[0], seekStream.read());
-
- // and now the pos has moved, again, seek will be OK
- seekStream.seek(1);
- seekStream.seek(1);
- // but trying to seek elsewhere now fails
- intercept(PathIOException.class,
- SelectInputStream.SEEK_UNSUPPORTED,
- () -> seekStream.seek(0));
- // positioned reads from the current location work.
- byte[] buffer = new byte[1];
- seekStream.readFully(seekStream.getPos(), buffer);
- // but positioned backwards fail.
- intercept(PathIOException.class,
- SelectInputStream.SEEK_UNSUPPORTED,
- () -> seekStream.readFully(0, buffer));
- // forward seeks are implemented as 1+ skip
- long target = seekStream.getPos() + seekRange;
- seek(seekStream, target);
- assertEquals("Seek position in " + seekStream,
- target, seekStream.getPos());
- // now do a read and compare values
- assertEquals("byte at seek position",
- dataset[(int) seekStream.getPos()], seekStream.read());
- assertEquals("Seek bytes skipped in " + streamStats,
- seekRange, streamStats.getBytesSkippedOnSeek());
- long offset;
- long increment = 64 * _1KB;
-
- // seek forward, comparing bytes
- for(offset = 32 * _1KB; offset < 256 * _1KB; offset += increment) {
- seek(seekStream, offset);
- assertEquals("Seek position in " + seekStream,
- offset, seekStream.getPos());
- // now do a read and compare values
- assertEquals("byte at seek position",
- dataset[(int) seekStream.getPos()], seekStream.read());
- }
- // there's no knowledge of how much data is left, but with Gzip
- // involved there can be a lot. To keep the test duration down,
- // this test, unlike the simpler one, doesn't try to read past the
- // EOF. Know this: it will be slow.
-
- LOG.info("Seek statistics {}", streamStats);
- }
- }
-
-}
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/ITestS3SelectMRJob.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/ITestS3SelectMRJob.java
deleted file mode 100644
index c1c7b89dce83d..0000000000000
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/ITestS3SelectMRJob.java
+++ /dev/null
@@ -1,216 +0,0 @@
-/*
- * 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.select;
-
-import java.io.IOException;
-import java.io.UncheckedIOException;
-import java.util.concurrent.atomic.AtomicLong;
-
-import org.apache.hadoop.fs.s3a.impl.ChangeDetectionPolicy;
-import org.apache.hadoop.fs.s3a.impl.ChangeDetectionPolicy.Source;
-import org.junit.Assume;
-import org.junit.Test;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.examples.WordCount;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.s3a.S3AFileSystem;
-import org.apache.hadoop.fs.s3a.S3ATestUtils;
-import org.apache.hadoop.fs.s3a.S3AUtils;
-import org.apache.hadoop.fs.s3a.commit.files.SuccessData;
-import org.apache.hadoop.fs.s3a.commit.staging.StagingCommitter;
-import org.apache.hadoop.io.IOUtils;
-import org.apache.hadoop.io.IntWritable;
-import org.apache.hadoop.io.Text;
-import org.apache.hadoop.mapred.JobConf;
-import org.apache.hadoop.mapreduce.Job;
-import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
-import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
-import org.apache.hadoop.util.DurationInfo;
-import org.apache.hadoop.util.functional.FutureIO;
-import org.apache.hadoop.yarn.conf.YarnConfiguration;
-import org.apache.hadoop.yarn.server.MiniYARNCluster;
-
-import static org.apache.hadoop.fs.s3a.commit.CommitConstants.FS_S3A_COMMITTER_NAME;
-import static org.apache.hadoop.fs.s3a.commit.CommitConstants.FS_S3A_COMMITTER_STAGING_UNIQUE_FILENAMES;
-import static org.apache.hadoop.fs.s3a.select.SelectConstants.*;
-
-/**
- * Run an MR job with a select query.
- * This is the effective end-to-end test which verifies:
- *
- * - Passing of select parameters through an MR job conf.
- * - Automatic pick-up of these parameter through TextInputFormat's use
- * of the mapreduce.lib.input.LineRecordReaderLineRecordReader.
- * - Issuing of S3 Select queries in mapper processes.
- * - Projection of columns in a select.
- * - Ability to switch to the Passthrough decompressor in an MR job.
- * - Saving of results through the S3A Staging committer.
- * - Basic validation of results.
- *
- * This makes it the most complex of the MR jobs in the hadoop-aws test suite.
- *
- * The query used is
- * {@link ITestS3SelectLandsat#SELECT_PROCESSING_LEVEL_NO_LIMIT},
- * which lists the processing level of all records in the source file,
- * and counts the number in each one by way of the normal word-count
- * routines.
- * This works because the SQL is projecting only the processing level.
- *
- * The result becomes something like (with tabs between fields):
- *
- * L1GT 370231
- * L1T 689526
- *
- */
-public class ITestS3SelectMRJob extends AbstractS3SelectTest {
-
- private final Configuration conf = new YarnConfiguration();
-
- private S3AFileSystem fs;
-
- private MiniYARNCluster yarnCluster;
-
- private Path rootPath;
-
- @Override
- public void setup() throws Exception {
- super.setup();
- fs = S3ATestUtils.createTestFileSystem(conf);
-
- ChangeDetectionPolicy changeDetectionPolicy =
- getLandsatFS().getChangeDetectionPolicy();
- Assume.assumeFalse("the standard landsat bucket doesn't have versioning",
- changeDetectionPolicy.getSource() == Source.VersionId
- && changeDetectionPolicy.isRequireVersion());
-
- rootPath = path("ITestS3SelectMRJob");
- Path workingDir = path("working");
- fs.setWorkingDirectory(workingDir);
- fs.mkdirs(new Path(rootPath, "input/"));
-
- yarnCluster = new MiniYARNCluster("ITestS3SelectMRJob", // testName
- 1, // number of node managers
- 1, // number of local log dirs per node manager
- 1); // number of hdfs dirs per node manager
- yarnCluster.init(conf);
- yarnCluster.start();
- }
-
- @Override
- public void teardown() throws Exception {
- if (yarnCluster != null) {
- yarnCluster.stop();
- }
- super.teardown();
- }
-
- @Test
- public void testLandsatSelect() throws Exception {
- final Path input = getLandsatGZ();
- final Path output = path("testLandsatSelect")
- .makeQualified(fs.getUri(), fs.getWorkingDirectory());
-
- final Job job = Job.getInstance(conf, "process level count");
- job.setJarByClass(WordCount.class);
- job.setMapperClass(WordCount.TokenizerMapper.class);
- job.setCombinerClass(WordCount.IntSumReducer.class);
- job.setReducerClass(WordCount.IntSumReducer.class);
- job.setOutputKeyClass(Text.class);
- job.setOutputValueClass(IntWritable.class);
- FileInputFormat.addInputPath(job, input);
- FileOutputFormat.setOutputPath(job, output);
-
- // job with use the staging committer
- final JobConf jobConf = (JobConf) job.getConfiguration();
- jobConf.set(FS_S3A_COMMITTER_NAME, StagingCommitter.NAME);
- jobConf.setBoolean(FS_S3A_COMMITTER_STAGING_UNIQUE_FILENAMES,
- false);
-
- final String query
- = ITestS3SelectLandsat.SELECT_PROCESSING_LEVEL_NO_LIMIT;
- inputMust(jobConf, SELECT_SQL,
- query);
- inputMust(jobConf, SELECT_INPUT_COMPRESSION, COMPRESSION_OPT_GZIP);
-
- // input settings
- inputMust(jobConf, SELECT_INPUT_FORMAT, SELECT_FORMAT_CSV);
- inputMust(jobConf, CSV_INPUT_HEADER, CSV_HEADER_OPT_USE);
-
- // output
- inputMust(jobConf, SELECT_OUTPUT_FORMAT, SELECT_FORMAT_CSV);
- inputMust(jobConf, CSV_OUTPUT_QUOTE_FIELDS,
- CSV_OUTPUT_QUOTE_FIELDS_AS_NEEEDED);
-
- // disable the gzip codec, so that the record readers do not
- // get confused
- enablePassthroughCodec(jobConf, ".gz");
-
- try (DurationInfo ignored = new DurationInfo(LOG, "SQL " + query)) {
- int exitCode = job.waitForCompletion(true) ? 0 : 1;
- assertEquals("Returned error code.", 0, exitCode);
- }
-
- // log the success info
- Path successPath = new Path(output, "_SUCCESS");
- SuccessData success = SuccessData.load(fs, successPath);
- LOG.info("Job _SUCCESS\n{}", success);
-
- // process the results by ver
- //
- LOG.info("Results for query \n{}", query);
- final AtomicLong parts = new AtomicLong(0);
- S3AUtils.applyLocatedFiles(fs.listFiles(output, false),
- (status) -> {
- Path path = status.getPath();
- // ignore _SUCCESS, any temp files in subdirectories...
- if (path.getName().startsWith("part-")) {
- parts.incrementAndGet();
- String result = readStringFromFile(path);
- LOG.info("{}\n{}", path, result);
- String[] lines = result.split("\n", -1);
- int l = lines.length;
- // add a bit of slack here in case some new processing
- // option was added.
- assertTrue("Wrong number of lines (" + l + ") in " + result,
- l > 0 && l < 15);
- }
- });
- assertEquals("More part files created than expected", 1, parts.get());
- }
-
- /**
- * Read a file; using Async IO for completeness and to see how
- * well the async IO works in practice.
- * Summary: checked exceptions cripple Async operations.
- */
- private String readStringFromFile(Path path) throws IOException {
- int bytesLen = (int)fs.getFileStatus(path).getLen();
- byte[] buffer = new byte[bytesLen];
- return FutureIO.awaitFuture(
- fs.openFile(path).build().thenApply(in -> {
- try {
- IOUtils.readFully(in, buffer, 0, bytesLen);
- return new String(buffer);
- } catch (IOException ex) {
- throw new UncheckedIOException(ex);
- }
- }));
- }
-}
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/ITestSelectUnsupported.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/ITestSelectUnsupported.java
new file mode 100644
index 0000000000000..1b115158294f2
--- /dev/null
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/ITestSelectUnsupported.java
@@ -0,0 +1,100 @@
+/*
+ * 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.select;
+
+import org.assertj.core.api.Assertions;
+import org.junit.Test;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.contract.ContractTestUtils;
+import org.apache.hadoop.fs.s3a.AbstractS3ATestBase;
+import org.apache.hadoop.fs.s3a.S3AFileSystem;
+import org.apache.hadoop.fs.s3a.s3guard.S3GuardTool;
+import org.apache.hadoop.util.ExitUtil;
+
+import static org.apache.hadoop.fs.s3a.select.SelectConstants.SELECT_SQL;
+import static org.apache.hadoop.fs.s3a.select.SelectConstants.SELECT_UNSUPPORTED;
+import static org.apache.hadoop.service.launcher.LauncherExitCodes.EXIT_UNSUPPORTED_VERSION;
+import static org.apache.hadoop.test.LambdaTestUtils.intercept;
+import static org.apache.hadoop.util.ExitUtil.disableSystemExit;
+
+/**
+ * Verify that s3 select is unsupported.
+ */
+public class ITestSelectUnsupported extends AbstractS3ATestBase {
+
+ /**
+ * S3 Select SQL statement.
+ */
+ private static final String STATEMENT = "SELECT *" +
+ " FROM S3Object s" +
+ " WHERE s._1 = 'foo'";
+
+ /**
+ * A {@code .must(SELECT_SQL, _)} option MUST raise {@code UnsupportedOperationException}.
+ */
+ @Test
+ public void testSelectOpenFileMustFailure() throws Throwable {
+
+ intercept(UnsupportedOperationException.class, SELECT_UNSUPPORTED, () ->
+ getFileSystem().openFile(methodPath())
+ .must(SELECT_SQL, STATEMENT)
+ .build()
+ .get());
+ }
+
+ /**
+ * A {@code .opt(SELECT_SQL, _)} option is ignored..
+ */
+ @Test
+ public void testSelectOpenFileMayIsIgnored() throws Throwable {
+
+ final Path path = methodPath();
+ final S3AFileSystem fs = getFileSystem();
+ ContractTestUtils.touch(fs, path);
+ fs.openFile(path)
+ .opt(SELECT_SQL, STATEMENT)
+ .build()
+ .get()
+ .close();
+ }
+
+ @Test
+ public void testPathCapabilityNotAvailable() throws Throwable {
+ describe("verify that the FS lacks the path capability");
+ Assertions.assertThat(getFileSystem().hasPathCapability(methodPath(), SELECT_SQL))
+ .describedAs("S3 Select reported as present")
+ .isFalse();
+ }
+
+ @Test
+ public void testS3GuardToolFails() throws Throwable {
+
+ // ensure that the command doesn't actually exit the VM.
+ disableSystemExit();
+ final ExitUtil.ExitException ex =
+ intercept(ExitUtil.ExitException.class, SELECT_UNSUPPORTED,
+ () -> S3GuardTool.main(new String[]{
+ "select", "-sql", STATEMENT
+ }));
+ Assertions.assertThat(ex.getExitCode())
+ .describedAs("exit code of exception")
+ .isEqualTo(EXIT_UNSUPPORTED_VERSION);
+ }
+}
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/StreamPublisher.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/StreamPublisher.java
deleted file mode 100644
index 461aef726876c..0000000000000
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/StreamPublisher.java
+++ /dev/null
@@ -1,89 +0,0 @@
-/*
- * 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.select;
-
-import java.util.Iterator;
-import java.util.concurrent.Executor;
-import java.util.stream.Stream;
-
-import org.reactivestreams.Subscriber;
-import org.reactivestreams.Subscription;
-import software.amazon.awssdk.core.async.SdkPublisher;
-
-/**
- * Publisher used to test the handling of asynchronous responses.
- * @param The type of published elements.
- */
-final class StreamPublisher implements SdkPublisher {
- private final Executor executor;
- private final Iterator iterator;
- private Boolean done = false;
-
- StreamPublisher(Stream data, Executor executor) {
- this.iterator = data.iterator();
- this.executor = executor;
- }
-
- StreamPublisher(Stream data) {
- this(data, Runnable::run);
- }
-
- @Override
- public void subscribe(Subscriber subscriber) {
- subscriber.onSubscribe(new Subscription() {
- @Override
- public void request(long n) {
- if (done) {
- return;
- }
-
- if (n < 1) {
- done = true;
- executor.execute(() -> subscriber.onError(new IllegalArgumentException()));
- return;
- }
-
- for (long i = 0; i < n; i++) {
- final T value;
- try {
- synchronized (iterator) {
- value = iterator.hasNext() ? iterator.next() : null;
- }
- } catch (Throwable e) {
- executor.execute(() -> subscriber.onError(e));
- break;
- }
-
- if (value == null) {
- done = true;
- executor.execute(subscriber::onComplete);
- break;
- } else {
- executor.execute(() -> subscriber.onNext(value));
- }
- }
- }
-
- @Override
- public void cancel() {
- done = true;
- }
- });
- }
-}
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/TestBlockingEnumeration.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/TestBlockingEnumeration.java
deleted file mode 100644
index 43bdcb062f0a3..0000000000000
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/TestBlockingEnumeration.java
+++ /dev/null
@@ -1,200 +0,0 @@
-/*
- * 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.select;
-
-import java.util.concurrent.Executor;
-import java.util.concurrent.Executors;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.stream.IntStream;
-import java.util.stream.Stream;
-
-import org.junit.Assert;
-import org.junit.Test;
-import software.amazon.awssdk.core.async.SdkPublisher;
-import software.amazon.awssdk.core.exception.SdkException;
-
-/**
- * Unit tests for {@link BlockingEnumeration}.
- */
-public final class TestBlockingEnumeration extends Assert {
-
- @Test
- public void containsElement() {
- SdkPublisher publisher = new StreamPublisher<>(Stream.of("foo"));
-
- BlockingEnumeration enumeration =
- new BlockingEnumeration<>(publisher, 1);
-
- assertTrue(enumeration.hasMoreElements());
- assertEquals("foo", enumeration.nextElement());
- assertFalse(enumeration.hasMoreElements());
- }
-
- @Test
- public void containsInjectedElement() {
- SdkPublisher publisher = new StreamPublisher<>(Stream.of("foo"));
-
- BlockingEnumeration enumeration =
- new BlockingEnumeration<>(publisher, 1, "bar");
-
- assertTrue(enumeration.hasMoreElements());
- assertEquals("bar", enumeration.nextElement());
- assertTrue(enumeration.hasMoreElements());
- assertEquals("foo", enumeration.nextElement());
- assertFalse(enumeration.hasMoreElements());
- }
-
- @Test
- public void throwsExceptionOnFirstElement() {
- SdkPublisher publisher = new StreamPublisher<>(
- Stream.of(0, 1)
- .map(i -> {
- throw SdkException.create("error!", null);
- }),
- Executors.newSingleThreadExecutor());
-
- BlockingEnumeration enumeration =
- new BlockingEnumeration<>(publisher, 1);
- assertThrows(SdkException.class, enumeration::hasMoreElements);
- }
-
- @Test
- public void throwsExceptionAfterInjectedElement() {
- SdkPublisher publisher = new StreamPublisher<>(
- Stream.of(0, 1)
- .peek(i -> {
- throw SdkException.create("error!", null);
- }),
- Executors.newSingleThreadExecutor());
-
- BlockingEnumeration enumeration =
- new BlockingEnumeration<>(publisher, 1, 99);
- assertTrue(enumeration.hasMoreElements());
- assertEquals(99, enumeration.nextElement().intValue());
- assertThrows(SdkException.class, enumeration::hasMoreElements);
- }
-
- @Test
- public void throwsNonSdkException() {
- SdkPublisher publisher = new StreamPublisher<>(
- Stream.of(0, 1)
- .peek(i -> {
- throw new RuntimeException("error!", null);
- }),
- Executors.newSingleThreadExecutor());
-
- BlockingEnumeration enumeration =
- new BlockingEnumeration<>(publisher, 1);
- SdkException exception = Assert.assertThrows(SdkException.class, enumeration::hasMoreElements);
- assertEquals(RuntimeException.class, exception.getCause().getClass());
- }
-
- @Test
- public void throwsError() {
- SdkPublisher publisher = new StreamPublisher<>(
- Stream.of(0, 1)
- .peek(i -> {
- throw new Error("error!", null);
- }),
- Executors.newSingleThreadExecutor());
-
- BlockingEnumeration enumeration =
- new BlockingEnumeration<>(publisher, 1);
- assertThrows(Error.class, enumeration::hasMoreElements);
- }
-
- @Test
- public void throwsExceptionOnSecondElement() {
- SdkPublisher publisher = new StreamPublisher<>(
- Stream.of(0, 1)
- .peek(i -> {
- if (i == 1) {
- throw SdkException.create("error!", null);
- }
- }),
- Executors.newSingleThreadExecutor());
-
- BlockingEnumeration enumeration =
- new BlockingEnumeration<>(publisher, 1);
- assertTrue(enumeration.hasMoreElements());
- assertEquals(0, enumeration.nextElement().intValue());
- assertThrows(SdkException.class, enumeration::hasMoreElements);
- }
-
- @Test
- public void noMoreElementsAfterThrow() {
- SdkPublisher publisher = new StreamPublisher<>(
- Stream.of(0, 1)
- .map(i -> {
- throw SdkException.create("error!", null);
- }),
- Executors.newSingleThreadExecutor());
-
- BlockingEnumeration enumeration =
- new BlockingEnumeration<>(publisher, 1);
- assertThrows(SdkException.class, enumeration::hasMoreElements);
- assertFalse(enumeration.hasMoreElements());
- }
-
- @Test
- public void buffersOnSameThread() {
- verifyBuffering(10, 3, Runnable::run);
- }
-
- @Test
- public void publisherOnDifferentThread() {
- verifyBuffering(5, 1, Executors.newSingleThreadExecutor());
- }
-
- @Test
- public void publisherOnDifferentThreadWithBuffer() {
- verifyBuffering(30, 10, Executors.newSingleThreadExecutor());
- }
-
- private static void verifyBuffering(int length, int bufferSize, Executor executor) {
- AtomicInteger emitted = new AtomicInteger();
- SdkPublisher publisher = new StreamPublisher<>(
- IntStream.range(0, length).boxed().peek(i -> emitted.incrementAndGet()),
- executor);
-
- BlockingEnumeration enumeration =
- new BlockingEnumeration<>(publisher, bufferSize);
-
- int pulled = 0;
- while (true) {
- try {
- int expected = Math.min(length, pulled + bufferSize);
- if (expected != emitted.get()) {
- Thread.sleep(10);
- }
- assertEquals(expected, emitted.get());
- } catch (InterruptedException e) {
- fail("Interrupted: " + e);
- }
-
- if (!enumeration.hasMoreElements()) {
- break;
- }
-
- int i = enumeration.nextElement();
- assertEquals(pulled, i);
- pulled++;
- }
- }
-}
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/TestSelectEventStreamPublisher.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/TestSelectEventStreamPublisher.java
deleted file mode 100644
index fdf3b5b725376..0000000000000
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/TestSelectEventStreamPublisher.java
+++ /dev/null
@@ -1,190 +0,0 @@
-/*
- * 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.select;
-
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.Executor;
-import java.util.concurrent.Executors;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.stream.Stream;
-
-import org.junit.Assert;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
-import software.amazon.awssdk.core.SdkBytes;
-import software.amazon.awssdk.core.async.SdkPublisher;
-import software.amazon.awssdk.core.exception.SdkException;
-import software.amazon.awssdk.http.AbortableInputStream;
-import software.amazon.awssdk.services.s3.model.SelectObjectContentEventStream;
-import software.amazon.awssdk.services.s3.model.SelectObjectContentResponse;
-
-/**
- * Unit tests for {@link SelectEventStreamPublisher}.
- */
-@RunWith(Parameterized.class)
-public final class TestSelectEventStreamPublisher extends Assert {
-
- @Parameterized.Parameters(name = "threading-{0}")
- public static Collection params() {
- return Arrays.asList(new Object[][]{
- {"main"},
- {"background"}
- });
- }
-
- private final String threading;
-
- public TestSelectEventStreamPublisher(String threading) {
- this.threading = threading;
- }
-
- private Executor createExecutor() {
- if (threading.equals("main")) {
- return Runnable::run;
- } else if (threading.equals("background")) {
- return Executors.newSingleThreadExecutor();
- } else {
- throw new IllegalArgumentException("Unknown: " + threading);
- }
- }
-
- @Test
- public void emptyRecordsInputStream() throws IOException {
- SelectEventStreamPublisher selectEventStreamPublisher =
- createSelectPublisher(Stream.of(
- SelectObjectContentEventStream.recordsBuilder()
- .payload(SdkBytes.fromByteArray(new byte[0]))
- .build()));
-
- try (AbortableInputStream inputStream =
- selectEventStreamPublisher.toRecordsInputStream(e -> {})) {
- assertEquals(-1, inputStream.read());
- }
- }
-
- @Test
- public void multipleRecords() throws IOException {
- SelectEventStreamPublisher selectEventStreamPublisher =
- createSelectPublisher(Stream.of(
- SelectObjectContentEventStream.recordsBuilder()
- .payload(SdkBytes.fromUtf8String("foo"))
- .build(),
- SelectObjectContentEventStream.recordsBuilder()
- .payload(SdkBytes.fromUtf8String("bar"))
- .build()));
-
- try (AbortableInputStream inputStream =
- selectEventStreamPublisher.toRecordsInputStream(e -> {})) {
- String result = readAll(inputStream);
- assertEquals("foobar", result);
- }
- }
-
- @Test
- public void skipsOtherEvents() throws IOException {
- SelectEventStreamPublisher selectEventStreamPublisher =
- createSelectPublisher(Stream.of(
- SelectObjectContentEventStream.recordsBuilder()
- .payload(SdkBytes.fromUtf8String("foo"))
- .build(),
- SelectObjectContentEventStream.progressBuilder()
- .build(),
- SelectObjectContentEventStream.statsBuilder()
- .build(),
- SelectObjectContentEventStream.recordsBuilder()
- .payload(SdkBytes.fromUtf8String("bar"))
- .build(),
- SelectObjectContentEventStream.endBuilder()
- .build()));
-
- try (AbortableInputStream inputStream =
- selectEventStreamPublisher.toRecordsInputStream(e -> {})) {
- String result = readAll(inputStream);
- assertEquals("foobar", result);
- }
- }
-
- @Test
- public void callsOnEndEvent() throws IOException {
- SelectEventStreamPublisher selectEventStreamPublisher =
- createSelectPublisher(Stream.of(
- SelectObjectContentEventStream.recordsBuilder()
- .payload(SdkBytes.fromUtf8String("foo"))
- .build(),
- SelectObjectContentEventStream.endBuilder()
- .build()));
-
- AtomicBoolean endEvent = new AtomicBoolean(false);
- try (AbortableInputStream inputStream =
- selectEventStreamPublisher.toRecordsInputStream(e -> endEvent.set(true))) {
- String result = readAll(inputStream);
- assertEquals("foo", result);
- }
-
- assertTrue(endEvent.get());
- }
-
- @Test
- public void handlesErrors() throws IOException {
- SelectEventStreamPublisher selectEventStreamPublisher =
- createSelectPublisher(Stream.of(
- SelectObjectContentEventStream.recordsBuilder()
- .payload(SdkBytes.fromUtf8String("foo"))
- .build(),
- SelectObjectContentEventStream.recordsBuilder()
- .payload(SdkBytes.fromUtf8String("bar"))
- .build())
- .map(e -> {
- throw SdkException.create("error!", null);
- }));
-
- try (AbortableInputStream inputStream =
- selectEventStreamPublisher.toRecordsInputStream(e -> {})) {
- assertThrows(SdkException.class, () -> readAll(inputStream));
- }
- }
-
- private SelectEventStreamPublisher createSelectPublisher(
- Stream stream) {
- SdkPublisher sdkPublisher =
- new StreamPublisher<>(stream, createExecutor());
- CompletableFuture future =
- CompletableFuture.completedFuture(null);
- SelectObjectContentResponse response =
- SelectObjectContentResponse.builder().build();
- return new SelectEventStreamPublisher(future, response, sdkPublisher);
- }
-
- private static String readAll(InputStream inputStream) throws IOException {
- try (ByteArrayOutputStream outputStream = new ByteArrayOutputStream()) {
- byte[] buffer = new byte[8096];
- int read;
- while ((read = inputStream.read(buffer, 0, buffer.length)) != -1) {
- outputStream.write(buffer, 0, read);
- }
- return outputStream.toString();
- }
- }
-}
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/test/MinimalWriteOperationHelperCallbacks.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/test/MinimalWriteOperationHelperCallbacks.java
index eee0c71950566..6838129bb30d1 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/test/MinimalWriteOperationHelperCallbacks.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/test/MinimalWriteOperationHelperCallbacks.java
@@ -18,12 +18,8 @@
package org.apache.hadoop.fs.s3a.test;
-import java.util.concurrent.CompletableFuture;
-
import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest;
import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadResponse;
-import software.amazon.awssdk.services.s3.model.SelectObjectContentRequest;
-import software.amazon.awssdk.services.s3.model.SelectObjectContentResponseHandler;
import org.apache.hadoop.fs.s3a.WriteOperationHelper;
@@ -33,18 +29,11 @@
public class MinimalWriteOperationHelperCallbacks
implements WriteOperationHelper.WriteOperationHelperCallbacks {
- @Override
- public CompletableFuture selectObjectContent(
- SelectObjectContentRequest request,
- SelectObjectContentResponseHandler th) {
- return null;
- }
-
@Override
public CompleteMultipartUploadResponse completeMultipartUpload(
CompleteMultipartUploadRequest request) {
return null;
}
-};
+}
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/CsvFile.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/tools/CsvFile.java
similarity index 98%
rename from hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/CsvFile.java
rename to hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/tools/CsvFile.java
index 17574372de429..a786f974e0c40 100644
--- a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/select/CsvFile.java
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/tools/CsvFile.java
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-package org.apache.hadoop.fs.s3a.select;
+package org.apache.hadoop.fs.s3a.tools;
import java.io.Closeable;
import java.io.IOException;
diff --git a/hadoop-tools/hadoop-aws/src/test/resources/core-site.xml b/hadoop-tools/hadoop-aws/src/test/resources/core-site.xml
index c99d7d43134cb..70b87ee275406 100644
--- a/hadoop-tools/hadoop-aws/src/test/resources/core-site.xml
+++ b/hadoop-tools/hadoop-aws/src/test/resources/core-site.xml
@@ -31,6 +31,13 @@
+