How will the rewrite impact the compatibility with Jina?

[JoanFM]

How will the rewrite affect the capacity of DocumentArrays to be streamed through Jina Flows and Executors?

[samsja]

Hey @JoanFM indeed this new version of DocArray will have an impact on Jina and the way we work with Executor. There will be breaking change but we will avoid them as much as possible and we believe this this will only enforce a better usage of Executor in general if we have a input schema for Executor.

Herei the current vision for this work

Jina side : Executor interoperability

The API above is much more flexible than the current Document implementation. This buys us better multimodal support as well as more natural vector DB integration. On the other hand, the Executor have less structure to rely on.
We intend to tackle this limitation in the following way:

• Every Executor expects a schema that it will work on. This could be self-defined, or an imported ‘default’ Document:

class MyExecSchema(Document):
    text: str
    embedding: Embedding


class MyExec(Executor):
    @requests
    def foo(docs: DocumentArray[MyExecSchema], *args, **kwargs):
        ...

• On the client side, the user can (but does not have to!) define a translation (schema_map, name not final) from their schema to the expected schema:

class ClientDoc(Document):
    text: str
    first_embedding: Embedding
    second_embedding: Embedding


doc = ClientDoc(...)
# map `text` to `text` and `first_embedding` to `embedding`
client.post(doc, schema_map={'MyExec': 'text:text,first_embedding:embedding'})
# the case of nested schema can be handle with dunder notation

• The worker runtime performs the schema translation, by simply renaming fields in the received document. That way the Executor only gets to see what it wants. We plan on doing automatic translation as well to avoid verbosity when it is not needed (see below)
• Defaults: If the client schema already matches the Executor schema, no translation is necessary and no schema map needs to be passed.
  If the schemas do not match and no map is provided, the runtime can do a best effort translation based on the types. We clearly document that mechanism and avoid any surprises.
• Backward compatibility: If an Executor fails to provide an expected schema, we assume the current (legacy) schema that already exists for every Document
• Executors can receive any schema that is superset of his defined schema. Clients send to the Flow, documents with a schema that is valid to all Executors.
  In the worker runtime, deserializing docs from a python object to protobuf should rely on the initial protobuf message, rather than creating a new one

Automatic translation details:

Lets say my input data follow this schema

class ImageTextDocument(Document):
    text: Text
    image: Image
    embedding: Embedding

and that my Executor follow this one

class MyPhoto(Document):
    vector: Embedding
    photo: Image
    description: Text


class PhotoEmbeddingExecutor(Executor):
    @requests
    def encode(self, docs: DocumentArray[MyPhoto], **kwargs):
        for doc_ in docs:
            doc.embedding = self.image_model(doc.photo)

They define actually the same underlying schema but with different field name. So the way would be to do

client.post(doc, schema_map={'MyExec': 'image:photo,embedding:vector,text:description'})

But this is too verbose for smth just translating the same schema. We will do that automatically., How ? We look at the field type and do a one by one group by.

What if the matching is not exact ? i.e what if we have the two following schema ?

class ClientDoc(Document):
    text: Text
    embedding1: Embedding
    embedding2: Embedding


class ExecutorDoc(Document):
    text: Text
    embedding: Embedding

If we have collision on a field (here two embeddings) we will take the first field that correspond (in this case embedding1). This is a deterministic algorithm because fields are ordered in pydantic

[JoanFM]

Jina has added compatibility with the newer docarray versions