Connector Success Criteria

[chloeminkyung]

Let's start with following table and confirm on the connector success criteria - following could be trait level checklist.

Category	Criteria
Connection	Can connect
	Test connection
	Test source schemas
	Validate schemas
Sources	Can use multiple sources
	Can add/remove sources from connection
	Source types are mapped to dozer types
Snapshot sync	Can sync all used sources
	Can continue sync
CDC handling	Can get data changes
	Can continue replication

[karolisg]

I have documented what we expect in connector and what is expected in every trait function implementation.

Trait

Every connector to external database needs to implement connector trait (/dozer-ingestion/src/connectors/mod.rs)

fn validate(&self, tables: Option<Vec<TableInfo>>) -> Result<(), ConnectorError>
This function is supposed to validate connector configuration and connection to database. It also validates tables, columns existence and user details/permissions.

fn validate_schemas(&self, tables: &[TableInfo]) -> Result<ValidationResults, ConnectorError>;
This function's purpose is to validate schemas, which are used as sources for data ingestion. It should return error when column type is not supported by existing dozer types.

fn get_schemas(&self,table_names: Option<Vec<TableInfo>>,) -> Result<Vec<SchemaWithChangesType>, ConnectorError>;
This function is used for getting mapped external database schema to dozer schema. Also, as result of schema definition, developer also should provide ReplicationChangesTrackingType (described below)

fn initialize(&mut self,ingestor: Arc<RwLock<Ingestor>>,tables: Option<Vec<TableInfo>>,) -> Result<(), ConnectorError>;
In this method, developer passes ingestor and tables information, which will later be used during ingestion process.

fn start(&self, from_seq: Option<(u64, u64)>) -> Result<(), ConnectorError>;
This function is responsible for all ingestion processes. It has single parameter, which is used to resume ingestion it was stopped. That tuple contains two values - (u64, u64), first value is lsn of transaction and second value is seq no of last consumer record in transaction. It is used to allow connector to continue from middle of transaction.

Replication changes tracking types

Type
FullChanges	Connector gets old record on delete/update operations
OnlyPK	Connector only gets PK of old record on delete/update operations
Nothing	Connector cannot get any info about old records. In other words, the table is append-only

Tables and columns selection

Selection of which tables and columns will be used in ingestion is defined in sources configuration.
Structure of that configuration part is this

    name: users
    connection: !Ref pg_data_connection
    table_name: userdata
    columns:
      - gender
      - email

From this configuration, table_name is used as table name in external database and name is used in dozer transformations.
Another property connection is reference to connection, which is already defined in connections configuration.
columns property is used to restrict list of used columns from external database. If this value is empty array, ingestion will fetch all columns of that table.

Schemas

Every external schema should be mapped using dozer types. The latest types definitions can be found at https://getdozer.io/docs/reference/data_types . If type is not supported, connector should return error, during schema validation step.
During ingestion data should be cast to same type as it was defined in schema.

Replay support

During pipeline start, start function receives tuple from_seq: (u64, u64). That tuple is used to tell last message lsn and seq no. One lsn is shared for all operations inside single transaction, while second parameter is used for determining how many messages were successfully processed from transaction.

Unit tests

Unit tests are only possible in places where connection to external database is not required. It is important to have unit tests for schema mapping and data casting to dozer types. More complex tests should be implemented using E2E tests.

E2E tests

More complex tests require to have connection real database. Such tests cases expect to have several things:

• Database infrastructure (preferably created in docker container(s))
• Connection configuration (with placeholders)
• It should be possible to run test cases without doing any manual modifications in database.

[mediuminvader]

@karolisg let's update the main text of the discussion with these new points, so we can iterate around a draft and come up with a final version.

[chubei]

• Maybe we should describe the arguments as well? For example, in validate, I can't see what the connector should do when tables is None.
• I still suggest we remove initialize and pass ingestor and tables to start. Actually I encountered this again when revising the graceful shutdown process. I'll post a PR or discussion to elaborate.
• In Replay support, start_seq is actually an Option, and it should be renamed to something like last_checkpoint because the connector shouldn't start from it. For example, when passed Some((1, 1)), the first record that the connector sends should be the one with sequence number (1, 2).