1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
|
#pragma once
#include <Storages/StorageS3Settings.h>
#include "clickhouse_config.h"
#if USE_AWS_S3
#include <memory>
#include <IO/HTTPCommon.h>
#include <IO/ParallelReadBuffer.h>
#include <IO/ReadBuffer.h>
#include <IO/ReadSettings.h>
#include <IO/ReadBufferFromFileBase.h>
#include <IO/WithFileName.h>
#include <aws/s3/model/GetObjectResult.h>
namespace DB
{
/**
* Perform S3 HTTP GET request and provide response to read.
*/
class ReadBufferFromS3 : public ReadBufferFromFileBase
{
private:
std::shared_ptr<const S3::Client> client_ptr;
String bucket;
String key;
String version_id;
const S3Settings::RequestSettings request_settings;
/// These variables are atomic because they can be used for `logging only`
/// (where it is not important to get consistent result)
/// from separate thread other than the one which uses the buffer for s3 reading.
std::atomic<off_t> offset = 0;
std::atomic<off_t> read_until_position = 0;
std::optional<Aws::S3::Model::GetObjectResult> read_result;
std::unique_ptr<ReadBuffer> impl;
Poco::Logger * log = &Poco::Logger::get("ReadBufferFromS3");
public:
ReadBufferFromS3(
std::shared_ptr<const S3::Client> client_ptr_,
const String & bucket_,
const String & key_,
const String & version_id_,
const S3Settings::RequestSettings & request_settings_,
const ReadSettings & settings_,
bool use_external_buffer = false,
size_t offset_ = 0,
size_t read_until_position_ = 0,
bool restricted_seek_ = false,
std::optional<size_t> file_size = std::nullopt);
~ReadBufferFromS3() override;
bool nextImpl() override;
off_t seek(off_t off, int whence) override;
off_t getPosition() override;
size_t getFileSize() override;
void setReadUntilPosition(size_t position) override;
void setReadUntilEnd() override;
size_t getFileOffsetOfBufferEnd() const override { return offset; }
bool supportsRightBoundedReads() const override { return true; }
String getFileName() const override { return bucket + "/" + key; }
size_t readBigAt(char * to, size_t n, size_t range_begin, const std::function<bool(size_t)> & progress_callback) override;
bool supportsReadAt() override { return true; }
private:
std::unique_ptr<ReadBuffer> initialize();
/// If true, if we destroy impl now, no work was wasted. Just for metrics.
bool atEndOfRequestedRangeGuess();
/// Call inside catch() block if GetObject fails. Bumps metrics, logs the error.
/// Returns true if the error looks retriable.
bool processException(Poco::Exception & e, size_t read_offset, size_t attempt) const;
Aws::S3::Model::GetObjectResult sendRequest(size_t range_begin, std::optional<size_t> range_end_incl) const;
bool readAllRangeSuccessfully() const;
ReadSettings read_settings;
bool use_external_buffer;
/// There is different seek policy for disk seek and for non-disk seek
/// (non-disk seek is applied for seekable input formats: orc, arrow, parquet).
bool restricted_seek;
bool read_all_range_successfully = false;
};
}
#endif
|
