Engine Contract

Domain: Transformations
Depends on: Data Model
RFCs: RFC-015

​

Abstract

This spec defines the contract between a and an - the sandboxed execution environment of engines, the communication interface and the operations engines must support, and how engine versions are pinned and deprecated to preserve reproducibility of transformations.

​

Execution Model

executes in a very restricted environment (a “sandbox”) that prevents it from accessing any external resources other than those explicitly provided by the . Isolating the from making network calls and accessing any random files ensures both that the being processed cannot be leaked, and that non-deterministic behavior is harder to introduce accidentally. Our isolation technology of choice is OCI containers - a lightweight mechanism that provides good performance and strong isolation guarantees. We rely on OCI-compatible images as the distribution mechanism for the engines.

WIP: Add exact podman and docker commands to reproduce the sandboxed environment

​

Communication Interface

This section describes the RPC mechanism used in communications between the and the . The requirements we put towards the RPC mechanism are:

• Use common network protocols
• Wide support in many programming languages
• Support streaming responses (e.g. for streaming back the operation status and logs)

Our RPC technology of choice is gRPC because:

• It’s cross-platform, mature, and available in many languages
• Natively supports FlatBuffers which is our metadata format of choice, letting us reuse object schemas
• Supports streaming
• For our narrow use case, its convenience is more important than interoperability, as we don’t expect to have many server and client implementations

implementations must expose the gRPC API on port 5544. Request and response messages are defined by the message schemas and serialized using FlatBuffers. Every operation responds with a stream of messages, where zero or more Progress messages are followed by exactly one terminal message:

• Success - operation completed successfully
• InvalidQuery - the query did not pass validation, with a message explaining the error
• InternalError - an unexpected error occurred, with a message and an optional backtrace with details

Note: When starting an container, the will consider it fully initialized when it can establish a TCP connection to the port 5544.

​

Data Exchange

The exchange of raw data happens out-of-band of the gRPC API. Currently, data and checkpoints are exchanged via files and object stores. The makes input data part files and the previous available to the and passes their paths in the request, along with the paths where the must write the output data and the new . In future, input and output may be also exchanged using memory-mapped files (mmap) containing data in the common in-memory format. This avoids any intermediate IO and minimizes the copying of data for the best performance.

​

Operations

Engine implementation should support the following operations:

• Execute query - Performs the next iteration of the transformation.
• Raw query - Executes a query over raw data, e.g. during preprocessing of data in ingest.
• Validate query - Validates the user-specified query for basic syntax and schema correctness.
• Migrate query - Updates the transformation state from one query to another.
• Derive Provenance - Explains the origin of some data produced in the past.

WIP: Validate query, Migrate query, and Derive Provenance operations do not yet have defined message schemas.

​

Execute Query

This operation is used by the to perform the next iteration of the transformation. The is designed to isolate the from the complexity of managing the and make processing look as close as possible to a regular stream processing. Inputs (TransformRequest):

• ID and alias of the output
• to assign to new records
• Offset to assign to the first new record
• Vocabulary of the output - mapping of system columns onto its schema
• Transformation
• Inputs - for every input :
  • ID and alias of the dataset
  • Alias under which the input is referred to in the
  • Vocabulary of the input dataset
  • Offset interval of the input that goes into this transaction
  • Schema and data files of the input
  • Explicit that should be injected into the stream to separate micro-batches for reproducibility
• Previous
• Locations where new data and the new should be written

Outputs (TransformResponse):

• Offset interval of the output , if any data was produced
• Output , if it was advanced
• New
• Operation progress and status

This operation must be idempotent. Input and output data are changelog streams that can contain (see Processing Model: Retractions and Corrections). The must:

• Consume input records in the order of their , as the order of retractions and corrections relative to the events they apply to is significant
• Take the operation type of input records into account and propagate retractions and corrections into the output
• Produce output that is a valid changelog stream (see Data Model) - in particular, it must not reorder records in a way that separates a correction pair or places a retraction before the event it retracts, and must write both events of a correction pair into the same output

Engines that process data in batches and parallelize work must preserve the order of records (e.g. by coalescing inputs into one partition). Such engines that only support simple record-by-record operations (like map and filter) must propagate the operation type column as-is, so that operations apply identically to appends, retractions, and corrections and produce a valid changelog stream. When an operation is committed by the , the will not see the same input data again. If due to the nature of the query (e.g. windowing or watermarking configuration) cannot fully process and discard the input data - it should use the to buffer it.

​

Raw Query

This operation is used by the to execute a over raw data that is not yet part of any dataset, e.g. during the preprocessing step of . Inputs (RawQueryRequest):

• Input data files to perform the query over - all files must have identical schema
• Transformation
• Location where the query result should be written

Outputs (RawQueryResponse):

• Number of records produced by the query
• Operation progress and status

​

Validate Query

This operation may be used by the when creating a new or when changing the of an existing one to validate the user-specified query for basic syntax and schema correctness. Inputs:

• of the input
• (optional) Previous in case of an existing modification

Outputs:

• of the result
• (alternatively) Validation errors

​

Migrate Query

This operation is used by the when data processing hits the point where one transformation is being replaced by another one. It gives the all information needed to handle the transition as gracefully as possible, e.g. by reconciling the existing with the new , or, at the very least, finalizing the processing of the old and clearing the before the new one starts to execute. Inputs:

• Transaction ID
• Previous
• Next
• Previous

Outputs:

• New
• (optional) that has been produced when finalizing the old

​

Derive Provenance

This operation is used to trace back a set of events in the output to input events that contributed to their values or their existence (see ).

WIP: The design of this operation is in progress.

​

Engine Versioning

As described in Components of Trust, to guarantee the reproducibility and verifiability of the results a transformation must be associated with an exact version of an that is used to perform it. We want to exclude any possibility that the code changes in the will break this guarantee. Whenever the uses an to execute a query it must specify the full digest of the OCI container image in the resulting . maintainers are therefore responsible for treating the images as immutable and ensure old versions are never lost. See also:

• Engine Deprecation

​

Engine Deprecation

Pinning the exact version used by every transformation guarantees the reproducibility and verifiability of the results (see Components of Trust). In the long run, however, this creates a problem where in order to verify a dataset the will have to use a very large number of different versions of an that might’ve accumulated over the years. This slows down the verification procedure and can result in a significant disk space required for storing these images. To mitigate this problem the offers the engine deprecation operation that works as follows:

1. The reconstructs the dataset from scratch by repeating all transformation in the .
2. Instead of the version specified in the block, it uses the latest (or provided) version.
3. If the results produced in every step match the recorded ones - we can safely consider that the newer version of the is fully compatible with the older one.
4. A special record is created to record this compatibility, therefore, allowing all users to use the latest version of the instead of downloading all the versions seen previously.

​

Message Schemas

Messages exchanged between the and the are defined by the following schemas: