| 171 | } |
| 172 | |
| 173 | void FileMetadataUtils::AddIcebergColumns(MemPool* mem_pool, Tuple** template_tuple, |
| 174 | std::map<const SlotId, const SlotDescriptor*>* slot_descs_written) { |
| 175 | using namespace org::apache::impala::fb; |
| 176 | TextConverter text_converter = CreateTextConverter(); |
| 177 | const FbSplitFileMetadata* file_metadata = file_desc_->file_metadata; |
| 178 | const FbIcebergSplitMetadata* ice_metadata = file_metadata->iceberg_metadata(); |
| 179 | auto transforms = ice_metadata->partition_keys(); |
| 180 | |
| 181 | const TupleDescriptor* tuple_desc = scan_node_->tuple_desc(); |
| 182 | if (*template_tuple == nullptr) { |
| 183 | *template_tuple = Tuple::Create(tuple_desc->byte_size(), mem_pool); |
| 184 | } |
| 185 | for (const SlotDescriptor* slot_desc : scan_node_->tuple_desc()->slots()) { |
| 186 | if (slot_desc->IsVirtual()) { |
| 187 | AddVirtualIcebergColumn(mem_pool, *template_tuple, *ice_metadata, slot_desc); |
| 188 | continue; |
| 189 | } |
| 190 | if (transforms == nullptr) continue; |
| 191 | const SchemaPath& path = slot_desc->col_path(); |
| 192 | if (path.size() != 1) continue; |
| 193 | const ColumnDescriptor& col_desc = |
| 194 | scan_node_->hdfs_table()->col_descs()[path.front()]; |
| 195 | int field_id = col_desc.field_id(); |
| 196 | for (int i = 0; i < transforms->size(); ++i) { |
| 197 | auto transform = transforms->Get(i); |
| 198 | if (transform->transform_type() != |
| 199 | FbIcebergTransformType::FbIcebergTransformType_IDENTITY) { |
| 200 | continue; |
| 201 | } |
| 202 | if (field_id != transform->source_id()) continue; |
| 203 | if (!text_converter.WriteSlot(slot_desc, *template_tuple, |
| 204 | (const char*)transform->transform_value()->data(), |
| 205 | transform->transform_value()->size(), |
| 206 | /* copy_string = */ true, /* need_escape = */ false, |
| 207 | mem_pool)) { |
| 208 | ErrorMsg error_msg(TErrorCode::GENERAL, |
| 209 | Substitute("Could not parse partition value for " |
| 210 | "column '$0' in file '$1'. Partition string is '$2' " |
| 211 | "NULL Partition key value is '$3'", |
| 212 | col_desc.name(), file_desc_->filename, |
| 213 | transform->transform_value()->data(), |
| 214 | scan_node_->hdfs_table()->null_partition_key_value())); |
| 215 | // Dates are stored as INTs in the partition data in Iceberg, so let's try |
| 216 | // to parse them as INTs. |
| 217 | if (col_desc.type().type == PrimitiveType::TYPE_DATE) { |
| 218 | int32_t* slot = (*template_tuple)->GetIntSlot(slot_desc->tuple_offset()); |
| 219 | StringParser::ParseResult parse_result; |
| 220 | *slot = StringParser::StringToInt<int32_t>( |
| 221 | (const char*)transform->transform_value()->data(), |
| 222 | transform->transform_value()->size(), |
| 223 | &parse_result); |
| 224 | if (parse_result == StringParser::ParseResult::PARSE_SUCCESS) { |
| 225 | (*template_tuple)->SetNotNull(slot_desc->null_indicator_offset()); |
| 226 | slot_descs_written->insert({slot_desc->id(), slot_desc}); |
| 227 | } else { |
| 228 | state_->LogError(error_msg); |
| 229 | } |
| 230 | } else { |
nothing calls this directly
no test coverage detected