gripleaf commented on code in PR #411:
URL: https://github.com/apache/paimon-cpp/pull/411#discussion_r4236954640
##########
src/paimon/core/utils/objects_file.h:
##########
@@ -256,22 +239,126 @@ Result<std::unique_ptr<InputStream>>
ObjectsFile<T>::OpenForRead(
}
template <typename T>
-Result<MemorySegment> ObjectsFile<T>::ReadFileSegment(
- const std::string& file_path, const std::optional<int64_t>& file_size)
const {
- PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<InputStream> input_stream,
- OpenForRead(file_path, file_size));
- PAIMON_ASSIGN_OR_RAISE(int64_t input_length, input_stream->Length());
+Result<std::shared_ptr<CacheValue>> ObjectsFile<T>::SerializeArrowBatches(
+ const std::string& file_name, std::optional<int64_t> file_size,
+ std::vector<std::shared_ptr<arrow::StructArray>>* batches,
+ std::optional<Status>* read_status) const {
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ std::shared_ptr<arrow::io::BufferOutputStream> output,
+ arrow::io::BufferOutputStream::Create(4096, arrow_pool_.get()));
+ auto write_options = arrow::ipc::IpcWriteOptions::Defaults();
+ write_options.memory_pool = arrow_pool_.get();
+ write_options.use_threads = false;
+ std::shared_ptr<arrow::ipc::RecordBatchWriter> writer;
+ Status cache_status;
+ *read_status = ReadUncachedArrowBatches(
+ file_name, file_size,
+ [&](const std::shared_ptr<arrow::StructArray>& batch) -> Status {
+ // The loading caller consumes these original batches, without an
IPC round trip.
+ batches->push_back(batch);
+ if (!cache_status.ok()) {
+ return Status::OK();
+ }
+ auto write_batch = [&]() -> Status {
+ // Preserve physical types, including ORC nanosecond
timestamps.
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ std::shared_ptr<arrow::RecordBatch> record_batch,
+ arrow::RecordBatch::FromStructArray(batch,
arrow_pool_.get()));
+ if (!writer) {
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ writer, arrow::ipc::MakeStreamWriter(output,
record_batch->schema(),
+ write_options));
+ }
+
PAIMON_RETURN_NOT_OK_FROM_ARROW(writer->WriteRecordBatch(*record_batch));
+ return Status::OK();
+ };
+ // Optional serialization failures must not interrupt the source
read.
+ cache_status = write_batch();
+ return Status::OK();
+ },
+ /*prepare_reader=*/nullptr);
+ PAIMON_RETURN_NOT_OK(read_status->value());
+ PAIMON_RETURN_NOT_OK(cache_status);
+ if (!writer) {
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
+ writer,
+ arrow::ipc::MakeStreamWriter(
+ output, arrow::schema(serializer_->GetDataType()->fields()),
write_options));
+ }
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(writer->Close());
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Buffer> buffer,
output->Finish());
+ if (buffer->size() > std::numeric_limits<int32_t>::max()) {
+ return Status::Invalid("Manifest Arrow cache entry exceeds the memory
segment limit");
Review Comment:
Thanks for catching this. Updated the cache to charge the retained Arrow
buffer capacity through CacheValue::GetMemoryUsage(), while keeping the segment
length equal to the logical IPC size. The built-in LRU now uses this charge,
and the guide documents the same requirement for external cache
implementations. Added regression tests covering retained allocation
accounting, admission, eviction, and buffer release.
--
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]
