Skip to content

Arrow batches and IPC

Ownership rules explain how to initialize, move, and release batches, schemas, buffers, and streams.

Declarations below are included directly from the public headers.

Batch construction and access

vgi_rpc/batch.h on GitHub

vgi_rpc/batch.h
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
#ifndef VGI_BATCH_H
#define VGI_BATCH_H
#include "vgi_rpc/ipc.h"
#ifdef __cplusplus
extern "C" {
#endif

struct VgiBatch {
    struct ArrowSchema schema;
    struct ArrowArray array;
    struct ArrowArrayView view;
};
void VgiBatchReset(struct VgiBatch *batch);
int VgiBatchInit(struct VgiBatch *batch, const struct ArrowSchema *schema,
                 struct ArrowError *error);
int VgiBatchDecode(struct VgiBatch *batch, struct ArrowBufferView bytes, struct ArrowError *error);
int VgiBatchView(struct VgiBatch *batch, struct ArrowError *error);
int64_t VgiFieldIndex(const struct ArrowSchema *schema, const char *name);
struct ArrowArray *VgiBatchColumn(struct VgiBatch *batch, const char *name);
const struct ArrowArrayView *VgiBatchGet(const struct VgiBatch *batch, const char *name);
struct ArrowStringView VgiBatchString(const struct VgiBatch *batch, const char *name);
struct ArrowBufferView VgiBatchBytes(const struct VgiBatch *batch, const char *name);
int64_t VgiBatchInt(const struct VgiBatch *batch, const char *name, int64_t fallback);
int VgiBatchIsNull(const struct VgiBatch *batch, const char *name);
int VgiBatchChild(const struct VgiBatch *batch, const char *name, struct VgiBatch *out,
                  struct ArrowError *error);
/* Setters append one value. FinishRow fills omitted fields with null when
 * nullable, otherwise an empty/default value; use Finish for explicit rows. */
int VgiBatchSetString(struct VgiBatch *batch, const char *name, const char *value);
int VgiBatchSetBytes(struct VgiBatch *batch, const char *name, const void *data, int64_t size);
int VgiBatchSetInt(struct VgiBatch *batch, const char *name, int64_t value);
int VgiBatchSetNull(struct VgiBatch *batch, const char *name);
int VgiBatchFinishRow(struct VgiBatch *batch, struct ArrowError *error);
int VgiBatchFinish(struct VgiBatch *batch, struct ArrowError *error);
int VgiArrayAppendString(struct ArrowArray *array, struct ArrowStringView value);
int VgiArrayAppendDefault(struct ArrowArray *array, const struct ArrowSchema *schema);
/* Copy one logical value, decoding dictionaries and casting compatible storage.
 * The destination must be an appending nanoarrow array with the given schema. */
int VgiArrayAppendValue(struct ArrowArray *out, const struct ArrowSchema *schema,
                        const struct ArrowArrayView *value, const struct ArrowSchema *value_schema,
                        int64_t index, struct ArrowError *error);
int VgiBatchCoerce(const struct VgiBatch *input, const struct ArrowSchema *schema,
                   struct VgiBatch *out, struct ArrowError *error);
struct ArrowStringView VgiValueString(const struct ArrowArrayView *value, int64_t index);
int VgiValueIsNumeric(const struct ArrowArrayView *value);
int VgiValueDouble(const struct ArrowArrayView *value, const struct ArrowSchema *schema,
                   int64_t index, double *out, struct ArrowError *error);
#ifdef __cplusplus
}
#endif
#endif

Complete IPC messages

vgi_rpc/ipc.h on GitHub

vgi_rpc/ipc.h
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
#ifndef VGI_IPC_H
#define VGI_IPC_H
#include "nanoarrow/nanoarrow.h"
#ifdef __cplusplus
extern "C" {
#endif

/* Encode exactly one batch as a complete IPC stream, including dictionaries.
 * Inputs are borrowed for the duration of the call. metadata is NULL or Arrow
 * C Data metadata (ArrowMetadataBuilder). out must be initialized and empty;
 * it remains empty on failure. Release successful output with ArrowBufferReset. */
int VgiIpcWriteBatch(const struct ArrowSchema *schema, const struct ArrowArray *array,
                     const char *metadata, struct ArrowBuffer *out, struct ArrowError *error);

/* Decode exactly one complete, uncompressed IPC stream containing one batch.
 * Input bytes are borrowed. Outputs own their data and must be zero-initialized
 * (metadata must be initialized and empty). On failure no outputs are retained.
 * Trailing bytes, multiple batches, missing EOS and truncated input are errors.
 * Release outputs with ArrowSchemaRelease, ArrowArrayRelease, ArrowBufferReset. */
int VgiIpcReadBatch(struct ArrowBufferView input, struct ArrowSchema *schema,
                    struct ArrowArray *array, struct ArrowBuffer *metadata,
                    struct ArrowError *error);
#ifdef __cplusplus
}
#endif
#endif

Incremental IPC streams

vgi_rpc/stream.h on GitHub

vgi_rpc/stream.h
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
#ifndef VGI_STREAM_H
#define VGI_STREAM_H
#include "nanoarrow/nanoarrow_ipc.h"
#ifdef __cplusplus
extern "C" {
#endif

/* Incremental IPC with per-record-batch metadata. Init moves the input/output
 * stream on success. Reset releases it; ownership of any returned array stays
 * with the caller. All structs and outputs must be zero-initialized. */
struct VgiIpcReader {
    struct ArrowArrayStream stream;
    void *private_data;
};
int VgiIpcReaderInit(struct VgiIpcReader *reader, struct ArrowIpcInputStream *input);
int VgiIpcReaderSchema(struct VgiIpcReader *reader, struct ArrowSchema *out,
                       struct ArrowError *error);
int VgiIpcReaderNext(struct VgiIpcReader *reader, struct ArrowArray *out,
                     struct ArrowBuffer *metadata, struct ArrowError *error);
void VgiIpcReaderReset(struct VgiIpcReader *reader);

struct VgiIpcWriter {
    struct ArrowIpcWriter writer;
    struct ArrowSchema schema;
    int finished;
};
int VgiIpcWriterInit(struct VgiIpcWriter *writer, const struct ArrowSchema *schema,
                     struct ArrowIpcOutputStream *output, struct ArrowError *error);
int VgiIpcWriterNext(struct VgiIpcWriter *writer, const struct ArrowArray *array,
                     const char *metadata, struct ArrowError *error);
int VgiIpcWriterFinish(struct VgiIpcWriter *writer, struct ArrowError *error);
void VgiIpcWriterReset(struct VgiIpcWriter *writer);

/* Schema IPC carries no batches and is used for function arguments/results. */
int VgiIpcWriteSchema(const struct ArrowSchema *schema, struct ArrowBuffer *out,
                      struct ArrowError *error);
int VgiIpcReadSchema(struct ArrowBufferView input, struct ArrowSchema *out,
                     struct ArrowError *error);
#ifdef __cplusplus
}
#endif
#endif

Schema factories

vgi_rpc/schema.h on GitHub

vgi_rpc/schema.h
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
#ifndef VGI_SCHEMA_INTERNAL_H
#define VGI_SCHEMA_INTERNAL_H
#include "nanoarrow/nanoarrow.h"

struct VgiSchemaNode {
    const char *format;
    const char *name;
    const char *metadata;
    int64_t flags;
    int64_t n_children;
    const struct VgiSchemaNode *const *children;
    const struct VgiSchemaNode *dictionary;
};
int VgiSchemaBuild(const struct VgiSchemaNode *node, struct ArrowSchema *out,
                   struct ArrowError *error);
#endif