Moving an array#

The consumer canmove theArrowDeviceArray structure by bitwise copyingor shallow member-wise copying. Then it MUST mark the source structure releasedby setting therelease member of the embeddedArrowArray structure toNULL, butwithout calling that release callback. This ensures that onlyone live copy of the struct is active at any given time and that lifetime iscorrectly communicated to the producer.

As usual, the release callback will be called on the destination structurewhen it is not needed anymore.

The ArrowDeviceArrayStream structure#

TheArrowDeviceArrayStream provides a device type that can access theresulting data along with the required callbacks to interact with astreaming source of Arrow arrays. It has the following fields:

ArrowDeviceTypedevice_type#

Mandatory. The device type that this stream produces data on. AllArrowDeviceArray s that are produced by this stream should have thesame device type as is set here. This is a convenience for the consumerto not have to check every array that is retrieved and instead allowshigher-level coding constructs for streams.

int(*ArrowDeviceArrayStream.get_schema)(structArrowDeviceArrayStream*,structArrowSchema*out)#

Mandatory. This callback allows the consumer to query the schema ofthe chunks of data in the stream. The schema is the same for all datachunks.

This callback must NOT be called on a releasedArrowDeviceArrayStream.

Return value: 0 on success, a non-zeroerror code otherwise.

int(*ArrowDeviceArrayStream.get_next)(structArrowDeviceArrayStream*,structArrowDeviceArray*out)#

Mandatory. This callback allows the consumer to get the next chunk ofdata in the stream.

This callback must NOT be called on a releasedArrowDeviceArrayStream.

The next chunk of data MUST be accessible from a device type matching theArrowDeviceArrayStream.device_type.

Return value: 0 on success, a non-zeroerror code otherwise.

On success, the consumer must check whether theArrowDeviceArray’sembeddedArrowArray is markedreleased.If the embeddedArrowDeviceArray.array is released, then the end of thestream has been reached. Otherwise, theArrowDeviceArray contains avalid data chunk.

constchar*(*ArrowDeviceArrayStream.get_last_error)(structArrowDeviceArrayStream*)#

Mandatory. This callback allows the consumer to get a textual descriptionof the last error.

This callback must ONLY be called if the last operation on theArrowDeviceArrayStream returned an error. It must NOT be called on areleasedArrowDeviceArrayStream.

Return value: a pointer to a NULL-terminated character string(UTF8-encoded). NULL can also be returned if no detailed description isavailable.

The returned pointer is only guaranteed to be valid until the next callof one of the stream’s callbacks. The character string it points to shouldbe copied to consumer-managed storage if it is intended to survive longer.

void(*ArrowDeviceArrayStream.release)(structArrowDeviceArrayStream*)#

Mandatory. A pointer to a producer-provided release callback.

void*ArrowDeviceArrayStream.private_data#

Optional. An opaque pointer to producer-provided private data.

Consumers MUST NOT process this member. Lifetime of this member ishandled by the producer, and especially by the release callback.

Result lifetimes#

The data returned by theget_schema andget_next callbacks must bereleased independently. Their lifetimes are not tied to that ofArrowDeviceArrayStream.

Stream lifetime#

Lifetime of the C stream is managed using a release callback with similarusage as inC data interface.

Thread safety#

The stream source is not assumed to be thread-safe. Consumers wanting tocallget_next from several threads should ensure those calls areserialized.

The ArrowAsyncDeviceStreamHandler structure#

The structure has the following fields:

int(*ArrowAsyncDeviceStreamHandler.on_schema)(structArrowAsyncDeviceStreamHandler*,structArrowSchema*)#

Mandatory. Handler for receiving the schema of the stream. All incoming records shouldmatch the provided schema. If successful, the function should return 0, otherwiseit should return anerrno-compatible error code.

If there is any extra contextual information that the producer wants to provide, it can setArrowAsyncProducer.additional_metadata to a non-NULL value. This is encoded in thesame format asArrowSchema.metadata. The lifetime of this metadata, if notNULL,should be tied to the lifetime of theArrowAsyncProducer object.

Unless theon_error handler is called, this will always get called exactly once and will bethe first method called on this object. As such the producerMUST populate theArrowAsyncProducermember before calling this function to allow the consumer to apply back-pressure and control the flow of data.The producer maintains ownership of theArrowAsyncProducer and must clean it upaftercalling the release callback on theArrowAsyncDeviceStreamHandler.

A producer that receives a non-zero result here must not subsequently call anything other thanthe release callback on this object.

int(*ArrowAsyncDeviceStreamHandler.on_next_task)(structArrowAsyncDeviceStreamHandler*,structArrowAsyncTask*,constchar*)#

