While writing my target, I am a little struck by how many different issues there are with making taps and targets interoperable. Has anyone thought about some sort of tag or badge system to make it easier to figure out if a tap and a target works together? Some of the issues I think about: - Does the target need a “key” property? - Can the target handle any and all column names, whatever the special characters? - Does the target handle records that do not include all attributes that are in the schema? - Deletes? I’m sure there are a bunch of other issues too, and pretty much the same questions can be used for the tap.

Hi, @Henning Holgersen. Thanks for raising these. I think the guiding philosophy is that targets should just never fail unless they must fail - or said another way, the target would make every reasonable attempt to serialize the data it receives, if it is able to do so. And then within that guidance, there are indeed additional capabilities or restrictions that each target should declare in their readme docs. The pipelinewise snowflake target, for instance, requires primary keys by default but in their readme, there's documentation on how to disable that restriction via settings. On the tap side, we do have several enumerated 'capabilities' such as incremental (state) and log-based replication. We don't have anything for whether there's a primary key for each stream, but this is mostly because the majority of taps are a hybrid. Each tap's catalog does advertise this, but it's just on a much lower grain because each stream can be different. (We can probably detect this though and get into the hub with a 'some', 'many', 'all', 'none' characterization. Other folks have asked for a list of supported streams for each tap to be on the hub, and that would be another way for us to document which streams do/don't have primary keys or incremental capability.)

I like the focus on targets, and because I am currently preoccupied with sql targets there is some generic sdk functionality we might create: automatically add columns that aren’t present in the record but are in the schema; create row hash if there is no key; and add a separate method for cleaning up column names. I know some of these are already in the roadmap.

automatically add columns that aren’t present in the record but are in the schema

💯 - This is something I definitely think we should expect. Would be good to log an issue on this so we're clear on requirements and add to our SQL milestone.

create row hash if there is no key

Awesome. I think this would be really valuable to a lot of folks. Do you mind logging an SDK issue or a new GitHub discussion with your thoughts on this?

add a separate method for cleaning up column names

I think @ken_payne recently added this (or is doing so presently). 🙂

automatically add columns that aren’t present in the record but are in the schema

I recently implemented this at the tap level (but agree that target level would probably be better), by creating an "empty record" and merging it with the tap generated record in the

post_process

step. Feel free to steal this code:

def empty_record(self) -> dict:
        """An empty record in schema shape up to flatten depth"""
        empty_record = self._empty_record(
            self.schema, 0, self.config.get("flattening_max_depth", 0)
        )
        return empty_record if empty_record is not None else {}    

    @staticmethod
    def _empty_record(
        schema_node: dict, level: int = 0, max_level: int = 0
    ) -> dict | None:
        """recursive helper of empty_record"""
        if level > max_level or "properties" not in schema_node:
            return None
        return {
            k: SlingStream._empty_record(v, level + 1, max_level)
            for k, v in schema_node["properties"].items()
        }

    @staticmethod
    def merge_records(record_1: dict, record_2: dict) -> dict:
        """deep merge of two records, record_2 takes precedence"""
        result: dict = deepcopy(record_1)
        for k, value in record_2.items():
            if isinstance(value, dict) and k in result and isinstance(result[k], dict):
                result[k] = SlingStream.merge_records(result[k], value)
            else:
                result[k] = value
        return result

    def post_process(self, row: dict, context: dict | None = None) -> dict:
        """adds empty nodes matching the schema to flatten depth"""
        return self.merge_records(self.empty_record(), row)