Data movement

[Jesse-Bakker]

Transactional consistency

To ensure transactional consistency, transactions need to be ordered
and, if supported by the sink, committed together. One option for this
is for the pipeline to act on

struct TableOperation {
    table: String,
    op: Operation
}
struct Transaction {
    ops: Vec<TableOperation>
}

Transformations can happen in parallel, as long as the ordering of
transformations is restored before pushing the data to the sink,
something like:

for tid in start.. {
    for input_channel in inputs {
      if input_channel.peek().tid == next_tid {
        commit(input_channel.recv());
        break;
      }
    }
}

Denormalization

There are two main shapes that denormalization may take. Lets take the
following example schema:

CREATE TABLE user (
  id INTEGER PRIMARY KEY,
  name VARCHAR NOT NULL
)

CREATE TABLE address (
  id INTEGER PRIMARY KEY,
  user_id INTEGER NOT NULL REFERENCES (user.id)
  address VARCHAR NOT NULL
)

To keep the Dozer pipeline as stateless as possible, denormalization
state is pushed to the sink database, and changes to the denormalized
sink table are applied in a trigger-like fashion.

There are two main ways this might be denormalized, depending on the
purpose and sink type. (these are terms I came up with, so don't try to google them)

Aggregation denormalization

Example json record:

{
    id: 1,
    name: "Santa Claus",
    addresses: [
        {
            id: 1,
            address: "North Pole"
        }
    ]
}

This is most common for sinks like Mongodb, and data might be replicated
to this sink for an application-specific cache. An insert into the
source address table would cause an update to be performed in mongodb
like:

db.user.updateOne(
    {id: { $eq: 1 }},
    {$push: {
        addresses: {
              id: 1,
              address: "North Pole"
        }
    }}
)

Expressing this denormalization in SQL would require implementing
something like json_agg.

Analytical denormalization

Example denormalized schema:

CREATE TABLE user_addresses (
  id INTEGER PRIMARY KEY,
  user_id INTEGER NOT NULL,
  name VARCHAR NOT NULL,
  address VARCHAR NOT NULL
)

An insert into the source address table would cause an update to be
performed in postgres like:

INSERT INTO user_addresses(id, user_id, name, address) (
    SELECT 1, user_id, name, "North Pole" 
    FROM user_addresses WHERE user_id = 1 
    LIMIT 1
);