Create deferrable SqlExecuteQueryOperator compatible with Postgres#65618
Create deferrable SqlExecuteQueryOperator compatible with Postgres#65618karenbraganz wants to merge 15 commits into
Conversation
|
Thanks for incorporating the review changes for import of conf @karenbraganz !!! Seeing that the 3 failing tests are related to docker failures and files not being found as below. Can you try rebasing and pushing the changes to repo, so as to ensure any bug fixes for unrelated failures are resolved by another open source contributor, automatically? I observed @jscheffl pushed no files however CI ran fine, from log above last on May 29th. Looks to me, possibly some change happened after that to cause these errors to start failing. FYI - @potiuk I see same error in failed CI run below. Looks to me the same issue. |
potiuk
added
the
ready for maintainer review
|
Sorry, I accidentally tangled up my branch and pushed the wrong commits, which is why all these reviews were requested. I have reverted the mistake. |
| if self.fetch_results: | ||
| # Fetch the raw rows with the built-in handler and return them with the cursor | ||
| # descriptions; the operator applies any user handler on the worker. | ||
| results = await hook.run_async( |
There was a problem hiding this comment.
What happens if the triggerer is killed in the middle of a run? Will this trigger be re-executed? Is there a chance that multiple instances of the same query are invoked?
There was a problem hiding this comment.
yes, this could result in a duplicate query run if the trigger is killed. I can look into whether this can be avoided.
There was a problem hiding this comment.
I wonder if we can initiate on the worker and then poll on the triggerer? Not sure there's a straightforward way to do that.
We could also potentially block non-read-only queries by default and include a parameter to do things like INSERT, UPDATE, DELETE, etc with a docstring explaining the risks?
There was a problem hiding this comment.
I looked into this. There isn't a straightforward way to implement this for Postgres. It requires the client side connection to remain alive in order to continue running the query. So we cannot open a connection from the worker to start the query, then close that, and re-open a new connection on the triggerer. Your second suggestion seems doable though:
We could also potentially block non-read-only queries by default and include a parameter to do things like INSERT, UPDATE, DELETE, etc with a docstring explaining the risks?
I can look into implementing it like that.
There was a problem hiding this comment.
I have added some logic for this. Still need to run tests and confirm that it is working as expected.
|
|
||
| class SQLExecuteQueryTrigger(BaseTrigger): | ||
|
|
||
| class SQLGenericTransferTrigger(BaseTrigger): |
There was a problem hiding this comment.
This is a breaking change, right? If so, we'll need to bump the major version of this provider.
There was a problem hiding this comment.
yes, this would qualify as a breaking change, so we should bump the major version FYI @kacpermuda
I do not anticipate a lot of disruption since only the trigger name is changed- not the operator name. This would only affect people who use this trigger in a custom operator.
There was a problem hiding this comment.
Are there any other breaking changes in this PR? I think we discussed that for the default deferrable value we'll have a workaround to avoid breaking changes for now, so doing it just so that we can rename a class seems a bit ... pointless ? Not sure.
There was a problem hiding this comment.
I don't think it's pointless -- If someone built their own deferrable operator with SQLExecuteQueryTrigger, it will break.
There was a problem hiding this comment.
I'm not saying making a major release is pointless, just wondering if renaming a class is needed, trying to think of a way we can add this deferrable functionality as purely additive
There was a problem hiding this comment.
@kacpermuda previously, the GenericTransfer operator used a trigger called the SqlExecuteQueryTrigger. Now that I am making the SqlExecuteQueryOperator deferrable, I thought it would be confusing to have a SqlExecuteQueryTrigger that is actually used by a different operator. I renamed the trigger used by the GenericTransfer operator to SqlGenericTransferTrigger and used the SqlExecuteQueryTrigger name for the trigger used by the SqlExecuteQueryOperator.
This name change is not required, but is good for consistency between the trigger and operator.
There was a problem hiding this comment.
Sure sure, that makes sense, just wanted to make sure we're doing all the changes consciously since they lead to major release 😄
| yield TriggerEvent({"status": "failure", "message": str(e)}) | ||
|
|
||
|
|
||
| class SQLExecuteQueryTrigger(BaseTrigger): |
There was a problem hiding this comment.
I would prefer to see the GenericTransfer also use this SQLExecuteQueryTrigger, which is better suited due to it's more async native nature. I don't like the fact that we have now a dedicated trigger for the GenericTransfer, I think it should be possible to reuse the new SQLExecuteQueryTrigger.
There was a problem hiding this comment.
Makes sense, I will look into how this can be implemented. @RNHTTR @kacpermuda we might be able to avoid releasing this in a major version if I do this since the trigger name for GenericTransfer won't change.
There was a problem hiding this comment.
This to me stays a major change independently of this.
6 participants
Was generative AI tooling used to co-author this PR?
This PR modifies the SqlExecuteQueryOperator to be deferrable. I have also created a SqlExecuteQueryTrigger and added asynchronous functions to the DbApiHook for use with the SqlExecuteQueryOperator when it is deferred.
I plan on making this compatible with several database types. This PR only adds compatibility with Postgres.
{pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.