be/src/format/jni/jni_reader.h
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 | | #pragma once |
19 | | |
20 | | #include <cstddef> |
21 | | #include <map> |
22 | | #include <memory> |
23 | | #include <string> |
24 | | #include <unordered_map> |
25 | | #include <unordered_set> |
26 | | #include <vector> |
27 | | |
28 | | #include "common/status.h" |
29 | | #include "format/generic_reader.h" |
30 | | #include "format/jni/jni_data_bridge.h" |
31 | | #include "runtime/runtime_profile.h" |
32 | | #include "util/jni-util.h" |
33 | | #include "util/jni_plugin_registry.h" |
34 | | #include "util/profile_collector.h" |
35 | | #include "util/string_util.h" |
36 | | |
37 | | namespace doris { |
38 | | class RuntimeProfile; |
39 | | class RuntimeState; |
40 | | class SlotDescriptor; |
41 | | class Block; |
42 | | } // namespace doris |
43 | | |
44 | | namespace doris { |
45 | | |
46 | | /** |
47 | | * JniReader is the base class for all JNI-based readers. It directly manages |
48 | | * the JNI lifecycle (open/read/close) for Java scanners that extend |
49 | | * org.apache.doris.jni.spi.JniScanner. |
50 | | * |
51 | | * Subclasses only need to: |
52 | | * 1. Build scanner_params/column_names in their constructor |
53 | | * 2. Pass them to JniReader's constructor |
54 | | * 3. Call open() in their init_reader() |
55 | | * |
56 | | * This class replaces the old JniConnector intermediary. |
57 | | */ |
58 | | class JniReader : public GenericReader { |
59 | | public: |
60 | | /** |
61 | | * Constructor for scan mode. |
62 | | * @param file_slot_descs Slot descriptors for the output columns |
63 | | * @param state Runtime state |
64 | | * @param profile Runtime profile for metrics |
65 | | * @param plugin_ref Plugin and factory that build the Java scanner (e.g. {"paimon", "paimon"}) |
66 | | * @param scanner_params Configuration map passed to the Java scanner factory |
67 | | * @param column_names Fields to read (also the required_fields in scanner_params) |
68 | | * @param self_split_weight Weight for this split (for profile conditition counter) |
69 | | */ |
70 | | JniReader(const std::vector<SlotDescriptor*>& file_slot_descs, RuntimeState* state, |
71 | | RuntimeProfile* profile, Jni::PluginRef plugin_ref, |
72 | | std::map<std::string, std::string> scanner_params, |
73 | | std::vector<std::string> column_names, int64_t self_split_weight = -1); |
74 | | |
75 | | /** |
76 | | * Constructor for table-schema-only mode (no data reading). |
77 | | * @param plugin_ref Plugin and factory that build the Java scanner |
78 | | * @param scanner_params Configuration map passed to the Java scanner factory |
79 | | */ |
80 | | JniReader(Jni::PluginRef plugin_ref, std::map<std::string, std::string> scanner_params); |
81 | | |
82 | 5 | ~JniReader() override = default; |
83 | | |
84 | | /** |
85 | | * Open the java scanner: set up profile counters, create Java object, |
86 | | * get method IDs, and call JniScanner#open. |
87 | | */ |
88 | | Status open(RuntimeState* state, RuntimeProfile* profile); |
89 | | |
90 | 0 | Status _get_columns_impl(std::unordered_map<std::string, DataTypePtr>* name_to_type) override { |
91 | 0 | for (const auto& desc : _file_slot_descs) { |
92 | 0 | name_to_type->emplace(desc->col_name(), desc->type()); |
93 | 0 | } |
94 | 0 | return Status::OK(); |
95 | 0 | } |
96 | | |
97 | | void set_batch_size(size_t batch_size) override; |
98 | 0 | size_t get_batch_size() const override { return _batch_size; } |
99 | | |
100 | | /** |
101 | | * Read next batch from Java scanner and fill the block. |
102 | | */ |
103 | | Status _do_get_next_block(Block* block, size_t* read_rows, bool* eof) override; |
104 | | |
105 | | /** |
106 | | * Close the scanner and release JNI resources. |
107 | | */ |
108 | | Status close() override; |
109 | | |
110 | | /** |
111 | | * Set column name to block index map from FileScanner to avoid repeated map creation. |
112 | | */ |
113 | | void set_col_name_to_block_idx( |
114 | 0 | const std::unordered_map<std::string, uint32_t>* col_name_to_block_idx) { |
115 | 0 | _col_name_to_block_idx = col_name_to_block_idx; |
116 | 0 | } |
117 | | |
118 | | protected: |
119 | | Status on_before_init_reader(ReaderInitContext* ctx) override; |
120 | | Status on_after_read_block(Block* block, size_t* read_rows) override; |
121 | | void _collect_profile_before_close() override; |
122 | | |
123 | | /** |
124 | | * Update scanner params and column names after construction. |
125 | | * Used by Avro which builds params in init_reader/init_schema_reader |
126 | | * rather than in the constructor. |
127 | | */ |
128 | | void _update_scanner_params(std::map<std::string, std::string> params, |
129 | 0 | std::vector<std::string> column_names) { |
130 | 0 | _scanner_params = std::move(params); |
131 | 0 | _column_names = std::move(column_names); |
132 | 0 | } |
133 | | |
134 | | const std::vector<SlotDescriptor*>& _file_slot_descs; |
135 | | RuntimeState* _state = nullptr; |
136 | | RuntimeProfile* _profile = nullptr; |
137 | | |
138 | | private: |
139 | | static const std::vector<SlotDescriptor*> _s_empty_slot_descs; |
140 | | |
141 | | Status _fill_partition_columns(Block* block, size_t num_rows); |
142 | | Status _init_jni_scanner(JNIEnv* env, int batch_size); |
143 | | Status _fill_block(Block* block, size_t num_rows); |
144 | | Status _get_statistics(JNIEnv* env, std::map<std::string, std::string>* result); |
145 | | |
146 | | Jni::PluginRef _plugin_ref; |
147 | | std::string _connector_name; |
148 | | std::map<std::string, std::string> _scanner_params; |
149 | | std::vector<std::string> _column_names; |
150 | | int32_t _self_split_weight = -1; |
151 | | bool _is_table_schema = false; |
152 | | |
153 | | RuntimeProfile::Counter* _open_scanner_time = nullptr; |
154 | | RuntimeProfile::Counter* _java_scan_time = nullptr; |
155 | | RuntimeProfile::Counter* _java_append_data_time = nullptr; |
156 | | RuntimeProfile::Counter* _java_create_vector_table_time = nullptr; |
157 | | RuntimeProfile::Counter* _fill_block_time = nullptr; |
158 | | RuntimeProfile::ConditionCounter* _max_time_split_weight_counter = nullptr; |
159 | | |
160 | | int64_t _jni_scanner_open_watcher = 0; |
161 | | int64_t _java_scan_watcher = 0; |
162 | | int64_t _fill_block_watcher = 0; |
163 | | |
164 | | size_t _has_read = 0; |
165 | | |
166 | | bool _closed = false; |
167 | | bool _scanner_opened = false; |
168 | | |
169 | | Jni::GlobalObject _jni_scanner_obj; |
170 | | // Resolved on the SPI base class and shared by every reader in the process, so this is a |
171 | | // borrowed pointer into storage that outlives any reader. |
172 | | const Jni::ScannerApi* _scanner_api = nullptr; |
173 | | |
174 | | JniDataBridge::TableMetaAddress _table_meta; |
175 | | size_t _batch_size = 0; |
176 | | |
177 | | // Column name to block index map, passed from FileScanner to avoid repeated map creation |
178 | | const std::unordered_map<std::string, uint32_t>* _col_name_to_block_idx = nullptr; |
179 | | std::unordered_map<std::string, std::tuple<std::string, const SlotDescriptor*>> |
180 | | _partition_values; |
181 | | std::unordered_map<std::string, bool> _partition_value_is_null; |
182 | | |
183 | 0 | void _set_meta(long meta_addr) { _table_meta.set_meta(meta_addr); } |
184 | | }; |
185 | | |
186 | | } // namespace doris |