Consider passing min and max constraints in the get_records() and get_batches() method signatures

[aaronsteers]

Related to:

• working mvp MeltanoLabs/tap-snowflake#2 (comment)
• working mvp MeltanoLabs/tap-snowflake#2 (comment)

Disclaimer: I realize this is likely a breaking change. We could time this during 1.0 and/or we could in the meanwhile have a _v2 version of the method if necessary.

Background

Developers as of now have to spend a lot of time in the docs to understand how to honor the bookmark location and replication method. We could make this much easier by simply passing min_replication_key_value and min_replication_key_value with the method signatures of both get_records() and get_batches().

Example

Our SQL implementation is a decent example:

Current code

This requires the developer to call get_starting_replication_key_value(). The SDK checks that we're not in full table mode when responding to get_starting_replication_key_value and replication_key - but it's not clear at all that they don't have to check those.

    def get_records(self, context: dict | None) -> Iterable[dict[str, Any]]:
        """Return a generator of record-type dictionary objects.

        If the stream has a replication_key value defined, records will be sorted by the
        incremental key. If the stream also has an available starting bookmark, the
        records will be filtered for values greater than or equal to the bookmark value.

        Args:
            context: If partition context is provided, will read specifically from this
                data slice.

        Yields:
            One dict per record.

        Raises:
            NotImplementedError: If partition is passed in context and the stream does
                not support partitioning.
        """
        if context:
            raise NotImplementedError(
                f"Stream '{self.name}' does not support partitioning."
            )

        table = self.connector.get_table(self.fully_qualified_name)
        query = table.select()
        if self.replication_key:
            replication_key_col = table.columns[self.replication_key]
            query = query.order_by(replication_key_col)

            start_val = self.get_starting_replication_key_value(context)
            if start_val:
                query = query.where(
                    sqlalchemy.text(":replication_key >= :start_val").bindparams(
                        replication_key=replication_key_col, start_val=start_val
                    )
                )

        for record in self.connector.connection.execute(query):
            yield dict(record)

Proposal (NEW)

• Add paradigm for stream filter constraints and max_records constraints #1119

Proposal (OLD)

This example doesn't require the developer to check anything except self.replication_key. Everything else is passed explicitly.

    def get_records(
        self,
        context: dict | None,
        start_position: Any | None,
        end_position: Any | None,
    ) -> Iterable[dict[str, Any]]:
        """Return a generator of record-type dictionary objects.

        If the stream has a replication_key value defined, records will be sorted by the
        incremental key. If the stream also has an available starting bookmark, the
        records will be filtered for values greater than or equal to the bookmark value.

        Args:
            context: If partition context is provided, will read specifically from this
                data slice.
            start_position: If specified, the minimum value for the replication_key
                column. Otherwise unbounded.
            end_position: If specified, the maximum value for the replication_key
                column. Otherwise unbounded.

        Yields:
            One dict per record.

        Raises:
            NotImplementedError: If partition is passed in context and the stream does
                not support partitioning.
        """
        if context:
            raise NotImplementedError(
                f"Stream '{self.name}' does not support partitioning."
            )

        table = self.connector.get_table(self.fully_qualified_name)
        query = table.select()
        if self.replication_key:
            query = query.order_by(self.replication_key)

        if start_position is not None or end_position is not None:
            if not self.replication_key:
                # Meltano won't send constraints if there's no replication key defined
                raise MeltanoError("...")

            if start_position:
                query.where(
                    sqlalchemy.text(":replication_key >= :start_val").bindparams(
                        replication_key=table.columns[self.replication_key],
                        start_val=start_position,
                    )
                )
            if end_position:
                query.where(
                    sqlalchemy.text(":replication_key >= :start_val").bindparams(
                        replication_key=table.columns[self.replication_key],
                        start_val=start_position,
                    )
                )

        for record in self.connector.connection.execute(query):
            yield dict(record)

For tangental reasons, the example above includes also and ending value constraint. We don't strictly have to add this now, but I think we probably should.

[edgarrmondragon]

@aaronsteers is the idea that in the default implementation

• start_position is passed by the SDK from the value of the get_starting_* method and
• end_position takes its value from the replication signpost?

It is indeed a breaking change, if the dev implemented its own get_records (not too common) or get_batches (has low adoption since it's a new feature).

[aaronsteers]

@aaronsteers is the idea that in the default implementation

• start_position is passed by the SDK from the value of the get_starting_* method and

Yes to this. 👍

• end_position takes its value from the replication signpost?

I was thinking 'yes', but on thinking about this more, probably not. After proposing this on the batch side, I realized there is a good reason not to this unless we are willing to run a 'final batch' also with the end_position included as start_position to the next call.

A quick diversion into the theoretical principals here:

1. We guarantee each record arrives at least once.
2. This alteration of using signpost as a filter gives us at least once eventually, which means no specific sync is guaranteed to have all records up to the point of being executed. This is definitely not what we want.
   • The missed records are the ones who's replication key value were modified to now() within the small gap of time between querying the min/max and retrieving the records. The chance of missing records in this window is small, but it's not zero, and there's an (extremely small but non-zero) chance that the same record could be modified in that window again during the next sync operation. Et cetera.

So, I think we have two choices, and this applies to get_batches() as well as get_records().

• Option 1: We run bounded call as the main operation and then we just always finish up with a final unbounded call.
  • The exception here is if end_date is requested explicitly, which is our signal that they are taking the responsibility and there's nothing for us "after" our end date restriction anyway.
  • This actually could work nicely for batch, but it adds complexity and means that (at least) two batches need to be loaded per stream.
• Option 2: We don't use signpost as the upper bound.
  • We would still use end_date as the upper bound if provided.
  • We can still populate the bookmark with our signpost value - it just won't be a constraint (same as today with get_records()).

[aaronsteers]

As it stands right now, I think it makes sense for us to use "Option 2" above and not bound the upper limit, but in terms of method signature, I still like the idea of making space for end_date configs in future.

[edgarrmondragon]

Yeah, Option 2 keeps the functionality most similar with get_records and still allows developers to configure and upper bound 👍

[aaronsteers]

In a new issue, I propose a similar approach as described here, but with a comprehensive covering of filter cases, and with reduced reliance on developers to inspect self.replication_key, self.replication_method etc.

• Add paradigm for stream filter constraints and max_records constraints #1119

[kgpayne]

In addition to min/max constraints, we could include limit too, for cases when the upstream system can return filtered responses (such as SQL LIMIT). This is important for testing, where we use an internal _MAX_RECORDS_LIMIT value to terminate a tap early for testing purposes. More details here.

[aaronsteers]

@kgpayne - Thanks for raising. I incorporated that also in a new proposal here:

• Add paradigm for stream filter constraints and max_records constraints #1119