[GitHub] [arrow] lidavidm commented on a change in pull request #9656: ARROW-11772: [C++] Provide reentrant IPC file reader

[GitBox]

lidavidm commented on a change in pull request #9656:
URL: https://github.com/apache/arrow/pull/9656#discussion_r606289729



##########
File path: cpp/src/arrow/ipc/reader.cc
##########
@@ -958,10 +959,98 @@ Result<std::shared_ptr<RecordBatchStreamReader>> 
RecordBatchStreamReader::Open(
 // ----------------------------------------------------------------------
 // Reader implementation
 
+// Common functions used in both the random-access file reader and the
+// asynchronous generator
 static inline FileBlock FileBlockFromFlatbuffer(const flatbuf::Block* block) {
   return FileBlock{block->offset(), block->metaDataLength(), 
block->bodyLength()};
 }
 
+static Result<std::unique_ptr<Message>> ReadMessageFromBlock(const FileBlock& 
block,
+                                                             
io::RandomAccessFile* file) {
+  if (!BitUtil::IsMultipleOf8(block.offset) ||
+      !BitUtil::IsMultipleOf8(block.metadata_length) ||
+      !BitUtil::IsMultipleOf8(block.body_length)) {
+    return Status::Invalid("Unaligned block in IPC file");
+  }
+
+  // TODO(wesm): this breaks integration tests, see ARROW-3256
+  // DCHECK_EQ((*out)->body_length(), block.body_length);
+
+  ARROW_ASSIGN_OR_RAISE(auto message,
+                        ReadMessage(block.offset, block.metadata_length, 
file));
+  return std::move(message);
+}
+
+static Future<std::shared_ptr<Message>> ReadMessageFromBlockAsync(
+    const FileBlock& block, io::RandomAccessFile* file, const io::IOContext& 
io_context) {
+  if (!BitUtil::IsMultipleOf8(block.offset) ||
+      !BitUtil::IsMultipleOf8(block.metadata_length) ||
+      !BitUtil::IsMultipleOf8(block.body_length)) {
+    return Status::Invalid("Unaligned block in IPC file");
+  }
+
+  // TODO(wesm): this breaks integration tests, see ARROW-3256
+  // DCHECK_EQ((*out)->body_length(), block.body_length);
+
+  return ReadMessageAsync(block.offset, block.metadata_length, file, 
io_context);
+}
+
+static Status ReadOneDictionary(Message* message, const IpcReadContext& 
context) {
+  CHECK_HAS_BODY(*message);
+  ARROW_ASSIGN_OR_RAISE(auto reader, Buffer::GetReader(message->body()));
+  DictionaryKind kind;
+  RETURN_NOT_OK(ReadDictionary(*message->metadata(), context, &kind, 
reader.get()));
+  if (kind != DictionaryKind::New) {
+    return Status::Invalid(
+        "Unsupported dictionary replacement or "
+        "dictionary delta in IPC file");
+  }
+  return Status::OK();
+}
+
+class RecordBatchFileReaderImpl;
+
+/// A generator of IPC messages (performs I/O for a record batch generator).
+///
+/// First all dictionary messages are yielded in order, then all record batch 
messages
+/// are yielded in order.
+AsyncGenerator<std::shared_ptr<Message>> MakeMessageGenerator(
+    std::shared_ptr<RecordBatchFileReaderImpl> state, const io::IOContext& 
io_context);

Review comment:
       It helps somewhat with S3 (with the effect lessening with more files) 
and hurts local files quite a bit, so I'll get rid of it. As Weston's noted, 
intra-file parallelism is going to need more work. 
   
   ![Local Median Scan Time 
(seconds)](https://user-images.githubusercontent.com/327919/113429824-24bb2c80-93a7-11eb-8144-70fe6ceea298.png)
   ![S3 Median Scan Time 
(seconds)](https://user-images.githubusercontent.com/327919/113429826-25ec5980-93a7-11eb-8322-e99b35dd42a0.png)
   




-- 
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.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org