fix: create schema and table on add_sink

[kgpayne]

Closes #1027

• bug: Tables not created if no records arrive with SQLSink #1027

---

📚 Documentation preview 📚: https://meltano-sdk--1036.org.readthedocs.build/en/1036/

[codecov]

Codecov Report

Merging #1036 (7139d08) into main (fab2cc3) will increase coverage by 0.26%.
The diff coverage is 90.47%.

@@            Coverage Diff             @@
##             main    #1036      +/-   ##
==========================================
+ Coverage   83.13%   83.39%   +0.26%     
==========================================
  Files          39       39              
  Lines        3747     3758      +11     
  Branches      628      628              
==========================================
+ Hits         3115     3134      +19     
+ Misses        470      464       -6     
+ Partials      162      160       -2     

Impacted Files	Coverage Δ
singer_sdk/streams/sql.py	80.27% <80.95%> (+1.04%)	⬆️
singer_sdk/sinks/core.py	80.26% <100.00%> (+0.26%)	⬆️
singer_sdk/sinks/sql.py	87.01% <100.00%> (+8.53%)	⬆️
singer_sdk/target_base.py	85.44% <100.00%> (+0.06%)	⬆️

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

[kgpayne]

I see codecov is failing. Wasn't exactly sure where best to add tests, but will look again in the morning 👍

[aaronsteers]

@kgpayne - Looks like we've all had a first pass at this, and I see you are making a couple iterations, so I'll mark as draft. Feel free to move back to ready status once you are ready for final review. 👍

[BuzzCutNorman]

@kgpayne I was wondering how should a developer set a target's default schema once this code is put in place? Say for argument's sake I created a target-postgres and set the default schema on connect with some code like this.

        @event.listens_for(engine, "connect", insert=True)
        def set_search_path(dbapi_connection, connection_record):
            existing_autocommit = dbapi_connection.autocommit
            dbapi_connection.autocommit = True
            cursor = dbapi_connection.cursor()
            cursor.execute(f"SET SESSION search_path='{self.config.get('schema') or 'public'}'")
            cursor.close()
            dbapi_connection.autocommit = existing_autocommit

I believe with this PR in place any blank table would be created in the tap's schema not in the target's default schema from the meltano.yml

[aaronsteers]

@BuzzCutNorman - Thanks for raising.

@kgpayne - There are probably several discussion points to unwind here, but what do you think of using a setting name of load_schema (aligned with internal Meltano naming) or default_target_schema (Pipelinewise convention)?

New discussion here where we can dive deep:

• How should the SDK deal with settings for `load_schema` and `schema_mappings` for SQL-based targets #1084 (comment)

If we can remove the reference to config.get('schema') and/or config["schema"] from this PR and apply a calculated schema when available (and just don't set a load schema otherwise?), that safely would push the new scope to a subsequent PR.

What do you think?

[kgpayne]

@aaronsteers @BuzzCutNorman

If we can remove the reference to config.get('schema') and/or config["schema"] from this PR...

We don't currently have any references to config.get('schema') in this PR or on main - that call is in BuzzCutNormans' example only. This PR already implements the behaviour you describe:

apply a calculated schema when available (and just don't set a load schema otherwise?)

I.e. the incoming schema_name property will try to derive a schema name (by splitting the stream name on -) and return None if the name is not splittable 👍

I posted feedback on the discussion too, but TLDR; is that we don't currently allow the specification of a schema/search path during the construction of the SQLAlchemy engine, but could easily do so with a connect_args setting (here or in another PR). This would then provide a default schema in cases where the schema_name property returns None, without the need for a custom snippet (like the one BuzzCutNorman suggested) or the overriding of create_sqlalchemy_engine on the SQLConnecter class 🙂 In future we could additionally support a load_schema to 'force' the schema to a user value irrespective of connection or derived schema names.

WDYT?

[aaronsteers]

We don't currently have any references to config.get('schema') in this PR or on main - that call is in BuzzCutNormans' example only.

This was my misreading then. Thanks very much for clarifying.

[kgpayne]

We don't currently have any references to config.get('schema') in this PR or on main - that call is in BuzzCutNormans' example only.

This was my misreading then. Thanks very much for clarifying.

@aaronsteers I had to check, as I thought we used it somewhere too 🙂 Whilst its common in SQL target implementations (as you point out in the discussion) it isn't (yet) in the SDK 👍

[aaronsteers]

@kgpayne - Looks good from my side.

@edgarrmondragon - Any other changes you'd want to see in this iteration?

[edgarrmondragon]

@kgpayne Looks good to me!

[aaronsteers]

@BuzzCutNorman re:

@kgpayne I was wondering how should a developer set a target's default schema once this code is put in place?

I think we should follow this PR up with this one:

• For SQL-based targets, add built-in handling for schema_mapping #1086

That would give a standard way for users to control the target load schemas. And presumably those behaviors would be built into the SQLStream.schema_name property (or a similar built-in featureset), such that the schema_name returned there would honor the user's default schema preference and/or a preference for schema (re)mappings.