Skip to content
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

Jump to bottom

GH-36674: [C++] Use anonymous namespace in arrow/ipc/reader.cc #36937

Merged
merged 2 commits into from
Aug 9, 2023
Merged

GH-36674: [C++] Use anonymous namespace in arrow/ipc/reader.cc #36937

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f93d6c2
[C++] Use anonymous namespace in arrow/IPC/reader.cc
pegasas Jul 30, 2023
91c3a09
Fix lint
pitrou Aug 9, 2023
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
81 changes: 40 additions & 41 deletions cpp/src/arrow/ipc/reader.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -109,8 +109,6 @@ Status InvalidMessageType(MessageType expected, MessageType actual) {
} \
} while (0)

} // namespace

// ----------------------------------------------------------------------
// Record batch read path

Expand Down Expand Up @@ -634,36 +632,14 @@ Status GetCompressionExperimental(const flatbuf::Message* message,
return Status::OK();
}

static Status ReadContiguousPayload(io::InputStream* file,
std::unique_ptr<Message>* message) {
Status ReadContiguousPayload(io::InputStream* file, std::unique_ptr<Message>* message) {
ARROW_ASSIGN_OR_RAISE(*message, ReadMessage(file));
if (*message == nullptr) {
return Status::Invalid("Unable to read metadata at offset");
}
return Status::OK();
}
pegasas marked this conversation as resolved.
Show resolved Hide resolved

Result<std::shared_ptr<RecordBatch>> ReadRecordBatch(
const std::shared_ptr<Schema>& schema, const DictionaryMemo* dictionary_memo,
const IpcReadOptions& options, io::InputStream* file) {
std::unique_ptr<Message> message;
RETURN_NOT_OK(ReadContiguousPayload(file, &message));
CHECK_HAS_BODY(*message);
ARROW_ASSIGN_OR_RAISE(auto reader, Buffer::GetReader(message->body()));
return ReadRecordBatch(*message->metadata(), schema, dictionary_memo, options,
reader.get());
}

Result<std::shared_ptr<RecordBatch>> ReadRecordBatch(
const Message& message, const std::shared_ptr<Schema>& schema,
const DictionaryMemo* dictionary_memo, const IpcReadOptions& options) {
CHECK_MESSAGE_TYPE(MessageType::RECORD_BATCH, message.type());
CHECK_HAS_BODY(message);
ARROW_ASSIGN_OR_RAISE(auto reader, Buffer::GetReader(message.body()));
return ReadRecordBatch(*message.metadata(), schema, dictionary_memo, options,
reader.get());
}

Result<RecordBatchWithMetadata> ReadRecordBatchInternal(
const Buffer& metadata, const std::shared_ptr<Schema>& schema,
const std::vector<bool>& inclusion_mask, IpcReadContext& context,
Expand Down Expand Up @@ -764,22 +740,6 @@ Status UnpackSchemaMessage(const Message& message, const IpcReadOptions& options
out_schema, field_inclusion_mask, swap_endian);
}

Result<std::shared_ptr<RecordBatch>> ReadRecordBatch(
const Buffer& metadata, const std::shared_ptr<Schema>& schema,
const DictionaryMemo* dictionary_memo, const IpcReadOptions& options,
io::RandomAccessFile* file) {
std::shared_ptr<Schema> out_schema;
// Empty means do not use
std::vector<bool> inclusion_mask;
IpcReadContext context(const_cast<DictionaryMemo*>(dictionary_memo), options, false);
RETURN_NOT_OK(GetInclusionMaskAndOutSchema(schema, context.options.included_fields,
&inclusion_mask, &out_schema));
ARROW_ASSIGN_OR_RAISE(
auto batch_and_custom_metadata,
ReadRecordBatchInternal(metadata, schema, inclusion_mask, context, file));
return batch_and_custom_metadata.batch;
}

Status ReadDictionary(const Buffer& metadata, const IpcReadContext& context,
DictionaryKind* kind, io::RandomAccessFile* file) {
const flatbuf::Message* message = nullptr;
Expand Down Expand Up @@ -851,6 +811,45 @@ Status ReadDictionary(const Message& message, const IpcReadContext& context,
return ReadDictionary(*message.metadata(), context, kind, reader.get());
}

} // namespace

Result<std::shared_ptr<RecordBatch>> ReadRecordBatch(
const Buffer& metadata, const std::shared_ptr<Schema>& schema,
const DictionaryMemo* dictionary_memo, const IpcReadOptions& options,
io::RandomAccessFile* file) {
std::shared_ptr<Schema> out_schema;
// Empty means do not use
std::vector<bool> inclusion_mask;
IpcReadContext context(const_cast<DictionaryMemo*>(dictionary_memo), options, false);
RETURN_NOT_OK(GetInclusionMaskAndOutSchema(schema, context.options.included_fields,
&inclusion_mask, &out_schema));
ARROW_ASSIGN_OR_RAISE(
auto batch_and_custom_metadata,
ReadRecordBatchInternal(metadata, schema, inclusion_mask, context, file));
return batch_and_custom_metadata.batch;
}

Result<std::shared_ptr<RecordBatch>> ReadRecordBatch(
const std::shared_ptr<Schema>& schema, const DictionaryMemo* dictionary_memo,
const IpcReadOptions& options, io::InputStream* file) {
std::unique_ptr<Message> message;
RETURN_NOT_OK(ReadContiguousPayload(file, &message));
CHECK_HAS_BODY(*message);
ARROW_ASSIGN_OR_RAISE(auto reader, Buffer::GetReader(message->body()));
return ReadRecordBatch(*message->metadata(), schema, dictionary_memo, options,
reader.get());
}

Result<std::shared_ptr<RecordBatch>> ReadRecordBatch(
const Message& message, const std::shared_ptr<Schema>& schema,
const DictionaryMemo* dictionary_memo, const IpcReadOptions& options) {
CHECK_MESSAGE_TYPE(MessageType::RECORD_BATCH, message.type());
CHECK_HAS_BODY(message);
ARROW_ASSIGN_OR_RAISE(auto reader, Buffer::GetReader(message.body()));
return ReadRecordBatch(*message.metadata(), schema, dictionary_memo, options,
reader.get());
}

// Streaming format decoder
class StreamDecoderInternal : public MessageDecoderListener {
public:
Expand Down
Loading