Copilot commented on code in PR #119: URL: https://github.com/apache/paimon-cpp/pull/119#discussion_r3472308325
########## src/paimon/core/table/system/options_system_table.cpp: ########## @@ -0,0 +1,156 @@ +/* + * 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/core/table/system/options_system_table.h" + +#include <map> +#include <memory> +#include <string> +#include <utility> +#include <vector> + +#include "arrow/api.h" +#include "arrow/c/bridge.h" +#include "paimon/common/metrics/metrics_impl.h" +#include "paimon/common/utils/arrow/mem_utils.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/core/schema/table_schema.h" +#include "paimon/core/table/system/system_table_scan.h" +#include "paimon/memory/memory_pool.h" +#include "paimon/read_context.h" +#include "paimon/result.h" +#include "paimon/status.h" +#include "paimon/table/source/table_read.h" + +namespace paimon { +namespace { + +std::shared_ptr<arrow::Schema> OptionsSchema() { + return arrow::schema({arrow::field("key", arrow::utf8(), /*nullable=*/false), + arrow::field("value", arrow::utf8(), /*nullable=*/false)}); +} + +class OptionsBatchReader : public BatchReader { + public: + OptionsBatchReader(std::map<std::string, std::string> options, + const std::shared_ptr<MemoryPool>& pool) + : arrow_pool_(GetArrowPool(pool)), options_(std::move(options)) {} + + Result<ReadBatch> NextBatch() override { + if (emitted_) { + return BatchReader::MakeEofBatch(); + } + emitted_ = true; + + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::unique_ptr<arrow::ArrayBuilder> key_array_builder, + arrow::MakeBuilder(arrow::utf8(), arrow_pool_.get())); + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::unique_ptr<arrow::ArrayBuilder> value_array_builder, + arrow::MakeBuilder(arrow::utf8(), arrow_pool_.get())); + auto* key_builder = dynamic_cast<arrow::StringBuilder*>(key_array_builder.get()); + auto* value_builder = dynamic_cast<arrow::StringBuilder*>(value_array_builder.get()); + if (key_builder == nullptr || value_builder == nullptr) { + return Status::Invalid("cannot create string builders for options system table"); + } + for (const auto& [key, value] : options_) { + PAIMON_RETURN_NOT_OK_FROM_ARROW(key_builder->Append(key)); + PAIMON_RETURN_NOT_OK_FROM_ARROW(value_builder->Append(value)); + } + std::shared_ptr<arrow::Array> key_array; + std::shared_ptr<arrow::Array> value_array; + PAIMON_RETURN_NOT_OK_FROM_ARROW(key_builder->Finish(&key_array)); + PAIMON_RETURN_NOT_OK_FROM_ARROW(value_builder->Finish(&value_array)); + auto struct_array = std::make_shared<arrow::StructArray>( + arrow::struct_(OptionsSchema()->fields()), key_array->length(), + std::vector<std::shared_ptr<arrow::Array>>{key_array, value_array}); + + auto c_array = std::make_unique<::ArrowArray>(); + auto c_schema = std::make_unique<::ArrowSchema>(); + PAIMON_RETURN_NOT_OK_FROM_ARROW( + arrow::ExportArray(*struct_array, c_array.get(), c_schema.get())); + return std::make_pair(std::move(c_array), std::move(c_schema)); + } + + std::shared_ptr<Metrics> GetReaderMetrics() const override { + return std::make_shared<MetricsImpl>(); + } + + void Close() override { + emitted_ = true; + } + + private: + std::unique_ptr<arrow::MemoryPool> arrow_pool_; + std::map<std::string, std::string> options_; + bool emitted_ = false; +}; + +class OptionsTableRead : public TableRead { + public: + OptionsTableRead(std::map<std::string, std::string> options, + const std::shared_ptr<MemoryPool>& memory_pool) + : TableRead(memory_pool), options_(std::move(options)) {} + + Result<std::unique_ptr<BatchReader>> CreateReader( + const std::vector<std::shared_ptr<Split>>& splits) override { + if (splits.size() != 1) { + return Status::Invalid("options system table expects a single split"); + } + for (const auto& split : splits) { + if (!std::dynamic_pointer_cast<SystemTableSplit>(split)) { + return Status::Invalid("unsupported split for options system table"); + } + } + return std::make_unique<OptionsBatchReader>(options_, GetMemoryPool()); + } + + Result<std::unique_ptr<BatchReader>> CreateReader( + const std::shared_ptr<Split>& split) override { + std::vector<std::shared_ptr<Split>> splits = {split}; + return CreateReader(splits); + } + + private: + std::map<std::string, std::string> options_; +}; + +} // namespace + +OptionsSystemTable::OptionsSystemTable(std::string table_path, + std::shared_ptr<TableSchema> table_schema) + : table_path_(std::move(table_path)), table_schema_(std::move(table_schema)) {} + +std::string OptionsSystemTable::Name() const { + return kName; +} + +Result<std::shared_ptr<arrow::Schema>> OptionsSystemTable::ArrowSchema() const { + return OptionsSchema(); +} + +Result<std::unique_ptr<TableScan>> OptionsSystemTable::NewScan( + const std::shared_ptr<ScanContext>& /*context*/) const { + return std::make_unique<SystemTableScan>(table_path_); +} + +Result<std::unique_ptr<TableRead>> OptionsSystemTable::NewRead( + const std::shared_ptr<ReadContext>& context) const { + return std::make_unique<OptionsTableRead>(table_schema_->Options(), context->GetMemoryPool()); +} Review Comment: OptionsSystemTable::NewRead() currently returns only table_schema_->Options(), ignoring runtime options passed via ReadContext (context->GetOptions()). This prevents expected option overrides (e.g. file.format / manifest.format) from being reflected in the options system table output. ########## src/paimon/core/table/sink/commit_message.cpp: ########## @@ -0,0 +1,101 @@ +/* + * 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/commit_message.h" + +#include <utility> + +#include "paimon/common/io/memory_segment_output_stream.h" +#include "paimon/common/memory/memory_segment_utils.h" +#include "paimon/core/table/sink/commit_message_impl.h" +#include "paimon/core/table/sink/commit_message_serializer.h" +#include "paimon/io/byte_array_input_stream.h" +#include "paimon/io/data_input_stream.h" +#include "paimon/memory/bytes.h" +#include "paimon/memory/memory_pool.h" +#include "paimon/result.h" +#include "paimon/status.h" + +namespace paimon { + +int32_t CommitMessage::CurrentVersion() { + return CommitMessageSerializer::CURRENT_VERSION; +} + +CommitMessage::~CommitMessage() = default; + +Result<std::string> CommitMessage::Serialize(const std::shared_ptr<CommitMessage>& commit_message, + const std::shared_ptr<MemoryPool>& pool) { + CommitMessageSerializer serializer(pool); + MemorySegmentOutputStream out(MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool); + PAIMON_RETURN_NOT_OK(serializer.Serialize(commit_message, &out)); Review Comment: CommitMessage::Serialize() does not validate the provided MemoryPool pointer. Many other public APIs return Status::Invalid when the memory pool is null; here a null pool would be forwarded into CommitMessageSerializer / MemorySegmentOutputStream and can lead to undefined behavior. ########## src/paimon/core/table/sink/commit_message.cpp: ########## @@ -0,0 +1,101 @@ +/* + * 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/commit_message.h" + +#include <utility> + +#include "paimon/common/io/memory_segment_output_stream.h" +#include "paimon/common/memory/memory_segment_utils.h" +#include "paimon/core/table/sink/commit_message_impl.h" +#include "paimon/core/table/sink/commit_message_serializer.h" +#include "paimon/io/byte_array_input_stream.h" +#include "paimon/io/data_input_stream.h" +#include "paimon/memory/bytes.h" +#include "paimon/memory/memory_pool.h" +#include "paimon/result.h" +#include "paimon/status.h" + +namespace paimon { + +int32_t CommitMessage::CurrentVersion() { + return CommitMessageSerializer::CURRENT_VERSION; +} + +CommitMessage::~CommitMessage() = default; + +Result<std::string> CommitMessage::Serialize(const std::shared_ptr<CommitMessage>& commit_message, + const std::shared_ptr<MemoryPool>& pool) { + CommitMessageSerializer serializer(pool); + MemorySegmentOutputStream out(MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool); + PAIMON_RETURN_NOT_OK(serializer.Serialize(commit_message, &out)); + PAIMON_UNIQUE_PTR<Bytes> bytes = + MemorySegmentUtils::CopyToBytes(out.Segments(), 0, out.CurrentSize(), pool.get()); + return std::string(bytes->data(), bytes->size()); +} + +Result<std::string> CommitMessage::SerializeList( + const std::vector<std::shared_ptr<CommitMessage>>& commit_messages, + const std::shared_ptr<MemoryPool>& pool) { + CommitMessageSerializer serializer(pool); + MemorySegmentOutputStream out(MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool); + PAIMON_RETURN_NOT_OK(serializer.SerializeList(commit_messages, &out)); Review Comment: CommitMessage::SerializeList() does not validate the provided MemoryPool pointer before using it for serialization. Passing a null pool can cause undefined behavior inside CommitMessageSerializer / MemorySegmentOutputStream. ########## src/paimon/core/table/sink/commit_message_serializer.cpp: ########## @@ -0,0 +1,256 @@ +/* + * 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/core/table/sink/commit_message_serializer.h" + +#include <algorithm> +#include <optional> +#include <utility> + +#include "fmt/format.h" +#include "paimon/common/data/binary_row.h" +#include "paimon/common/io/memory_segment_output_stream.h" +#include "paimon/common/utils/serialization_utils.h" +#include "paimon/core/index/index_file_meta_serializer.h" +#include "paimon/core/index/index_file_meta_v1_deserializer.h" +#include "paimon/core/index/index_file_meta_v2_deserializer.h" +#include "paimon/core/index/index_file_meta_v3_deserializer.h" +#include "paimon/core/io/compact_increment.h" +#include "paimon/core/io/data_file_meta_09_serializer.h" +#include "paimon/core/io/data_file_meta_10_serializer.h" +#include "paimon/core/io/data_file_meta_12_serializer.h" +#include "paimon/core/io/data_file_meta_first_row_id_legacy_serializer.h" +#include "paimon/core/io/data_file_meta_serializer.h" +#include "paimon/core/io/data_increment.h" +#include "paimon/core/table/sink/commit_message_impl.h" +#include "paimon/core/utils/object_serializer.h" +#include "paimon/io/data_input_stream.h" + +namespace paimon { +class MemoryPool; + +const int32_t CommitMessageSerializer::CURRENT_VERSION = 11; + +CommitMessageSerializer::CommitMessageSerializer(const std::shared_ptr<MemoryPool>& pool) + : memory_pool_(pool), + data_file_serializer_(std::make_unique<DataFileMetaSerializer>(pool)), + index_entry_serializer_(std::make_unique<IndexFileMetaSerializer>(pool)) {} + +CommitMessageSerializer::~CommitMessageSerializer() = default; + +Status CommitMessageSerializer::Serialize(const std::shared_ptr<CommitMessage>& obj, + MemorySegmentOutputStream* out) { + auto message = std::dynamic_pointer_cast<CommitMessageImpl>(obj); + if (message == nullptr) { + return Status::Invalid("failed to cast commit message to commit message impl"); + } + PAIMON_RETURN_NOT_OK(SerializationUtils::SerializeBinaryRow(message->Partition(), out)); + out->WriteValue<int32_t>(message->Bucket()); + std::optional<int32_t> total_buckets = message->TotalBuckets(); + if (total_buckets == std::nullopt) { + out->WriteValue<bool>(false); + } else { + out->WriteValue<bool>(true); + out->WriteValue<int32_t>(total_buckets.value()); + } + // data increment + PAIMON_RETURN_NOT_OK( + data_file_serializer_->SerializeList(message->GetNewFilesIncrement().NewFiles(), out)); + PAIMON_RETURN_NOT_OK( + data_file_serializer_->SerializeList(message->GetNewFilesIncrement().DeletedFiles(), out)); + PAIMON_RETURN_NOT_OK(data_file_serializer_->SerializeList( + message->GetNewFilesIncrement().ChangelogFiles(), out)); + PAIMON_RETURN_NOT_OK(index_entry_serializer_->SerializeList( + message->GetNewFilesIncrement().NewIndexFiles(), out)); + PAIMON_RETURN_NOT_OK(index_entry_serializer_->SerializeList( + message->GetNewFilesIncrement().DeletedIndexFiles(), out)); + // compact increment + PAIMON_RETURN_NOT_OK( + data_file_serializer_->SerializeList(message->GetCompactIncrement().CompactBefore(), out)); + PAIMON_RETURN_NOT_OK( + data_file_serializer_->SerializeList(message->GetCompactIncrement().CompactAfter(), out)); + PAIMON_RETURN_NOT_OK( + data_file_serializer_->SerializeList(message->GetCompactIncrement().ChangelogFiles(), out)); + PAIMON_RETURN_NOT_OK(index_entry_serializer_->SerializeList( + message->GetCompactIncrement().NewIndexFiles(), out)); + PAIMON_RETURN_NOT_OK(index_entry_serializer_->SerializeList( + message->GetCompactIncrement().DeletedIndexFiles(), out)); + + return Status::OK(); +} + +Status CommitMessageSerializer::SerializeList( + const std::vector<std::shared_ptr<CommitMessage>>& commit_messages, + MemorySegmentOutputStream* out) { + out->WriteValue<int32_t>(commit_messages.size()); + for (const auto& commit_message : commit_messages) { + PAIMON_RETURN_NOT_OK(Serialize(commit_message, out)); + } + return Status::OK(); +} + +Result<std::vector<std::shared_ptr<CommitMessage>>> CommitMessageSerializer::DeserializeList( + int32_t version, DataInputStream* in) { + PAIMON_ASSIGN_OR_RAISE(int32_t length, in->ReadValue<int32_t>()); + std::vector<std::shared_ptr<CommitMessage>> commit_messages; + for (int32_t i = 0; i < length; i++) { + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<CommitMessage> commit_message, + Deserialize(version, in)); + commit_messages.push_back(commit_message); + } + return commit_messages; +} + +Result<std::shared_ptr<CommitMessage>> CommitMessageSerializer::Deserialize( + int32_t version, + const ObjectSerializer<std::shared_ptr<DataFileMeta>>* data_file_meta_serializer, + const ObjectSerializer<std::shared_ptr<IndexFileMeta>>* index_entry_serializer, + DataInputStream* in) { + PAIMON_ASSIGN_OR_RAISE(BinaryRow partition, + SerializationUtils::DeserializeBinaryRow(in, memory_pool_.get())); + PAIMON_ASSIGN_OR_RAISE(int32_t bucket, in->ReadValue<int32_t>()); + + std::optional<int32_t> total_buckets; + if (version >= 7) { + PAIMON_ASSIGN_OR_RAISE(bool total_buckets_exist, in->ReadValue<bool>()); + if (total_buckets_exist) { + PAIMON_ASSIGN_OR_RAISE(total_buckets, in->ReadValue<int32_t>()); + } + } + if (version >= 10) { + // data increment + PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<DataFileMeta>> new_files, + data_file_meta_serializer->DeserializeList(in)); + PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<DataFileMeta>> deleted_files, + data_file_meta_serializer->DeserializeList(in)); + PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<DataFileMeta>> changelog_files, + data_file_meta_serializer->DeserializeList(in)); + PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<IndexFileMeta>> new_data_index, + index_entry_serializer->DeserializeList(in)); + PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<IndexFileMeta>> deleted_data_index, + index_entry_serializer->DeserializeList(in)); + // compact increment + PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<DataFileMeta>> before, + data_file_meta_serializer->DeserializeList(in)); + PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<DataFileMeta>> after, + data_file_meta_serializer->DeserializeList(in)); + PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<DataFileMeta>> changelog, + data_file_meta_serializer->DeserializeList(in)); + PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<IndexFileMeta>> new_compact_index, + index_entry_serializer->DeserializeList(in)); + PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<IndexFileMeta>> deleted_compact_index, + index_entry_serializer->DeserializeList(in)); + return std::make_shared<CommitMessageImpl>( + partition, bucket, total_buckets, + DataIncrement(std::move(new_files), std::move(deleted_files), + std::move(changelog_files), std::move(new_data_index), + std::move(deleted_data_index)), + CompactIncrement(std::move(before), std::move(after), std::move(changelog), + std::move(new_compact_index), std::move(deleted_compact_index))); + } + // data increment + PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<DataFileMeta>> new_files, + data_file_meta_serializer->DeserializeList(in)); + PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<DataFileMeta>> deleted_files, + data_file_meta_serializer->DeserializeList(in)); + PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<DataFileMeta>> changelog_files, + data_file_meta_serializer->DeserializeList(in)); + // compact increment + PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<DataFileMeta>> before, + data_file_meta_serializer->DeserializeList(in)); + PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<DataFileMeta>> after, + data_file_meta_serializer->DeserializeList(in)); + PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<DataFileMeta>> changelog, + data_file_meta_serializer->DeserializeList(in)); + // index increment + PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<IndexFileMeta>> new_index, + index_entry_serializer->DeserializeList(in)); + PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<IndexFileMeta>> deleted_index, + index_entry_serializer->DeserializeList(in)); + + DataIncrement data_increment(std::move(new_files), std::move(deleted_files), + std::move(changelog_files)); + CompactIncrement compact_increment(std::move(before), std::move(after), std::move(changelog)); + + if (compact_increment.IsEmpty()) { + data_increment.AddNewIndexFiles(std::move(new_index)); + data_increment.AddDeletedIndexFiles(std::move(deleted_index)); + } else { + compact_increment.AddNewIndexFiles(std::move(new_index)); + compact_increment.AddDeletedIndexFiles(std::move(deleted_index)); + } + return std::make_shared<CommitMessageImpl>(partition, bucket, total_buckets, data_increment, + compact_increment); +} + +Result<std::shared_ptr<CommitMessage>> CommitMessageSerializer::Deserialize(int32_t version, + DataInputStream* in) { + if (version == CURRENT_VERSION) { + return Deserialize(version, data_file_serializer_.get(), index_entry_serializer_.get(), in); + } else if (version == 9 || version == 10) { + auto index_entry_v3_deserializer = + std::make_unique<IndexFileMetaV3Deserializer>(memory_pool_); + return Deserialize(version, data_file_serializer_.get(), index_entry_v3_deserializer.get(), + in); + } else if (version == 8) { + auto data_file_meta_first_row_id_legacy_serializer = + std::make_unique<DataFileMetaFirstRowIdLegacySerializer>(memory_pool_); + auto index_entry_v2_deserializer = + std::make_unique<IndexFileMetaV2Deserializer>(memory_pool_); + return Deserialize(version, data_file_meta_first_row_id_legacy_serializer.get(), + index_entry_v2_deserializer.get(), in); + } else if (version == 6 || version == 7) { + auto index_entry_v2_deserializer = + std::make_unique<IndexFileMetaV2Deserializer>(memory_pool_); + auto data_file_meta_12_serializer = + std::make_unique<DataFileMeta12Serializer>(memory_pool_); + return Deserialize(version, data_file_meta_12_serializer.get(), + index_entry_v2_deserializer.get(), in); + } else if (version == 5) { + auto index_entry_v2_deserializer = + std::make_unique<IndexFileMetaV2Deserializer>(memory_pool_); + auto data_file_meta_10_serializer = + std::make_unique<DataFileMeta10Serializer>(memory_pool_); + return Deserialize(version, data_file_meta_10_serializer.get(), + index_entry_v2_deserializer.get(), in); + } else if (version == 4) { + auto index_entry_v1_deserializer = + std::make_unique<IndexFileMetaV1Deserializer>(memory_pool_); + auto data_file_meta_10_serializer = + std::make_unique<DataFileMeta10Serializer>(memory_pool_); + return Deserialize(version, data_file_meta_10_serializer.get(), + index_entry_v1_deserializer.get(), in); + } else if (version == 3) { + auto data_file_meta_09_serializer = + std::make_unique<DataFileMeta09Serializer>(memory_pool_); + auto index_entry_v1_deserializer = + std::make_unique<IndexFileMetaV1Deserializer>(memory_pool_); + return Deserialize(version, data_file_meta_09_serializer.get(), + index_entry_v1_deserializer.get(), in); + } else if (version <= 2) { + return Status::NotImplemented("deserialize 08 not implemented"); + } else { Review Comment: The error message for unsupported legacy commit message versions (<= 2) is unclear: "deserialize 08 not implemented" doesn’t indicate what version was requested or what "08" refers to. This makes troubleshooting difficult for callers. ########## src/paimon/core/table/sink/commit_message.cpp: ########## @@ -0,0 +1,101 @@ +/* + * 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/commit_message.h" + +#include <utility> + +#include "paimon/common/io/memory_segment_output_stream.h" +#include "paimon/common/memory/memory_segment_utils.h" +#include "paimon/core/table/sink/commit_message_impl.h" +#include "paimon/core/table/sink/commit_message_serializer.h" +#include "paimon/io/byte_array_input_stream.h" +#include "paimon/io/data_input_stream.h" +#include "paimon/memory/bytes.h" +#include "paimon/memory/memory_pool.h" +#include "paimon/result.h" +#include "paimon/status.h" + +namespace paimon { + +int32_t CommitMessage::CurrentVersion() { + return CommitMessageSerializer::CURRENT_VERSION; +} + +CommitMessage::~CommitMessage() = default; + +Result<std::string> CommitMessage::Serialize(const std::shared_ptr<CommitMessage>& commit_message, + const std::shared_ptr<MemoryPool>& pool) { + CommitMessageSerializer serializer(pool); + MemorySegmentOutputStream out(MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool); + PAIMON_RETURN_NOT_OK(serializer.Serialize(commit_message, &out)); + PAIMON_UNIQUE_PTR<Bytes> bytes = + MemorySegmentUtils::CopyToBytes(out.Segments(), 0, out.CurrentSize(), pool.get()); + return std::string(bytes->data(), bytes->size()); +} + +Result<std::string> CommitMessage::SerializeList( + const std::vector<std::shared_ptr<CommitMessage>>& commit_messages, + const std::shared_ptr<MemoryPool>& pool) { + CommitMessageSerializer serializer(pool); + MemorySegmentOutputStream out(MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool); + PAIMON_RETURN_NOT_OK(serializer.SerializeList(commit_messages, &out)); + PAIMON_UNIQUE_PTR<Bytes> bytes = + MemorySegmentUtils::CopyToBytes(out.Segments(), 0, out.CurrentSize(), pool.get()); + return std::string(bytes->data(), bytes->size()); +} + +Result<std::shared_ptr<CommitMessage>> CommitMessage::Deserialize( + int32_t version, const char* buffer, int32_t length, const std::shared_ptr<MemoryPool>& pool) { + if (buffer == nullptr) { + return Status::Invalid("buffer is null pointer"); + } + if (length <= 0) { + return Status::Invalid("length is equal or less than zero"); + } + CommitMessageSerializer serializer(pool); + auto input_stream = std::make_shared<ByteArrayInputStream>(buffer, length); + DataInputStream in(input_stream); + return serializer.Deserialize(version, &in); +} Review Comment: CommitMessage::Deserialize() validates buffer/length but does not validate the MemoryPool pointer before constructing CommitMessageSerializer. A null pool can lead to undefined behavior during deserialization. ########## src/paimon/core/table/sink/commit_message.cpp: ########## @@ -0,0 +1,101 @@ +/* + * 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/commit_message.h" + +#include <utility> + +#include "paimon/common/io/memory_segment_output_stream.h" +#include "paimon/common/memory/memory_segment_utils.h" +#include "paimon/core/table/sink/commit_message_impl.h" +#include "paimon/core/table/sink/commit_message_serializer.h" +#include "paimon/io/byte_array_input_stream.h" +#include "paimon/io/data_input_stream.h" +#include "paimon/memory/bytes.h" +#include "paimon/memory/memory_pool.h" +#include "paimon/result.h" +#include "paimon/status.h" + +namespace paimon { + +int32_t CommitMessage::CurrentVersion() { + return CommitMessageSerializer::CURRENT_VERSION; +} + +CommitMessage::~CommitMessage() = default; + +Result<std::string> CommitMessage::Serialize(const std::shared_ptr<CommitMessage>& commit_message, + const std::shared_ptr<MemoryPool>& pool) { + CommitMessageSerializer serializer(pool); + MemorySegmentOutputStream out(MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool); + PAIMON_RETURN_NOT_OK(serializer.Serialize(commit_message, &out)); + PAIMON_UNIQUE_PTR<Bytes> bytes = + MemorySegmentUtils::CopyToBytes(out.Segments(), 0, out.CurrentSize(), pool.get()); + return std::string(bytes->data(), bytes->size()); +} + +Result<std::string> CommitMessage::SerializeList( + const std::vector<std::shared_ptr<CommitMessage>>& commit_messages, + const std::shared_ptr<MemoryPool>& pool) { + CommitMessageSerializer serializer(pool); + MemorySegmentOutputStream out(MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool); + PAIMON_RETURN_NOT_OK(serializer.SerializeList(commit_messages, &out)); + PAIMON_UNIQUE_PTR<Bytes> bytes = + MemorySegmentUtils::CopyToBytes(out.Segments(), 0, out.CurrentSize(), pool.get()); + return std::string(bytes->data(), bytes->size()); +} + +Result<std::shared_ptr<CommitMessage>> CommitMessage::Deserialize( + int32_t version, const char* buffer, int32_t length, const std::shared_ptr<MemoryPool>& pool) { + if (buffer == nullptr) { + return Status::Invalid("buffer is null pointer"); + } + if (length <= 0) { + return Status::Invalid("length is equal or less than zero"); + } + CommitMessageSerializer serializer(pool); + auto input_stream = std::make_shared<ByteArrayInputStream>(buffer, length); + DataInputStream in(input_stream); + return serializer.Deserialize(version, &in); +} + +Result<std::vector<std::shared_ptr<CommitMessage>>> CommitMessage::DeserializeList( + int32_t version, const char* buffer, int32_t length, const std::shared_ptr<MemoryPool>& pool) { + if (buffer == nullptr) { + return Status::Invalid("buffer is null pointer"); + } + if (length <= 0) { + return Status::Invalid("length is equal or less than zero"); + } + CommitMessageSerializer serializer(pool); + auto input_stream = std::make_shared<ByteArrayInputStream>(buffer, length); + DataInputStream in(input_stream); + return serializer.DeserializeList(version, &in); +} Review Comment: CommitMessage::DeserializeList() validates buffer/length but does not validate the MemoryPool pointer before constructing CommitMessageSerializer. A null pool can lead to undefined behavior during deserialization. -- 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]
