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 <memory> |
24 | | #include <unordered_set> |
25 | | #include <vector> |
26 | | |
27 | | #include "core/arena.h" |
28 | | #include "core/block/column_with_type_and_name.h" |
29 | | #include "core/block/columns_with_type_and_name.h" |
30 | | #include "core/column/column.h" |
31 | | #include "core/column/column_string.h" |
32 | | #include "core/data_type/data_type.h" |
33 | | #include "core/data_type/primitive_type.h" |
34 | | #include "core/data_type_serde/data_type_serde.h" |
35 | | #include "core/string_ref.h" |
36 | | #include "core/value/jsonb_value.h" |
37 | | #include "runtime/descriptors.h" |
38 | | #include "storage/tablet/tablet_schema.h" |
39 | | #include "util/defer_op.h" |
40 | | #include "util/jsonb_document.h" |
41 | | #include "util/jsonb_stream.h" |
42 | | #include "util/jsonb_writer.h" |
43 | | |
44 | | namespace doris { |
45 | | |
46 | | void JsonbSerializeUtil::block_to_jsonb(const TabletSchema& schema, const Block& block, |
47 | | ColumnString& dst, int num_cols, |
48 | | const DataTypeSerDeSPtrs& serdes, |
49 | 1.46k | const std::unordered_set<int32_t>& row_store_cids) { |
50 | 1.46k | auto num_rows = block.rows(); |
51 | 1.46k | Arena arena; |
52 | 1.46k | assert(num_cols <= block.columns()); |
53 | 1.45k | DataTypeSerDe::FormatOptions options; |
54 | 1.45k | auto tz = cctz::utc_time_zone(); |
55 | 1.45k | options.timezone = &tz; |
56 | 1.91M | for (int i = 0; i < num_rows; ++i) { |
57 | 1.91M | JsonbWriterT<JsonbOutStream> jsonb_writer; |
58 | 1.91M | jsonb_writer.writeStartObject(); |
59 | 38.9M | for (int j = 0; j < num_cols; ++j) { |
60 | 37.0M | const auto& column = block.get_by_position(j).column; |
61 | 37.0M | const auto& tablet_column = *schema.columns()[j]; |
62 | | // ignore row store columns |
63 | 37.0M | if (tablet_column.is_row_store_column()) { |
64 | 1.91M | continue; |
65 | 1.91M | } |
66 | | // TODO improve performance for checking column in group |
67 | 35.1M | if (row_store_cids.empty() || row_store_cids.contains(tablet_column.unique_id())) { |
68 | 23.5M | serdes[j]->write_one_cell_to_jsonb(*column, jsonb_writer, arena, |
69 | 23.5M | tablet_column.unique_id(), i, options); |
70 | 23.5M | } |
71 | 35.1M | } |
72 | 1.91M | jsonb_writer.writeEndObject(); |
73 | 1.91M | dst.insert_data(jsonb_writer.getOutput()->getBuffer(), jsonb_writer.getOutput()->getSize()); |
74 | 1.91M | } |
75 | 1.45k | } |
76 | | |
77 | | // batch rows |
78 | | Status JsonbSerializeUtil::jsonb_to_block( |
79 | | const DataTypeSerDeSPtrs& serdes, const ColumnString& jsonb_column, |
80 | | const std::unordered_map<uint32_t, uint32_t>& col_id_to_idx, Block& dst, |
81 | | const std::vector<std::string>& default_values, |
82 | 343 | const std::unordered_set<int>& include_cids) { |
83 | 343 | auto dst_columns_guard = dst.mutate_columns_scoped(); |
84 | 343 | MutableColumns& dst_columns = dst_columns_guard.mutable_columns(); |
85 | 2.60k | for (int i = 0; i < jsonb_column.size(); ++i) { |
86 | 2.26k | StringRef jsonb_data = jsonb_column.get_data_at(i); |
87 | 2.26k | RETURN_IF_ERROR(jsonb_to_columns(serdes, jsonb_data.data, jsonb_data.size, col_id_to_idx, |
88 | 2.26k | dst_columns, default_values, include_cids)); |
89 | 2.26k | } |
90 | 343 | return Status::OK(); |
91 | 343 | } |
92 | | |
93 | | Status JsonbSerializeUtil::jsonb_to_columns( |
94 | | const DataTypeSerDeSPtrs& serdes, const char* data, size_t size, |
95 | | const std::unordered_map<uint32_t, uint32_t>& col_id_to_idx, MutableColumns& dst_columns, |
96 | | const std::vector<std::string>& default_values, |
97 | 6.90k | const std::unordered_set<int>& include_cids) { |
98 | 6.90k | const JsonbDocument* pdoc = nullptr; |
99 | 6.90k | RETURN_IF_ERROR(JsonbDocument::checkAndCreateDocument(data, size, &pdoc)); |
100 | 6.90k | const JsonbDocument& doc = *pdoc; |
101 | 6.90k | DCHECK(!dst_columns.empty()); |
102 | 6.90k | size_t num_rows = dst_columns[0]->size(); |
103 | 6.90k | size_t filled_columns = 0; |
104 | 115k | for (auto it = doc->begin(); it != doc->end(); ++it) { |
105 | 109k | auto col_it = col_id_to_idx.find(it->getKeyId()); |
106 | 109k | if (col_it != col_id_to_idx.end() && |
107 | 109k | (include_cids.empty() || include_cids.contains(it->getKeyId()))) { |
108 | 94.7k | auto& dst_column = dst_columns[col_it->second]; |
109 | 94.7k | serdes[col_it->second]->read_one_cell_from_jsonb(*dst_column, it->value()); |
110 | 94.7k | ++filled_columns; |
111 | 94.7k | } |
112 | 109k | } |
113 | 6.90k | if (filled_columns >= dst_columns.size()) { |
114 | 5.70k | return Status::OK(); |
115 | 5.70k | } |
116 | 16.6k | auto fill_column = [&](size_t pos, size_t old_num_rows) { |
117 | 16.6k | auto& dst_column = dst_columns[pos]; |
118 | 16.6k | if (dst_column->size() < old_num_rows + 1) { |
119 | 1.22k | DCHECK(dst_column->size() == old_num_rows); |
120 | 1.22k | Status st = Status::OK(); |
121 | 1.22k | if (default_values[pos].empty()) { |
122 | 1.18k | dst_column->insert_default(); |
123 | 1.18k | } else { |
124 | 39 | Slice value(default_values[pos].data(), default_values[pos].size()); |
125 | 39 | DataTypeSerDe::FormatOptions opt; |
126 | 39 | opt.converted_from_string = true; |
127 | 39 | st = serdes[pos]->deserialize_one_cell_from_json(*dst_column, value, opt); |
128 | 39 | } |
129 | 1.22k | RETURN_IF_ERROR(st); |
130 | 1.22k | DCHECK(dst_column->size() == num_rows + 1); |
131 | 1.22k | return Status::OK(); |
132 | 1.22k | } |
133 | 16.6k | DCHECK(dst_column->size() == num_rows + 1); |
134 | 15.4k | return Status::OK(); |
135 | 16.6k | }; |
136 | | // fill missing column |
137 | 1.20k | if (!include_cids.empty()) { |
138 | 18 | for (auto cid : include_cids) { |
139 | 18 | auto col_it = col_id_to_idx.find(cid); |
140 | 18 | if (col_it == col_id_to_idx.end()) { |
141 | 2 | continue; |
142 | 2 | } |
143 | 16 | RETURN_IF_ERROR(fill_column(static_cast<size_t>(col_it->second), num_rows)); |
144 | 16 | } |
145 | 1.19k | } else { |
146 | 17.7k | for (size_t i = 0; i < dst_columns.size(); ++i) { |
147 | 16.5k | RETURN_IF_ERROR(fill_column(i, num_rows)); |
148 | 16.5k | } |
149 | 1.19k | } |
150 | 1.20k | return Status::OK(); |
151 | 1.20k | } |
152 | | |
153 | | // single row |
154 | | Status JsonbSerializeUtil::jsonb_to_block( |
155 | | const DataTypeSerDeSPtrs& serdes, const char* data, size_t size, |
156 | | const std::unordered_map<uint32_t, uint32_t>& col_id_to_idx, Block& dst, |
157 | | const std::vector<std::string>& default_values, |
158 | 0 | const std::unordered_set<int>& include_cids) { |
159 | 0 | auto dst_columns_guard = dst.mutate_columns_scoped(); |
160 | 0 | MutableColumns& dst_columns = dst_columns_guard.mutable_columns(); |
161 | 0 | return jsonb_to_columns(serdes, data, size, col_id_to_idx, dst_columns, default_values, |
162 | 0 | include_cids); |
163 | 0 | } |
164 | | |
165 | | } // namespace doris |