Mandatory. Handler to be called when a new record is available for processing. Theschema for each record should be the same as the schema thaton_schema was called with.If successfully handled, the function should return 0, otherwise it should return anerrno-compatible error code.

Rather than passing the record itself it receives anArrowAsyncTask instead to facilitatebetter consumer-focused thread control as far as receiving the data. A call to this functionsimply indicates that data is available via the provided task.

The producer signals the end of the stream by passingNULL for theArrowAsyncTaskpointer instead of a valid address. This task object is only valid during the lifetime ofthis function call. If the consumer wants to use the task beyond the scope of this method, itmust copy or move its contents to a new ArrowAsyncTask object.

Theconstchar* parameter exists for producers to provide any extra contextual informationthey want. This is encoded in the same format asArrowSchema.metadata. If notNULL,the lifetime is only the scope of the call to this function. A consumer who wants to maintainthe additional metadata beyond the lifetime of this callMUST copy the value themselves.

A producerMUST NOT call this concurrently from multiple threads.

TheArrowAsyncProducer.request callback must be called to start receiving calls to thishandler.

void(*ArrowAsyncDeviceStreamHandler.on_error)(structArrowAsyncDeviceStreamHandler,int,constchar*,constchar*)#

Mandatory. Handler to be called when an error is encountered by the producer. After callingthis, therelease callback will be called as the last call on this struct. The parametersare anerrno-compatible error code and an optional error message and metadata.

If the message and metadata are notNULL, their lifetime is only valid during the scopeof this call. A consumer who wants to maintain these values past the return of this functionMUST copy the values themselves.

If the metadata parameter is notNULL, to provide key-value error metadata, then it shouldbe encoded identically to the way that metadata is encoded inArrowSchema.metadata.

It is valid for this to be called by a producer with or without a preceding call toArrowAsyncProducer.request. This callbackMUST NOT call any methods of anArrowAsyncProducer object.

void(*ArrowAsyncDeviceStreamHandler.release)(structArrowAsyncDeviceStreamHandler*)#

Mandatory. A pointer to a consumer-provided release callback for the handler.

It is valid for this to be called by a producer with or without a preceding call toArrowAsyncProducer.request. This must not call any methods of anArrowAsyncProducerobject.

structArrowAsyncProducerArrowAsyncDeviceStreamHandler.producer#

Mandatory. The producer object that the consumer will use to request additional data or cancel.

This objectMUST be populated by the producer before calling theArrowAsyncDeviceStreamHandler.on_schemacallback. The producer maintains ownership of this object and must clean it upafter callingthe release callback on theArrowAsyncDeviceStreamHandler.

The consumerCANNOT assume that this is valid until theon_schema callback is called.

void*ArrowAsyncDeviceStreamHandler.private_data#

Optional. An opaque pointer to consumer-provided private data.

ProducersMUST NOT process this member. Lifetime of this member is handled bythe consumer, and especially by the release callback.

The ArrowAsyncTask structure#

The purpose of using a Task object rather than passing the array directly to theon_nextcallback is to allow for more complex and efficient thread handling. Utilizing a Taskobject allows for a producer to separate the “decoding” logic from the I/O, enabling aconsumer to avoid transferring data between CPU cores (e.g. from one L1/L2 cache to another).

This producer-provided structure has the following fields:

int(*ArrowArrayTask.extract_data)(structArrowArrayTask*,structArrowDeviceArray*)#

Mandatory. A callback to populate the providedArrowDeviceArray with the available data.The order ofArrowAsyncTasks provided by the producer enables a consumer to know the order ofthe data to process. If the consumer does not care about the data that is owned by this task,it must still callextract_data so that the producer can perform any required cleanup.NULLshould be passed as the device array pointer to indicate that the consumer doesn’t want theactual data, letting the task perform necessary cleanup.

If a non-zero value is returned from this, it should be followed only by the producer callingtheon_error callback of theArrowAsyncDeviceStreamHandler. Because calling this methodis likely to be separate from the current control flow, returning a non-zero value to signalan error occurring allows the current thread to decide handle the case accordingly, while stillallowing all error logging and handling to be centralized in theArrowAsyncDeviceStreamHandler.on_error callback.

Rather than having a separate release callback, any required cleanup should be performed as partof the invocation of this callback. Ownership of the Array is given to the pointer passed in asa parameter, and this array must be released separately.

It is only valid to call this method exactly once.

void*ArrowArrayTask.private_data#

Optional. An opaque pointer to producer-provided private data.

ConsumersMUST NOT process this member. Lifetime of this member is handled bythe producer who created this object, and should be cleaned up if necessary duringthe call toArrowArrayTask.extract_data.

The ArrowAsyncProducer structure#

This producer-provided and managed object has the following fields:

