thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-11-30 11:32:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

Added JSONEachRowWithProgressRowOutputFormat.

Browse Source
This commit is contained in:
Nikolai Kochetov 2019-08-09 18:58:07 +03:00
parent 2ae3db7920
commit 23e2d17d9d
5 changed files with 65 additions and 75 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
dbms/src/Formats/JSONEachRowWithProgressRowOutputStream.cpp
View File

@ -1,47 +0,0 @@
#include <IO/WriteHelpers.h>
#include <IO/WriteBufferValidUTF8.h>
#include <Formats/JSONEachRowWithProgressRowOutputStream.h>
#include <Formats/FormatFactory.h>
#include <Formats/BlockOutputStreamFromRowOutputStream.h>
namespace DB
{
void JSONEachRowWithProgressRowOutputStream::writeRowStartDelimiter()
{
writeCString("{\"row\":{", ostr);
}
void JSONEachRowWithProgressRowOutputStream::writeRowEndDelimiter()
{
writeCString("}}\n", ostr);
field_number = 0;
}
void JSONEachRowWithProgressRowOutputStream::onProgress(const Progress & value)
{
progress.incrementPiecewiseAtomically(value);
writeCString("{\"progress\":", ostr);
progress.writeJSON(ostr);
writeCString("}\n", ostr);
}
void registerOutputFormatJSONEachRowWithProgress(FormatFactory & factory)
{
factory.registerOutputFormat("JSONEachRowWithProgress", [](
WriteBuffer & buf,
const Block & sample,
const Context &,
const FormatSettings & format_settings)
{
return std::make_shared<BlockOutputStreamFromRowOutputStream>(
std::make_shared<JSONEachRowWithProgressRowOutputStream>(buf, sample, format_settings), sample);
});
}
}

27
dbms/src/Formats/JSONEachRowWithProgressRowOutputStream.h
View File

@ -1,27 +0,0 @@
#pragma once
#include <IO/Progress.h>
#include <Formats/JSONEachRowRowOutputStream.h>
namespace DB
{
/** The stream for outputting data in JSON format, by object per line
* that includes progress rows. Does not validate UTF-8.
*/
class JSONEachRowWithProgressRowOutputStream : public JSONEachRowRowOutputStream
{
public:
using JSONEachRowRowOutputStream::JSONEachRowRowOutputStream;
void writeRowStartDelimiter() override;
void writeRowEndDelimiter() override;
void onProgress(const Progress & value) override;
private:
Progress progress;
};
}

3
dbms/src/Processors/Formats/Impl/JSONEachRowRowOutputFormat.h
View File

@ -29,8 +29,9 @@ protected:
void consumeTotals(Chunk) override {}
void consumeExtremes(Chunk) override {}
private:
size_t field_number = 0;
private:
Names fields;
FormatSettings settings;

43
dbms/src/Processors/Formats/Impl/JSONEachRowWithProgressRowOutputFormat.cpp Normal file
View File

@ -0,0 +1,43 @@
#include <IO/WriteHelpers.h>
#include <IO/WriteBufferValidUTF8.h>
#include <Processors/Formats/Impl/JSONEachRowWithProgressRowOutputFormat.h>
#include <Formats/FormatFactory.h>
namespace DB
{
void JSONEachRowWithProgressRowOutputFormat::writeRowStartDelimiter()
{
writeCString("{\"row\":{", out);
}
void JSONEachRowWithProgressRowOutputFormat::writeRowEndDelimiter()
{
writeCString("}}\n", out);
field_number = 0;
}
void JSONEachRowWithProgressRowOutputFormat::onProgress(const Progress & value)
{
progress.incrementPiecewiseAtomically(value);
writeCString("{\"progress\":", out);
progress.writeJSON(out);
writeCString("}\n", out);
}
void registerOutputFormatProcessorJSONEachRow(FormatFactory & factory)
{
factory.registerOutputFormatProcessor("JSONEachRowWithProgress", [](
WriteBuffer & buf,
const Block & sample,
const Context &,
const FormatSettings & format_settings)
{
return std::make_shared<JSONEachRowWithProgressRowOutputFormat>(buf, sample, format_settings);
});
}
}

20
dbms/src/Processors/Formats/Impl/JSONEachRowWithProgressRowOutputFormat.h Normal file
View File

@ -0,0 +1,20 @@
#pragma once
#include <Processors/Formats/Impl/JSONEachRowRowOutputFormat.h>
namespace DB
{
class JSONEachRowWithProgressRowOutputFormat : public JSONEachRowRowOutputFormat
{
public:
using JSONEachRowRowOutputFormat::JSONEachRowRowOutputFormat;
void writeRowStartDelimiter() override;
void writeRowEndDelimiter() override;
void onProgress(const Progress & value) override;
private:
Progress progress;
};
}