ClickHouse/src/Processors/Formats/IOutputFormat.h

95 lines
2.9 KiB
C++
Raw Normal View History

#pragma once
#include <string>
#include <Processors/IProcessor.h>
#include <Processors/RowsBeforeLimitCounter.h>
2019-02-19 18:41:18 +00:00
#include <IO/Progress.h>
namespace DB
{
class WriteBuffer;
/** Output format have three inputs and no outputs. It writes data from WriteBuffer.
*
* First input is for main resultset, second is for "totals" and third is for "extremes".
* It's not necessarily to connect "totals" or "extremes" ports (they may remain dangling).
*
* Data from input ports are pulled in order: first, from main input, then totals, then extremes.
*
* By default, data for "totals" and "extremes" is ignored.
*/
class IOutputFormat : public IProcessor
{
public:
enum PortKind { Main = 0, Totals = 1, Extremes = 2 };
protected:
WriteBuffer & out;
Chunk current_chunk;
2019-02-19 18:41:18 +00:00
PortKind current_block_kind = PortKind::Main;
2019-02-07 18:51:53 +00:00
bool has_input = false;
2019-02-19 18:41:18 +00:00
bool finished = false;
bool finalized = false;
2020-08-08 00:47:03 +00:00
/// Flush data on each consumed chunk. This is intended for interactive applications to output data as soon as it's ready.
bool auto_flush = false;
RowsBeforeLimitCounterPtr rows_before_limit_counter;
virtual void consume(Chunk) = 0;
virtual void consumeTotals(Chunk) {}
virtual void consumeExtremes(Chunk) {}
2019-02-19 18:41:18 +00:00
virtual void finalize() {}
public:
2019-08-03 11:02:40 +00:00
IOutputFormat(const Block & header_, WriteBuffer & out_);
Status prepare() override;
void work() override;
2019-02-19 18:41:18 +00:00
/// Flush output buffers if any.
virtual void flush();
void setAutoFlush() { auto_flush = true; }
2019-02-19 18:41:18 +00:00
/// Value for rows_before_limit_at_least field.
virtual void setRowsBeforeLimit(size_t /*rows_before_limit*/) {}
/// Counter to calculate rows_before_limit_at_least in processors pipeline.
void setRowsBeforeLimitCounter(RowsBeforeLimitCounterPtr counter) { rows_before_limit_counter.swap(counter); }
2019-02-19 18:41:18 +00:00
/// Notify about progress. Method could be called from different threads.
/// Passed value are delta, that must be summarized.
virtual void onProgress(const Progress & /*progress*/) {}
2019-02-19 18:41:18 +00:00
/// Content-Type to set when sending HTTP response.
virtual std::string getContentType() const { return "text/plain; charset=UTF-8"; }
InputPort & getPort(PortKind kind) { return *std::next(inputs.begin(), kind); }
/// Compatible to IBlockOutputStream interface
2020-04-27 18:15:55 +00:00
void write(const Block & block);
2019-08-05 09:35:46 +00:00
virtual void doWritePrefix() {}
virtual void doWriteSuffix() { finalize(); }
void setTotals(const Block & totals) { consumeTotals(Chunk(totals.getColumns(), totals.rows())); }
void setExtremes(const Block & extremes) { consumeExtremes(Chunk(extremes.getColumns(), extremes.rows())); }
size_t getResultRows() const { return result_rows; }
size_t getResultBytes() const { return result_bytes; }
private:
/// Counters for consumed chunks. Are used for QueryLog.
size_t result_rows = 0;
size_t result_bytes = 0;
bool prefix_written = false;
};
}