ArrowDeviceTypeArrowAsyncProducer.device_type#

Mandatory. The device type that this producer will provide data on. AllArrowDeviceArray structs that are produced by this producer should have thesame device type as is set here.

void(*ArrowAsyncProducer.request)(structArrowAsyncProducer*,uint64_t)#

Mandatory. This function must be called by a consumer to start receiving calls toArrowAsyncDeviceStreamHandler.on_next_task. ItMUST be valid to callthis synchronously from withinArrowAsyncDeviceStreamHandler.on_next_taskorArrowAsyncDeviceStreamHandler.on_schema. As a result, this functionMUST NOT synchronously callon_next_task oron_error to avoid recursiveand reentrant callbacks.

Aftercancel is called, additional calls to this function must be a NOP, but allowed.

While not cancelled, calling this function registers the given number of additionalarrays/batches to be produced by the producer. A producer should only callthe appropriateon_next_task callback up to a maximum of the total sum of calls tothis method before propagating back-pressure / waiting.

Any error encountered by calling request must be propagated by calling theon_errorcallback of theArrowAsyncDeviceStreamHandler.

It is invalid to call this function with a value ofn that is<=0. Producers shoulderror (e.g. callon_error) if receiving such a value forn.

void(*ArrowAsyncProducer.cancel)(structArrowAsyncProducer*)#

Mandatory. This function signals to the producer that it musteventually stop callingon_next_task. Calls tocancel must be idempotent and thread-safe. After callingit once, subsequent callsMUST be a NOP. ThisMUST NOT call any consumer-side handlersother thanon_error.

It is not required that callingcancel affect the producerimmediately, only that itmust eventually stop callingon_next_task and then subsequently callreleaseon the async handler object. As such, a consumerMUST be prepared to receive one or morecalls toon_next_task oron_error even after callingcancel if there are stillrequested arrays pending.

Successful cancellingMUST NOT result in a producer callingArrowAsyncDeviceStreamHandler.on_error, instead it should finish out any remainingtasks (callingon_next_task accordingly) and eventually just callrelease.

Any error encountered during handling a call to cancel must be reported via theon_errorcallback on the async stream handler.

constchar*ArrowAsyncProducer.additional_metadata#

Optional. An additional metadata string to provide any extra context to the consumer. ThisMUSTeither beNULL or a valid string that is encoded in the same way asArrowSchema.metadata.As an example, a producer could utilize this metadata to provide the total number of rows and/or batchesin the stream if known.

If notNULL itMUST be valid for at least the lifetime of this object.

void*ArrowAsyncProducer.private_data#

Optional. An opaque pointer to producer-provided specific data.

ConsumersMUST NOT process this member, the lifetime is owned by the producerthat constructed this object.

Error Handling#

Unlike the regular C Stream interface, the Async interface allows for errors to flow inboth directions. As a result, error handling can be slightly more complex. Thus this specdesignates the following rules:

• If the producer encounters an error during processing, it should call theon_errorcallback, and then callrelease after it returns.

• Ifon_schema oron_next_task returns a non-zero integer value, the producershould notcall theon_error callback, but instead should eventually callrelease at some pointbefore or after any logging or processing of the error code.

Result lifetimes#

TheArrowSchema passed to theon_schema callback must be released independently,with the object itself needing to be moved to a consumer ownedArrowSchema object. TheArrowSchema* passed as a parameter to the callbackMUST NOT be stored and kept.

TheArrowAsyncTask object provided toon_next_task is owned by the producer andwill be cleaned up during the invocation of callingextract_data on it. If the consumerdoesn’t care about the data, it should passNULL instead of a validArrowDeviceArray*.

Theconstchar* errormessage andmetadata which are passed toon_errorare only valid within the scope of theon_error function itself. They must be copiedif it is necessary for them to exist after it returns.

Stream Handler Lifetime#

Lifetime of the async stream handler is managed using a release callback with similarusage as inC data interface.

ArrowAsyncProducer Lifetime#

The lifetime of theArrowAsyncProducer is owned by the producer itself and shouldbe managed by it. ItMUST be populated before calling any methods other thanreleaseandMUST remain valid at least until just before callingrelease on the stream handler object.

Thread safety#

All handler functions on theArrowAsyncDeviceStreamHandler should only be called in aserialized manner, but are not guaranteed to be called from the same thread every time. Aproducer should wait for handler callbacks to return before calling the next handler callback,and before calling therelease callback.

Back-pressure is managed by the consumer making calls toArrowAsyncProducer.requestto indicate how many arrays it is ready to receive.

TheArrowAsyncDeviceStreamHandler object should be able to handle callbacks as soon asit is passed to the producer, any initialization should be performed before it is provided.