I would also like to see some discussion in this AIP about how the data is going to be serialised to and from the database instances (obviously Connexion is involved, but I presume more transformation code is needed than that) and the potential slowdown this would cause.
In my experience, a somewhat direct ORM mapping like this is going to result in considerably slower times for any complex operation that's touching a few hundred rows. Is there a reason this is being proposed for the scheduler code, too? In my mind, the best approach to multitenancy would be to remove all user-supplied code from the scheduler and leave it with direct DB access, rather than trying to indirect all scheduler access through another API layer. Andrew On Thu, Dec 2, 2021 at 10:29 AM Jarek Potiuk <ja...@potiuk.com> wrote: > Yeah - I thik Ash you are completely right we need some more > "detailed" clarification. > > I believe, I know what you are - rightfully - afraid of (re impact on > the code), and maybe we have not done a good job on explaining it with > some of our assumptions we had when we worked on it with Mateusz. > > Simply it was not clear that our aim is to absolutely minimise the > impact on the "internal DB transactions" done in schedulers and > workers. The idea is that change will at most result in moving an > execution of the transactions to another process but not changing what > the DB transactions do internally. Actually this was one of the reason > for the "alternative" approach (you can see it in the document) we > discussed about - hijack "sqlalchemy session" - this is far too low > level and the aim of the "DB-API" is NOT to replace direct DB calls > (Hence we need to figure out a better name). The API is there to > provide "scheduler logic" API and "REST access to Airflow primitives > like dags/tasks/variables/connections" etc.. > > As an example (which we briefly talked about in slack) the > "_run_mini_scheduler_on_child_tasks" case > ( > https://github.com/apache/airflow/blob/main/airflow/jobs/local_task_job.py#L225-L274 > ) > is an example (that we would put in the doc). As we thought of it - > this is a "single DB-API operation". Those are not Pure REST calls of > course, they are more RPC-like calls. That is why even initially I > thought of separating the API completely. But since there are a lot of > common "primitive" calls that we can re-use, I think having a separate > DB-API component which will re-use connexion implementation, replacing > authentication with the custom worker <> DB-API authentication is the > way to go. And yes if we agree on the general idea, we need to choose > the best way on how to best "connect" the REST API we have with the > RPC-kind of API we need for some cases in workers. But we wanted to > make sure we are on the same page with the direction. And yes it means > that DB-API will potentially have to handle quite a number of DB > operations (and that it has to be replicable and scalable as well) - > but DB-API will be "stateless" similarly as the webserver is, so it > will be scalable by definition. And yest performance tests will be > part of POC - likely even before we finally ask for votes there. > > So in short: > * no modification or impact on current scheduler behaviour when DB > Isolation is disabled > * only higher level methods will be moved out to DB-API and we will > reuse existing "REST" APIS where it makes sense > * we aim to have "0" changes to the logic of processing - both in Dag > Processing logic and DB API. We think with this architecture we > proposed it's perfectly doable > > I hope this clarifies a bit, and once we agree on general direction, > we will definitely work on adding more details and clarification (we > actually already have a lot of that but we just wanted to start with > explaining the idea and going into more details later when we are sure > there are no "high-level" blockers from the community. > > J, > > On Thu, Dec 2, 2021 at 4:46 PM Ash Berlin-Taylor <a...@apache.org> wrote: > > > > I just provided a general idea for the approach - but if you want me to > put more examples then I am happy to do that > > > > > > Yes please. > > > > It is too general for me and I can't work out what effect it would > actually have on the code base, especially how it would look with the > config option to enable/disable direct db access. > > > > -ash > > > > On Thu, Dec 2 2021 at 16:36:57 +0100, Mateusz Henc > <mh...@google.com.INVALID> wrote: > > > > Hi, > > I am sorry if it is not clear enough, let me try to explain it here, so > maybe it gives more light on the idea. > > See my comments below > > > > On Thu, Dec 2, 2021 at 3:39 PM Ash Berlin-Taylor <a...@apache.org> wrote: > >> > >> I'm sorry to say it, but this proposal right just doesn't contain > enough detail to say what the actual changes to the code would be, and what > the impact would be > >> > >> To take the one example you have so far: > >> > >> > >> def get_dag_run(self, dag_id, execution_date): > >> return self.db_client.get_dag_run(dag_id,execution_date) > >> > >> So form this snippet I'm guessing it would be used like this: > >> > >> dag_run = db_client.get_dag_run(dag_id, execution_date) > >> > >> What type of object is returned? > > > > > > As it replaces: > > dag_run = session.query(DagRun) > > .filter(DagRun.dag_id == dag_id, DagRun.execution_date == > execution_date) > > .first() > > > > then the type of the object will be exactly the same (DagRun) . > > > >> > >> > >> Do we need one API method per individual query we have in the source? > > > > > > No, as explained by the sentence: > > > > The method may be extended, accepting more optional parameters to avoid > having too many similar implementations. > > > > > >> > >> > >> Which components would use this new mode when it's enabled? > > > > > > You may read: > > Airflow Database APi is a new independent component of Airflow. It > allows isolating some components (Worker, DagProcessor and Triggerer) from > direct access to DB. > > > >> > >> But what you haven't said the first thing about is what _other_ changes > would be needed in the code. To take a fairly simple example: > >> > >> dag_run = db_client.get_dag_run(dag_id, execution_date) > >> dag_run.queued_at = timezone.now() > >> # How do I save this? > >> > >> In short, you need to put a lot more detail into this before we can > even have an idea of the full scope of the change this proposal would > involve, and what code changes would be needed for compnents to work with > and without this setting enabled. > > > > > > For this particular example - it depends on the intention of the code > author > > - If this should be in transaction - then I would actually introduce new > method like enqueue_dag_run(...) that would run these two steps on Airflow > DB API side > > - if not then, maybe just the "update_dag_run" method accepting the > whole "dag_run" object and saving it to the DB. > > > > In general - we could take naive approach, eg replace code: > > dag_run = session.query(DagRun) > > .filter(DagRun.dag_id == dag_id, DagRun.execution_date == > execution_date) > > .first() > > with: > > if self.db_isolation: > > dag_run = session.query(DagRun) > > .filter(DagRun.dag_id == dag_id, DagRun.execution_date == > execution_date) > > .first() > > else: > > dag_run = db_client.get_dag_run(self, dag_id, execution_date) > > > > The problem is that Airflow DB API would need to have the same > implementation for the query - so duplicated code. That's why we propose > moving this code to the DBClient which is also used by the Airflow DB > API(in DB direct mode). > > > > I know there are many places where the code is much more complicated > than a single query, but they must be handled one-by-one, during the > implementation, otherwise this AIP would be way too big. > > > > I just provided a general idea for the approach - but if you want me to > put more examples then I am happy to do that > > > > Best regards, > > Mateusz Henc > > > >> > >> On Thu, Dec 2 2021 at 14:23:56 +0100, Mateusz Henc > <mh...@google.com.INVALID> wrote: > >> > >> Hi, > >> I just added a new AIP for running some Airflow components in > DB-isolation mode, without direct access to the Airflow Database, but they > will use a new API for thi purpose. > >> > >> PTAL: > >> > https://cwiki.apache.org/confluence/display/AIRFLOW/AIP-44+Airflow+Database+API > >> > >> Open question: > >> I called it "Airflow Database API" - however I feel it could be more > than just an access layer for the database. So if you have a better name, > please let me know, I am happy to change it. > >> > >> Best regards, > >> Mateusz Henc >
