vdk-core: add new managed db_connection_execute_operation hook

[antoniivanov]

Currently we cannot track the full execution of a query (before and
after it) . This is necessary in order to be able to generate proper
lineage events (query start and end) or to track its duration (we had to
currently had to add this to vdk-core
(#804) but that's
really better implemented as plugin (think aspect-oriented programming).

In the future we should consider adding hook for fetch (fetchMany,
fetchAll) as well as it's there where the data is returned (plugins can
take stats - count number of rows, validate sensitive columns, do result
validation tests, etc.)

Testing Done: unit tests, functional test.

Signed-off-by: Antoni Ivanov aivanov@vmware.com

[ivakoleva]

Could be simpler for API usage, yet local context with yielding is possible this way,
we agreed to merge

[antoniivanov]

Could be simpler for API usage

I don't think we have an idea for simpler API usage than this?

The other possible API could be to have two hooks

db_connection_execute_before_operation(cursor, query)
db_connection_execute_after_operation(cursor, query)

But in that case, a simple use case of tracking the duration of query becomes very hard to implement
Maybe like this

class QueryDuratinPlugin:
     queries: Dict # keep state of queries

     def db_connection_execute_before_operation(self, cursor) 
            # we do not have id because query is not yet executed so we have to generate it
            id = generate_id(query) 
            query = update_query_comments_with_id(query)
            self.queries[id] = time.time() 
   
    def db_connection_execute_after_operation(self, cursor) 
            id = parse_id_from_comments(query) # we do not have id because query is not yet executed 
            start = self.queries[id]
            end = time.time()
            log.info("Duration: {end - start}")
   

while now with this hook it is:

            @hookimpl(hookwrapper=True)
            db_connection_execute_operation(execution_cursor: ExecutionCursor) -> Optional[int]: 
                start = time.time()
                outcome = yield # we yield the execution to other implementations (including default one) 
                end = time.time()
                log.info(f" duration: {end - start}. ")