https://linen.dev logo
Join Slack
Powered by
Hey team, I want to know in the meltano why there ...
# contributing
k
Hey team, I want to know in the meltano why there is no overlap period option for key-base-incremental-replication.
e
Hi @Kate Liu đź‘‹ What do you mean by "overlap period"?
k
Hi Edgar, Thank you for your reply. The overlap period means the replication-key: update_at - 2 hours. So the relocation start time = updated_at - 2 hours and then end time = current _timestamp.
d
A couple things: @Kate Liu If I understand it correctly, the tap will use the last updated_at it read as the state for the next import. It's not looking at the current time - it's looking at the greatest updated_at value. So, say the first run reads: row 1, 1pm row 2, 2pm row 3, 3pm It then stores 3pm as its state. Then we add some more data: row 4, 4pm row 5, 5pm Then we start the next run at, say, 11pm (or whenever, really), we'll start from 3pm and read: row 4, 4pm row 5, 5pm That said, @Edgar RamĂ­rez (Arch.dev) there is an overlap case that I have been wondering about. So, say the first run reads: row 1, 1pm row 2, 2pm row 3, 3pm It then stores 3pm as its state. Then we add some more data: row 4, 2:58pm row 5, 2:59pm and then the next run misses row 4 and 5. Why were they missed? Well, maybe the first run didn't have 3 rows, it had 45 million rows. During that time, row 4 and 5 were inserted. Due to some edge case, we inserted data with an earlier timestamp. Maybe we were using client dates instead of Postgres dates. Maybe a transaction took too long. Maybe it's some weird glitch. I'd feel more comfortable adding, say, a 1 hour overlap, and re-ingesting the data from 2pm, catching the anomaly and deduping on the primary key.
đź‘Ť 1
d
Hey @Kate Liu, @dylan_just, I could be wrong, but I believe what you set in the replication key is used to store state as @dylan_just rightly pointed out but in the next run from your example it would be looking at values >= state so it should bring row 3 , 4 and 5 in the next run
for reference, from singer sdk
Copy code
def apply_query_filters(
        self,
        query: selectable.Select,
        table: sa.Table,
        *,
        context: Context | None = None,
    ) -> selectable.Select:
        """Apply WHERE and ORDER BY clauses to the query.

        By default, this method applies a replication filter to the query
        and orders the results by the replication key, if a replication key is set.

        Args:
            query: The SQLAlchemy Select object.
            table: The SQLAlchemy Table object.
            context: The context object.

        Returns:
            A SQLAlchemy Select object.
        """
        if self.replication_key:
            column = table.columns[self.replication_key]
            order_by = (
                sa.nulls_first(column.asc())
                if self.supports_nulls_first
                else column.asc()
            )
            query = query.order_by(order_by)

            start_val = self.get_starting_replication_key_value(context)
            if start_val is not None:
                query = query.where(column >= start_val)

        return query
k
@dylan_just yes, that's the issue one of our project has. Your description is correct. Because the query timestamp and the source system time maybe have a gap. it maybe just 1second or 10min . If the meltano tap can add the parameter overlap, we can eliminate this application system and its database time gap problem which causes the replication missing data issue.
@Dron Adhikari But the replication key is 100% follow the incremental sequence. For example replication_key is id Row 1 id = 1 Row2 id =2 Row 3 id = 5 Row 4 id = 3 The long transition commit can be one of scenario that happens the replication increasing is not followed the sequence.
d
You could run a "meltano state merge" command to rewind the state a bit between runs.
k
Can you please share the meltano document link about the rewind?
d
https://docs.meltano.com/reference/command-line-interface#state You want to call 
meltano state merge
and pass it the tap and a state json string that sets the state to where you want it. You'll need to calculate what that state should be, though. You could do this: • 
meltano state read
• Parse the result • Subtract 2 hours from that • 
meltano state merge
to set the new value So, it's not an automatic thing like an "overlap" setting, but it should help with this scenario. Anyway, this process of using "merge" to set the state to an earlier value is what I mean by "rewinding" the state. It can be useful in other scenarios e.g. if you want to re-ingest a table from scratch, or if you added a column to your source data and want to re-ingest old rows to capture that.
d
@Kate Liu, for the issue you mentioned of long transaction, thats a valid case, An overlap period can help with that However for pipelines like this we solve it in 2 ways 1. We apply overlap period by using SQL style query taps where we calculate the overlap date by ourselves, we have to build some customisation on our side for this 2. we usually look for columns that actually record the timestamp when the record was written to the source DB or add a similar system timestamp if possible in a column like 
___updated_at_timestamp__
,_ and set this column for state replication key, it allows for taking into account the rows always with respect to when they were actually last written in to the source
đź‘Ť 1
8 Views
Open in Slack
PreviousNext
Powered by