/root/doris/be/src/vec/jsonb/serialize.cpp
Line | Count | Source (jump to first uncovered line) |
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 <memory> |
24 | | #include <unordered_set> |
25 | | #include <vector> |
26 | | |
27 | | #include "olap/tablet_schema.h" |
28 | | #include "runtime/descriptors.h" |
29 | | #include "runtime/jsonb_value.h" |
30 | | #include "runtime/primitive_type.h" |
31 | | #include "runtime/types.h" |
32 | | #include "util/jsonb_document.h" |
33 | | #include "util/jsonb_stream.h" |
34 | | #include "util/jsonb_writer.h" |
35 | | #include "vec/columns/column.h" |
36 | | #include "vec/columns/column_string.h" |
37 | | #include "vec/common/arena.h" |
38 | | #include "vec/common/string_ref.h" |
39 | | #include "vec/core/column_with_type_and_name.h" |
40 | | #include "vec/core/columns_with_type_and_name.h" |
41 | | #include "vec/data_types/data_type.h" |
42 | | #include "vec/data_types/serde/data_type_serde.h" |
43 | | #include "vec/io/reader_buffer.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 | auto num_rows = block.rows(); |
52 | 4 | Arena pool; |
53 | 4 | assert(num_cols <= block.columns()); |
54 | 1.03k | for (int i = 0; i < num_rows; ++i) { |
55 | 1.03k | JsonbWriterT<JsonbOutStream> jsonb_writer; |
56 | 1.03k | jsonb_writer.writeStartObject(); |
57 | 10.2k | for (int j = 0; j < num_cols; ++j) { |
58 | 9.22k | const auto& column = block.get_by_position(j).column; |
59 | 9.22k | const auto& tablet_column = *schema.columns()[j]; |
60 | | // ignore row store columns |
61 | 9.22k | if (tablet_column.is_row_store_column()) { |
62 | 0 | continue; |
63 | 0 | } |
64 | | // TODO improve performance for checking column in group |
65 | 9.22k | if (row_store_cids.empty() || row_store_cids.contains(tablet_column.unique_id())) { |
66 | 9.22k | serdes[j]->write_one_cell_to_jsonb(*column, jsonb_writer, &pool, |
67 | 9.22k | tablet_column.unique_id(), i); |
68 | 9.22k | } |
69 | 9.22k | } |
70 | 1.03k | jsonb_writer.writeEndObject(); |
71 | 1.03k | dst.insert_data(jsonb_writer.getOutput()->getBuffer(), jsonb_writer.getOutput()->getSize()); |
72 | 1.03k | } |
73 | 4 | } |
74 | | |
75 | | // batch rows |
76 | | void JsonbSerializeUtil::jsonb_to_block(const DataTypeSerDeSPtrs& serdes, |
77 | | const ColumnString& jsonb_column, |
78 | | const std::unordered_map<uint32_t, uint32_t>& col_id_to_idx, |
79 | | Block& dst, const std::vector<std::string>& default_values, |
80 | 4 | const std::unordered_set<int>& include_cids) { |
81 | 1.03k | for (int i = 0; i < jsonb_column.size(); ++i) { |
82 | 1.03k | StringRef jsonb_data = jsonb_column.get_data_at(i); |
83 | 1.03k | jsonb_to_block(serdes, jsonb_data.data, jsonb_data.size, col_id_to_idx, dst, default_values, |
84 | 1.03k | include_cids); |
85 | 1.03k | } |
86 | 4 | } |
87 | | |
88 | | // single row |
89 | | void JsonbSerializeUtil::jsonb_to_block(const DataTypeSerDeSPtrs& serdes, const char* data, |
90 | | size_t size, |
91 | | const std::unordered_map<uint32_t, uint32_t>& col_id_to_idx, |
92 | | Block& dst, const std::vector<std::string>& default_values, |
93 | 1.03k | const std::unordered_set<int>& include_cids) { |
94 | 1.03k | auto pdoc = JsonbDocument::createDocument(data, size); |
95 | 1.03k | JsonbDocument& doc = *pdoc; |
96 | 1.03k | size_t num_rows = dst.rows(); |
97 | 1.03k | size_t filled_columns = 0; |
98 | 10.2k | for (auto it = doc->begin(); it != doc->end(); ++it) { |
99 | 9.22k | auto col_it = col_id_to_idx.find(it->getKeyId()); |
100 | 9.22k | if (col_it != col_id_to_idx.end() && |
101 | 9.22k | (include_cids.empty() || include_cids.contains(it->getKeyId()))) { |
102 | 9.22k | MutableColumnPtr dst_column = |
103 | 9.22k | dst.get_by_position(col_it->second).column->assume_mutable(); |
104 | 9.22k | serdes[col_it->second]->read_one_cell_from_jsonb(*dst_column, it->value()); |
105 | 9.22k | ++filled_columns; |
106 | 9.22k | } |
107 | 9.22k | } |
108 | 1.03k | if (filled_columns >= dst.columns()) { |
109 | 1.03k | return; |
110 | 1.03k | } |
111 | 0 | auto fill_column = [&](Block& dst, int pos, size_t old_num_rows) { |
112 | 0 | MutableColumnPtr dst_column = dst.get_by_position(pos).column->assume_mutable(); |
113 | 0 | if (dst_column->size() < old_num_rows + 1) { |
114 | 0 | DCHECK(dst_column->size() == old_num_rows); |
115 | 0 | if (default_values[pos].empty()) { |
116 | 0 | dst_column->insert_default(); |
117 | 0 | } else { |
118 | 0 | Slice value(default_values[pos].data(), default_values[pos].size()); |
119 | 0 | DataTypeSerDe::FormatOptions opt; |
120 | 0 | opt.converted_from_string = true; |
121 | 0 | static_cast<void>( |
122 | 0 | serdes[pos]->deserialize_one_cell_from_json(*dst_column, value, opt)); |
123 | 0 | } |
124 | 0 | } |
125 | 0 | DCHECK(dst_column->size() == num_rows + 1); |
126 | 0 | }; |
127 | | // fill missing column |
128 | 0 | if (!include_cids.empty()) { |
129 | 0 | for (auto cid : include_cids) { |
130 | 0 | auto col_it = col_id_to_idx.find(cid); |
131 | 0 | if (col_it == col_id_to_idx.end()) { |
132 | 0 | continue; |
133 | 0 | } |
134 | 0 | fill_column(dst, col_it->second, num_rows); |
135 | 0 | } |
136 | 0 | } else { |
137 | 0 | for (int i = 0; i < dst.columns(); ++i) { |
138 | 0 | fill_column(dst, i, num_rows); |
139 | 0 | } |
140 | 0 | } |
141 | 0 | } |
142 | | |
143 | | } // namespace doris::vectorized |