feat: symbolically represent IPC message bodies in IR

guides/quick-run-through.livemd

@@ -1,5 +1,7 @@
 <!-- -*- mode: markdown; fill-column: 80; -*- -->
-<!---
+
+<!--
+-
   Licensed to the Apache Software Foundation (ASF) under one
   or more contributor license agreements.  See the NOTICE file
   distributed with this work for additional information
@@ -22,7 +24,7 @@
 
 ```elixir
 Mix.install([
-  {:arrow, git: "https://github.com/Benjamin-Philip/arrow.git"}
+  {:arrow, git: "https://github.com/apache/arrow-erlang.git"}
 ])
 ```
 
@@ -159,20 +161,12 @@ Class2RecordBatch = arrow_ipc_record_batch:from_erlang(Class2Columns),
 Class3RecordBatch = arrow_ipc_record_batch:from_erlang(Class3Columns).
 ```
 
-Each of these RecordBatches need a body:
-
-```erlang
-Class1Msg = arrow_ipc_message:body_from_erlang(Class1Columns),
-Class2Msg = arrow_ipc_message:body_from_erlang(Class2Columns),
-Class3Msg = arrow_ipc_message:body_from_erlang(Class3Columns).
-```
-
-Which can then be combined into a message:
+We can then create a Message from each Record Batch:
 
 ```erlang
