This is an automated email from the ASF dual-hosted git repository.
kou pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-erlang.git
The following commit(s) were added to refs/heads/main by this push:
new 24790a4 feat: symbolically represent IPC message bodies in IR (#115)
24790a4 is described below
commit 24790a4e24e457441114bf8b57b41cb601b7f049
Author: Benjamin Philip <[email protected]>
AuthorDate: Fri Aug 7 05:46:18 2026 +0530
feat: symbolically represent IPC message bodies in IR (#115)
## What issue does this PR close?
Closes #112.
The body of a message needs to represented symbolically in Arrow JSON.
We
previously serialized it within our internal representation, making it
impossible to produce Arrow JSON from the IR.
## What's Changed
We now leave the body as-is in the IR and only serialize it on `to_ipc`.
---------
Co-authored-by: Copilot Autofix powered by AI
<[email protected]>
---
guides/quick-run-through.livemd | 22 ++++++++--------------
src/arrow_ipc_message.erl | 19 +++++++++++--------
src/arrow_ipc_message.hrl | 2 +-
src/arrow_ipc_record_batch.erl | 18 ++++++++++++++++--
test/arrow_ipc_marks_data.hrl | 4 ++--
test/arrow_ipc_message_SUITE.erl | 2 +-
test/arrow_ipc_record_batch_SUITE.erl | 6 +++++-
7 files changed, 44 insertions(+), 29 deletions(-)
diff --git a/guides/quick-run-through.livemd b/guides/quick-run-through.livemd
index a3d6220..b781f97 100644
--- a/guides/quick-run-through.livemd
+++ b/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):
diff --git a/src/arrow_ipc_message.erl b/src/arrow_ipc_message.erl
index c27eb14..95adb42 100644
--- a/src/arrow_ipc_message.erl
+++ b/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>>.
diff --git a/src/arrow_ipc_message.hrl b/src/arrow_ipc_message.hrl
index d818269..08bd187 100644
--- a/src/arrow_ipc_message.hrl
+++ b/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
}).
diff --git a/src/arrow_ipc_record_batch.erl b/src/arrow_ipc_record_batch.erl
index 6d1cd77..6e79919 100644
--- a/src/arrow_ipc_record_batch.erl
+++ b/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.
diff --git a/test/arrow_ipc_marks_data.hrl b/test/arrow_ipc_marks_data.hrl
index 39669f7..cb6ac56 100644
--- a/test/arrow_ipc_marks_data.hrl
+++ b/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 %%
diff --git a/test/arrow_ipc_message_SUITE.erl b/test/arrow_ipc_message_SUITE.erl
index 27c928e..6116804 100644
--- a/test/arrow_ipc_message_SUITE.erl
+++ b/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 %%
diff --git a/test/arrow_ipc_record_batch_SUITE.erl
b/test/arrow_ipc_record_batch_SUITE.erl
index 2195e4f..f564f4a 100644
--- a/test/arrow_ipc_record_batch_SUITE.erl
+++ b/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)).
