Skip to content

Commit 456a2f9

Browse files
vaibhav5140vaibhav5140
committed
Enabled tests[IOStats]
1 parent 8104378 commit 456a2f9

File tree

9 files changed

+130
-109
lines changed

9 files changed

+130
-109
lines changed

‎hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/streams/AnalyticsRequestCallback.java‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,25 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
119
package org.apache.hadoop.fs.s3a.impl.streams;
220

321
import org.apache.hadoop.fs.s3a.statistics.S3AInputStreamStatistics;
22+
import org.apache.hadoop.fs.statistics.DurationTracker;
423
import software.amazon.s3.analyticsaccelerator.util.RequestCallback;
524

625
/**
@@ -20,6 +39,9 @@ public AnalyticsRequestCallback(S3AInputStreamStatistics statistics) {
2039
@Override
2140
public void onGetRequest() {
2241
statistics.incrementAnalyticsGetRequests();
42+
// Update ACTION_HTTP_GET_REQUEST statistic
43+
DurationTracker tracker = statistics.initiateGetRequest();
44+
tracker.close();
2345
}
2446

2547
@Override

‎hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/impl/streams/AnalyticsStream.java‎

Lines changed: 31 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -49,13 +49,17 @@ public class AnalyticsStream extends ObjectInputStream implements StreamCapabili
4949
private S3SeekableInputStream inputStream;
5050
private long lastReadCurrentPos = 0;
5151
private volatile boolean closed;
52+
private final long contentLength;
53+
private final long lengthLimit;
5254

5355
public static final Logger LOG = LoggerFactory.getLogger(AnalyticsStream.class);
5456

5557
public AnalyticsStream(final ObjectReadParameters parameters,
5658
final S3SeekableInputStreamFactory s3SeekableInputStreamFactory) throws IOException {
5759
super(InputStreamType.Analytics, parameters);
5860
S3ObjectAttributes s3Attributes = parameters.getObjectAttributes();
61+
this.contentLength = s3Attributes.getLen();
62+
this.lengthLimit = s3Attributes.getLen();
5963
this.inputStream = s3SeekableInputStreamFactory.createStream(S3URI.of(s3Attributes.getBucket(),
6064
s3Attributes.getKey()), buildOpenStreamInformation(parameters));
6165
getS3AStreamStatistics().streamOpened(InputStreamType.Analytics);
@@ -64,6 +68,9 @@ public AnalyticsStream(final ObjectReadParameters parameters,
6468
@Override
6569
public int read() throws IOException {
6670
throwIfClosed();
71+
if (getPos() >= lengthLimit) {
72+
return -1; // EOF reached due to length limit
73+
}
6774
getS3AStreamStatistics().readOperationStarted(getPos(), 1);
6875

6976
int bytesRead;
@@ -75,7 +82,7 @@ public int read() throws IOException {
7582
}
7683

7784
if (bytesRead != -1) {
78-
getS3AStreamStatistics().bytesRead(1);
85+
incrementBytesRead(1);
7986
}
8087

8188
return bytesRead;
@@ -124,7 +131,7 @@ public int readTail(byte[] buf, int off, int len) throws IOException {
124131
}
125132

126133
if (bytesRead > 0) {
127-
getS3AStreamStatistics().bytesRead(bytesRead);
134+
incrementBytesRead(bytesRead);
128135
}
129136

130137
return bytesRead;
@@ -133,18 +140,25 @@ public int readTail(byte[] buf, int off, int len) throws IOException {
133140
@Override
134141
public int read(byte[] buf, int off, int len) throws IOException {
135142
throwIfClosed();
136-
getS3AStreamStatistics().readOperationStarted(getPos(), len);
143+
long pos = getPos();
144+
if (pos >= lengthLimit) {
145+
return -1; // EOF reached due to length limit
146+
}
147+
148+
// Limit read length to not exceed the length limit
149+
int maxRead = (int) Math.min(len, lengthLimit - pos);
150+
getS3AStreamStatistics().readOperationStarted(pos, maxRead);
137151

138152
int bytesRead;
139153
try {
140-
bytesRead = inputStream.read(buf, off, len);
154+
bytesRead = inputStream.read(buf, off, maxRead);
141155
} catch (IOException ioe) {
142156
onReadFailure(ioe);
143157
throw ioe;
144158
}
145159

146160
if (bytesRead > 0) {
147-
getS3AStreamStatistics().bytesRead(bytesRead);
161+
incrementBytesRead(bytesRead);
148162
}
149163

150164
return bytesRead;
@@ -260,4 +274,16 @@ protected void throwIfClosed() throws IOException {
260274
throw new IOException(getKey() + ": " + FSExceptionMessages.STREAM_IS_CLOSED);
261275
}
262276
}
277+
278+
/**
279+
* Increment the bytes read counter if there is a stats instance
280+
* and the number of bytes read is more than zero.
281+
* @param bytesRead number of bytes read
282+
*/
283+
private void incrementBytesRead(long bytesRead) {
284+
getS3AStreamStatistics().bytesRead(bytesRead);
285+
if (getContext().getStats() != null && bytesRead > 0) {
286+
getContext().getStats().incrementBytesRead(bytesRead);
287+
}
288+
}
263289
}

