lxy-9602 commented on code in PR #411:
URL: https://github.com/apache/paimon-cpp/pull/411#discussion_r4236526826
##########
src/paimon/core/manifest/manifest_file.cpp:
##########
@@ -133,6 +134,66 @@ Status ManifestFile::ReadBucketEntries(const std::string&
file_name, int32_t buc
});
}
+Status ManifestFile::ReadRowRangeEntries(
+ const std::string& file_name, const RowRangeIndex& row_ranges,
+ const std::function<Result<bool>(const ManifestEntry&)>& filter,
+ std::optional<int64_t> file_size, std::vector<ManifestEntry>* entries)
const {
+ return ReadArrowBatches(
+ file_name, file_size,
+ [this, &row_ranges, &filter,
+ entries](const std::shared_ptr<arrow::StructArray>& batch) -> Status {
+ // ManifestMetaReader has aligned both the entry and its nested
file schema. Probe
+ // the two range columns without allocating DataFileMeta, stats or
binary keys.
+ constexpr int32_t kFileFieldIndex = 5;
+ constexpr int32_t kRowCountFieldIndex = 2;
+ constexpr int32_t kFirstRowIdFieldIndex = 18;
Review Comment:
Please put this definition into `ManifestEntry` and `DataFileMeta`. Right
now, the (de)serialization logic and the fast-short-circuiting logic are
scattered across different places. Ideally, the position/location information
should live together with the data structure definition.
##########
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:
`OwnedBuffer` keeps the full underlying allocation, but we only record the
cache segment length as `buffer->size()`. With Arrow 17, the buffer may
double-expand, and `Finish()` does not shrink capacity.
For example, an ~5 MiB IPC may keep an ~8 MiB allocation, while the cache
only accounts for ~5 MiB.
As a result, resident memory can exceed the configured budget, while the
eviction logic may still think it’s within limit.
Recommendation: shrink the buffer before inserting into the cache, or charge
the cache using the actual capacity.
--
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]
