Skip to content

GH-51100: [C++] Deprecate IPC tensor/sparse tensor messages#51102

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
pitrou wants to merge 1 commit into
base:main
Choose a base branch
from
Draft

GH-51100: [C++] Deprecate IPC tensor/sparse tensor messages #51102

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions c_glib/arrow-glib/input-stream.cpp
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -285,7 +285,9 @@ garrow_input_stream_read_tensor(GArrowInputStream *input_stream, GError **error)
{
auto arrow_input_stream = garrow_input_stream_get_raw(input_stream);

ARROW_SUPPRESS_DEPRECATION_WARNING
auto arrow_tensor = arrow::ipc::ReadTensor(arrow_input_stream.get());
ARROW_UNSUPPRESS_DEPRECATION_WARNING
if (garrow::check(error, arrow_tensor, "[input-stream][read-tensor]")) {
return garrow_tensor_new_raw(&(arrow_tensor.ValueOrDie()));
} else {
Expand Down
2 changes: 2 additions & 0 deletions c_glib/arrow-glib/output-stream.cpp
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -211,10 +211,12 @@ garrow_output_stream_write_tensor(GArrowOutputStream *stream,
auto arrow_tensor = garrow_tensor_get_raw(tensor);
int32_t metadata_length;
int64_t body_length;
ARROW_SUPPRESS_DEPRECATION_WARNING
auto status = arrow::ipc::WriteTensor(*arrow_tensor,
arrow_stream.get(),
&metadata_length,
&body_length);
ARROW_UNSUPPRESS_DEPRECATION_WARNING
if (garrow::check(error, status, "[output-stream][write-tensor]")) {
return metadata_length + body_length;
} else {
Expand Down
2 changes: 2 additions & 0 deletions cpp/src/arrow/ipc/generate_tensor_fuzz_corpus.cc
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -52,12 +52,14 @@ Result<std::shared_ptr<Buffer>> MakeSerializedBuffer(
}

Result<std::shared_ptr<Buffer>> SerializeTensor(const std::shared_ptr<Tensor>& tensor) {
ARROW_SUPPRESS_DEPRECATION_WARNING
return MakeSerializedBuffer(
[&](const std::shared_ptr<io::BufferOutputStream>& sink) -> Status {
int32_t metadata_length;
int64_t body_length;
return ipc::WriteTensor(*tensor, sink.get(), &metadata_length, &body_length);
});
ARROW_UNSUPPRESS_DEPRECATION_WARNING
}

Result<std::vector<std::shared_ptr<Tensor>>> Tensors() {
Expand Down
4 changes: 4 additions & 0 deletions cpp/src/arrow/ipc/reader.cc
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -2276,7 +2276,9 @@ Result<std::shared_ptr<Schema>> ReadSchema(const Message& message,
Result<std::shared_ptr<Tensor>> ReadTensor(io::InputStream* file) {
std::unique_ptr<Message> message;
RETURN_NOT_OK(ReadContiguousPayload(file, &message));
ARROW_SUPPRESS_DEPRECATION_WARNING
return ReadTensor(*message);
ARROW_UNSUPPRESS_DEPRECATION_WARNING
}

Result<std::shared_ptr<Tensor>> ReadTensor(const Message& message) {
Expand DownExpand Up@@ -2969,7 +2971,9 @@ Status FuzzIpcTensorStream(const uint8_t* data, int64_t size) {
std::shared_ptr<Tensor> tensor;

while (true) {
ARROW_SUPPRESS_DEPRECATION_WARNING
ARROW_ASSIGN_OR_RAISE(tensor, ReadTensor(&buffer_reader));
ARROW_UNSUPPRESS_DEPRECATION_WARNING
if (tensor == nullptr) {
break;
}
Expand Down
18 changes: 14 additions & 4 deletions cpp/src/arrow/ipc/reader.h
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -36,8 +36,7 @@
#include "arrow/util/macros.h"
#include "arrow/util/visibility.h"

namespace arrow {
namespace ipc {
namespace arrow::ipc {

class DictionaryMemo;
struct IpcPayload;
Expand DownExpand Up@@ -589,27 +588,39 @@ Result<std::shared_ptr<RecordBatch>> ReadRecordBatch(
///
/// \param[in] file an InputStream pointed at the start of the message
/// \return the read tensor
ARROW_DEPRECATED(
"Tensor-specific IPC messages are deprecated in 26.0.0. "
"Use FixedShapeTensor extension arrays instead.")
ARROW_EXPORT
Result<std::shared_ptr<Tensor>> ReadTensor(io::InputStream* file);

/// \brief EXPERIMENTAL: Read arrow::Tensor from IPC message
///
/// \param[in] message a Message containing the tensor metadata and body
/// \return the read tensor
ARROW_DEPRECATED(
"Tensor-specific IPC messages are deprecated in 26.0.0. "
"Use FixedShapeTensor extension arrays instead.")
ARROW_EXPORT
Result<std::shared_ptr<Tensor>> ReadTensor(const Message& message);

/// \brief EXPERIMENTAL: Read arrow::SparseTensor as encapsulated IPC message in file
///
/// \param[in] file an InputStream pointed at the start of the message
/// \return the read sparse tensor
ARROW_DEPRECATED(
"Tensor-specific IPC messages are deprecated in 26.0.0. "
"Use FixedShapeTensor extension arrays instead.")
ARROW_EXPORT
Result<std::shared_ptr<SparseTensor>> ReadSparseTensor(io::InputStream* file);

/// \brief EXPERIMENTAL: Read arrow::SparseTensor from IPC message
///
/// \param[in] message a Message containing the tensor metadata and body
/// \return the read sparse tensor
ARROW_DEPRECATED(
"Tensor-specific IPC messages are deprecated in 26.0.0. "
"Use FixedShapeTensor extension arrays instead.")
ARROW_EXPORT
Result<std::shared_ptr<SparseTensor>> ReadSparseTensor(const Message& message);

Expand DownExpand Up@@ -639,5 +650,4 @@ Status FuzzIpcFile(const uint8_t* data, int64_t size);

} // namespace internal

} // namespace ipc
} // namespace arrow
} // namespace arrow::ipc
24 changes: 24 additions & 0 deletions cpp/src/arrow/ipc/tensor_test.cc
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -72,15 +72,19 @@ class TestTensorRoundTrip : public BaseTensorTest {

ASSERT_OK(mmap_->Seek(0));

ARROW_SUPPRESS_DEPRECATION_WARNING
ASSERT_OK(WriteTensor(tensor, mmap_.get(), &metadata_length, &body_length));
ARROW_UNSUPPRESS_DEPRECATION_WARNING

const int64_t expected_body_length = elem_size * tensor.size();
ASSERT_EQ(expected_body_length, body_length);

ASSERT_OK(mmap_->Seek(0));

std::shared_ptr<Tensor> result;
ARROW_SUPPRESS_DEPRECATION_WARNING
ASSERT_OK_AND_ASSIGN(result, ReadTensor(mmap_.get()));
ARROW_UNSUPPRESS_DEPRECATION_WARNING

ASSERT_EQ(result->data()->size(), expected_body_length);
ASSERT_TRUE(tensor.Equals(*result));
Expand DownExpand Up@@ -115,7 +119,9 @@ TEST_F(TestTensorRoundTrip, BasicRoundtrip) {
CheckTensorRoundTrip(t_zero_length_dim);

int64_t serialized_size;
ARROW_SUPPRESS_DEPRECATION_WARNING
ASSERT_OK(GetTensorSize(t0, &serialized_size));
ARROW_UNSUPPRESS_DEPRECATION_WARNING
ASSERT_TRUE(serialized_size > static_cast<int64_t>(size * sizeof(int64_t)));

// ARROW-2840: Check that padding/alignment minded
Expand DownExpand Up@@ -151,8 +157,10 @@ class TestSparseTensorRoundTrip : public BaseTensorTest {

ASSERT_OK(mmap_->Seek(0));

ARROW_SUPPRESS_DEPRECATION_WARNING
ASSERT_OK(
WriteSparseTensor(sparse_tensor, mmap_.get(), &metadata_length, &body_length));
ARROW_UNSUPPRESS_DEPRECATION_WARNING

const auto& sparse_index =
checked_cast<const SparseCOOIndex&>(*sparse_tensor.sparse_index());
Expand All@@ -166,7 +174,9 @@ class TestSparseTensorRoundTrip : public BaseTensorTest {
ASSERT_OK(mmap_->Seek(0));

std::shared_ptr<SparseTensor> result;
ARROW_SUPPRESS_DEPRECATION_WARNING
ASSERT_OK_AND_ASSIGN(result, ReadSparseTensor(mmap_.get()));
ARROW_UNSUPPRESS_DEPRECATION_WARNING
ASSERT_EQ(SparseTensorFormat::COO, result->format_id());

const auto& resulted_sparse_index =
Expand All@@ -192,8 +202,10 @@ class TestSparseTensorRoundTrip : public BaseTensorTest {

ASSERT_OK(mmap_->Seek(0));

ARROW_SUPPRESS_DEPRECATION_WARNING
ASSERT_OK(
WriteSparseTensor(sparse_tensor, mmap_.get(), &metadata_length, &body_length));
ARROW_UNSUPPRESS_DEPRECATION_WARNING

const auto& sparse_index =
checked_cast<const SparseIndexType&>(*sparse_tensor.sparse_index());
Expand All@@ -209,7 +221,9 @@ class TestSparseTensorRoundTrip : public BaseTensorTest {
ASSERT_OK(mmap_->Seek(0));

std::shared_ptr<SparseTensor> result;
ARROW_SUPPRESS_DEPRECATION_WARNING
ASSERT_OK_AND_ASSIGN(result, ReadSparseTensor(mmap_.get()));
ARROW_UNSUPPRESS_DEPRECATION_WARNING

constexpr auto expected_format_id =
std::is_same<SparseIndexType, SparseCSRIndex>::value ? SparseTensorFormat::CSR
Expand All@@ -233,8 +247,10 @@ class TestSparseTensorRoundTrip : public BaseTensorTest {

ASSERT_OK(mmap_->Seek(0));

ARROW_SUPPRESS_DEPRECATION_WARNING
ASSERT_OK(
WriteSparseTensor(sparse_tensor, mmap_.get(), &metadata_length, &body_length));
ARROW_UNSUPPRESS_DEPRECATION_WARNING

const auto& sparse_index =
checked_cast<const SparseCSFIndex&>(*sparse_tensor.sparse_index());
Expand All@@ -259,7 +275,9 @@ class TestSparseTensorRoundTrip : public BaseTensorTest {
ASSERT_OK(mmap_->Seek(0));

std::shared_ptr<SparseTensor> result;
ARROW_SUPPRESS_DEPRECATION_WARNING
ASSERT_OK_AND_ASSIGN(result, ReadSparseTensor(mmap_.get()));
ARROW_UNSUPPRESS_DEPRECATION_WARNING
ASSERT_EQ(SparseTensorFormat::CSF, result->format_id());

const auto& resulted_sparse_index =
Expand DownExpand Up@@ -565,6 +583,7 @@ IpcPayload MakeSparseTensorPayload(const std::shared_ptr<Message>& message,
} // namespace

TEST(TestSparseCSFIndex, RejectInconsistentBufferCounts) {
ARROW_SUPPRESS_DEPRECATION_WARNING
// ndim == 1 is not a valid CSF index (it has no indptr buffers), and used to
// reach SparseCSFIndex's constructor with an empty indptr vector.
ASSERT_OK_AND_ASSIGN(auto message,
Expand All@@ -587,9 +606,11 @@ TEST(TestSparseCSFIndex, RejectInconsistentBufferCounts) {
/*num_indices_buffers=*/2,
/*axis_order_size=*/3));
ASSERT_RAISES(Invalid, ReadSparseTensor(*message));
ARROW_UNSUPPRESS_DEPRECATION_WARNING
}

TEST(TestSparseCSFIndex, RejectInconsistentPayloadBufferCounts) {
ARROW_SUPPRESS_DEPRECATION_WARNING
ASSERT_OK_AND_ASSIGN(auto message,
MakeCSFSparseTensorMessage(/*shape=*/{4}, /*num_indptr_buffers=*/0,
/*num_indices_buffers=*/1,
Expand All@@ -603,6 +624,7 @@ TEST(TestSparseCSFIndex, RejectInconsistentPayloadBufferCounts) {
/*axis_order_size=*/3));
ASSERT_RAISES(Invalid,
internal::ReadSparseTensorPayload(MakeSparseTensorPayload(message, 4)));
ARROW_UNSUPPRESS_DEPRECATION_WARNING
}

TEST(TestSparseCSXIndex, RejectIndptrLengthOverflow) {
Expand All@@ -618,6 +640,7 @@ TEST(TestSparseCSXIndex, RejectIndptrLengthOverflow) {
TEST(TestSparseTensor, RejectNegativeShapeAndNonZeroLength) {
// A negative non_zero_length must be rejected by GetSparseTensorMetadata,
// otherwise the negative size product bypasses the index buffer-size guards.
ARROW_SUPPRESS_DEPRECATION_WARNING
ASSERT_OK_AND_ASSIGN(
auto message, MakeCSFSparseTensorMessage(/*shape=*/{4, 4}, /*num_indptr_buffers=*/1,
/*num_indices_buffers=*/2,
Expand All@@ -631,6 +654,7 @@ TEST(TestSparseTensor, RejectNegativeShapeAndNonZeroLength) {
/*num_indices_buffers=*/2,
/*axis_order_size=*/2));
ASSERT_RAISES(Invalid, ReadSparseTensor(*message));
ARROW_UNSUPPRESS_DEPRECATION_WARNING
}

} // namespace test
Expand Down
4 changes: 4 additions & 0 deletions cpp/src/arrow/ipc/writer.cc
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -1119,7 +1119,9 @@ Status GetSparseTensorPayload(const SparseTensor& sparse_tensor, MemoryPool* poo
Result<std::unique_ptr<Message>> GetSparseTensorMessage(const SparseTensor& sparse_tensor,
MemoryPool* pool) {
IpcPayload payload;
ARROW_SUPPRESS_DEPRECATION_WARNING
RETURN_NOT_OK(GetSparseTensorPayload(sparse_tensor, pool, &payload));
ARROW_UNSUPPRESS_DEPRECATION_WARNING
return std::unique_ptr<Message>(
new Message(std::move(payload.metadata), std::move(payload.body_buffers[0])));
}
Expand DownExpand Up@@ -1154,7 +1156,9 @@ Status GetTensorSize(const Tensor& tensor, int64_t* size) {
int32_t metadata_length = 0;
int64_t body_length = 0;
io::MockOutputStream dst;
ARROW_SUPPRESS_DEPRECATION_WARNING
RETURN_NOT_OK(WriteTensor(tensor, &dst, &metadata_length, &body_length));
ARROW_UNSUPPRESS_DEPRECATION_WARNING
*size = dst.GetExtentBytesWritten();
return Status::OK();
}
Expand Down
44 changes: 22 additions & 22 deletions cpp/src/arrow/ipc/writer.h
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -23,33 +23,16 @@
#include <memory>
#include <vector>

#include "arrow/io/type_fwd.h"
#include "arrow/ipc/dictionary.h" // IWYU pragma: export
#include "arrow/ipc/message.h"
#include "arrow/ipc/options.h"
#include "arrow/result.h"
#include "arrow/type_fwd.h"
#include "arrow/util/macros.h"
#include "arrow/util/visibility.h"

namespace arrow {

class Array;
class Buffer;
class MemoryManager;
class MemoryPool;
class RecordBatch;
class Schema;
class Status;
class Table;
class Tensor;
class SparseTensor;

namespace io {

class OutputStream;

} // namespace io

namespace ipc {
namespace arrow::ipc {

/// \brief Intermediate data structure with metadata header, and zero
/// or more buffers for the message body.
Expand DownExpand Up@@ -284,6 +267,9 @@ Status GetRecordBatchSize(const RecordBatch& batch, const IpcWriteOptions& optio
/// \param[in] tensor the tensor to write
/// \param[out] size the size of the complete encapsulated message
/// \return Status
ARROW_DEPRECATED(
"Tensor-specific IPC messages are deprecated in 26.0.0. "
"Use FixedShapeTensor extension arrays instead.")
ARROW_EXPORT
Status GetTensorSize(const Tensor& tensor, int64_t* size);

Expand All@@ -293,6 +279,9 @@ Status GetTensorSize(const Tensor& tensor, int64_t* size);
/// \param[in] tensor the Tensor to write
/// \param[in] pool MemoryPool to allocate space for metadata
/// \return the resulting Message
ARROW_DEPRECATED(
"Tensor-specific IPC messages are deprecated in 26.0.0. "
"Use FixedShapeTensor extension arrays instead.")
ARROW_EXPORT
Result<std::unique_ptr<Message>> GetTensorMessage(const Tensor& tensor, MemoryPool* pool);

Expand All@@ -312,6 +301,9 @@ Result<std::unique_ptr<Message>> GetTensorMessage(const Tensor& tensor, MemoryPo
/// \param[out] metadata_length the actual metadata length, including padding
/// \param[out] body_length the actual message body length
/// \return Status
ARROW_DEPRECATED(
"Tensor-specific IPC messages are deprecated in 26.0.0. "
"Use FixedShapeTensor extension arrays instead.")
ARROW_EXPORT
Status WriteTensor(const Tensor& tensor, io::OutputStream* dst, int32_t* metadata_length,
int64_t* body_length);
Expand All@@ -327,6 +319,9 @@ Status WriteTensor(const Tensor& tensor, io::OutputStream* dst, int32_t* metadat
/// \param[in] sparse_tensor the SparseTensor to write
/// \param[in] pool MemoryPool to allocate space for metadata
/// \return the resulting Message
ARROW_DEPRECATED(
"Tensor-specific IPC messages are deprecated in 26.0.0. "
"Use FixedShapeTensor extension arrays instead.")
ARROW_EXPORT
Result<std::unique_ptr<Message>> GetSparseTensorMessage(const SparseTensor& sparse_tensor,
MemoryPool* pool);
Expand All@@ -341,6 +336,9 @@ Result<std::unique_ptr<Message>> GetSparseTensorMessage(const SparseTensor& spar
/// \param[out] metadata_length the actual metadata length, including padding
/// \param[out] body_length the actual message body length
/// \return Status
ARROW_DEPRECATED(
"Tensor-specific IPC messages are deprecated in 26.0.0. "
"Use FixedShapeTensor extension arrays instead.")
ARROW_EXPORT
Status WriteSparseTensor(const SparseTensor& sparse_tensor, io::OutputStream* dst,
int32_t* metadata_length, int64_t* body_length);
Expand DownExpand Up@@ -413,6 +411,9 @@ Status WriteIpcPayload(const IpcPayload& payload, const IpcWriteOptions& options
/// \param[in,out] pool for any required temporary memory allocations
/// \param[out] out the returned IpcPayload
/// \return Status
ARROW_DEPRECATED(
"Tensor-specific IPC messages are deprecated in 26.0.0. "
"Use FixedShapeTensor extension arrays instead.")
ARROW_EXPORT
Status GetSparseTensorPayload(const SparseTensor& sparse_tensor, MemoryPool* pool,
IpcPayload* out);
Expand DownExpand Up@@ -471,5 +472,4 @@ Result<std::unique_ptr<RecordBatchWriter>> OpenRecordBatchWriter(
const IpcWriteOptions& options = IpcWriteOptions::Defaults());

} // namespace internal
} // namespace ipc
} // namespace arrow
} // namespace arrow::ipc
Loading