‎hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AAnalyticsAcceleratorStreamReading.java‎

Lines changed: 48 additions & 71 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,8 @@
4242
import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY;
4343
import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY_PARQUET;
4444
import static org.apache.hadoop.fs.Options.OpenFileOptions.FS_OPTION_OPENFILE_READ_POLICY_WHOLE_FILE;
45+
import static org.apache.hadoop.fs.contract.ContractTestUtils.writeDataset;
46+
import static org.apache.hadoop.fs.contract.ContractTestUtils.dataset;
4547
import static org.apache.hadoop.fs.s3a.Constants.ANALYTICS_ACCELERATOR_CONFIGURATION_PREFIX;
4648
import static org.apache.hadoop.fs.s3a.S3ATestUtils.enableAnalyticsAccelerator;
4749
import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
@@ -54,6 +56,8 @@
5456
import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_ANALYTICS_HEAD_REQUESTS;
5557
import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_ANALYTICS_OPENED;
5658
import static org.apache.hadoop.fs.statistics.StreamStatisticNames.ANALYTICS_STREAM_FACTORY_CLOSED;
59+
import static org.apache.hadoop.io.Sizes.S_1K;
60+
import static org.apache.hadoop.io.Sizes.S_1M;
5761
import static org.apache.hadoop.test.LambdaTestUtils.intercept;
5862

