QoL: Destructive schema sync after manual column dropping

[anuunchin]

Description

This PR adds a new pipeline function that syncs the dlt schema with the destination (not vice versa) by removing a column from the schema if that column has been manually deleted in the destination.

The motivation behind this is that rather than offering a cli command that drops columns - where we need to have separate drop_columns functions due to dialect differences and thus adds additional maintenance overhead - we delegate the dropping part to the user and instead allow them to sync the dlt schema in those scenarios.

Related PRs:

#2754

Further:

This should be extended to table drop syncs as well.

Note:

This is essentially solving the problem when the user manually drops things in the destination and the dlt pipeline breaks.

[netlify]

✅ Deploy Preview for dlt-hub-docs canceled.

Name	Link
🔨 Latest commit	d2ad80e
🔍 Latest deploy log	https://app.netlify.com/projects/dlt-hub-docs/deploys/68bac554550c830008d938d7

[anuunchin]

I just realized the incremental setting is also saved outside of the schema, so when the user

• manually drops a column from destination with an incremental setting
• syncs the schema destructively

They will need to also unset the incremental with apply_hints(incremental=dlt.sources.incremental.EMPTY)

[anuunchin]

We just get columns from schema because it's a dummy

[anuunchin]

It's impossible to get the actual columns from filesystem files without table format unless we read entire files. Also, it's unlikely that the user manually deletes specific columns from filesystem files I think 👀 . Therefore, we should raise NotImplemented instead of doing basically nothing.

[rudolfix]

This is very good idea but we need to approach it in more systematic way:

1. (almost) all of our destinations have

def get_storage_table(self, table_name: str) -> Tuple[str, TTableSchemaColumns]:

and/or

def get_storage_tables(
        self, table_names: Iterable[str]
    ) -> Iterable[Tuple[str, TTableSchemaColumns]]: 

implemented. this will reflect the storage to get table schema out of it. you can use it to compare with the pipeline schema.

2. let's formalize it: add mixin class like WithTableReflection in the same manner WithStateSync is done. get_storage_tables is the more general method so you can add only this to the mixin
3. Now add this mixing to all JobClientBase implementations for which you want to support our new sync schema.

When the above is done we are able to actually compute schema diff.

Top level interface:

1. we have sync_schema that will do a regular schema migration (add missing columns and tables in the destination`
2. we need another method which is the reverse: it will delete columns and tables in the schema that are not present in the destination and then do the schema sync above
3. the method above should have a dry run mode - where we do not change the pipeline schema and we do not sync it
4. it should make sure if destination_client() implements WithTableReflection before continuing
5. it should allow to select tables to be affected

when this is done we can think about extending cli ie dlt pipeline <name> schema command

[rudolfix]

this should be merged into get_storage_tables which is already present in the filesystem. it will be really helpful.

note: both are operating on Arrow schemas so look how get_storage_table works in lance... you do not need to implement TypeMapper just look at lancedb implementation

[rudolfix]

this is overall a good idea to add to JobClient but we should be more selective. Look at my top level review. We need 1. mixin class 2. it needs to be optional

[rudolfix]

some changes needed

[anuunchin]

Just realized that for parquet files we can also just use pyarrow and read actual metadata 👀 , but I still don't think people drop columns in parquet files...