Documentation
¶
Overview ¶
Package ipc moves data between kuma and other Arrow implementations.
The Arrow C data interface is how two libraries in one process hand each other a column without copying it. It is a pair of C structs, one describing the type and one holding the pointers, and the type half of it is a string: "l" is an int64, "tsu:UTC" is a microsecond timestamp in UTC, "+l" is a list. Format and Type are the two directions of that mapping, and EncodeMetadata and DecodeMetadata are the small binary blob that carries the key and value pairs attached to a field.
Export and Import are the other half, the values. They work on a Layout, which is what the C struct holds once the pointers are Go slices: a length, an offset, a null count and the buffers. Neither of them copies a value. Import borrows the buffers it is given, which is what the C data interface is for, so the array it returns is only alive for as long as the memory behind those buffers is.
The C structs themselves are CSchema and CArray, with ExportField, ExportArray, ImportField and ImportArray between them and kuma. Those need cgo and live behind a build tag, so that a program that only reads Parquet does not pay for a C toolchain and still cross compiles. Everything else here is pure Go and works on any platform, including the ones with no C toolchain at all.
EncodeSchema and DecodeSchema are the other way two libraries hand each other a table, which is to write it down. A schema on the wire is an Arrow IPC message, meaning FlatBuffers, and this package reads and writes that itself rather than pulling in a generated reader, because a message off a socket is somebody else's bytes and the generated readers for this format are not bounds checked.
EncodeBatch and DecodeBatch are the values that go with a schema. A record batch is a message describing where every buffer of every column sits, followed by the buffers themselves, and the schema it belongs to is something the reader is expected to have already. Nothing is copied on the way in, so the arrays a batch decodes into point at the bytes it was decoded from.
Writer and Reader are those two messages put together into the Arrow IPC stream format, which is a schema, a record batch for every batch, and a marker saying there are no more. That is what one process sends another over a socket, and it is read from front to back because that is the only way anything arriving over a socket can be read.
FileWriter and FileReader are the Arrow IPC file format, which is that same stream with a magic number in front of it and a footer behind it. The footer holds a block per batch saying where in the file it starts and how long it is, so a reader that can seek does not have to walk the batches it does not want. FileReader takes an io.ReaderAt and the size, the way archive/zip does, and hands out batches by number in any order.
A dictionary encoded column travels split in two. The record batch carries the indices and the values are sent once in a message of their own, so a file of ten thousand batches of country codes holds the two hundred and fifty strings once. Both readers put the two halves back together, and the columns they hand out share the one copy of the values. The writers take the column and do the splitting, so nothing outside this package has to know about the identifiers the format ties the halves together with.
Both interfaces are checked against pyarrow, in both directions, by the tests in testdata/pyarrow. The C one runs two libraries in one process and compares buffer addresses, and the message one passes files.
What is not here yet: arrays of the nested types, dictionary deltas, a compressed body, and the arrow-go bridge.
Stability: tier 1, stable.
Index ¶
- Variables
- func DecodeMetadata(b []byte) (dtype.Metadata, error)
- func DecodeSchema(b []byte) (dtype.Schema, error)
- func EncodeBatch(s dtype.Schema, b Batch) ([]byte, error)
- func EncodeMetadata(m dtype.Metadata) ([]byte, error)
- func EncodeSchema(s dtype.Schema) ([]byte, error)
- func ExportArray(a *array.Array, out *CArray) error
- func ExportField(f dtype.Field, out *CSchema) error
- func Format(t dtype.DataType) (string, error)
- func Import(format string, l Layout) (*array.Array, error)
- func ImportField(s *CSchema) (dtype.Field, error)
- func ReleaseArray(a *CArray)
- func ReleaseSchema(s *CSchema)
- func Type(format string, children []dtype.Field) (dtype.DataType, error)
- type Batch
- type CArray
- type CSchema
- type FileReader
- type FileWriter
- type Imported
- type Layout
- type Reader
- type Writer
Examples ¶
Constants ¶
This section is empty.
Variables ¶
var ( // ErrFormat is returned when a format string is not one the C data // interface defines, or is one that names a type kuma has no equivalent // for, such as a union. ErrFormat = errors.New("bad format string") // ErrType is returned when a kuma type has no format string, which means // the type was not built by this module or holds a parameter that is not a // real type, such as a time32 counting nanoseconds. ErrType = errors.New("type has no format string") // ErrChildren is returned when the number of child fields does not match // what the format string needs. A list has one child and a map has one // child that is a struct of two. ErrChildren = errors.New("wrong number of children") // ErrMetadata is returned when a metadata blob is truncated, claims a // length that is not there, or is too large to be written. ErrMetadata = errors.New("bad metadata") // ErrBuffers is returned when the buffers of an array are not the ones its // type needs: the wrong number of them, one too short for the values it has // to hold, or offsets that point outside the data. ErrBuffers = errors.New("bad buffers") // ErrMessage is returned when a piece of Arrow IPC metadata is malformed: // an offset that points outside the message it lives in, a length that runs // past the end, a vtable that is not there. The metadata is FlatBuffers, and // a FlatBuffer arriving over a socket is somebody else's bytes, so every // number in it is checked before it is used. ErrMessage = errors.New("bad message") // ErrClosed is returned by a write to a stream or a file whose end of stream // marker has already gone out. There is nowhere left to put a batch, and in // a file the footer saying where the batches are is already behind it, so // this is a mistake in the calling code rather than anything the bytes did. ErrClosed = errors.New("the stream is closed") // ErrUnsupported is returned for the shapes this package understands and // cannot build yet: the nested types, since the array package has no nested // arrays to put them in, a dictionary that arrives as a delta, and a file // whose dictionary was written twice. ErrUnsupported = errors.New("not supported yet") )
The errors this package returns. They are comparable with errors.Is, and the message carries the format string or the type that caused it, since a mismatch at this boundary is nearly always a producer and a consumer that disagree about one column out of two hundred.
Functions ¶
func DecodeMetadata ¶
DecodeMetadata reads the blob EncodeMetadata writes.
An empty or nil blob is no metadata and not an error, which is what a null metadata member arrives as. Anything else that does not describe a whole set of pairs is an error rather than as much as could be read, since a truncated blob means the two sides disagree about the layout and the rest of the schema is not to be trusted either.
The keys and values are copied. The blob usually points into memory the producer owns and will free, and metadata that turns to nonsense after a release callback runs would be a bug nobody could reproduce.
func DecodeSchema ¶ added in v0.0.2
DecodeSchema reads an encapsulated Arrow IPC schema message.
The bytes are somebody else's, so everything in them is checked: the message has to be a schema, the version has to be one kuma reads, and every offset has to point inside the message. A column of a type kuma has no equivalent for is an error naming the type and the column, since that is what the person holding the file needs to know.
Anything after the message is ignored, so a schema read out of the front of a stream does not have to be cut out of it first.
func EncodeBatch ¶ added in v0.0.3
EncodeBatch returns the Arrow IPC record batch message for b, with the values after it.
The result is an encapsulated message, framed the way a schema is, followed by the body it describes. Every buffer in the body starts on an eight byte boundary and the body is padded out to one, so a reader can use the buffers where they lie and whatever comes next starts on the alignment the format promises.
The columns have to be the types the schema says they are, and there have to be as many of them as the schema has fields. The nested types cannot be written yet, and the error names the column that could not be.
A dictionary encoded column is written as its indices. The values travel in a dictionary batch of their own, which is a message this does not write, so a caller doing its own framing has to write those as well and the Writer and the FileWriter are the ones that do. Either form of the column is taken: the dictionary encoded column itself, which is what array.NewDictionary builds, or the indices on their own, which is what DecodeBatch hands back.
A column carrying an offset, which is what slicing a batch out of a longer one gives, is trimmed on the way out. The format has no per column offset, so the values in front of the first one are not written, and a validity bitmap that begins part way through a byte is shifted rather than sliced.
func EncodeMetadata ¶
EncodeMetadata packs metadata into the blob the C data interface expects in the metadata member of a schema.
Empty metadata encodes as nil, not as a count of zero. The C data interface says the member is a null pointer when there is nothing to say, and a blob holding only a zero would be a pointer the consumer has to read to learn nothing.
The keys and values are bytes rather than text. Nothing checks that they are valid UTF-8, since Arrow does not require it and a producer that puts a serialized message in a value is doing something the format allows.
func EncodeSchema ¶ added in v0.0.2
EncodeSchema returns the Arrow IPC schema message for s.
The result is an encapsulated message: the continuation bytes, the length, and the metadata padded out to eight. That is what pyarrow's schema.serialize writes and what its read_schema reads, so the bytes go straight into any other implementation without anything wrapped around them first.
A type kuma can hold but Arrow cannot name, and there are none today, would be an error here rather than a column that quietly changes type.
func ExportArray ¶ added in v0.0.2
ExportArray fills out with the C array for a.
No value is copied. The buffer pointers point into the array's own memory, which is pinned until the consumer calls the release callback, so the array has to stay reachable in Go until then as well. Pinning stops the collector moving the buffers, not the caller dropping the last reference to the frame they came from.
The view layout gets the buffer of buffer sizes the C ABI asks for, built here, since a Go slice knows its own length and a C pointer does not.
The type does not travel with the values. A consumer needs the schema too, which is what ExportField is for.
func ExportField ¶ added in v0.0.2
ExportField fills out with the C schema for f.
The strings and the child structs are allocated with malloc and are owned by the release callback, which the consumer calls when it is finished. The consumer owns the struct out points at, both before and after: this fills it in and never allocates or frees it.
Nested types export their children, so a struct of three fields or a map of string to int64 comes out whole. That is worth having even though the values of those types cannot be exported yet, because a schema travels on its own: it is what a reader asks for before it decides which columns to read.
A dictionary type exports the index type as the format string and the value type as the dictionary member, which is how the C data interface splits it.
On error nothing is left allocated and out is a released schema.
func Format ¶
Format returns the Arrow C data interface format string for t.
The string describes t and nothing about the values, so a nullable column and a non-nullable one of the same type give the same string. Nullability is a flag on the schema struct rather than part of the type, the same way it is a flag on dtype.Field.
The nested types give the format string of the container only: a list of int64 is "+l", and the int64 travels separately as the child. That is how the C structs are laid out, with the children in their own array, and it is why there is no format string anywhere that names two types at once.
A dictionary is the one type whose format string is not its own. The C data interface puts the index type in the format string and the value type in the dictionary member of the schema, so Format reports the format of the index, and an exporter has to fill in the dictionary member itself. Passing the result back to Type gives the index type back rather than the dictionary, which is correct: the string never held the value type to begin with.
Example ¶
package main
import (
"fmt"
"github.com/tamnd/kuma/dtype"
"github.com/tamnd/kuma/ipc"
)
func main() {
types := []dtype.DataType{
dtype.Int64,
dtype.String,
dtype.Timestamp{Unit: dtype.Microsecond, Zone: "Europe/London"},
dtype.Decimal128{Precision: 18, Scale: 2},
dtype.List{Elem: dtype.Int64},
}
for _, t := range types {
format, err := ipc.Format(t)
if err != nil {
fmt.Println(err)
continue
}
fmt.Printf("%-40s %s\n", t, format)
}
}
Output: int64 l string vu timestamp[us, tz=Europe/London] tsu:Europe/London decimal128(18, 2) d:18,2 list<int64> +l
func Import ¶
Import builds a kuma array out of the buffers of an incoming one.
It takes the format string rather than the type, because the format string is what says which layout the buffers are in. The three text layouts all become one kuma type and only the format string tells them apart, so a caller that has already called Type has thrown that away.
Nothing is copied except the views of a column that arrived in one of the offset layouts, which are sixteen bytes per value and have to be built because kuma has no offsets. The values themselves are still borrowed, in that case and in every other. That means the array shares the producer's memory, so whatever release callback comes with those buffers must not run until the array is gone.
The null count in the layout is ignored and the nulls are counted from the bitmap, since a producer is allowed to send a count of minus one for a count it has not worked out.
A dictionary encoded column does not arrive here as one. Its format string names the index type, so this builds the indices, and the caller that read the dictionary member of the schema is the one that knows better.
Example ¶
A column of strings from another library usually arrives in the offset layout, which kuma does not store. Import converts it to views, which is sixteen bytes per value, and leaves the text itself where it is.
package main
import (
"encoding/binary"
"fmt"
"github.com/tamnd/kuma/ipc"
)
func main() {
// What pyarrow would hand over for the three values below: the offsets of
// each value in the data buffer, and the data buffer.
offsets := []byte{}
for _, n := range []uint32{0, 3, 6, 12} {
offsets = binary.NativeEndian.AppendUint32(offsets, n)
}
data := []byte("onetwothree!")
a, err := ipc.Import("u", ipc.Layout{
Length: 3,
Buffers: [][]byte{nil, offsets, data},
})
if err != nil {
fmt.Println(err)
return
}
for i := range a.Len() {
fmt.Printf("%d %s\n", i, a.Bytes(i))
}
// Going back out is the layout kuma stores, so the format string changes
// and nothing is copied.
format, err := ipc.Format(a.DType())
if err != nil {
fmt.Println(err)
return
}
l, err := ipc.Export(a)
if err != nil {
fmt.Println(err)
return
}
fmt.Printf("%s in %d buffers\n", format, len(l.Buffers))
}
Output: 0 one 1 two 2 three! vu in 2 buffers
func ImportField ¶ added in v0.0.2
ImportField reads a C schema and returns the field it describes.
The schema is released before this returns, whether or not it succeeded, because everything in it has been copied into Go by then. That is the contract of the interface: a consumer that is handed a struct owns it.
The struct itself is not freed. Whoever allocated it frees it.
func ReleaseArray ¶ added in v0.0.2
func ReleaseArray(a *CArray)
ReleaseArray releases an array struct, whoever produced it.
func ReleaseSchema ¶ added in v0.0.2
func ReleaseSchema(s *CSchema)
ReleaseSchema and ReleaseArray hand a struct back to whoever produced it by calling its own release callback. They do nothing for a null pointer or for a struct that has already been released, which is what makes them safe to defer.
A Go caller needs them for two things: a struct this package exported and then did not manage to send, and a struct another library handed over that this package is finished with. The memory the struct itself sits in belongs to whoever allocated it either way, and neither of these frees that.
func Type ¶
Type returns the kuma type a format string names.
The children are the types of the members the format string does not name, already converted: one field for a list, one field per member for a struct, and for a map one field whose type is a two field struct of the key and the value. Everything else takes no children and ignores them. An importer walks the C structs from the leaves up, so by the time it asks about a list it already has the element type in hand, which is why this takes the children rather than trying to parse a tree out of a string that cannot describe one.
The result is validated, so a producer that writes a decimal with a precision of ninety nine is rejected here rather than three operations later.
Text and bytes are the one place where this is not the inverse of Format. The C data interface has three layouts for each, meaning 32 bit offsets, 64 bit offsets and views, and kuma stores exactly one of them. An import materializes whatever arrived into the view layout, so "u", "U" and "vu" all become dtype.String and "z", "Z" and "vz" all become dtype.Binary. A caller that needs to know which layout the incoming buffers are in has the format string in front of it and should read that rather than the type.
A dictionary encoded column does not arrive as a format string. The format names the index type and the value type hangs off the dictionary member of the schema, so an importer calls Type twice and builds the dtype.Dictionary itself.
Example ¶
A list is two calls, one for the element and one for the list, because the format string of a list says only that it is a list. An importer works its way up from the leaves, which is the order the C structs are laid out in.
package main
import (
"fmt"
"github.com/tamnd/kuma/dtype"
"github.com/tamnd/kuma/ipc"
)
func main() {
elem, err := ipc.Type("l", nil)
if err != nil {
fmt.Println(err)
return
}
list, err := ipc.Type("+l", []dtype.Field{{Name: "item", Type: elem, Nullable: true}})
if err != nil {
fmt.Println(err)
return
}
fmt.Println(list)
// The three text layouts all arrive as one type, since kuma stores text
// one way and an import converts whatever it is given.
for _, format := range []string{"u", "U", "vu"} {
t, err := ipc.Type(format, nil)
if err != nil {
fmt.Println(err)
continue
}
fmt.Printf("%s is %s\n", format, t)
}
}
Output: list<int64> u is string U is string vu is string
Types ¶
type Batch ¶ added in v0.0.3
type Batch struct {
// Length is the number of rows.
Length int
// Columns are the values, one array per field of the schema, in the
// order the schema has them.
Columns []*array.Array
}
Batch is one record batch: some rows of every column of a schema.
Length is on the struct rather than taken from the columns because a batch of no columns still has a number of rows, which is what a count of a table with no projection reads. Every column has to be that long.
func DecodeBatch ¶ added in v0.0.3
DecodeBatch reads one encapsulated Arrow IPC record batch message and the body after it, and returns the batch along with whatever follows the pair of them.
The schema is the one the batch belongs to, which the format expects a reader to have already: a batch on the wire says how many values each column has and where its buffers are, and nothing at all about what they mean.
Nothing is copied. The arrays point into b, so they are alive for as long as those bytes are unmodified, the same bargain Import makes.
A dictionary encoded column comes back as its indices, of the index type the schema names, since the values are in a dictionary batch and one message on its own has no way to know them. The Reader and the FileReader put the two together, which is what a caller who wants the column rather than the message should be using.
The bytes are somebody else's, so everything in them is checked. Every buffer has to lie inside the body, every column has to have as many values as the batch says it has rows, and the buffers have to add up to what the schema needs. A compressed body and the nested types are refused rather than half read.
Text and bytes are the one place a schema does not say enough. Arrow has four layouts for each and kuma collapses them into one type, so this reads the layout out of the batch instead: a column with a variadic buffer count is the view layout kuma writes, and one without is the offset layout everybody else does. A batch that mixes the two is refused, since there is nothing in either message saying which column is which.
type CArray ¶ added in v0.0.2
type CArray = C.struct_ArrowArray
CArray is the ArrowArray struct of the C data interface, which holds the values: a length, an offset, a null count and the buffer pointers.
type CSchema ¶ added in v0.0.2
type CSchema = C.struct_ArrowSchema
CSchema is the ArrowSchema struct of the C data interface, which describes one field: its name, its type, whether it is nullable and its metadata.
It is an alias for the cgo type, so a caller with a struct of its own can convert a pointer to it through unsafe.Pointer. Two packages that both run cgo get two Go types for the same C struct, and that conversion is how they meet. The struct itself is the same memory either way.
type FileReader ¶ added in v0.0.4
type FileReader struct {
// contains filtered or unexported fields
}
FileReader reads the Arrow IPC file format.
The footer is read when the reader is made, which is the schema and where every batch in the file is, and after that a batch is read by number. Nothing else is touched: reading the last batch of a file reads the footer and that batch. That is what the format is for, and it is the difference between a file and a stream.
The schema is the one in the footer rather than the one in front of the batches. Both are written by the same writer and say the same thing, and the footer is the one a reader already has in hand.
Each batch is read into a buffer of its own and its arrays point into that buffer, the same bargain the rest of the package makes. A batch costs the memory of the batch for as long as it is held and nothing once it is dropped.
func NewFileReader ¶ added in v0.0.4
func NewFileReader(r io.ReaderAt, size int64) (*FileReader, error)
NewFileReader reads the footer of a file and returns a reader for the batches it indexes. The size is how many bytes the file has, which the caller knows and an io.ReaderAt does not, the same way a zip reader is opened.
The dictionaries are read here as well. They are needed by any batch that reads from them, so a reader that put it off would be doing it on whichever batch happened to be asked for first, and there is one of them per column rather than one per batch.
func (*FileReader) Batch ¶ added in v0.0.4
func (r *FileReader) Batch(i int) (Batch, error)
Batch reads batch i, which is a read of that batch and nothing else. The batches of a file can be read in any order and more than once.
func (*FileReader) NumBatches ¶ added in v0.0.4
func (r *FileReader) NumBatches() int
NumBatches is how many record batches the file holds.
func (*FileReader) Schema ¶ added in v0.0.4
func (r *FileReader) Schema() dtype.Schema
Schema is the schema every batch in the file belongs to.
type FileWriter ¶ added in v0.0.4
type FileWriter struct {
// contains filtered or unexported fields
}
FileWriter writes the Arrow IPC file format.
The schema goes out when the writer is made and every batch has to match it, the same as a stream. Close writes the footer, which is the index of everything written before it, and a file without one is not a file: the batches are all on the disk and there is nothing saying where. So Close matters more here than it does on a stream, and its error more still.
An error from the underlying writer sticks, since the file is broken at that point. An error about a batch does not, since nothing went out.
The values of a dictionary encoded column go out once, in front of the first batch that reads from them, and the footer says where they are. A file has one dictionary per column and no way to replace it, so a batch arriving with different values is an error rather than a second dictionary.
Example ¶
package main
import (
"bytes"
"fmt"
"github.com/tamnd/kuma/array"
"github.com/tamnd/kuma/dtype"
"github.com/tamnd/kuma/ipc"
)
func main() {
s := dtype.Schema{Fields: []dtype.Field{
{Name: "id", Type: dtype.Int64},
{Name: "symbol", Type: dtype.String},
}}
var buf bytes.Buffer
w, err := ipc.NewFileWriter(&buf, s)
if err != nil {
fmt.Println(err)
return
}
for _, ids := range [][]int64{{1, 2}, {3}} {
symbols := make([]string, len(ids))
for i, id := range ids {
symbols[i] = fmt.Sprintf("S%d", id)
}
err = w.Write(ipc.Batch{Length: len(ids), Columns: []*array.Array{
array.Of(ids...), array.OfStrings(symbols...),
}})
if err != nil {
fmt.Println(err)
return
}
}
if err = w.Close(); err != nil {
fmt.Println(err)
return
}
// The last batch of the file, without reading the ones in front of it.
file := buf.Bytes()
r, err := ipc.NewFileReader(bytes.NewReader(file), int64(len(file)))
if err != nil {
fmt.Println(err)
return
}
b, err := r.Batch(r.NumBatches() - 1)
if err != nil {
fmt.Println(err)
return
}
for i := range b.Length {
fmt.Println(b.Columns[0].Value[int64](i), string(b.Columns[1].Bytes(i)))
}
}
Output: 3 S3
func NewFileWriter ¶ added in v0.0.4
NewFileWriter starts a file on w and writes the magic and the schema. The error is either a schema Arrow cannot name or whatever w said about the write.
func (*FileWriter) Close ¶ added in v0.0.4
func (w *FileWriter) Close() error
Close writes the end of stream marker and the footer after it. It does not close the writer underneath, which belongs to whoever opened it.
Closing twice is not an error, so a deferred Close after an explicit one reports what the explicit one did rather than writing a second footer.
func (*FileWriter) Schema ¶ added in v0.0.4
func (w *FileWriter) Schema() dtype.Schema
Schema is the schema every batch in the file belongs to.
func (*FileWriter) Write ¶ added in v0.0.4
func (w *FileWriter) Write(b Batch) error
Write appends one record batch and remembers where it went. The columns have to be the types the schema says they are and there have to be as many of them as it has fields.
A dictionary encoded column has to arrive as the column and not as its indices, since the values are what this has to write, and every batch of one column has to arrive with the same values.
type Imported ¶ added in v0.0.2
type Imported struct {
// Field is what the schema said: the name, the type, the nullability and
// the metadata.
Field dtype.Field
// Array is the values.
Array *array.Array
// contains filtered or unexported fields
}
Imported is a kuma array over memory that belongs to whoever produced it.
The values are not copied, so the array is only valid until Release is called, and Release must be called or the producer never learns that it can free anything. The usual shape is a call and a defer on the line after it.
Release sets Array to nil rather than leaving it pointing at memory that has been handed back, so that code holding it past the release fails on the spot instead of reading whatever moved in.
func ImportArray ¶ added in v0.0.2
ImportArray reads a C schema and a C array and returns a kuma array over the producer's own memory.
Both structs are taken over. The schema is released before this returns, since the type is copied into Go. The array is released by Imported.Release, which the caller has to get to, and is released here if the import fails.
Nothing is copied except the views of a column that arrived in one of the offset layouts. See Import for what that means and why.
Nested and dictionary encoded arrays are not supported yet, because the array package has nowhere to put them. Their schemas import fine, so a reader can still see what a file holds before it decides that it cannot read one column of it.
type Layout ¶
type Layout struct {
// Length is the number of values.
Length int
// Offset is how many values at the front of the buffers are not part of
// this array. It is in values rather than bytes, and for a Bool column
// that means bits.
Offset int
// NullCount is how many of the values are missing. Export fills it in.
// Import ignores it and counts for itself, since kuma keeps the count on
// the array and a producer is allowed to say that it does not know.
NullCount int
// Buffers are the raw bytes, borrowed rather than owned.
Buffers [][]byte
}
Layout is one array the way the C data interface describes it, with the pointers turned into slices. The type is not in here because the type travels separately, in a schema that usually covers a whole table.
Buffers is in the order the format string implies, which is the order the C struct has them in: the validity bitmap first, then the values. A validity buffer of no bytes means every value is present, which is how a null pointer arrives. A Null array has no buffers at all. The three text and byte layouts have a third buffer, and the view layout has one buffer per data block after the views, so a column of strings can be any number of buffers long.
The buffer of buffer sizes that the C ABI appends to the view layout is not here. It is there so that a consumer reading the C struct knows how long each data block is, and a Go slice already knows, so the cgo layer builds it on the way out and drops it on the way in.
func Export ¶
Export describes a kuma array in C data interface terms.
Nothing is copied. The slices in the result point into the array's own buffers, so they stay valid for as long as the array does and no longer, and a caller that hands them to another library has to keep the array alive until that library is finished with them.
The nested types are not here, because the array package has no nested arrays to export yet.
type Reader ¶ added in v0.0.3
type Reader struct {
// contains filtered or unexported fields
}
Reader reads the Arrow IPC stream format.
The schema is read when the reader is made, so a stream whose first message is not one is refused before any values are looked at. After that Next reads a batch at a time until the stream ends or something in it does not hold together.
The arrays in a batch point into the bytes that batch was read from, which is a buffer of its own rather than one the reader keeps reusing. Holding on to a batch after reading the next one is fine and costs the memory of the batch being held.
The dictionaries are the exception, and they are shared on purpose. The values of a dictionary encoded column arrive once and every batch after them points at that one array, which is the whole reason the encoding is worth having.
func NewReader ¶ added in v0.0.3
NewReader reads the schema off the front of a stream and returns a reader for the batches after it.
func (*Reader) All ¶ added in v0.0.3
All is the batches of the stream, for a range loop. It stops at the end of the stream or at the first error, which is then in Err, so a loop over this still has to check Err afterwards.
func (*Reader) Err ¶ added in v0.0.3
Err is the error that stopped the reader, or nil if the stream simply ended.
func (*Reader) Next ¶ added in v0.0.3
Next reads the next batch and reports whether there is one. It returns false at the end of the stream and at the first error, which Err then holds.
The dictionary batches in between are read here rather than handed out. They are not rows and a caller looping over the batches of a stream has nothing to do with one, so the loop stops at the record batches and the values are already on the columns by then.
type Writer ¶ added in v0.0.3
type Writer struct {
// contains filtered or unexported fields
}
Writer writes the Arrow IPC stream format.
The schema goes out when the writer is made and every batch has to match it. Close writes the end of stream marker, and a stream without one is a stream that was cut off, so it is worth deferring.
An error from the underlying writer sticks: the stream is broken at that point and nothing after it would be readable. An error about a batch does not, since the stream is untouched and a caller can fix the batch and try again.
The values of a dictionary encoded column go out in a message of their own, in front of the first batch that reads from them and again whenever a batch arrives holding different ones. A batch that shares its dictionary with the one before it, which is what reading a file gives, writes the values once.
Example ¶
package main
import (
"bytes"
"fmt"
"github.com/tamnd/kuma/array"
"github.com/tamnd/kuma/dtype"
"github.com/tamnd/kuma/ipc"
)
func main() {
s := dtype.Schema{Fields: []dtype.Field{
{Name: "id", Type: dtype.Int64},
{Name: "symbol", Type: dtype.String},
}}
var buf bytes.Buffer
w, err := ipc.NewWriter(&buf, s)
if err != nil {
fmt.Println(err)
return
}
for _, ids := range [][]int64{{1, 2}, {3}} {
symbols := make([]string, len(ids))
for i, id := range ids {
symbols[i] = fmt.Sprintf("S%d", id)
}
err = w.Write(ipc.Batch{Length: len(ids), Columns: []*array.Array{
array.Of(ids...), array.OfStrings(symbols...),
}})
if err != nil {
fmt.Println(err)
return
}
}
if err = w.Close(); err != nil {
fmt.Println(err)
return
}
r, err := ipc.NewReader(&buf)
if err != nil {
fmt.Println(err)
return
}
for b := range r.All() {
for i := range b.Length {
fmt.Println(b.Columns[0].Value[int64](i), string(b.Columns[1].Bytes(i)))
}
}
if err := r.Err(); err != nil {
fmt.Println(err)
}
}
Output: 1 S1 2 S2 3 S3
func NewWriter ¶ added in v0.0.3
NewWriter starts a stream over w and writes the schema, which is the first message of one. The error is either a schema Arrow cannot name or whatever w said about the write.
func (*Writer) Close ¶ added in v0.0.3
Close writes the end of stream marker. It does not close the writer underneath, which belongs to whoever opened it.
Closing twice is not an error, so a deferred Close after an explicit one reports what the explicit one did rather than something new.
func (*Writer) Write ¶ added in v0.0.3
Write appends one record batch. The columns have to be the types the schema says they are and there have to be as many of them as it has fields.
A dictionary encoded column has to arrive as the column and not as its indices, since the values are what this has to write and the indices on their own do not have them.
Source Files
Directories