5963
/**
@@ -154,6 +158,9 @@ public void testMalformedParquetFooter() throws IOException {
154158
}
155159

156160
verifyStatisticCounterValue(ioStats, STREAM_READ_ANALYTICS_OPENED, 1);
161+
verifyStatisticCounterValue(ioStats, STREAM_READ_ANALYTICS_GET_REQUESTS, 1);
162+
// S3A passes in the meta data on file open, we expect AAL to make no HEAD requests
163+
verifyStatisticCounterValue(ioStats, STREAM_READ_ANALYTICS_HEAD_REQUESTS, 0);
157164
}
158165

159166
/**
@@ -195,6 +202,7 @@ public void testMultiRowGroupParquet() throws Throwable {
195202
}
196203

197204
verifyStatisticCounterValue(ioStats, STREAM_READ_ANALYTICS_OPENED, 1);
205+
// S3A passes in the meta-data(content length) on file open, we expect AAL to make no HEAD requests
198206
verifyStatisticCounterValue(ioStats, STREAM_READ_ANALYTICS_HEAD_REQUESTS, 0);
199207
}
200208

@@ -215,32 +223,48 @@ public void testInvalidConfigurationThrows() throws Exception {
215223
() -> S3SeekableInputStreamConfiguration.fromConfiguration(connectorConfiguration));
216224
}
217225

226+
/**
227+
*
228+
* TXT files are classified as SEQUENTIAL format and use SequentialPrefetcher(requests the entire 10MB file)
229+
* RangeOptimiser splits ranges larger than maxRangeSizeBytes (8MB) using partSizeBytes (8MB)
230+
* The 10MB range gets split into: [0-8MB) and [8MB-10MB)
231+
* Each split range becomes a separate Block, resulting in 2 GET requests:
232+
*/
218233
@Test
219234
public void testLargeFileMultipleGets() throws Throwable {
220235
describe("Large file should trigger multiple GET requests");
221236

222-
Path dest = writeThenReadFile("large-test-file.txt", 10 * 1024 * 1024); // 10MB
223-
237+
Path dest = path("large-test-file.txt");
238+
byte[] data = dataset(10 * S_1M, 256, 255);
239+
writeDataset(getFileSystem(), dest, data, 10 * S_1M, 1024, true);
224240

241+
byte[] buffer = new byte[S_1M * 10];
225242
try (FSDataInputStream inputStream = getFileSystem().open(dest)) {
226243
IOStatistics ioStats = inputStream.getIOStatistics();
227-
inputStream.readFully(new byte[(int) getFileSystem().getFileStatus(dest).getLen()]);
244+
inputStream.readFully(buffer);
228245

229246
verifyStatisticCounterValue(ioStats, STREAM_READ_ANALYTICS_GET_REQUESTS, 2);
247+
// Because S3A passes in the meta-data(content length) on file open, we expect AAL to make no HEAD requests
248+
verifyStatisticCounterValue(ioStats, STREAM_READ_ANALYTICS_HEAD_REQUESTS, 0);
230249
}
231250
}
232251

233252
@Test
234253
public void testSmallFileSingleGet() throws Throwable {
235254
describe("Small file should trigger only one GET request");
236255

237-
Path dest = writeThenReadFile("small-test-file.txt", 1 * 1024 * 1024); // 1KB
256+
Path dest = path("small-test-file.txt");
257+
byte[] data = dataset(S_1M, 256, 255);
258+
writeDataset(getFileSystem(), dest, data, S_1M, 1024, true);
238259

260+
byte[] buffer = new byte[S_1M];
239261
try (FSDataInputStream inputStream = getFileSystem().open(dest)) {
240262
IOStatistics ioStats = inputStream.getIOStatistics();
241-
inputStream.readFully(new byte[(int) getFileSystem().getFileStatus(dest).getLen()]);
263+
inputStream.readFully(buffer);
242264

243265
verifyStatisticCounterValue(ioStats, STREAM_READ_ANALYTICS_GET_REQUESTS, 1);
266+
// Because S3A passes in the meta-data(content length) on file open, we expect AAL to make no HEAD requests
267+
verifyStatisticCounterValue(ioStats, STREAM_READ_ANALYTICS_HEAD_REQUESTS, 0);
244268
}
245269
}
246270

