Relax 64 byte padding. Do not write RecordBatch embedded in Message for now

Change-Id: Ib493e982ff6c2d1502212985cec858a1e341c874

cpp/src/arrow/ipc/adapter.cc

@@ -48,15 +48,6 @@ namespace flatbuf = org::apache::arrow::flatbuf;
 
 namespace ipc {
 
-namespace {
-Status CheckMultipleOf64(int64_t size) {
-  if (BitUtil::IsMultipleOf64(size)) { return Status::OK(); }
-  return Status::Invalid(
-      "Attempted to write a buffer that "
-      "wasn't a multiple of 64 bytes");
-}
-}
-
 static bool IsPrimitive(const DataType* type) {
   DCHECK(type != nullptr);
   switch (type->type) {
@@ -224,7 +215,7 @@ class RecordBatchWriter {
 
 #ifndef NDEBUG
     RETURN_NOT_OK(dst->Tell(&current_position));
-    DCHECK(BitUtil::IsMultipleOf64(current_position));
+    DCHECK(BitUtil::IsMultipleOf8(current_position));
 #endif
 
     // Now write the buffers
@@ -246,7 +237,7 @@ class RecordBatchWriter {
 
 #ifndef NDEBUG
     RETURN_NOT_OK(dst->Tell(&current_position));
-    DCHECK(BitUtil::IsMultipleOf64(current_position));
+    DCHECK(BitUtil::IsMultipleOf8(current_position));
 #endif
 
     return Status::OK();
@@ -411,7 +402,9 @@ class RecordBatchReader {
 
   Status GetBuffer(int buffer_index, std::shared_ptr<Buffer>* out) {
     BufferMetadata metadata = metadata_->buffer(buffer_index);
-    RETURN_NOT_OK(CheckMultipleOf64(metadata.length));
+    if (!BitUtil::IsMultipleOf8(metadata.length)) {
+      return Status::Invalid("Expected buffer to be a multiple of 8 bytes");
+    }
     return file_->ReadAt(metadata.offset, metadata.length, out);
   }
 
@@ -441,14 +434,7 @@ Status ReadRecordBatchMetadata(int64_t offset, int32_t metadata_length,
     return Status::Invalid(ss.str());
   }
 
-  std::shared_ptr<Message> message;
-  RETURN_NOT_OK(Message::Open(buffer, sizeof(int32_t), &message));
-
-  if (message->type() != Message::RECORD_BATCH) {
-    return Status::Invalid("Metadata message is not a record batch");
-  }
-
-  *metadata = std::make_shared<RecordBatchMetadata>(message);
+  *metadata = std::make_shared<RecordBatchMetadata>(buffer, sizeof(int32_t));
   return Status::OK();
 }
 

cpp/src/arrow/ipc/metadata-internal.cc

@@ -291,10 +291,23 @@ Status MessageBuilder::SetRecordBatch(int32_t length, int64_t body_length,
 Status WriteRecordBatchMetadata(int32_t length, int64_t body_length,
     const std::vector<flatbuf::FieldNode>& nodes,
     const std::vector<flatbuf::Buffer>& buffers, std::shared_ptr<Buffer>* out) {
-  MessageBuilder message;
-  RETURN_NOT_OK(message.SetRecordBatch(length, body_length, nodes, buffers));
-  RETURN_NOT_OK(message.Finish());
-  return message.GetBuffer(out);
+  flatbuffers::FlatBufferBuilder fbb;
+
+  auto batch = flatbuf::CreateRecordBatch(fbb, length, fbb.CreateVectorOfStructs(nodes),
+      fbb.CreateVectorOfStructs(buffers));
+
+  fbb.Finish(batch);
+
+  int32_t size = fbb.GetSize();
+
+  auto result = std::make_shared<PoolBuffer>();
+  RETURN_NOT_OK(result->Resize(size));
+
+  uint8_t* dst = result->mutable_data();
+  memcpy(dst, fbb.GetBufferPointer(), size);
+
+  *out = result;
+  return Status::OK();
 }
 
 Status MessageBuilder::Finish() {

cpp/src/arrow/ipc/metadata.cc

@@ -184,6 +184,18 @@ RecordBatchMetadata::RecordBatchMetadata(const std::shared_ptr<Message>& message
   impl_.reset(new RecordBatchMetadataImpl(message->impl_->header()));
 }
 
+RecordBatchMetadata::RecordBatchMetadata(const std::shared_ptr<Buffer>& buffer, int64_t offset) {
+  message_ = nullptr;
+  buffer_ = buffer;
+
+  const flatbuf::RecordBatch* metadata = flatbuffers::GetRoot<flatbuf::RecordBatch>(
+      buffer->data() + offset);
+
+  // TODO(wesm): validate table
+
+  impl_.reset(new RecordBatchMetadataImpl(metadata));
+}
+
 RecordBatchMetadata::~RecordBatchMetadata() {}
 
 // TODO(wesm): Copying the flatbuffer data isn't great, but this will do for

cpp/src/arrow/ipc/metadata.h

@@ -100,6 +100,9 @@ struct BufferMetadata {
 class ARROW_EXPORT RecordBatchMetadata {
  public:
   explicit RecordBatchMetadata(const std::shared_ptr<Message>& message);
+
+  RecordBatchMetadata(const std::shared_ptr<Buffer>& message, int64_t offset);
+
   ~RecordBatchMetadata();
 
   FieldMetadata field(int i) const;
@@ -112,6 +115,7 @@ class ARROW_EXPORT RecordBatchMetadata {
  private:
   // Parent, owns the flatbuffer data
   std::shared_ptr<Message> message_;
+  std::shared_ptr<Buffer> buffer_;
 
   class RecordBatchMetadataImpl;
   std::unique_ptr<RecordBatchMetadataImpl> impl_;

cpp/src/arrow/util/bit-util.h

@@ -78,6 +78,10 @@ static inline bool IsMultipleOf64(int64_t n) {
   return (n & 63) == 0;
 }
 
+static inline bool IsMultipleOf8(int64_t n) {
+  return (n & 7) == 0;
+}
+
 inline int64_t RoundUpToMultipleOf64(int64_t num) {
   // TODO(wesm): is this definitely needed?
   // DCHECK_GE(num, 0);

integration/integration_test.py

@@ -15,6 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 
+import argparse
 import glob
 import itertools
 import os
@@ -53,10 +54,11 @@ def run_cmd(cmd):
 
 class IntegrationRunner(object):
 
-    def __init__(self, json_files, testers):
+    def __init__(self, json_files, testers, debug=False):
         self.json_files = json_files
         self.testers = testers
         self.temp_dir = tempfile.mkdtemp()
+        self.debug = debug
 
     def run(self):
         for producer, consumer in itertools.product(self.testers,
@@ -122,9 +124,15 @@ class CPPTester(Tester):
 
     name = 'C++'
 
+    def __init__(self, debug=False):
+        self.debug = debug
+
     def _run(self, arrow_path=None, json_path=None, command='VALIDATE'):
         cmd = [self.CPP_INTEGRATION_EXE, '--integration']
 
+        if self.debug:
+            cmd = ['gdb', '--args'] + cmd
+
         if arrow_path is not None:
             cmd.append('--arrow=' + arrow_path)
 
@@ -146,13 +154,19 @@ def get_json_files():
     return glob.glob(glob_pattern)
 
 
-def run_all_tests():
-    testers = [JavaTester(), CPPTester()]
+def run_all_tests(debug=False):
+    testers = [JavaTester(), CPPTester(debug=debug)]
     json_files = get_json_files()
 
-    runner = IntegrationRunner(json_files, testers)
+    runner = IntegrationRunner(json_files, testers, debug=debug)
     runner.run()
 
 
 if __name__ == '__main__':
-    run_all_tests()
+    parser = argparse.ArgumentParser(description='Arrow integration test CLI')
+    parser.add_argument('--debug', dest='debug', action='store_true',
+                        default=False,
+                        help='Run executables in debug mode as relevant')
+
+    args = parser.parse_args()
+    run_all_tests(debug=args.debug)