/root/doris/be/src/vec/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 "vec/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 "olap/tablet_schema.h" |
29 | | #include "runtime/descriptors.h" |
30 | | #include "runtime/jsonb_value.h" |
31 | | #include "runtime/primitive_type.h" |
32 | | #include "runtime/types.h" |
33 | | #include "util/jsonb_document.h" |
34 | | #include "util/jsonb_stream.h" |
35 | | #include "util/jsonb_writer.h" |
36 | | #include "vec/columns/column.h" |
37 | | #include "vec/columns/column_string.h" |
38 | | #include "vec/common/arena.h" |
39 | | #include "vec/common/string_ref.h" |
40 | | #include "vec/core/column_with_type_and_name.h" |
41 | | #include "vec/core/columns_with_type_and_name.h" |
42 | | #include "vec/data_types/data_type.h" |
43 | | #include "vec/data_types/serde/data_type_serde.h" |
44 | | |
45 | | namespace doris::vectorized { |
46 | | |
47 | | void JsonbSerializeUtil::block_to_jsonb(const TabletSchema& schema, const Block& block, |
48 | | ColumnString& dst, int num_cols, |
49 | | const DataTypeSerDeSPtrs& serdes, |
50 | 4 | const std::unordered_set<int32_t>& row_store_cids) { |
51 | 4 | block_to_jsonb(schema, block, dst, num_cols, serdes, row_store_cids, 0, block.rows()); |
52 | 4 | } |
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 | 4 | size_t row_pos, size_t num_rows) { |
59 | 4 | static_cast<void>(block_to_jsonb(schema, block, dst, num_cols, serdes, row_store_cids, row_pos, |
60 | 4 | num_rows, std::numeric_limits<size_t>::max())); |
61 | 4 | } |
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 | 4 | size_t row_pos, size_t num_rows, size_t max_bytes) { |
68 | 4 | Arena arena; |
69 | 4 | assert(num_cols <= block.columns()); |
70 | 4 | assert(row_pos + num_rows <= block.rows()); |
71 | 4 | DataTypeSerDe::FormatOptions options; |
72 | 4 | auto tz = cctz::utc_time_zone(); |
73 | 4 | options.timezone = &tz; |
74 | 4 | size_t written_rows = 0; |
75 | 1.03k | for (size_t i = row_pos; i < row_pos + num_rows; ++i) { |
76 | 1.03k | JsonbWriterT<JsonbOutStream> jsonb_writer; |
77 | 1.03k | jsonb_writer.writeStartObject(); |
78 | 10.2k | for (int j = 0; j < num_cols; ++j) { |
79 | 9.22k | const auto& column = block.get_by_position(j).column; |
80 | 9.22k | const auto& tablet_column = *schema.columns()[j]; |
81 | | // ignore row store columns |
82 | 9.22k | if (tablet_column.is_row_store_column()) { |
83 | 0 | continue; |
84 | 0 | } |
85 | | // TODO improve performance for checking column in group |
86 | 9.22k | if (row_store_cids.empty() || row_store_cids.contains(tablet_column.unique_id())) { |
87 | 9.22k | serdes[j]->write_one_cell_to_jsonb(*column, jsonb_writer, arena, |
88 | 9.22k | tablet_column.unique_id(), i, options); |
89 | 9.22k | } |
90 | 9.22k | } |
91 | 1.03k | jsonb_writer.writeEndObject(); |
92 | 1.03k | dst.insert_data(jsonb_writer.getOutput()->getBuffer(), jsonb_writer.getOutput()->getSize()); |
93 | 1.03k | ++written_rows; |
94 | 1.03k | if (dst.byte_size() >= max_bytes) { |
95 | 0 | break; |
96 | 0 | } |
97 | 1.03k | } |
98 | 4 | return written_rows; |
99 | 4 | } |
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 | 4 | const std::unordered_set<int>& include_cids) { |
107 | 1.03k | for (int i = 0; i < jsonb_column.size(); ++i) { |
108 | 1.03k | StringRef jsonb_data = jsonb_column.get_data_at(i); |
109 | 1.03k | RETURN_IF_ERROR(jsonb_to_block(serdes, jsonb_data.data, jsonb_data.size, col_id_to_idx, dst, |
110 | 1.03k | default_values, include_cids)); |
111 | 1.03k | } |
112 | 4 | return Status::OK(); |
113 | 4 | } |
114 | | |
115 | | // single row |
116 | | Status JsonbSerializeUtil::jsonb_to_block( |
117 | | const DataTypeSerDeSPtrs& serdes, const char* data, size_t size, |
118 | | const std::unordered_map<uint32_t, uint32_t>& col_id_to_idx, Block& dst, |
119 | | const std::vector<std::string>& default_values, |
120 | 1.03k | const std::unordered_set<int>& include_cids) { |
121 | 1.03k | const JsonbDocument* pdoc = nullptr; |
122 | 1.03k | RETURN_IF_ERROR(JsonbDocument::checkAndCreateDocument(data, size, &pdoc)); |
123 | 1.03k | const JsonbDocument& doc = *pdoc; |
124 | 1.03k | size_t num_rows = dst.rows(); |
125 | 1.03k | size_t filled_columns = 0; |
126 | 10.2k | for (auto it = doc->begin(); it != doc->end(); ++it) { |
127 | 9.22k | auto col_it = col_id_to_idx.find(it->getKeyId()); |
128 | 9.22k | if (col_it != col_id_to_idx.end() && |
129 | 9.22k | (include_cids.empty() || include_cids.contains(it->getKeyId()))) { |
130 | 9.22k | MutableColumnPtr dst_column = |
131 | 9.22k | dst.get_by_position(col_it->second).column->assume_mutable(); |
132 | 9.22k | serdes[col_it->second]->read_one_cell_from_jsonb(*dst_column, it->value()); |
133 | 9.22k | ++filled_columns; |
134 | 9.22k | } |
135 | 9.22k | } |
136 | 1.03k | if (filled_columns >= dst.columns()) { |
137 | 1.03k | return Status::OK(); |
138 | 1.03k | } |
139 | 0 | auto fill_column = [&](Block& dst, int pos, size_t old_num_rows) { |
140 | 0 | MutableColumnPtr dst_column = dst.get_by_position(pos).column->assume_mutable(); |
141 | 0 | if (dst_column->size() < old_num_rows + 1) { |
142 | 0 | DCHECK(dst_column->size() == old_num_rows); |
143 | 0 | if (default_values[pos].empty()) { |
144 | 0 | dst_column->insert_default(); |
145 | 0 | } else { |
146 | 0 | Slice value(default_values[pos].data(), default_values[pos].size()); |
147 | 0 | DataTypeSerDe::FormatOptions opt; |
148 | 0 | opt.converted_from_string = true; |
149 | 0 | RETURN_IF_ERROR( |
150 | 0 | serdes[pos]->deserialize_one_cell_from_json(*dst_column, value, opt)); |
151 | 0 | } |
152 | 0 | } |
153 | 0 | DCHECK(dst_column->size() == num_rows + 1); |
154 | 0 | return Status::OK(); |
155 | 0 | }; |
156 | | // fill missing column |
157 | 0 | if (!include_cids.empty()) { |
158 | 0 | for (auto cid : include_cids) { |
159 | 0 | auto col_it = col_id_to_idx.find(cid); |
160 | 0 | if (col_it == col_id_to_idx.end()) { |
161 | 0 | continue; |
162 | 0 | } |
163 | 0 | RETURN_IF_ERROR(fill_column(dst, col_it->second, num_rows)); |
164 | 0 | } |
165 | 0 | } else { |
166 | 0 | for (int i = 0; i < dst.columns(); ++i) { |
167 | 0 | RETURN_IF_ERROR(fill_column(dst, i, num_rows)); |
168 | 0 | } |
169 | 0 | } |
170 | 0 | return Status::OK(); |
171 | 0 | } |
172 | | |
173 | | } // namespace doris::vectorized |