Skip to content

fix(ingest): limit number of upstreams generated by sql parsing aggre… #11267

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hsheth2 merged 6 commits into from
Aug 30, 2024
Merged

fix(ingest): limit number of upstreams generated by sql parsing aggre… #11267

Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
c2070c6
fix(ingest): limit number of upstreams generated by sql parsing aggre…
mayurinehate Aug 29, 2024
3d18b9a
Merge branch 'master' into feature/ing-696
mayurinehate Aug 29, 2024
c90bb44
Update metadata-ingestion/src/datahub/sql_parsing/sql_parsing_aggrega…
mayurinehate Aug 30, 2024
ec55b14
Update metadata-ingestion/src/datahub/sql_parsing/sql_parsing_aggrega…
mayurinehate Aug 30, 2024
dff2f7e
add counters to report trimmed lineage
mayurinehate Aug 30, 2024
a56fef4
Merge branch 'master' into feature/ing-696
mayurinehate Aug 30, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 24 additions & 0 deletions metadata-ingestion/src/datahub/sql_parsing/sql_parsing_aggregator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,6 +72,8 @@ class QueryLogSetting(enum.Enum):
_DEFAULT_QUERY_LOG_SETTING = QueryLogSetting[
os.getenv("DATAHUB_SQL_AGG_QUERY_LOG") or QueryLogSetting.DISABLED.name
]
MAX_UPSTREAM_TABLES_COUNT = 300
MAX_FINEGRAINEDLINEAGE_COUNT = 2000


@dataclasses.dataclass
Expand Down Expand Up @@ -229,6 +231,8 @@ class SqlAggregatorReport(Report):
num_unique_query_fingerprints: Optional[int] = None
num_urns_with_lineage: Optional[int] = None
num_lineage_skipped_due_to_filters: int = 0
num_table_lineage_trimmed_due_to_large_size: int = 0
num_column_lineage_trimmed_due_to_large_size: int = 0

# Queries.
num_queries_entities_generated: int = 0
Expand Down Expand Up @@ -1154,6 +1158,26 @@ def _gen_lineage_for_downstream(
confidenceScore=queries_map[query_id].confidence_score,
)
)

if len(upstream_aspect.upstreams) > MAX_UPSTREAM_TABLES_COUNT:
logger.warning(
f"Too many upstream tables for {downstream_urn}: {len(upstream_aspect.upstreams)}"
f"Keeping only {MAX_UPSTREAM_TABLES_COUNT} table level upstreams/"
)
upstream_aspect.upstreams = upstream_aspect.upstreams[
:MAX_UPSTREAM_TABLES_COUNT
]
self.report.num_table_lineage_trimmed_due_to_large_size += 1
if len(upstream_aspect.fineGrainedLineages) > MAX_FINEGRAINEDLINEAGE_COUNT:
logger.warning(
Copy link
Contributor

@hsheth2 hsheth2 Aug 29, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

would be good to add some counters to the report for these

f"Too many upstream columns for {downstream_urn}: {len(upstream_aspect.fineGrainedLineages)}"
f"Keeping only {MAX_FINEGRAINEDLINEAGE_COUNT} column level upstreams/"
)
upstream_aspect.fineGrainedLineages = upstream_aspect.fineGrainedLineages[
:MAX_FINEGRAINEDLINEAGE_COUNT
]
self.report.num_column_lineage_trimmed_due_to_large_size += 1

upstream_aspect.fineGrainedLineages = (
upstream_aspect.fineGrainedLineages or None
)
Expand Down