SteNicholas commented on code in PR #376: URL: https://github.com/apache/paimon-cpp/pull/376#discussion_r4278790216
########## src/paimon/format/vortex/vortex_format_writer.cpp: ########## @@ -0,0 +1,173 @@ +/* + * 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/format/vortex/vortex_format_writer.h" + +#include <utility> + +#include "arrow/c/bridge.h" +#include "arrow/memory_pool.h" +#include "arrow/type.h" +#include "paimon/common/metrics/metrics_impl.h" +#include "paimon/common/utils/arrow/arrow_utils.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/core/schema/schema_validation.h" +#include "paimon/format/vortex/vortex_ffi_util.h" +#include "paimon/fs/file_system.h" + +namespace paimon::vortex { + +VortexFormatWriter::VortexFormatWriter(std::shared_ptr<OutputStream> output, + std::shared_ptr<arrow::Schema> schema, VxSessionPtr session, + std::shared_ptr<VortexOutputContext> output_context, + vx_callback_sink* sink, + std::shared_ptr<arrow::MemoryPool> arrow_pool) + : output_(std::move(output)), + schema_(std::move(schema)), + arrow_pool_(std::move(arrow_pool)), + session_(std::move(session)), + output_context_(std::move(output_context)), + sink_(sink), + metrics_(std::make_shared<MetricsImpl>()) {} + +Result<std::unique_ptr<VortexFormatWriter>> VortexFormatWriter::Create( + const std::shared_ptr<OutputStream>& output, const std::shared_ptr<arrow::Schema>& schema, + const std::shared_ptr<arrow::MemoryPool>& arrow_pool) { + if (output == nullptr || schema == nullptr || arrow_pool == nullptr) { + return Status::Invalid("Vortex writer requires non-null output, schema and arrow pool"); + } + PAIMON_RETURN_NOT_OK(SchemaValidation::ValidateVortexArrowSchema(schema)); + VxSessionPtr session(vx_session_new(), vx_session_free); + if (session == nullptr) { + return Status::IOError("failed to create Vortex session"); + } + ::ArrowSchema ffi_schema = {}; + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*schema, &ffi_schema)); + vx_error* error = nullptr; + // vx_dtype_from_arrow_schema consumes ffi_schema on both success and failure. + VxDtypePtr dtype(vx_dtype_from_arrow_schema(&ffi_schema, &error), vx_dtype_free); + if (dtype == nullptr) { + return VortexFfiError("convert Arrow schema to Vortex dtype", error); + } + error = nullptr; + auto output_context = std::make_shared<VortexOutputContext>(output); + vx_callback_sink* sink = vx_callback_sink_open( + session.get(), VortexOutputContext::MakeCallbacks(output_context), dtype.get(), &error); + if (sink == nullptr) { + return VortexCallbackError("open Vortex callback sink", error, + output_context->GetCallbackStatus()); + } + return std::unique_ptr<VortexFormatWriter>(new VortexFormatWriter( + output, schema, std::move(session), std::move(output_context), sink, arrow_pool)); +} + +VortexFormatWriter::~VortexFormatWriter() { + if (sink_ != nullptr) { + // Not finished: abort so Vortex drops the sink without a footer; the caller discards the + // partial output. + output_context_->MarkAborted(); + vx_callback_sink_abort(sink_); + sink_ = nullptr; + output_context_->WaitForIdleCallbacks(); + } +} + +Status VortexFormatWriter::AddBatch(::ArrowArray* batch) { + if (batch == nullptr) { + return Status::Invalid("Vortex writer batch is nullptr"); + } + if (finished_ || sink_ == nullptr) { + return Status::Invalid("cannot add a batch after Vortex writer is finished"); + } + // arrow-rs cannot represent a sliced (offset > 0) top-level struct: importing one trips a slice + // assertion that aborts the process, so rebase such a batch to offset 0 first. Offset-0 batches + // are handed over untouched. + ::ArrowArray* vortex_batch = batch; + ::ArrowArray rebased_batch{}; + if (batch->offset != 0) { + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + std::shared_ptr<arrow::Array> imported, + arrow::ImportArray(batch, arrow::struct_(schema_->fields()))); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> normalized, + ArrowUtils::NormalizeArrayOffsets(imported, arrow_pool_.get())); + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*normalized, &rebased_batch)); Review Comment: **Retain the normalization pool with the exported buffers.** Rebasing a sliced batch can allocate validity or offset buffers from `arrow_pool_`, but `ExportArray` retains the buffers without retaining their memory pool. On abort, dropping the Rust task only requests cancellation; its queued arrays can survive the writer even when `WaitForIdleCallbacks()` sees no active I/O. `VortexWriterBuilder::Build` creates a separate Arrow pool adaptor for this writer, so its destruction leaves a dangling pool pointer in those buffers, causing use-after-free when Rust later releases them. An isolated probe of the pinned cancellation primitive confirmed that resource destruction happens after task destruction. Please attach `AddArrowArrayLifetime` to `rebased_batch` before handing it to Rust, or fully drain task-owned arrays before destroying the pool. ########## src/paimon/format/vortex/vortex_format_writer.cpp: ########## @@ -0,0 +1,173 @@ +/* + * 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/format/vortex/vortex_format_writer.h" + +#include <utility> + +#include "arrow/c/bridge.h" +#include "arrow/memory_pool.h" +#include "arrow/type.h" +#include "paimon/common/metrics/metrics_impl.h" +#include "paimon/common/utils/arrow/arrow_utils.h" +#include "paimon/common/utils/arrow/status_utils.h" +#include "paimon/core/schema/schema_validation.h" +#include "paimon/format/vortex/vortex_ffi_util.h" +#include "paimon/fs/file_system.h" + +namespace paimon::vortex { + +VortexFormatWriter::VortexFormatWriter(std::shared_ptr<OutputStream> output, + std::shared_ptr<arrow::Schema> schema, VxSessionPtr session, + std::shared_ptr<VortexOutputContext> output_context, + vx_callback_sink* sink, + std::shared_ptr<arrow::MemoryPool> arrow_pool) + : output_(std::move(output)), + schema_(std::move(schema)), + arrow_pool_(std::move(arrow_pool)), + session_(std::move(session)), + output_context_(std::move(output_context)), + sink_(sink), + metrics_(std::make_shared<MetricsImpl>()) {} + +Result<std::unique_ptr<VortexFormatWriter>> VortexFormatWriter::Create( + const std::shared_ptr<OutputStream>& output, const std::shared_ptr<arrow::Schema>& schema, + const std::shared_ptr<arrow::MemoryPool>& arrow_pool) { + if (output == nullptr || schema == nullptr || arrow_pool == nullptr) { + return Status::Invalid("Vortex writer requires non-null output, schema and arrow pool"); + } + PAIMON_RETURN_NOT_OK(SchemaValidation::ValidateVortexArrowSchema(schema)); + VxSessionPtr session(vx_session_new(), vx_session_free); + if (session == nullptr) { + return Status::IOError("failed to create Vortex session"); + } + ::ArrowSchema ffi_schema = {}; + PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*schema, &ffi_schema)); + vx_error* error = nullptr; + // vx_dtype_from_arrow_schema consumes ffi_schema on both success and failure. + VxDtypePtr dtype(vx_dtype_from_arrow_schema(&ffi_schema, &error), vx_dtype_free); + if (dtype == nullptr) { + return VortexFfiError("convert Arrow schema to Vortex dtype", error); + } + error = nullptr; + auto output_context = std::make_shared<VortexOutputContext>(output); + vx_callback_sink* sink = vx_callback_sink_open( + session.get(), VortexOutputContext::MakeCallbacks(output_context), dtype.get(), &error); + if (sink == nullptr) { + return VortexCallbackError("open Vortex callback sink", error, + output_context->GetCallbackStatus()); + } + return std::unique_ptr<VortexFormatWriter>(new VortexFormatWriter( + output, schema, std::move(session), std::move(output_context), sink, arrow_pool)); +} + +VortexFormatWriter::~VortexFormatWriter() { + if (sink_ != nullptr) { + // Not finished: abort so Vortex drops the sink without a footer; the caller discards the + // partial output. + output_context_->MarkAborted(); + vx_callback_sink_abort(sink_); + sink_ = nullptr; + output_context_->WaitForIdleCallbacks(); + } +} + +Status VortexFormatWriter::AddBatch(::ArrowArray* batch) { + if (batch == nullptr) { + return Status::Invalid("Vortex writer batch is nullptr"); + } + if (finished_ || sink_ == nullptr) { + return Status::Invalid("cannot add a batch after Vortex writer is finished"); + } + // arrow-rs cannot represent a sliced (offset > 0) top-level struct: importing one trips a slice + // assertion that aborts the process, so rebase such a batch to offset 0 first. Offset-0 batches + // are handed over untouched. + ::ArrowArray* vortex_batch = batch; + ::ArrowArray rebased_batch{}; + if (batch->offset != 0) { + PAIMON_ASSIGN_OR_RAISE_FROM_ARROW( + std::shared_ptr<arrow::Array> imported, + arrow::ImportArray(batch, arrow::struct_(schema_->fields()))); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> normalized, + ArrowUtils::NormalizeArrayOffsets(imported, arrow_pool_.get())); Review Comment: **Normalize nested offsets even when the batch offset is zero.** A valid outer batch can have offset 0 while containing a sliced `ROW<ROW<INT>>` column. This guard then skips normalization, and `vx_array_from_arrow` reaches an assertion in Arrow Rust's nested-struct import instead of returning a `Status`. I reproduced `assertion failed: end <= self.len()` with the pinned Arrow 58.3.0 import path: wrapping a sliced nested struct in a fresh offset-zero outer struct fails, while recursively normalizing the same data succeeds. Please normalize recursively regardless of the outer offset (the existing helper already detects whether normalization is needed), and add a regression test for this layout. -- 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]
