-
Notifications
You must be signed in to change notification settings - Fork 6.6k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #52853 from Avogar/http-valid-json-on-exception
Output valid JSON/XML on excetpion during HTTP query execution
- Loading branch information
Showing
32 changed files
with
1,163 additions
and
79 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,85 @@ | ||
| #include <IO/PeekableWriteBuffer.h> | ||
|
|
||
| namespace DB | ||
| { | ||
|
|
||
| PeekableWriteBuffer::PeekableWriteBuffer(DB::WriteBuffer & sub_buf_) : BufferWithOwnMemory(0), sub_buf(sub_buf_) | ||
| { | ||
| Buffer & sub_working = sub_buf.buffer(); | ||
| BufferBase::set(sub_working.begin() + sub_buf.offset(), sub_working.size() - sub_buf.offset(), 0); | ||
| } | ||
|
|
||
| void PeekableWriteBuffer::nextImpl() | ||
| { | ||
| if (checkpoint) | ||
| { | ||
| if (write_to_own_memory) | ||
| { | ||
| size_t prev_size = position() - memory.data(); | ||
| size_t new_size = memory.size() * 2; | ||
| memory.resize(new_size); | ||
| BufferBase::set(memory.data(), memory.size(), prev_size); | ||
| return; | ||
| } | ||
|
|
||
| if (memory.size() == 0) | ||
| memory.resize(DBMS_DEFAULT_BUFFER_SIZE); | ||
|
|
||
| sub_buf.position() = position(); | ||
| BufferBase::set(memory.data(), memory.size(), 0); | ||
| write_to_own_memory = true; | ||
| return; | ||
| } | ||
|
|
||
| sub_buf.position() = position(); | ||
| sub_buf.next(); | ||
| BufferBase::set(sub_buf.buffer().begin(), sub_buf.buffer().size(), sub_buf.offset()); | ||
| } | ||
|
|
||
|
|
||
| void PeekableWriteBuffer::dropCheckpoint() | ||
| { | ||
| assert(checkpoint); | ||
| checkpoint = std::nullopt; | ||
| /// If we have saved data in own memory, write it to sub-buf. | ||
| if (write_to_own_memory) | ||
| { | ||
| try | ||
| { | ||
| sub_buf.next(); | ||
| sub_buf.write(memory.data(), position() - memory.data()); | ||
| Buffer & sub_working = sub_buf.buffer(); | ||
| BufferBase::set(sub_working.begin(), sub_working.size(), sub_buf.offset()); | ||
| write_to_own_memory = false; | ||
| } | ||
| catch (...) | ||
| { | ||
| /// If exception happened during writing to sub buffer, we should | ||
| /// update buffer to not leave it in invalid state. | ||
| Buffer & sub_working = sub_buf.buffer(); | ||
| BufferBase::set(sub_working.begin(), sub_working.size(), sub_buf.offset()); | ||
| write_to_own_memory = false; | ||
| } | ||
| } | ||
|
|
||
| } | ||
|
|
||
| void PeekableWriteBuffer::rollbackToCheckpoint(bool drop) | ||
| { | ||
| assert(checkpoint); | ||
|
|
||
| /// Just ignore all data written after checkpoint. | ||
| if (write_to_own_memory) | ||
| { | ||
| Buffer & sub_working = sub_buf.buffer(); | ||
| BufferBase::set(sub_working.begin(), sub_working.size(), sub_buf.offset()); | ||
| write_to_own_memory = false; | ||
| } | ||
|
|
||
| position() = *checkpoint; | ||
|
|
||
| if (drop) | ||
| checkpoint = std::nullopt; | ||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,59 @@ | ||
| #pragma once | ||
| #include <IO/WriteBuffer.h> | ||
| #include <IO/BufferWithOwnMemory.h> | ||
| #include <stack> | ||
|
|
||
| namespace DB | ||
| { | ||
|
|
||
| namespace ErrorCodes | ||
| { | ||
| extern const int LOGICAL_ERROR; | ||
| } | ||
|
|
||
| /// Similar to PeekableReadBuffer. | ||
| /// Allows to set checkpoint at some position in stream and come back to this position later. | ||
| /// When next() is called, saves data between checkpoint and current position to own memory instead of writing it to sub-buffer. | ||
| /// So, all the data after checkpoint won't be written in sub-buffer until checkpoint is dropped. | ||
| /// Rollback to checkpoint means that all data after checkpoint will be ignored and not sent to sub-buffer. | ||
| /// Sub-buffer should not be accessed directly during the lifetime of peekable buffer (unless | ||
| /// you reset() the state of peekable buffer after each change of underlying buffer) | ||
| /// If position() of peekable buffer is explicitly set to some position before checkpoint | ||
| /// (e.g. by istr.position() = prev_pos), behavior is undefined. | ||
| class PeekableWriteBuffer : public BufferWithOwnMemory<WriteBuffer> | ||
| { | ||
| friend class PeekableWriteBufferCheckpoint; | ||
| public: | ||
| explicit PeekableWriteBuffer(WriteBuffer & sub_buf_); | ||
|
|
||
| /// Sets checkpoint at current position | ||
| ALWAYS_INLINE inline void setCheckpoint() | ||
| { | ||
| if (checkpoint) | ||
| throw Exception(ErrorCodes::LOGICAL_ERROR, "PeekableWriteBuffer does not support recursive checkpoints."); | ||
|
|
||
| checkpoint.emplace(pos); | ||
| } | ||
|
|
||
| /// Forget checkpoint and send all data from checkpoint to position to sub-buffer. | ||
| void dropCheckpoint(); | ||
|
|
||
| /// Sets position at checkpoint and forget all data written from checkpoint to position. | ||
| /// All pointers (such as this->buffer().end()) may be invalidated | ||
| void rollbackToCheckpoint(bool drop = false); | ||
|
|
||
| void finalizeImpl() override | ||
| { | ||
| assert(!checkpoint); | ||
| sub_buf.position() = position(); | ||
| } | ||
|
|
||
| private: | ||
| void nextImpl() override; | ||
|
|
||
| WriteBuffer & sub_buf; | ||
| bool write_to_own_memory = false; | ||
| std::optional<Position> checkpoint = std::nullopt; | ||
| }; | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.