Do not embed metadata length in WriteDataHeader

Change-Id: Ibe6461c0e7673072471e3a613e6d6934759ed6e4

cpp/src/arrow/ipc/adapter.cc

@@ -190,7 +190,7 @@ class RecordBatchWriter {
     // determine the data header size then request a buffer such that you can
     // construct the flatbuffer data accessor object (see arrow::ipc::Message)
     std::shared_ptr<Buffer> data_header;
-    RETURN_NOT_OK(WriteDataHeader(
+    RETURN_NOT_OK(WriteRecordBatchMetadata(
         num_rows_, position - start_position, field_nodes_, buffer_meta_, &data_header));
 
     // Write the data header at the end

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

@@ -37,20 +37,6 @@ namespace flatbuf = org::apache::arrow::flatbuf;
 
 namespace ipc {
 
-const std::shared_ptr<DataType> BOOL = std::make_shared<BooleanType>();
-const std::shared_ptr<DataType> INT8 = std::make_shared<Int8Type>();
-const std::shared_ptr<DataType> INT16 = std::make_shared<Int16Type>();
-const std::shared_ptr<DataType> INT32 = std::make_shared<Int32Type>();
-const std::shared_ptr<DataType> INT64 = std::make_shared<Int64Type>();
-const std::shared_ptr<DataType> UINT8 = std::make_shared<UInt8Type>();
-const std::shared_ptr<DataType> UINT16 = std::make_shared<UInt16Type>();
-const std::shared_ptr<DataType> UINT32 = std::make_shared<UInt32Type>();
-const std::shared_ptr<DataType> UINT64 = std::make_shared<UInt64Type>();
-const std::shared_ptr<DataType> FLOAT = std::make_shared<FloatType>();
-const std::shared_ptr<DataType> DOUBLE = std::make_shared<DoubleType>();
-const std::shared_ptr<DataType> STRING = std::make_shared<StringType>();
-const std::shared_ptr<DataType> BINARY = std::make_shared<BinaryType>();
-
 static Status IntFromFlatbuffer(
     const flatbuf::Int* int_data, std::shared_ptr<DataType>* out) {
   if (int_data->bitWidth() > 64) {
@@ -62,16 +48,16 @@ static Status IntFromFlatbuffer(
 
   switch (int_data->bitWidth()) {
     case 8:
-      *out = int_data->is_signed() ? INT8 : UINT8;
+      *out = int_data->is_signed() ? int8() : uint8();
       break;
     case 16:
-      *out = int_data->is_signed() ? INT16 : UINT16;
+      *out = int_data->is_signed() ? int16() : uint16();
       break;
     case 32:
-      *out = int_data->is_signed() ? INT32 : UINT32;
+      *out = int_data->is_signed() ? int32() : uint32();
       break;
     case 64:
-      *out = int_data->is_signed() ? INT64 : UINT64;
+      *out = int_data->is_signed() ? int64() : uint64();
       break;
     default:
       return Status::NotImplemented("Integers not in cstdint are not implemented");
@@ -81,10 +67,12 @@ static Status IntFromFlatbuffer(
 
 static Status FloatFromFlatuffer(
     const flatbuf::FloatingPoint* float_data, std::shared_ptr<DataType>* out) {
-  if (float_data->precision() == flatbuf::Precision_SINGLE) {
-    *out = FLOAT;
+  if (float_data->precision() == flatbuf::Precision_HALF) {
+    *out = float16();
+  } else if (float_data->precision() == flatbuf::Precision_SINGLE) {
+    *out = float32();
   } else {
-    *out = DOUBLE;
+    *out = float64();
   }
   return Status::OK();
 }
@@ -100,13 +88,13 @@ static Status TypeFromFlatbuffer(flatbuf::Type type, const void* type_data,
       return FloatFromFlatuffer(
           static_cast<const flatbuf::FloatingPoint*>(type_data), out);
     case flatbuf::Type_Binary:
-      *out = BINARY;
+      *out = binary();
       return Status::OK();
     case flatbuf::Type_Utf8:
-      *out = STRING;
+      *out = utf8();
       return Status::OK();
     case flatbuf::Type_Bool:
-      *out = BOOL;
+      *out = boolean();
       return Status::OK();
     case flatbuf::Type_Decimal:
     case flatbuf::Type_Timestamp:
@@ -300,7 +288,7 @@ Status MessageBuilder::SetRecordBatch(int32_t length, int64_t body_length,
   return Status::OK();
 }
 
-Status WriteDataHeader(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;
@@ -317,17 +305,13 @@ Status MessageBuilder::Finish() {
 }
 
 Status MessageBuilder::GetBuffer(std::shared_ptr<Buffer>* out) {
-  // The message buffer is suffixed by the size of the complete flatbuffer as
-  // int32_t
-  // <uint8_t*: flatbuffer data><int32_t: flatbuffer size>
   int32_t size = fbb_.GetSize();
 
   auto result = std::make_shared<PoolBuffer>();
-  RETURN_NOT_OK(result->Resize(size + sizeof(int32_t)));
+  RETURN_NOT_OK(result->Resize(size));
 
   uint8_t* dst = result->mutable_data();
   memcpy(dst, fbb_.GetBufferPointer(), size);
-  memcpy(dst + size, reinterpret_cast<int32_t*>(&size), sizeof(int32_t));
 
   *out = result;
   return Status::OK();

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

@@ -70,7 +70,7 @@ class MessageBuilder {
   flatbuffers::FlatBufferBuilder fbb_;
 };
 
-Status WriteDataHeader(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);