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
109
|
#include <IO/SeekableReadBuffer.h>
namespace DB
{
namespace ErrorCodes
{
extern const int CANNOT_READ_FROM_ISTREAM;
}
namespace
{
template <typename CustomData>
class SeekableReadBufferWrapper : public SeekableReadBuffer
{
public:
SeekableReadBufferWrapper(SeekableReadBuffer & in_, CustomData && custom_data_)
: SeekableReadBuffer(in_.buffer().begin(), in_.buffer().size(), in_.offset())
, in(in_)
, custom_data(std::move(custom_data_))
{
}
private:
SeekableReadBuffer & in;
CustomData custom_data;
bool nextImpl() override
{
in.position() = position();
if (!in.next())
{
set(in.position(), 0);
return false;
}
BufferBase::set(in.buffer().begin(), in.buffer().size(), in.offset());
return true;
}
off_t seek(off_t off, int whence) override
{
in.position() = position();
off_t new_pos = in.seek(off, whence);
BufferBase::set(in.buffer().begin(), in.buffer().size(), in.offset());
return new_pos;
}
off_t getPosition() override
{
in.position() = position();
return in.getPosition();
}
};
}
std::unique_ptr<SeekableReadBuffer> wrapSeekableReadBufferReference(SeekableReadBuffer & ref)
{
return std::make_unique<SeekableReadBufferWrapper<nullptr_t>>(ref, nullptr);
}
std::unique_ptr<SeekableReadBuffer> wrapSeekableReadBufferPointer(SeekableReadBufferPtr ptr)
{
return std::make_unique<SeekableReadBufferWrapper<SeekableReadBufferPtr>>(*ptr, SeekableReadBufferPtr{ptr});
}
size_t copyFromIStreamWithProgressCallback(std::istream & istr, char * to, size_t n, const std::function<bool(size_t)> & progress_callback, bool * out_cancelled)
{
const size_t chunk = DBMS_DEFAULT_BUFFER_SIZE;
if (out_cancelled)
*out_cancelled = false;
size_t copied = 0;
while (copied < n)
{
size_t to_copy = std::min(chunk, n - copied);
istr.read(to + copied, to_copy);
size_t gcount = istr.gcount();
copied += gcount;
bool cancelled = false;
if (gcount && progress_callback)
cancelled = progress_callback(copied);
if (gcount != to_copy)
{
if (!istr.eof())
throw Exception(
ErrorCodes::CANNOT_READ_FROM_ISTREAM,
"{} at offset {}",
istr.fail() ? "Cannot read from istream" : "Unexpected state of istream",
copied);
break;
}
if (cancelled)
{
if (out_cancelled != nullptr)
*out_cancelled = true;
break;
}
}
return copied;
}
}
|
