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
|
#include "ProtobufRowOutputFormat.h"
#if USE_PROTOBUF
# include <Formats/FormatFactory.h>
# include <Core/Block.h>
# include <Formats/FormatSettings.h>
# error #include <Formats/ProtobufSchemas.h>
# include <Formats/ProtobufSerializer.h>
# include <Formats/ProtobufWriter.h>
# error #include <google/protobuf/descriptor.h>
namespace DB
{
namespace ErrorCodes
{
extern const int NO_ROW_DELIMITER;
}
ProtobufRowOutputFormat::ProtobufRowOutputFormat(
WriteBuffer & out_,
const Block & header_,
const ProtobufSchemaInfo & schema_info_,
const FormatSettings & settings_,
bool with_length_delimiter_)
: IRowOutputFormat(header_, out_)
, writer(std::make_unique<ProtobufWriter>(out))
, serializer(ProtobufSerializer::create(
header_.getNames(),
header_.getDataTypes(),
*ProtobufSchemas::instance().getMessageTypeForFormatSchema(schema_info_.getSchemaInfo(), ProtobufSchemas::WithEnvelope::No),
with_length_delimiter_,
/* with_envelope = */ false,
settings_.protobuf.output_nullables_with_google_wrappers,
*writer))
, allow_multiple_rows(with_length_delimiter_ || settings_.protobuf.allow_multiple_rows_without_delimiter)
{
}
void ProtobufRowOutputFormat::write(const Columns & columns, size_t row_num)
{
if (!allow_multiple_rows && !first_row)
throw Exception(ErrorCodes::NO_ROW_DELIMITER,
"The ProtobufSingle format can't be used "
"to write multiple rows because this format doesn't have any row delimiter.");
if (row_num == 0)
serializer->setColumns(columns.data(), columns.size());
serializer->writeRow(row_num);
}
void registerOutputFormatProtobuf(FormatFactory & factory)
{
for (bool with_length_delimiter : {false, true})
{
factory.registerOutputFormat(
with_length_delimiter ? "Protobuf" : "ProtobufSingle",
[with_length_delimiter](WriteBuffer & buf,
const Block & header,
const FormatSettings & settings)
{
return std::make_shared<ProtobufRowOutputFormat>(
buf, header, ProtobufSchemaInfo(settings, "Protobuf", header, settings.protobuf.use_autogenerated_schema),
settings, with_length_delimiter);
});
}
}
}
#else
namespace DB
{
class FormatFactory;
void registerOutputFormatProtobuf(FormatFactory &) {}
}
#endif
|
