be/src/util/jsonb/serialize.cpp
Line | Count | Source |
1 | | // Licensed to the Apache Software Foundation (ASF) under one |
2 | | // or more contributor license agreements. See the NOTICE file |
3 | | // distributed with this work for additional information |
4 | | // regarding copyright ownership. The ASF licenses this file |
5 | | // to you under the Apache License, Version 2.0 (the |
6 | | // "License"); you may not use this file except in compliance |
7 | | // with the License. You may obtain a copy of the License at |
8 | | // |
9 | | // http://www.apache.org/licenses/LICENSE-2.0 |
10 | | // |
11 | | // Unless required by applicable law or agreed to in writing, |
12 | | // software distributed under the License is distributed on an |
13 | | // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
14 | | // KIND, either express or implied. See the License for the |
15 | | // specific language governing permissions and limitations |
16 | | // under the License. |
17 | | |
18 | | #include "util/jsonb/serialize.h" |
19 | | |
20 | | #include <assert.h> |
21 | | |
22 | | #include <algorithm> |
23 | | #include <limits> |
24 | | #include <memory> |
25 | | #include <unordered_set> |
26 | | #include <vector> |
27 | | |
28 | | #include "core/arena.h" |
29 | | #include "core/block/column_with_type_and_name.h" |
30 | | #include "core/block/columns_with_type_and_name.h" |
31 | | #include "core/column/column.h" |
32 | | #include "core/column/column_string.h" |
33 | | #include "core/data_type/data_type.h" |
34 | | #include "core/data_type/primitive_type.h" |
35 | | #include "core/data_type_serde/data_type_serde.h" |
36 | | #include "core/string_ref.h" |
37 | | #include "core/value/jsonb_value.h" |
38 | | #include "runtime/descriptors.h" |
39 | | #include "storage/tablet/tablet_schema.h" |
40 | | #include "util/defer_op.h" |
41 | | #include "util/jsonb_document.h" |
42 | | #include "util/jsonb_stream.h" |
43 | | #include "util/jsonb_writer.h" |
44 | | |
45 | | namespace doris { |
46 | | |
47 | | void JsonbSerializeUtil::block_to_jsonb(const TabletSchema& schema, const Block& block, |
48 | | ColumnString& dst, int num_cols, |
49 | | const DataTypeSerDeSPtrs& serdes, |
50 | 272 | const std::unordered_set<int32_t>& row_store_cids) { |
51 | 272 | block_to_jsonb(schema, block, dst, num_cols, serdes, row_store_cids, 0, block.rows()); |
52 | 272 | } |
53 | | |
54 | | void JsonbSerializeUtil::block_to_jsonb(const TabletSchema& schema, const Block& block, |
55 | | ColumnString& dst, int num_cols, |
56 | | const DataTypeSerDeSPtrs& serdes, |
57 | | const std::unordered_set<int32_t>& row_store_cids, |
58 | 272 | size_t row_pos, size_t num_rows) { |
59 | 272 | static_cast<void>(block_to_jsonb(schema, block, dst, num_cols, serdes, row_store_cids, row_pos, |
60 | 272 | num_rows, std::numeric_limits<size_t>::max())); |
61 | 272 | } |
62 | | |
63 | | size_t JsonbSerializeUtil::block_to_jsonb(const TabletSchema& schema, const Block& block, |
64 | | ColumnString& dst, int num_cols, |
65 | | const DataTypeSerDeSPtrs& serdes, |
66 | | const std::unordered_set<int32_t>& row_store_cids, |
67 | 2.40k | size_t row_pos, size_t num_rows, size_t max_bytes) { |
68 | 2.40k | Arena arena; |
69 | 2.40k | assert(num_cols <= block.columns()); |
70 | 2.40k | assert(row_pos + num_rows <= block.rows()); |
71 | 2.40k | DataTypeSerDe::FormatOptions options; |
72 | 2.40k | auto tz = cctz::utc_time_zone(); |
73 | 2.40k | options.timezone = &tz; |
74 | 2.40k | size_t written_rows = 0; |
75 | 1.91M | for (size_t i = row_pos; i < row_pos + num_rows; ++i) { |
76 | 1.91M | JsonbWriterT<JsonbOutStream> jsonb_writer; |
77 | 1.91M | jsonb_writer.writeStartObject(); |
78 | 38.9M | for (int j = 0; j < num_cols; ++j) { |
79 | 37.0M | const auto& column = block.get_by_position(j).column; |
80 | 37.0M | const auto& tablet_column = *schema.columns()[j]; |
81 | | // ignore row store columns |
82 | 37.0M | if (tablet_column.is_row_store_column()) { |
83 | 1.91M | continue; |
84 | 1.91M | } |
85 | | // TODO improve performance for checking column in group |
86 | 35.1M | if (row_store_cids.empty() || row_store_cids.contains(tablet_column.unique_id())) { |
87 | 23.6M | serdes[j]->write_one_cell_to_jsonb(*column, jsonb_writer, arena, |
88 | 23.6M | tablet_column.unique_id(), i, options); |
89 | 23.6M | } |
90 | 35.1M | } |
91 | 1.91M | jsonb_writer.writeEndObject(); |
92 | 1.91M | dst.insert_data(jsonb_writer.getOutput()->getBuffer(), jsonb_writer.getOutput()->getSize()); |
93 | 1.91M | ++written_rows; |
94 | 1.91M | if (dst.byte_size() >= max_bytes) { |
95 | 0 | break; |
96 | 0 | } |
97 | 1.91M | } |
98 | 2.40k | return written_rows; |
99 | 2.40k | } |
100 | | |
101 | | // batch rows |
102 | | Status JsonbSerializeUtil::jsonb_to_block( |
103 | | const DataTypeSerDeSPtrs& serdes, const ColumnString& jsonb_column, |
104 | | const std::unordered_map<uint32_t, uint32_t>& col_id_to_idx, Block& dst, |
105 | | const std::vector<std::string>& default_values, |
106 | 342 | const std::unordered_set<int>& include_cids) { |
107 | 342 | auto dst_columns_guard = dst.mutate_columns_scoped(); |
108 | 342 | MutableColumns& dst_columns = dst_columns_guard.mutable_columns(); |
109 | 2.60k | for (int i = 0; i < jsonb_column.size(); ++i) { |
110 | 2.26k | StringRef jsonb_data = jsonb_column.get_data_at(i); |
111 | 2.26k | RETURN_IF_ERROR(jsonb_to_columns(serdes, jsonb_data.data, jsonb_data.size, col_id_to_idx, |
112 | 2.26k | dst_columns, default_values, include_cids)); |
113 | 2.26k | } |
114 | 342 | return Status::OK(); |
115 | 342 | } |
116 | | |
117 | | Status JsonbSerializeUtil::jsonb_to_columns( |
118 | | const DataTypeSerDeSPtrs& serdes, const char* data, size_t size, |
119 | | const std::unordered_map<uint32_t, uint32_t>& col_id_to_idx, MutableColumns& dst_columns, |
120 | | const std::vector<std::string>& default_values, |
121 | 6.89k | const std::unordered_set<int>& include_cids) { |
122 | 6.89k | const JsonbDocument* pdoc = nullptr; |
123 | 6.89k | RETURN_IF_ERROR(JsonbDocument::checkAndCreateDocument(data, size, &pdoc)); |
124 | 6.89k | const JsonbDocument& doc = *pdoc; |
125 | 6.89k | DCHECK(!dst_columns.empty()); |
126 | 6.89k | size_t num_rows = dst_columns[0]->size(); |
127 | 6.89k | size_t filled_columns = 0; |
128 | 116k | for (auto it = doc->begin(); it != doc->end(); ++it) { |
129 | 109k | auto col_it = col_id_to_idx.find(it->getKeyId()); |
130 | 109k | if (col_it != col_id_to_idx.end() && |
131 | 109k | (include_cids.empty() || include_cids.contains(it->getKeyId()))) { |
132 | 94.7k | auto& dst_column = dst_columns[col_it->second]; |
133 | 94.7k | serdes[col_it->second]->read_one_cell_from_jsonb(*dst_column, it->value()); |
134 | 94.7k | ++filled_columns; |
135 | 94.7k | } |
136 | 109k | } |
137 | 6.89k | if (filled_columns >= dst_columns.size()) { |
138 | 5.69k | return Status::OK(); |
139 | 5.69k | } |
140 | 16.6k | auto fill_column = [&](size_t pos, size_t old_num_rows) { |
141 | 16.6k | auto& dst_column = dst_columns[pos]; |
142 | 16.6k | if (dst_column->size() < old_num_rows + 1) { |
143 | 1.22k | DCHECK(dst_column->size() == old_num_rows); |
144 | 1.22k | Status st = Status::OK(); |
145 | 1.22k | if (default_values[pos].empty()) { |
146 | 1.18k | dst_column->insert_default(); |
147 | 1.18k | } else { |
148 | 39 | Slice value(default_values[pos].data(), default_values[pos].size()); |
149 | 39 | DataTypeSerDe::FormatOptions opt; |
150 | 39 | opt.converted_from_string = true; |
151 | 39 | st = serdes[pos]->deserialize_one_cell_from_json(*dst_column, value, opt); |
152 | 39 | } |
153 | 1.22k | RETURN_IF_ERROR(st); |
154 | 1.22k | DCHECK(dst_column->size() == num_rows + 1); |
155 | 1.22k | return Status::OK(); |
156 | 1.22k | } |
157 | 16.6k | DCHECK(dst_column->size() == num_rows + 1); |
158 | 15.4k | return Status::OK(); |
159 | 16.6k | }; |
160 | | // fill missing column |
161 | 1.20k | if (!include_cids.empty()) { |
162 | 18 | for (auto cid : include_cids) { |
163 | 18 | auto col_it = col_id_to_idx.find(cid); |
164 | 18 | if (col_it == col_id_to_idx.end()) { |
165 | 2 | continue; |
166 | 2 | } |
167 | 16 | RETURN_IF_ERROR(fill_column(static_cast<size_t>(col_it->second), num_rows)); |
168 | 16 | } |
169 | 1.19k | } else { |
170 | 17.8k | for (size_t i = 0; i < dst_columns.size(); ++i) { |
171 | 16.6k | RETURN_IF_ERROR(fill_column(i, num_rows)); |
172 | 16.6k | } |
173 | 1.19k | } |
174 | 1.20k | return Status::OK(); |
175 | 1.20k | } |
176 | | |
177 | | // single row |
178 | | Status JsonbSerializeUtil::jsonb_to_block( |
179 | | const DataTypeSerDeSPtrs& serdes, const char* data, size_t size, |
180 | | const std::unordered_map<uint32_t, uint32_t>& col_id_to_idx, Block& dst, |
181 | | const std::vector<std::string>& default_values, |
182 | 0 | const std::unordered_set<int>& include_cids) { |
183 | 0 | auto dst_columns_guard = dst.mutate_columns_scoped(); |
184 | 0 | MutableColumns& dst_columns = dst_columns_guard.mutable_columns(); |
185 | 0 | return jsonb_to_columns(serdes, data, size, col_id_to_idx, dst_columns, default_values, |
186 | 0 | include_cids); |
187 | 0 | } |
188 | | |
189 | | } // namespace doris |