@@ -249,93 +273,46 @@ public void testSmallFileSingleGet() throws Throwable {
249273
public void testRandomSeekPatternGets() throws Throwable {
250274
describe("Random seek pattern should optimize GET requests");
251275

252-
Path dest = writeThenReadFile("seek-test.txt", 100 * 1024);
276+
Path dest = path("seek-test.txt");
277+
byte[] data = dataset(5 * S_1M, 256, 255);
278+
writeDataset(getFileSystem(), dest, data, 5 * S_1M, 1024, true);
253279

280+
byte[] buffer = new byte[S_1M];
254281
try (FSDataInputStream inputStream = getFileSystem().open(dest)) {
255282
IOStatistics ioStats = inputStream.getIOStatistics();
256283

257-
inputStream.seek(1000);
258-
inputStream.read(new byte[100]);
259-
260-
inputStream.seek(50000);
261-
inputStream.read(new byte[100]);
262-
263-
inputStream.seek(90000);
264-
inputStream.read(new byte[100]);
284+
inputStream.read(buffer);
285+
inputStream.seek(2 * S_1M);
286+
inputStream.read(new byte[512 * S_1K]);
287+
inputStream.seek(3 * S_1M);
288+
inputStream.read(new byte[512 * S_1K]);
265289

266290
verifyStatisticCounterValue(ioStats, STREAM_READ_ANALYTICS_GET_REQUESTS, 1);
267-
}
268-
}
269-
270-
@Test
271-
public void testAALNeverMakesHeadRequests() throws Throwable {
272-
describe("Prove AAL never makes HEAD requests - S3A provides all metadata");
273-
274-
Path dest = writeThenReadFile("no-head-test.txt", 1024 * 1024); // 1MB
275-
276-
try (FSDataInputStream inputStream = getFileSystem().open(dest)) {
277-
IOStatistics ioStats = inputStream.getIOStatistics();
278-
inputStream.read(new byte[1024]);
279-
280291
verifyStatisticCounterValue(ioStats, STREAM_READ_ANALYTICS_HEAD_REQUESTS, 0);
281-
verifyStatisticCounterValue(ioStats, STREAM_READ_ANALYTICS_OPENED, 1);
282-
283-
ObjectInputStream objectInputStream = (ObjectInputStream) inputStream.getWrappedStream();
284-
Assertions.assertThat(objectInputStream.streamType()).isEqualTo(InputStreamType.Analytics);
285-
286292
}
287293
}
288294

289295

290296
@Test
291-
public void testParquetReadingNoHeadRequests() throws Throwable {
292-
describe("Parquet-optimized reading should not trigger AAL HEAD requests");
293-
294-
Path dest = path("parquet-head-test.parquet");
295-
File file = new File("src/test/resources/multi_row_group.parquet");
296-
Path sourcePath = new Path(file.toURI().getPath());
297-
getFileSystem().copyFromLocalFile(false, true, sourcePath, dest);
298-
299-
try (FSDataInputStream stream = getFileSystem().openFile(dest)
300-
.must(FS_OPTION_OPENFILE_READ_POLICY, FS_OPTION_OPENFILE_READ_POLICY_PARQUET)
301-
.build().get()) {
302-
303-
FileStatus fileStatus = getFileSystem().getFileStatus(dest);
304-
stream.readFully(new byte[(int) fileStatus.getLen()]);
297+
public void testSequentialStreamsNoDuplicateGets() throws Throwable {
298+
describe("Sequential streams reading same object should not duplicate GETs");
305299

306-
IOStatistics stats = stream.getIOStatistics();
307-
308-
verifyStatisticCounterValue(stats, STREAM_READ_ANALYTICS_HEAD_REQUESTS, 0);
309-
verifyStatisticCounterValue(stats, STREAM_READ_ANALYTICS_OPENED, 1);
310-
311-
verifyStatisticCounterValue(stats, STREAM_READ_ANALYTICS_GET_REQUESTS, 1);
312-
}
313-
}
314-
315-
316-
@Test
317-
public void testConcurrentStreamsNoDuplicateGets() throws Throwable {
318-
describe("Concurrent streams reading same object should not duplicate GETs");
319-
320-
Path dest = writeThenReadFile("concurrent-test.txt", 1 * 1024 * 1024);
300+
Path dest = path("sequential-test.txt");
301+
byte[] data = dataset(S_1M, 256, 255);
302+
writeDataset(getFileSystem(), dest, data, S_1M, 1024, true);
321303

304+
byte[] buffer = new byte[1024];
322305
try (FSDataInputStream stream1 = getFileSystem().open(dest);
323306
FSDataInputStream stream2 = getFileSystem().open(dest)) {
324307

325-
byte[] buffer1 = new byte[1024];
326-
byte[] buffer2 = new byte[1024];
327-
328-
stream1.read(buffer1);
329-
stream2.read(buffer2);
308+
stream1.read(buffer);
309+
stream2.read(buffer);
330310

331311
IOStatistics stats1 = stream1.getIOStatistics();
332312
IOStatistics stats2 = stream2.getIOStatistics();
333313

334-
long totalGets = stats1.counters().getOrDefault(
335-
STREAM_READ_ANALYTICS_GET_REQUESTS, 0L) +
336-
stats2.counters().getOrDefault(
337-
STREAM_READ_ANALYTICS_GET_REQUESTS, 0L);
338-
Assertions.assertThat(totalGets).isEqualTo(1);
314+
verifyStatisticCounterValue(stats1, STREAM_READ_ANALYTICS_GET_REQUESTS, 1);
315+
verifyStatisticCounterValue(stats2, STREAM_READ_ANALYTICS_GET_REQUESTS, 0);
339316
}
340317
}
341318
}

‎hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AIOStatisticsContext.java‎

Lines changed: 1 addition & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -81,10 +81,7 @@ protected Configuration createConfiguration() {
8181
public void setup() throws Exception {
8282
super.setup();
8383
executor = HadoopExecutors.newFixedThreadPool(SMALL_THREADS);
84-
// Analytics accelerator currently does not support IOStatisticsContext, this will be added as
85-
// part of https://issues.apache.org/jira/browse/HADOOP-19364
86-
skipIfAnalyticsAcceleratorEnabled(getConfiguration(),
87-
"Analytics Accelerator currently does not support IOStatisticsContext");
84+
8885

8986
}
9087

‎hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AMetrics.java‎

Lines changed: 1 addition & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -52,10 +52,7 @@ public void testMetricsRegister()
5252

5353
@Test
5454
public void testStreamStatistics() throws IOException {
55-
// Analytics accelerator currently does not support IOStatistics, this will be added as
56-
// part of https://issues.apache.org/jira/browse/HADOOP-19364
57-
skipIfAnalyticsAcceleratorEnabled(getConfiguration(),
58-
"Analytics Accelerator currently does not support stream statistics");
55+
5956

6057
S3AFileSystem fs = getFileSystem();
6158
Path file = path("testStreamStatistics");

‎hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/ITestCommitOperationCost.java‎

Lines changed: 1 addition & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -174,10 +174,7 @@ private void abortActiveStream() throws IOException {
174174
public void testCostOfCreatingMagicFile() throws Throwable {
175175
describe("Files created under magic paths skip existence checks and marker deletes");
176176

177-
// Analytics accelerator currently does not support IOStatistics, this will be added as
178-
// part of https://issues.apache.org/jira/browse/HADOOP-19364
179-
skipIfAnalyticsAcceleratorEnabled(getConfiguration(),
180-
"Analytics Accelerator currently does not support stream statistics");
177+
181178
S3AFileSystem fs = getFileSystem();
182179
Path destFile = methodSubPath("file.txt");
183180
fs.delete(destFile.getParent(), true);
@@ -255,10 +252,6 @@ public void testCostOfCreatingMagicFile() throws Throwable {
255252
public void testCostOfSavingLoadingPendingFile() throws Throwable {
256253
describe("Verify costs of saving .pending file under a magic path");
257254

258-
// Analytics accelerator currently does not support IOStatistics, this will be added as
259-
// part of https://issues.apache.org/jira/browse/HADOOP-19364
260-
skipIfAnalyticsAcceleratorEnabled(getConfiguration(),
261-
"Analytics Accelerator currently does not support stream statistics");
262255
S3AFileSystem fs = getFileSystem();
263256
Path partDir = methodSubPath("file.pending");
264257
Path destFile = new Path(partDir, "file.pending");

‎hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/fileContext/ITestS3AFileContextStatistics.java‎

Lines changed: 1 addition & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -46,10 +46,7 @@ public class ITestS3AFileContextStatistics extends FCStatisticsBaseTest {
4646
@BeforeEach
4747
public void setUp() throws Exception {
4848
conf = new Configuration();
49-
// Analytics accelerator currently does not support IOStatistics, this will be added as
50-
// part of https://issues.apache.org/jira/browse/HADOOP-19364
51-
skipIfAnalyticsAcceleratorEnabled(conf,
52-
"Analytics Accelerator currently does not support stream statistics");
49+
5350
fc = S3ATestUtils.createTestFileContext(conf);
5451
testRootPath = fileContextTestHelper.getTestRootPath(fc, "test");
5552
fc.mkdir(testRootPath,

0 commit comments

Comments
 (0)