#pragma once #include #include #include 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; PortKind current_block_kind = PortKind::Main; bool has_input = false; bool finished = false; bool finalized = false; virtual void consume(Chunk) = 0; virtual void consumeTotals(Chunk) {} virtual void consumeExtremes(Chunk) {} virtual void finalize() {} public: IOutputFormat(const Block & header, WriteBuffer & out); Status prepare() override; void work() override; /// Flush output buffers if any. virtual void flush(); /// Value for rows_before_limit_at_least field. virtual void setRowsBeforeLimit(size_t /*rows_before_limit*/) {} /// Notify about progress. Method could be called from different threads. /// Passed value are delta, that must be summarized. virtual void onProgress(const Progress & /*progress*/) {} /// 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); } public: /// Compatible to IBlockOutputStream interface void write(const Block & block) { consume(Chunk(block.getColumns(), block.rows())); } void doWritePrefix() {} void doWriteSuffix() { finalize(); } void setTotals(const Block & totals) { consume(Chunk(totals.getColumns(), totals.rows())); } void setExtremes(const Block & extremes) { consume(Chunk(extremes.getColumns(), extremes.rows())); } }; }