be/src/format_v2/table/hive_reader.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 "format_v2/table/hive_reader.h" |
19 | | |
20 | | #include <algorithm> |
21 | | #include <cctype> |
22 | | #include <string_view> |
23 | | |
24 | | #include "common/consts.h" |
25 | | #include "format_v2/column_mapper.h" |
26 | | #include "format_v2/file_reader.h" |
27 | | #include "runtime/runtime_state.h" |
28 | | |
29 | | namespace doris::format::hive { |
30 | | namespace { |
31 | | |
32 | 9 | TFileFormatType::type format_type_from_context(const format::ProjectedColumnBuildContext& context) { |
33 | 9 | DORIS_CHECK(context.scan_params != nullptr); |
34 | 9 | if (context.range != nullptr && context.range->__isset.format_type) { |
35 | 0 | return context.range->format_type; |
36 | 0 | } |
37 | 9 | return context.scan_params->format_type; |
38 | 9 | } |
39 | | |
40 | 9 | bool use_column_position_mapping(const format::ProjectedColumnBuildContext& context) { |
41 | 9 | if (context.runtime_state == nullptr || context.scan_params == nullptr) { |
42 | 0 | return false; |
43 | 0 | } |
44 | 9 | switch (format_type_from_context(context)) { |
45 | 7 | case TFileFormatType::FORMAT_PARQUET: |
46 | 7 | return !context.runtime_state->query_options().hive_parquet_use_column_names; |
47 | 1 | case TFileFormatType::FORMAT_ORC: |
48 | 1 | return !context.runtime_state->query_options().hive_orc_use_column_names; |
49 | 1 | default: |
50 | 1 | return false; |
51 | 9 | } |
52 | 9 | } |
53 | | |
54 | | bool is_file_column_position_slot(const TFileScanSlotInfo& slot_info, |
55 | 6 | const std::string& column_name) { |
56 | 6 | if (column_name.starts_with(BeConsts::GLOBAL_ROWID_COL) || |
57 | 6 | column_name == BeConsts::ICEBERG_ROWID_COL) { |
58 | 0 | return false; |
59 | 0 | } |
60 | 6 | if (slot_info.__isset.is_file_slot) { |
61 | 6 | return slot_info.is_file_slot; |
62 | 6 | } |
63 | 0 | return !slot_info.__isset.category || slot_info.category != TColumnCategory::PARTITION_KEY; |
64 | 6 | } |
65 | | |
66 | 3 | bool is_hive1_orc_column_name(std::string_view name) { |
67 | 3 | if (name.size() <= 4 || name.substr(0, 4) != "_col") { |
68 | 0 | return false; |
69 | 0 | } |
70 | 3 | return std::ranges::all_of(name.substr(4), [](unsigned char c) { return std::isdigit(c); }); |
71 | 3 | } |
72 | | |
73 | 1 | bool is_hive1_orc_file_schema(const std::vector<format::ColumnDefinition>& file_schema) { |
74 | 1 | bool has_data_column = false; |
75 | 3 | for (const auto& field : file_schema) { |
76 | 3 | if (field.column_type != format::ColumnType::DATA_COLUMN) { |
77 | 0 | continue; |
78 | 0 | } |
79 | 3 | has_data_column = true; |
80 | 3 | if (!is_hive1_orc_column_name(field.name)) { |
81 | 0 | return false; |
82 | 0 | } |
83 | 3 | } |
84 | 1 | return has_data_column; |
85 | 1 | } |
86 | | |
87 | 3 | bool is_file_column_for_hive1_position_mapping(const format::ColumnDefinition& column) { |
88 | 3 | if (column.column_type != format::ColumnType::DATA_COLUMN || column.is_partition_key) { |
89 | 0 | return false; |
90 | 0 | } |
91 | 3 | return !column.name.starts_with(BeConsts::GLOBAL_ROWID_COL) && |
92 | 3 | column.name != BeConsts::ICEBERG_ROWID_COL; |
93 | 3 | } |
94 | | |
95 | 6 | void add_name_mapping(std::vector<std::string>* name_mapping, const std::string& alias) { |
96 | 6 | DORIS_CHECK(name_mapping != nullptr); |
97 | 6 | if (std::ranges::find(*name_mapping, alias) == name_mapping->end()) { |
98 | 3 | name_mapping->push_back(alias); |
99 | 3 | } |
100 | 6 | } |
101 | | |
102 | | } // namespace |
103 | | |
104 | 2 | Status HiveReader::prepare_split(const format::SplitReadOptions& options) { |
105 | 2 | if (options.current_split_format != _format) { |
106 | 1 | return Status::InternalError( |
107 | 1 | "Hive scan expects all splits to use the same file format, " |
108 | 1 | "initialized_format={}, current_split_format={}", |
109 | 1 | static_cast<int>(_format), static_cast<int>(options.current_split_format)); |
110 | 1 | } |
111 | 1 | return format::TableReader::prepare_split(options); |
112 | 2 | } |
113 | | |
114 | 6 | format::TableColumnMappingMode HiveReader::mapping_mode() const { |
115 | | // Hive-specific behavior: choose the column matching mode based on file format and the |
116 | | // matching session variable. |
117 | | // - hive_orc_use_column_names / hive_parquet_use_column_names == true |
118 | | // => BY_NAME (modern Hive default, match by column name) |
119 | | // - those options == false |
120 | | // => BY_INDEX (mainly for Hive1 ORC `_col0` / `_col1`, match by top-level position; |
121 | | // Parquet exposes the same switch for consistency) |
122 | | // TableReader updates `_format` in prepare_split(), so this is evaluated per split. |
123 | 6 | DORIS_CHECK(_runtime_state != nullptr); |
124 | 6 | const auto& query_options = _runtime_state->query_options(); |
125 | 6 | bool use_column_names = true; |
126 | 6 | if (_format == format::FileFormat::ORC) { |
127 | 1 | use_column_names = query_options.hive_orc_use_column_names; |
128 | 5 | } else if (_format == format::FileFormat::PARQUET) { |
129 | 2 | use_column_names = query_options.hive_parquet_use_column_names; |
130 | 3 | } else if (_format == format::FileFormat::CSV || _format == format::FileFormat::TEXT || |
131 | 3 | _format == format::FileFormat::JSON) { |
132 | | // Hive CSV/TEXT/JSON readers synthesize a file-local schema from FE-provided file slots |
133 | | // because these formats do not carry embedded column names or field ids. The scan params' |
134 | | // format-specific attributes still tell the physical reader how to read values, while the |
135 | | // table-level mapper can safely match the synthesized file schema by table column name. |
136 | 3 | use_column_names = true; |
137 | 3 | } else { |
138 | 0 | DORIS_CHECK(false) << "HiveReader does not support this file reader format"; |
139 | 0 | } |
140 | | |
141 | 6 | return use_column_names ? format::TableColumnMappingMode::BY_NAME |
142 | 6 | : format::TableColumnMappingMode::BY_INDEX; |
143 | 6 | } |
144 | | |
145 | | Status HiveReader::annotate_projected_column(const TFileScanSlotInfo& slot_info, |
146 | | format::ProjectedColumnBuildContext* context, |
147 | 7 | format::ColumnDefinition* column) const { |
148 | 7 | RETURN_IF_ERROR(format::TableReader::annotate_projected_column(slot_info, context, column)); |
149 | 7 | DORIS_CHECK(context != nullptr); |
150 | 7 | DORIS_CHECK(column != nullptr); |
151 | 7 | if (!use_column_position_mapping(*context) || |
152 | 7 | !is_file_column_position_slot(slot_info, column->name)) { |
153 | 2 | return Status::OK(); |
154 | 2 | } |
155 | 5 | const auto* scan_params = context->scan_params; |
156 | 5 | DORIS_CHECK(scan_params != nullptr); |
157 | 5 | if (!scan_params->__isset.column_idxs || |
158 | 5 | context->next_file_column_idx >= scan_params->column_idxs.size()) { |
159 | 1 | return Status::InvalidArgument( |
160 | 1 | "Hive positional column mapping is missing file index for column '{}', " |
161 | 1 | "required file slot ordinal={}, column_idxs_size={}", |
162 | 1 | column->name, context->next_file_column_idx, |
163 | 1 | scan_params->__isset.column_idxs ? scan_params->column_idxs.size() : 0); |
164 | 1 | } |
165 | 4 | const auto file_index = scan_params->column_idxs[context->next_file_column_idx]; |
166 | 4 | if (file_index < 0) { |
167 | 0 | return Status::InvalidArgument( |
168 | 0 | "Hive positional column mapping has negative file index {} for column '{}'", |
169 | 0 | file_index, column->name); |
170 | 0 | } |
171 | 4 | column->identifier = Field::create_field<TYPE_INT>(file_index); |
172 | 4 | ++context->next_file_column_idx; |
173 | 4 | return Status::OK(); |
174 | 4 | } |
175 | | |
176 | | Status HiveReader::validate_projected_columns( |
177 | 2 | const format::ProjectedColumnBuildContext& context) const { |
178 | 2 | if (!use_column_position_mapping(context)) { |
179 | 0 | return Status::OK(); |
180 | 0 | } |
181 | 2 | DORIS_CHECK(context.scan_params != nullptr); |
182 | 2 | if (context.scan_params->__isset.column_idxs && |
183 | 2 | context.next_file_column_idx != context.scan_params->column_idxs.size()) { |
184 | 0 | return Status::InvalidArgument( |
185 | 0 | "Hive positional column mapping has unused file indexes: consumed={}, " |
186 | 0 | "column_idxs_size={}", |
187 | 0 | context.next_file_column_idx, context.scan_params->column_idxs.size()); |
188 | 0 | } |
189 | 2 | return Status::OK(); |
190 | 2 | } |
191 | | |
192 | 1 | Status HiveReader::annotate_file_schema(std::vector<format::ColumnDefinition>* file_schema) { |
193 | 1 | DORIS_CHECK(file_schema != nullptr); |
194 | 1 | if (_format != format::FileFormat::ORC || _runtime_state == nullptr || |
195 | 1 | !_runtime_state->query_options().hive_orc_use_column_names || |
196 | 1 | !is_hive1_orc_file_schema(*file_schema)) { |
197 | 0 | return Status::OK(); |
198 | 0 | } |
199 | | |
200 | 1 | DORIS_CHECK(_scan_params != nullptr); |
201 | 1 | if (!_scan_params->__isset.column_idxs) { |
202 | 0 | return Status::InvalidArgument( |
203 | 0 | "Hive ORC Hive1-style name mapping is missing positional column indexes"); |
204 | 0 | } |
205 | | |
206 | 1 | size_t next_file_column_idx = 0; |
207 | 3 | for (const auto& table_column : _projected_columns) { |
208 | 3 | if (!is_file_column_for_hive1_position_mapping(table_column)) { |
209 | 0 | continue; |
210 | 0 | } |
211 | 3 | if (next_file_column_idx >= _scan_params->column_idxs.size()) { |
212 | 0 | return Status::InvalidArgument( |
213 | 0 | "Hive ORC Hive1-style name mapping is missing file index for column '{}', " |
214 | 0 | "required file slot ordinal={}, column_idxs_size={}", |
215 | 0 | table_column.name, next_file_column_idx, _scan_params->column_idxs.size()); |
216 | 0 | } |
217 | 3 | const auto file_index = _scan_params->column_idxs[next_file_column_idx]; |
218 | 3 | if (file_index < 0) { |
219 | 0 | return Status::InvalidArgument( |
220 | 0 | "Hive ORC Hive1-style name mapping has negative file index {} for column '{}'", |
221 | 0 | file_index, table_column.name); |
222 | 0 | } |
223 | 3 | ++next_file_column_idx; |
224 | 3 | if (static_cast<size_t>(file_index) >= file_schema->size()) { |
225 | 0 | continue; |
226 | 0 | } |
227 | | |
228 | 3 | auto& file_column = (*file_schema)[static_cast<size_t>(file_index)]; |
229 | 3 | add_name_mapping(&file_column.name_mapping, table_column.name); |
230 | 3 | if (table_column.has_identifier_name()) { |
231 | 3 | add_name_mapping(&file_column.name_mapping, table_column.get_identifier_name()); |
232 | 3 | } |
233 | 3 | for (const auto& alias : table_column.name_mapping) { |
234 | 0 | add_name_mapping(&file_column.name_mapping, alias); |
235 | 0 | } |
236 | 3 | } |
237 | 1 | if (next_file_column_idx != _scan_params->column_idxs.size()) { |
238 | 0 | return Status::InvalidArgument( |
239 | 0 | "Hive ORC Hive1-style name mapping has unused file indexes: consumed={}, " |
240 | 0 | "column_idxs_size={}", |
241 | 0 | next_file_column_idx, _scan_params->column_idxs.size()); |
242 | 0 | } |
243 | 1 | return Status::OK(); |
244 | 1 | } |
245 | | |
246 | | } // namespace doris::format::hive |