IPC Format

Зачем Arrow IPC

Arrow определяет формат в памяти — но память одного процесса недоступна другому. Чтобы передать Arrow-данные между процессами, между нодами кластера, или сохранить на диск — нужна сериализация.

Arrow IPC (Inter-Process Communication) — это сериализация, спроектированная для zero-copy десериализации: получатель может работать с данными напрямую из полученного буфера без парсинга, копирования и перекодирования.

FlatBuffers: метаданные без парсинга

Arrow IPC использует FlatBuffers (Google) для сериализации метаданных (схема, типы, lengths). FlatBuffers — бинарный формат с O(1) доступом к полям без распаковки.

В отличие от Protobuf:

• FlatBuffers не требует десериализации — данные читаются напрямую из буфера через vtable offsets
• Нет аллокации памяти при чтении
• Доступ к полю — два чтения из памяти (vtable → value), не парсинг потока байтов

Arrow использует FlatBuffers только для метаданных (Schema, RecordBatch header, Dictionary header). Сами данные (buffers) лежат рядом — без FlatBuffers-обёртки.

Encapsulated Message Format

Каждое IPC-сообщение состоит из трёх частей:

1. Continuation marker — 4 байта: 0xFFFFFFFF
2. Metadata — FlatBuffers-сериализованный Message (header type + header + body length)
3. Body — конкатенация data buffers с padding

Message Types

Arrow IPC определяет четыре типа сообщений:

RecordBatch: nodes и buffers

RecordBatch header содержит два массива:

nodes — по одному FieldNode на каждый массив (включая дочерние). Порядок: depth-first traversal по Schema.

• length — количество элементов в массиве
• null_count — количество NULL-значений

buffers — по одному Buffer descriptor на каждый физический буфер.

• offset — смещение от начала body
• length — длина буфера в байтах

Пример: Struct<name: Utf8, age: Int32> с 3 элементами:

nodes:
 [0] Struct: length=3, null_count=1
 [1] Utf8: length=3, null_count=0
 [2] Int32: length=3, null_count=0

buffers:
 [0] Struct validity: offset=0, length=1
 [1] Utf8 validity: offset=64, length=1
 [2] Utf8 offsets: offset=128, length=16
 [3] Utf8 data: offset=192, length=8
 [4] Int32 validity: offset=256, length=1
 [5] Int32 data: offset=320, length=12

Порядок bufers — depth-first, для каждого массива: validity → offsets → data.

Stream Format

Arrow IPC Stream — последовательность сообщений, пригодная для потоковой передачи (pipe, socket, HTTP chunked):

Свойства потокового формата:

• Однопроходное чтение — получатель не возвращается назад
• Произвольное количество RecordBatch — каждый самодостаточен
• Нет random access — чтобы прочитать batch N, нужно пропустить batches 0..N-1
• Нет footer — размер потока неизвестен заранее

File Format (IPC File / Feather v2)

Arrow IPC File добавляет magic number и footer для random access:

Footer содержит массив Block структур для каждого RecordBatch:

Block {
 offset: Int64, // смещение от начала файла
 metadataLength: Int32, // длина metadata (без body)
 bodyLength: Int64 // длина body
}

Это даёт O(1) random access к любому RecordBatch по индексу — без чтения предшествующих батчей.

DictionaryBatch: потоковые словари

Dictionary-encoded колонки требуют отправки словаря до RecordBatch, который на него ссылается.

Протокол:

1. Schema содержит field с dictionary type и dictionary id
2. Перед RecordBatch идёт DictionaryBatch с тем же id и данными словаря
3. RecordBatch содержит только indices array — ссылки на словарь

Delta dictionaries: если isDelta = true, новые значения добавляются к существующему словарю (indices со старыми значениями остаются валидными). Полезно для потоков, где новые категории появляются по мере поступления данных.

// Поток с delta dictionary:
Schema(status: Dictionary<Int8, Utf8>, id=0)
DictionaryBatch(id=0, data=["active", "pending"]) // initial
RecordBatch(indices=[0, 1, 0, 0]) // 4 rows
DictionaryBatch(id=0, isDelta=true, data=["closed"]) // adds index 2
RecordBatch(indices=[0, 2, 1, 0]) // uses new value

IPC Compression

С Arrow 1.0 IPC поддерживает сжатие буферов внутри RecordBatch. Каждый buffer в body может быть сжат независимо.

Формат сжатого буфера:

1. Uncompressed length — 8 байт (Int64, LE). Если -1 — буфер не сжат
2. Compressed data — сжатый блок

Поддерживаемые кодеки:

• LZ4_FRAME — быстрая компрессия/декомпрессия, умеренный ratio
• ZSTD — лучший ratio при умеренной скорости

Zero-Copy: как это работает

Zero-copy десериализация — главное преимущество Arrow IPC. Получатель создаёт Arrow-массивы, указывающие прямо в полученный буфер:

1. Получить буфер (mmap файла, recv из сокета, shared memory)
2. Прочитать metadata через FlatBuffers — без аллокации
3. Для каждого buffer descriptor — вычислить указатель: base + offset
4. Создать Arrow Array, ссылающийся на эти указатели

Ни одного memcpy. Ни одного malloc (кроме тонких обёрток Array). Данные остаются в исходном буфере.

# Python: zero-copy через mmap
import pyarrow as pa
import mmap

with open("data.arrow", "rb") as f:
 mm = mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ)
 reader = pa.ipc.open_file(mm)
 batch = reader.get_batch(0) # указатели в mmap, не копия

Stream vs File: когда что

Ключевые выводы

1. FlatBuffers — metadata без парсинга: O(1) доступ к полям, zero allocation при чтении
2. Encapsulated message — continuation marker (0xFFFFFFFF) + metadata length + FlatBuffers header + body
3. Stream — Schema → DictionaryBatch* → RecordBatch* → EOS. Однопроходное чтение, без random access
4. File (Feather v2) — ARROW1 + messages + Footer + ARROW1. Random access через Block descriptors в footer
5. DictionaryBatch — словари передаются отдельно, delta dictionaries для потоковых обновлений
6. IPC compression (LZ4/ZSTD) — уменьшает размер за счёт потери zero-copy при чтении
7. Zero-copy — получатель строит Arrow arrays как указатели в исходный буфер, без memcpy