| /* |
| * 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 "paimon/common/data/serializer/row_compacted_serializer.h" |
| |
| #include "paimon/common/data/binary_array.h" |
| #include "paimon/common/data/binary_array_writer.h" |
| #include "paimon/common/data/binary_map.h" |
| #include "paimon/common/data/binary_row_writer.h" |
| #include "paimon/common/data/data_define.h" |
| #include "paimon/common/data/generic_row.h" |
| #include "paimon/common/data/serializer/binary_serializer_utils.h" |
| #include "paimon/common/utils/date_time_utils.h" |
| #include "paimon/common/utils/fields_comparator.h" |
| |
| namespace paimon { |
| Result<std::unique_ptr<RowCompactedSerializer>> RowCompactedSerializer::Create( |
| const std::shared_ptr<arrow::Schema>& schema, const std::shared_ptr<MemoryPool>& pool) { |
| std::vector<InternalRow::FieldGetterFunc> getters(schema->num_fields()); |
| std::vector<RowCompactedSerializer::FieldWriter> writers(schema->num_fields()); |
| std::vector<RowCompactedSerializer::FieldReader> readers(schema->num_fields()); |
| for (int32_t i = 0; i < schema->num_fields(); i++) { |
| auto field_type = schema->field(i)->type(); |
| PAIMON_ASSIGN_OR_RAISE(getters[i], |
| InternalRow::CreateFieldGetter(i, field_type, /*use_view=*/true)); |
| PAIMON_ASSIGN_OR_RAISE(writers[i], CreateFieldWriter(field_type, pool)); |
| PAIMON_ASSIGN_OR_RAISE(readers[i], CreateFieldReader(field_type, pool)); |
| } |
| return std::unique_ptr<RowCompactedSerializer>(new RowCompactedSerializer( |
| schema, std::move(getters), std::move(writers), std::move(readers), pool)); |
| } |
| |
| Result<int32_t> RowCompactedSerializer::CompareField(const FieldInfo& field_info, |
| RowReader* reader1, RowReader* reader2) { |
| auto type = field_info.type_id; |
| switch (type) { |
| case arrow::Type::type::BOOL: { |
| auto val1 = reader1->ReadValue<bool>(); |
| auto val2 = reader2->ReadValue<bool>(); |
| return val1 == val2 ? 0 : (val1 < val2 ? -1 : 1); |
| } |
| case arrow::Type::type::INT8: { |
| auto val1 = reader1->ReadValue<char>(); |
| auto val2 = reader2->ReadValue<char>(); |
| return val1 == val2 ? 0 : (val1 < val2 ? -1 : 1); |
| } |
| case arrow::Type::type::INT16: { |
| auto val1 = reader1->ReadValue<int16_t>(); |
| auto val2 = reader2->ReadValue<int16_t>(); |
| return val1 == val2 ? 0 : (val1 < val2 ? -1 : 1); |
| } |
| case arrow::Type::type::INT32: |
| case arrow::Type::type::DATE32: { |
| auto val1 = reader1->ReadValue<int32_t>(); |
| auto val2 = reader2->ReadValue<int32_t>(); |
| return val1 == val2 ? 0 : (val1 < val2 ? -1 : 1); |
| } |
| case arrow::Type::type::INT64: { |
| auto val1 = reader1->ReadValue<int64_t>(); |
| auto val2 = reader2->ReadValue<int64_t>(); |
| return val1 == val2 ? 0 : (val1 < val2 ? -1 : 1); |
| } |
| case arrow::Type::type::FLOAT: { |
| auto val1 = reader1->ReadValue<float>(); |
| auto val2 = reader2->ReadValue<float>(); |
| return FieldsComparator::CompareFloatingPoint(val1, val2); |
| } |
| case arrow::Type::type::DOUBLE: { |
| auto val1 = reader1->ReadValue<double>(); |
| auto val2 = reader2->ReadValue<double>(); |
| return FieldsComparator::CompareFloatingPoint(val1, val2); |
| } |
| case arrow::Type::type::STRING: |
| case arrow::Type::type::BINARY: { |
| PAIMON_ASSIGN_OR_RAISE(std::string_view val1, reader1->ReadStringView()); |
| PAIMON_ASSIGN_OR_RAISE(std::string_view val2, reader2->ReadStringView()); |
| int32_t cmp = val1.compare(val2); |
| return cmp == 0 ? 0 : (cmp > 0 ? 1 : -1); |
| } |
| case arrow::Type::type::TIMESTAMP: { |
| PAIMON_ASSIGN_OR_RAISE(Timestamp val1, reader1->ReadTimestamp(field_info.precision)); |
| PAIMON_ASSIGN_OR_RAISE(Timestamp val2, reader2->ReadTimestamp(field_info.precision)); |
| return val1 == val2 ? 0 : (val1 < val2 ? -1 : 1); |
| } |
| case arrow::Type::type::DECIMAL: { |
| PAIMON_ASSIGN_OR_RAISE(Decimal val1, |
| reader1->ReadDecimal(field_info.precision, field_info.scale)); |
| PAIMON_ASSIGN_OR_RAISE(Decimal val2, |
| reader2->ReadDecimal(field_info.precision, field_info.scale)); |
| int32_t cmp = val1.CompareTo(val2); |
| return cmp == 0 ? 0 : (cmp > 0 ? 1 : -1); |
| } |
| default: |
| return Status::NotImplemented( |
| fmt::format("Do not support comparing type {} in CompareField", |
| static_cast<int32_t>(field_info.type_id))); |
| } |
| } |
| |
| Result<MemorySlice::SliceComparator> RowCompactedSerializer::CreateSliceComparator( |
| const std::shared_ptr<arrow::Schema>& schema, const std::shared_ptr<MemoryPool>& pool) { |
| int32_t bit_set_in_bytes = RowCompactedSerializer::CalculateBitSetInBytes(schema->num_fields()); |
| auto row_reader1 = std::make_shared<RowReader>(bit_set_in_bytes, pool); |
| auto row_reader2 = std::make_shared<RowReader>(bit_set_in_bytes, pool); |
| |
| std::vector<FieldInfo> field_infos(schema->num_fields()); |
| for (int32_t i = 0; i < schema->num_fields(); i++) { |
| auto field_type = schema->field(i)->type(); |
| field_infos[i].type_id = field_type->id(); |
| if (field_type->id() == arrow::Type::type::TIMESTAMP) { |
| auto timestamp_type = |
| arrow::internal::checked_pointer_cast<arrow::TimestampType>(field_type); |
| assert(timestamp_type); |
| field_infos[i].precision = DateTimeUtils::GetPrecisionFromType(timestamp_type); |
| } else if (field_type->id() == arrow::Type::type::DECIMAL) { |
| auto decimal_type = |
| arrow::internal::checked_pointer_cast<arrow::Decimal128Type>(field_type); |
| assert(decimal_type); |
| field_infos[i].precision = decimal_type->precision(); |
| field_infos[i].scale = decimal_type->scale(); |
| } |
| } |
| |
| auto comparator = [row_reader1, row_reader2, field_infos]( |
| const MemorySlice& slice1, const MemorySlice& slice2) -> Result<int32_t> { |
| row_reader1->PointTo(slice1.GetSegment(), slice1.Offset()); |
| row_reader2->PointTo(slice2.GetSegment(), slice2.Offset()); |
| for (int32_t i = 0; i < static_cast<int32_t>(field_infos.size()); i++) { |
| bool is_null1 = row_reader1->IsNullAt(i); |
| bool is_null2 = row_reader2->IsNullAt(i); |
| if (!is_null1 || !is_null2) { |
| if (is_null1) { |
| return -1; |
| } else if (is_null2) { |
| return 1; |
| } else { |
| PAIMON_ASSIGN_OR_RAISE( |
| int32_t comp, |
| CompareField(field_infos[i], row_reader1.get(), row_reader2.get())); |
| if (comp != 0) { |
| return comp; |
| } |
| } |
| } |
| } |
| return 0; |
| }; |
| return std::function<Result<int32_t>(const MemorySlice&, const MemorySlice&)>(comparator); |
| } |
| |
| Result<std::shared_ptr<Bytes>> RowCompactedSerializer::SerializeToBytes(const InternalRow& row) { |
| if (!row_writer_) { |
| row_writer_ = std::make_unique<RowWriter>(CalculateBitSetInBytes(getters_.size()), pool_); |
| } |
| row_writer_->Reset(); |
| PAIMON_ASSIGN_OR_RAISE(const RowKind* row_kind, row.GetRowKind()); |
| row_writer_->WriteRowKind(*row_kind); |
| for (size_t i = 0; i < getters_.size(); i++) { |
| VariantType field = getters_[i](row); |
| PAIMON_RETURN_NOT_OK(writers_[i](i, field, row_writer_.get())); |
| } |
| return row_writer_->CopyBuffer(); |
| } |
| |
| Result<std::unique_ptr<InternalRow>> RowCompactedSerializer::Deserialize( |
| const std::shared_ptr<Bytes>& bytes) { |
| if (!row_reader_) { |
| row_reader_ = std::make_unique<RowReader>(CalculateBitSetInBytes(getters_.size()), pool_); |
| } |
| row_reader_->PointTo(bytes); |
| auto row = std::make_unique<GenericRow>(getters_.size()); |
| PAIMON_ASSIGN_OR_RAISE(const RowKind* row_kind, row_reader_->ReadRowKind()); |
| row->SetRowKind(row_kind); |
| for (size_t i = 0; i < readers_.size(); i++) { |
| PAIMON_ASSIGN_OR_RAISE(VariantType field, readers_[i](i, row_reader_.get())); |
| row->SetField(i, field); |
| } |
| row->AddDataHolder(bytes); |
| return row; |
| } |
| |
| RowCompactedSerializer::RowCompactedSerializer( |
| const std::shared_ptr<arrow::Schema>& schema, |
| std::vector<InternalRow::FieldGetterFunc>&& getters, |
| std::vector<RowCompactedSerializer::FieldWriter>&& writers, |
| std::vector<RowCompactedSerializer::FieldReader>&& readers, |
| const std::shared_ptr<MemoryPool>& pool) |
| : pool_(pool), |
| schema_(schema), |
| getters_(std::move(getters)), |
| writers_(std::move(writers)), |
| readers_(std::move(readers)) {} |
| |
| Result<RowCompactedSerializer::FieldReader> RowCompactedSerializer::CreateFieldReader( |
| const std::shared_ptr<arrow::DataType>& field_type, const std::shared_ptr<MemoryPool>& pool) { |
| arrow::Type::type type = field_type->id(); |
| RowCompactedSerializer::FieldReader field_reader; |
| switch (type) { |
| case arrow::Type::type::BOOL: { |
| field_reader = [](int32_t pos, RowReader* reader) -> Result<VariantType> { |
| return VariantType(reader->ReadValue<bool>()); |
| }; |
| break; |
| } |
| case arrow::Type::type::INT8: { |
| field_reader = [](int32_t pos, RowReader* reader) -> Result<VariantType> { |
| return VariantType(reader->ReadValue<char>()); |
| }; |
| break; |
| } |
| case arrow::Type::type::INT16: { |
| field_reader = [](int32_t pos, RowReader* reader) -> Result<VariantType> { |
| return VariantType(reader->ReadValue<int16_t>()); |
| }; |
| break; |
| } |
| case arrow::Type::type::INT32: |
| case arrow::Type::type::DATE32: { |
| field_reader = [](int32_t pos, RowReader* reader) -> Result<VariantType> { |
| return VariantType(reader->ReadValue<int32_t>()); |
| }; |
| break; |
| } |
| case arrow::Type::type::INT64: { |
| field_reader = [](int32_t pos, RowReader* reader) -> Result<VariantType> { |
| return VariantType(reader->ReadValue<int64_t>()); |
| }; |
| break; |
| } |
| case arrow::Type::type::FLOAT: { |
| field_reader = [](int32_t pos, RowReader* reader) -> Result<VariantType> { |
| return VariantType(reader->ReadValue<float>()); |
| }; |
| break; |
| } |
| case arrow::Type::type::DOUBLE: { |
| field_reader = [](int32_t pos, RowReader* reader) -> Result<VariantType> { |
| return VariantType(reader->ReadValue<double>()); |
| }; |
| break; |
| } |
| case arrow::Type::type::STRING: { |
| field_reader = [](int32_t pos, RowReader* reader) -> Result<VariantType> { |
| PAIMON_ASSIGN_OR_RAISE(VariantType value, reader->ReadStringView()); |
| return value; |
| }; |
| break; |
| } |
| case arrow::Type::type::BINARY: { |
| field_reader = [](int32_t pos, RowReader* reader) -> Result<VariantType> { |
| PAIMON_ASSIGN_OR_RAISE(VariantType value, reader->ReadBinary()); |
| return value; |
| }; |
| break; |
| } |
| case arrow::Type::type::TIMESTAMP: { |
| auto timestamp_type = |
| arrow::internal::checked_pointer_cast<arrow::TimestampType>(field_type); |
| int32_t precision = DateTimeUtils::GetPrecisionFromType(timestamp_type); |
| field_reader = [precision](int32_t pos, RowReader* reader) -> Result<VariantType> { |
| PAIMON_ASSIGN_OR_RAISE(VariantType value, reader->ReadTimestamp(precision)); |
| return value; |
| }; |
| break; |
| } |
| case arrow::Type::type::DECIMAL: { |
| auto* decimal_type = |
| arrow::internal::checked_cast<arrow::Decimal128Type*>(field_type.get()); |
| assert(decimal_type); |
| auto precision = decimal_type->precision(); |
| auto scale = decimal_type->scale(); |
| field_reader = [precision, scale](int32_t pos, |
| RowReader* reader) -> Result<VariantType> { |
| PAIMON_ASSIGN_OR_RAISE(VariantType value, reader->ReadDecimal(precision, scale)); |
| return value; |
| }; |
| break; |
| } |
| case arrow::Type::type::LIST: { |
| field_reader = [](int32_t pos, RowReader* reader) -> Result<VariantType> { |
| PAIMON_ASSIGN_OR_RAISE(VariantType value, reader->ReadArray()); |
| return value; |
| }; |
| break; |
| } |
| case arrow::Type::type::MAP: { |
| field_reader = [](int32_t pos, RowReader* reader) -> Result<VariantType> { |
| PAIMON_ASSIGN_OR_RAISE(VariantType value, reader->ReadMap()); |
| return value; |
| }; |
| break; |
| } |
| case arrow::Type::type::STRUCT: { |
| auto* struct_type = arrow::internal::checked_cast<arrow::StructType*>(field_type.get()); |
| assert(struct_type); |
| PAIMON_ASSIGN_OR_RAISE( |
| std::shared_ptr<RowCompactedSerializer> serializer, |
| RowCompactedSerializer::Create(arrow::schema(struct_type->fields()), pool)); |
| field_reader = [serializer](int32_t pos, RowReader* reader) -> Result<VariantType> { |
| PAIMON_ASSIGN_OR_RAISE(VariantType value, reader->ReadRow(serializer)); |
| return value; |
| }; |
| break; |
| } |
| default: |
| return Status::Invalid( |
| fmt::format("type {} not support in FieldReader in row compacted serializer", |
| field_type->ToString())); |
| } |
| |
| RowCompactedSerializer::FieldReader ret = |
| [field_reader](int32_t pos, RowReader* reader) -> Result<VariantType> { |
| if (reader->IsNullAt(pos)) { |
| return VariantType(NullType()); |
| } |
| return field_reader(pos, reader); |
| }; |
| return ret; |
| } |
| |
| Result<RowCompactedSerializer::FieldWriter> RowCompactedSerializer::CreateFieldWriter( |
| const std::shared_ptr<arrow::DataType>& field_type, const std::shared_ptr<MemoryPool>& pool) { |
| arrow::Type::type type = field_type->id(); |
| RowCompactedSerializer::FieldWriter field_writer; |
| switch (type) { |
| case arrow::Type::type::BOOL: { |
| field_writer = [](int32_t pos, const VariantType& field, RowWriter* writer) -> Status { |
| return writer->WriteValue<bool>(DataDefine::GetVariantValue<bool>(field)); |
| }; |
| break; |
| } |
| case arrow::Type::type::INT8: { |
| field_writer = [](int32_t pos, const VariantType& field, RowWriter* writer) -> Status { |
| return writer->WriteValue<char>(DataDefine::GetVariantValue<char>(field)); |
| }; |
| break; |
| } |
| case arrow::Type::type::INT16: { |
| field_writer = [](int32_t pos, const VariantType& field, RowWriter* writer) -> Status { |
| return writer->WriteValue<int16_t>(DataDefine::GetVariantValue<int16_t>(field)); |
| }; |
| break; |
| } |
| case arrow::Type::type::INT32: |
| case arrow::Type::type::DATE32: { |
| field_writer = [](int32_t pos, const VariantType& field, RowWriter* writer) -> Status { |
| return writer->WriteValue<int32_t>(DataDefine::GetVariantValue<int32_t>(field)); |
| }; |
| break; |
| } |
| case arrow::Type::type::INT64: { |
| field_writer = [](int32_t pos, const VariantType& field, RowWriter* writer) -> Status { |
| return writer->WriteValue<int64_t>(DataDefine::GetVariantValue<int64_t>(field)); |
| }; |
| break; |
| } |
| case arrow::Type::type::FLOAT: { |
| field_writer = [](int32_t pos, const VariantType& field, RowWriter* writer) -> Status { |
| return writer->WriteValue<float>(DataDefine::GetVariantValue<float>(field)); |
| }; |
| break; |
| } |
| case arrow::Type::type::DOUBLE: { |
| field_writer = [](int32_t pos, const VariantType& field, RowWriter* writer) -> Status { |
| return writer->WriteValue<double>(DataDefine::GetVariantValue<double>(field)); |
| }; |
| break; |
| } |
| case arrow::Type::type::STRING: { |
| field_writer = [](int32_t pos, const VariantType& field, RowWriter* writer) -> Status { |
| const auto* view = DataDefine::GetVariantPtr<std::string_view>(field); |
| if (view) { |
| return writer->WriteStringView(*view); |
| } |
| return writer->WriteString(DataDefine::GetVariantValue<BinaryString>(field)); |
| }; |
| break; |
| } |
| case arrow::Type::type::BINARY: { |
| field_writer = [](int32_t pos, const VariantType& field, RowWriter* writer) -> Status { |
| const auto* view = DataDefine::GetVariantPtr<std::string_view>(field); |
| if (view) { |
| return writer->WriteStringView(*view); |
| } |
| return writer->WriteBinary( |
| DataDefine::GetVariantValue<std::shared_ptr<Bytes>>(field).get()); |
| }; |
| break; |
| } |
| case arrow::Type::type::TIMESTAMP: { |
| auto timestamp_type = |
| arrow::internal::checked_pointer_cast<arrow::TimestampType>(field_type); |
| int32_t precision = DateTimeUtils::GetPrecisionFromType(timestamp_type); |
| field_writer = [precision](int32_t pos, const VariantType& field, |
| RowWriter* writer) -> Status { |
| return writer->WriteTimestamp(DataDefine::GetVariantValue<Timestamp>(field), |
| precision); |
| }; |
| break; |
| } |
| case arrow::Type::type::DECIMAL: { |
| auto* decimal_type = |
| arrow::internal::checked_cast<arrow::Decimal128Type*>(field_type.get()); |
| assert(decimal_type); |
| auto precision = decimal_type->precision(); |
| field_writer = [precision](int32_t pos, const VariantType& field, |
| RowWriter* writer) -> Status { |
| return writer->WriteDecimal(DataDefine::GetVariantValue<Decimal>(field), precision); |
| }; |
| break; |
| } |
| case arrow::Type::type::LIST: { |
| field_writer = [field_type](int32_t pos, const VariantType& field, |
| RowWriter* writer) -> Status { |
| return writer->WriteArray( |
| DataDefine::GetVariantValue<std::shared_ptr<InternalArray>>(field), field_type); |
| }; |
| break; |
| } |
| case arrow::Type::type::MAP: { |
| field_writer = [field_type](int32_t pos, const VariantType& field, |
| RowWriter* writer) -> Status { |
| return writer->WriteMap( |
| DataDefine::GetVariantValue<std::shared_ptr<InternalMap>>(field), field_type); |
| }; |
| break; |
| } |
| case arrow::Type::type::STRUCT: { |
| auto struct_type = arrow::internal::checked_pointer_cast<arrow::StructType>(field_type); |
| assert(struct_type); |
| PAIMON_ASSIGN_OR_RAISE( |
| std::shared_ptr<RowCompactedSerializer> serializer, |
| RowCompactedSerializer::Create(arrow::schema(struct_type->fields()), pool)); |
| field_writer = [serializer](int32_t pos, const VariantType& field, |
| RowWriter* writer) -> Status { |
| return writer->WriteRow( |
| DataDefine::GetVariantValue<std::shared_ptr<InternalRow>>(field), serializer); |
| }; |
| break; |
| } |
| default: |
| return Status::Invalid( |
| fmt::format("type {} not support in FieldWriter in row compacted serializer", |
| field_type->ToString())); |
| } |
| |
| RowCompactedSerializer::FieldWriter ret = [field_writer](int32_t pos, const VariantType& field, |
| RowWriter* writer) -> Status { |
| if (DataDefine::IsVariantNull(field)) { |
| writer->SetNullAt(pos); |
| return Status::OK(); |
| } |
| return field_writer(pos, field, writer); |
| }; |
| return ret; |
| } |
| |
| RowCompactedSerializer::RowWriter::RowWriter(int32_t header_size_in_bytes, |
| const std::shared_ptr<MemoryPool>& pool) |
| : header_size_in_bytes_(header_size_in_bytes), pool_(pool) { |
| int32_t initial_capacity = std::max(64, header_size_in_bytes); |
| SetBuffer(std::make_shared<Bytes>(initial_capacity, pool_.get())); |
| position_ = header_size_in_bytes_; |
| } |
| |
| Status RowCompactedSerializer::RowWriter::WriteDecimal(const Decimal& value, int32_t precision) { |
| if (Decimal::IsCompact(precision)) { |
| return WriteValue<int64_t>(value.ToUnscaledLong()); |
| } else { |
| auto value_bytes = value.ToUnscaledBytes(); |
| return WriteBinary(&value_bytes); |
| } |
| } |
| |
| Status RowCompactedSerializer::RowWriter::WriteTimestamp(const Timestamp& value, |
| int32_t precision) { |
| if (Timestamp::IsCompact(precision)) { |
| return WriteValue<int64_t>(value.GetMillisecond()); |
| } else { |
| PAIMON_RETURN_NOT_OK(WriteValue<int64_t>(value.GetMillisecond())); |
| return WriteUnsignedInt(value.GetNanoOfMillisecond()); |
| } |
| } |
| |
| Status RowCompactedSerializer::RowWriter::WriteRow( |
| const std::shared_ptr<InternalRow>& value, |
| const std::shared_ptr<RowCompactedSerializer>& serializer) { |
| PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Bytes> bytes, serializer->SerializeToBytes(*value)); |
| return WriteBinary(bytes.get()); |
| } |
| |
| Status RowCompactedSerializer::RowWriter::WriteArray(const std::shared_ptr<InternalArray>& value, |
| const std::shared_ptr<arrow::DataType>& type) { |
| PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<BinaryArray> binary_array, |
| BinarySerializerUtils::WriteBinaryArray(value, type, pool_.get())); |
| return WriteSegment(binary_array->GetSegment(), binary_array->GetOffset(), |
| binary_array->GetSizeInBytes()); |
| } |
| |
| Status RowCompactedSerializer::RowWriter::WriteMap(const std::shared_ptr<InternalMap>& value, |
| const std::shared_ptr<arrow::DataType>& type) { |
| PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<BinaryMap> binary_map, |
| BinarySerializerUtils::WriteBinaryMap(value, type, pool_.get())); |
| return WriteSegment(binary_map->GetSegment(), binary_map->GetOffset(), |
| binary_map->GetSizeInBytes()); |
| } |
| |
| std::shared_ptr<Bytes> RowCompactedSerializer::RowWriter::CopyBuffer() const { |
| return Bytes::CopyOf(*buffer_, position_, pool_.get()); |
| } |
| |
| Status RowCompactedSerializer::RowWriter::WriteUnsignedInt(int32_t value) { |
| EnsureCapacity(VarLengthIntUtils::kMaxVarIntSize); |
| PAIMON_ASSIGN_OR_RAISE(int32_t len, |
| VarLengthIntUtils::EncodeInt(value, buffer_->data() + position_)); |
| position_ += len; |
| return Status::OK(); |
| } |
| |
| Status RowCompactedSerializer::RowWriter::WriteSegment(const MemorySegment& segment, int32_t off, |
| int32_t len) { |
| PAIMON_RETURN_NOT_OK(WriteUnsignedInt(len)); |
| EnsureCapacity(len); |
| MemorySegmentUtils::CopyToBytes({segment}, off, buffer_.get(), position_, len); |
| position_ += len; |
| return Status::OK(); |
| } |
| |
| void RowCompactedSerializer::RowWriter::EnsureCapacity(int32_t size) { |
| if (static_cast<int32_t>(buffer_->size()) - position_ < size) { |
| Grow(size); |
| } |
| } |
| |
| void RowCompactedSerializer::RowWriter::Grow(int32_t min_capacity_add) { |
| auto current_len = static_cast<int32_t>(buffer_->size()); |
| int32_t new_len = std::max(current_len * 2, current_len + min_capacity_add); |
| |
| auto new_buffer = std::make_shared<Bytes>(new_len, pool_.get()); |
| memcpy(new_buffer->data(), buffer_->data(), current_len); |
| SetBuffer(std::move(new_buffer)); |
| } |
| |
| void RowCompactedSerializer::RowWriter::SetBuffer(std::shared_ptr<Bytes> new_buffer) { |
| buffer_ = std::move(new_buffer); |
| segment_ = MemorySegment::Wrap(buffer_); |
| } |
| |
| void RowCompactedSerializer::RowReader::PointTo(const std::shared_ptr<Bytes>& bytes) { |
| MemorySegment seg = MemorySegment::Wrap(bytes); |
| PointTo(seg, 0); |
| } |
| |
| void RowCompactedSerializer::RowReader::PointTo(const MemorySegment& segment, int32_t offset) { |
| segment_ = segment; |
| |
| offset_ = offset; |
| position_ = offset + header_size_in_bytes_; |
| } |
| |
| Result<const RowKind*> RowCompactedSerializer::RowReader::ReadRowKind() const { |
| char b = segment_.Get(offset_); |
| return RowKind::FromByteValue(static_cast<int8_t>(b)); |
| } |
| |
| Result<std::string_view> RowCompactedSerializer::RowReader::ReadStringView() { |
| PAIMON_ASSIGN_OR_RAISE(int32_t length, ReadUnsignedInt()); |
| std::string_view str(segment_.Data() + position_, length); |
| position_ += length; |
| return str; |
| } |
| |
| Result<std::shared_ptr<Bytes>> RowCompactedSerializer::RowReader::ReadBinary() { |
| PAIMON_ASSIGN_OR_RAISE(int32_t length, ReadUnsignedInt()); |
| auto bytes = std::make_shared<Bytes>(length, pool_.get()); |
| segment_.Get(position_, bytes.get(), /*offset=*/0, length); |
| position_ += length; |
| return bytes; |
| } |
| |
| Result<Decimal> RowCompactedSerializer::RowReader::ReadDecimal(int32_t precision, int32_t scale) { |
| if (Decimal::IsCompact(precision)) { |
| return Decimal::FromUnscaledLong(ReadValue<int64_t>(), precision, scale); |
| } else { |
| PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Bytes> bytes, ReadBinary()); |
| return Decimal::FromUnscaledBytes(precision, scale, bytes.get()); |
| } |
| } |
| |
| Result<Timestamp> RowCompactedSerializer::RowReader::ReadTimestamp(int32_t precision) { |
| if (Timestamp::IsCompact(precision)) { |
| return Timestamp::FromEpochMillis(ReadValue<int64_t>()); |
| } |
| auto milliseconds = ReadValue<int64_t>(); |
| PAIMON_ASSIGN_OR_RAISE(int32_t nanos_of_millisecond, ReadUnsignedInt()); |
| return Timestamp::FromEpochMillis(milliseconds, nanos_of_millisecond); |
| } |
| |
| Result<std::shared_ptr<InternalArray>> RowCompactedSerializer::RowReader::ReadArray() { |
| auto value = std::make_shared<BinaryArray>(); |
| PAIMON_ASSIGN_OR_RAISE(int32_t length, ReadUnsignedInt()); |
| value->PointTo(segment_, position_, length); |
| position_ += length; |
| return value; |
| } |
| |
| Result<std::shared_ptr<InternalMap>> RowCompactedSerializer::RowReader::ReadMap() { |
| auto value = std::make_shared<BinaryMap>(); |
| PAIMON_ASSIGN_OR_RAISE(int32_t length, ReadUnsignedInt()); |
| value->PointTo(segment_, position_, length); |
| position_ += length; |
| return value; |
| } |
| |
| Result<std::shared_ptr<InternalRow>> RowCompactedSerializer::RowReader::ReadRow( |
| const std::shared_ptr<RowCompactedSerializer>& serializer) { |
| PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<Bytes> bytes, ReadBinary()); |
| PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<InternalRow> result, serializer->Deserialize(bytes)); |
| return result; |
| } |
| |
| } // namespace paimon |