-Class1Msg = arrow_ipc_message:from_erlang(Class1RecordBatch, Class1Body),
-Class2Msg = arrow_ipc_message:from_erlang(Class2RecordBatch, Class3Body),
-Class3Msg = arrow_ipc_message:from_erlang(Class3RecordBatch, Class3Body).
+Class1Msg = arrow_ipc_message:from_erlang(Class1RecordBatch, Class1Columns),
+Class2Msg = arrow_ipc_message:from_erlang(Class2RecordBatch, Class2Columns),
+Class3Msg = arrow_ipc_message:from_erlang(Class3RecordBatch, Class3Columns).
 ```
 
 If we wanted to, we could serialize these messages into an [Encapsulated message](https://arrow.apache.org/docs/format/Columnar.html#encapsulated-message-format):

src/arrow_ipc_message.erl

@@ -33,7 +33,7 @@ to represent a message. Metadata such as:
 3.  `body_length`: The length of the body in bytes
 4.  `custom_metadata`: A list of custom metadata in key-value format
 5.  `body`: The actual body. Can be undefined (in the case of Schema)
-    or a binary (in the case of Record Batch).
+    or a `[t:arrow_array:array/0]` (in the case of Record Batch).
 
 Currently, changing the version and custom metadata are not supported, but they
 have been added for forwards compatibility.
@@ -68,21 +68,24 @@ for more info:
 -type key_value() :: #{key => string(), value => string()}.
 
 -doc """
-Creates a message given a data header.
+Creates a message given a schema data header.
 """.
--spec from_erlang(Header :: arrow_ipc_schema:schema() | arrow_ipc_record_batch:record_batch()) ->
+-spec from_erlang(Header :: arrow_ipc_schema:schema()) ->
     Message :: message().
 from_erlang(Header) ->
     #message{header = Header, body_length = 0}.
 
 -doc """
-Creates a message given a data header and a body.
+Creates a message given a record batch data header and a body.
 """.
 -spec from_erlang(
-    Header :: arrow_ipc_schema:schema() | arrow_ipc_record_batch:record_batch(), Body :: binary()
+    Header :: arrow_ipc_record_batch:record_batch(),
+    Body :: [arrow_array:array()]
 ) -> Message :: message().
 from_erlang(Header, Body) ->
-    #message{header = Header, body = Body, body_length = byte_size(Body)}.
+    #message{
+        header = Header, body = Body, body_length = arrow_ipc_record_batch:body_length(Header)
+    }.
 
 -doc """
 Serializes a message into the Encapsulated Message Format.
@@ -98,8 +101,8 @@ to_ipc(Message) ->
         case Message#message.body of
             undefined ->
                 <<>>;
-            Bin ->
-                Bin
+            Arrays ->
+                body_from_erlang(Arrays)
         end,
 
     <<Continuation/binary, MetadataSize/binary, Metadata/binary, Body/binary>>.

src/arrow_ipc_message.hrl

@@ -27,5 +27,5 @@
 
     %% This field is unique to arrow.
     %% The rest are from the flatbuffers definitions.
-    body :: binary() | undefined
+    body :: [arrow_array:array()] | undefined
 }).

src/arrow_ipc_record_batch.erl

@@ -44,7 +44,7 @@ comapatibility.
 You can find RecordBatches in the Arrow spec
 [here](https://arrow.apache.org/docs/format/Columnar.html#recordbatch-message).
 """.
--export([from_erlang/1]).
+-export([from_erlang/1, body_length/1]).
 -export_type([field_node/0, buffer/0, record_batch/0]).
 
 -include("arrow_ipc_record_batch.hrl").
@@ -55,7 +55,7 @@ You can find RecordBatches in the Arrow spec
 -type record_batch() :: #record_batch{}.
 
 -doc """
-Creates a RecordBatch given a list of arrays
+Creates a RecordBatch given a list of arrays.
 """.
 -spec from_erlang(Arrays :: [arrow_array:array()]) -> RecordBatch :: record_batch().
 from_erlang(Arrays) ->
@@ -114,3 +114,17 @@ buffer_data(Buffer, CurOffset) ->
         #{offset => CurOffset, length => Buffer#buffer.length},
         arrow_buffer:size(Buffer) + CurOffset
     }.
+
+-doc """
+Returns the body length of a Record Batch.
+""".
+-spec body_length(RecordBatch :: record_batch()) -> non_neg_integer().
+body_length(RecordBatch) ->
+    case RecordBatch#record_batch.buffers of
+        [] ->
+            0;
+        Buffers ->
+            #{offset := Offset, length := Length} = lists:last(Buffers),
+            End = Offset + Length,
+            End + arrow_utils:pad_len(End)
+    end.

test/arrow_ipc_marks_data.hrl

@@ -61,9 +61,9 @@
 ).
 -define(Columns, [?ID, ?Name, ?Age, ?AnnualMarks]).
 -define(RecordBatch, arrow_ipc_record_batch:from_erlang(?Columns)).
--define(Body, <<<<(arrow_array:to_arrow(Array))/binary>> || Array <- ?Columns>>).
--define(RecordBatchMsg, arrow_ipc_message:from_erlang(?RecordBatch, ?Body)).
+-define(RecordBatchMsg, arrow_ipc_message:from_erlang(?RecordBatch, ?Columns)).
 -define(RecordBatchEMF, arrow_ipc_message:to_ipc(?RecordBatchMsg)).
+-define(Body, <<<<(arrow_array:to_arrow(Array))/binary>> || Array <- ?Columns>>).
 
 %%%%%%%%%%%%%%%%
 %% IPC Stream %%

test/arrow_ipc_message_SUITE.erl

@@ -68,7 +68,7 @@ valid_custom_metadata_on_from_erlang(_Config) ->
 
 valid_body_on_from_erlang(_Config) ->
     ?assertEqual((?SchemaMsg)#message.body, undefined),
-    ?assertEqual((?RecordBatchMsg)#message.body, ?Body).
+    ?assertEqual((?RecordBatchMsg)#message.body, ?Columns).
 
 %%%%%%%%%%%%%%
 %% to_ipc/1 %%

test/arrow_ipc_record_batch_SUITE.erl

@@ -32,7 +32,8 @@ all() ->
         valid_length_on_from_erlang,
         valid_nodes_on_from_erlang,
         valid_buffers_on_from_erlang,
-        valid_compression_on_from_erlang
+        valid_compression_on_from_erlang,
+        body_length
     ].
 
 valid_length_on_from_erlang(_Config) ->
@@ -62,3 +63,6 @@ valid_buffers_on_from_erlang(_Config) ->
 
 valid_compression_on_from_erlang(_Config) ->
     ?assertEqual((?RecordBatch)#record_batch.compression, undefined).
+
+body_length(_Config) ->
+    ?assertEqual(arrow_ipc_record_batch:body_length(?RecordBatch), byte_size(?Body)).