jacktengg commented on code in PR #67627: URL: https://github.com/apache/doris/pull/67627#discussion_r4110400627
########## be/src/core/data_type_serde/data_type_uuid_serde.cpp: ########## @@ -0,0 +1,361 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "core/data_type_serde/data_type_uuid_serde.h" + +#include <arrow/builder.h> +#include <arrow/extension/uuid.h> + +#include <cstring> +#include <utility> + +#include "common/config.h" +#include "core/column/column_const.h" +#include "core/data_type_serde/arrow_validation.h" +#include "core/data_type_serde/orc_serde_utils.h" +#include "core/data_type_serde/parquet_decode_source.h" +#include "core/value/uuid_value.h" +#include "util/jsonb_writer.h" + +namespace doris { + +namespace { + +class UUIDParquetConsumer final : public ParquetFixedValueConsumer, + public ParquetBinaryValueConsumer { +public: + explicit UUIDParquetConsumer(IColumn& column) : _column(assert_cast<ColumnUUID&>(column)) {} + + Status consume(const uint8_t* values, size_t num_values, size_t value_width) override { + if (value_width != UUIDValue::BINARY_LENGTH) { + return Status::Corruption("Parquet UUID requires 16 bytes, got {}", value_width); + } + auto& data = _column.get_data(); + const auto offset = data.size(); + data.resize(offset + num_values); + for (size_t row = 0; row < num_values; ++row) { + data[offset + row] = UUIDValue::from_big_endian(values + row * value_width); + } + return Status::OK(); + } + + Status consume(const StringRef* values, size_t num_values) override { + return Status::Corruption("Parquet UUID requires FIXED_LEN_BYTE_ARRAY(16)"); + } + +private: + ColumnUUID& _column; +}; + +Status validate_parquet_uuid(const ParquetDecodeContext& context) { + if (context.physical_type != ParquetPhysicalType::FIXED_LEN_BYTE_ARRAY || + std::cmp_not_equal(context.type_length, UUIDValue::BINARY_LENGTH)) { + return Status::Corruption("Parquet UUID requires FIXED_LEN_BYTE_ARRAY(16)"); + } + return Status::OK(); +} + +} // namespace + +Status DataTypeUUIDSerDe::read_parquet_dictionary(IColumn& column, ParquetDecodeSource& source, + const ParquetDecodeContext& context) const { + RETURN_IF_ERROR(validate_parquet_uuid(context)); + UUIDParquetConsumer consumer(column); + return source.decode_dictionary(consumer, consumer); +} + +Status DataTypeUUIDSerDe::read_column_from_parquet(IColumn& column, ParquetDecodeSource& source, + const ParquetDecodeContext& context, + size_t num_values, + ParquetMaterializationState& state) const { + RETURN_IF_ERROR(validate_parquet_uuid(context)); + if (context.encoding != ParquetValueEncoding::DICTIONARY) { + UUIDParquetConsumer consumer(column); + return source.decode_fixed_values(num_values, consumer); + } + if (state.dictionary_generation != source.dictionary_generation()) { + state.typed_dictionary = column.clone_empty(); + RETURN_IF_ERROR(read_parquet_dictionary(*state.typed_dictionary, source, context)); + DORIS_CHECK_EQ(state.typed_dictionary->size(), source.dictionary_size()); + state.dictionary_generation = source.dictionary_generation(); + } + return state.materialize_dictionary(column, source, num_values); +} + +Status DataTypeUUIDSerDe::write_column_to_mysql_binary(const IColumn& column, + MysqlRowBinaryBuffer& result, + int64_t row_idx, bool col_const, + const FormatOptions& options) const { + const auto& data = assert_cast<const ColumnUUID&>(column).get_data(); + const auto col_index = index_check_const(row_idx, col_const); + const auto uuid = UUIDValue::to_string(data[col_index]); + if (UNLIKELY(result.push_string(uuid.data(), uuid.size()) != 0)) { + return Status::InternalError("pack mysql buffer failed."); + } + return Status::OK(); +} + +void DataTypeUUIDSerDe::read_one_cell_from_jsonb(IColumn& column, const JsonbValue* arg) const { + const auto* value = arg->unpack<JsonbBinaryVal>(); + column.deserialize_and_insert_from_arena(value->getBlob()); +} + +void DataTypeUUIDSerDe::write_one_cell_to_jsonb(const IColumn& column, + JsonbWriterT<JsonbOutStream>& result, Arena& arena, + int col_id, int64_t row_num, + const FormatOptions& options) const { + result.writeKey(cast_set<JsonbKeyValue::keyid_type>(col_id)); + const char* begin = nullptr; + StringRef value = column.serialize_value_into_arena(row_num, arena, begin); + result.writeStartBinary(); + result.writeBinary(value.data, value.size); + result.writeEndBinary(); +} + +Status DataTypeUUIDSerDe::serialize_one_cell_to_json(const IColumn& column, int64_t row_num, + BufferWritable& bw, + FormatOptions& options) const { + if (_nesting_level > 1) { + bw.write('"'); + } + RETURN_IF_ERROR(serialize_one_cell_to_hive_text(column, row_num, bw, options)); + if (_nesting_level > 1) { + bw.write('"'); + } + return Status::OK(); +} + +Status DataTypeUUIDSerDe::serialize_one_cell_to_hive_text( + const IColumn& column, int64_t row_num, BufferWritable& bw, FormatOptions& options, + int hive_text_complex_type_delimiter_level) const { + // Hive text has no JSON quotes, including UUID elements inside complex values. + auto [column_ptr, real_row_num] = check_column_const_set_readability(column, row_num); + const auto value = assert_cast<const ColumnUUID&>(*column_ptr).get_element(real_row_num); + const auto uuid = UUIDValue::to_string(value); + bw.write(uuid.data(), uuid.size()); + return Status::OK(); +} + +Status DataTypeUUIDSerDe::deserialize_one_cell_from_json(IColumn& column, Slice& slice, + const FormatOptions& options) const { + if (_nesting_level > 1) { + slice.trim_quote(); + } + return deserialize_one_cell_from_hive_text(column, slice, options); +} + +Status DataTypeUUIDSerDe::deserialize_one_cell_from_hive_text( + IColumn& column, Slice& slice, const FormatOptions& options, + int hive_text_complex_type_delimiter_level) const { + StringRef input(slice.data, slice.size); + return from_string(input, column, options); +} + +Status DataTypeUUIDSerDe::deserialize_column_from_hive_text_vector( + IColumn& column, std::vector<Slice>& slices, uint64_t* num_deserialized, + const FormatOptions& options, int hive_text_complex_type_delimiter_level) const { + DESERIALIZE_COLUMN_FROM_HIVE_TEXT_VECTOR(); + return Status::OK(); +} + +Status DataTypeUUIDSerDe::write_column_to_pb(const IColumn& column, PValues& result, int64_t start, Review Comment: fixed -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
