yujunjun/datahub
2
0
Fork 0
mirror of https://github.com/datahub-project/datahub.git synced 2025-09-20 06:33:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(ingest): profiling add upper bound on combined query size (#3762)

Browse Source
This commit is contained in:
Harshal Sheth 2021-12-16 20:34:46 -05:00 committed by GitHub
parent e6f8c1c17c
commit df2cb94ed8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 7 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
metadata-ingestion/src/datahub/utilities/sqlalchemy_query_combiner.py
View File

@ -23,6 +23,8 @@ logger: logging.Logger = logging.getLogger(__name__)
P = ParamSpec("P") P = ParamSpec("P")
MAX_QUERIES_TO_COMBINE_AT_ONCE = 40
# We need to make sure that only one query combiner attempts to patch # We need to make sure that only one query combiner attempts to patch
# the SQLAlchemy execute method at a time so that they don't interfere. # the SQLAlchemy execute method at a time so that they don't interfere.
@ -286,6 +288,11 @@ class SQLAlchemyQueryCombiner:
full_queue = self._get_queue(main_greenlet) full_queue = self._get_queue(main_greenlet)
pending_queue = {k: v for k, v in full_queue.items() if not v.done} pending_queue = {k: v for k, v in full_queue.items() if not v.done}
pending_queue = dict(
itertools.islice(pending_queue.items(), MAX_QUERIES_TO_COMBINE_AT_ONCE)
)
if pending_queue: if pending_queue:
queue_item = next(iter(pending_queue.values())) queue_item = next(iter(pending_queue.values()))