fix(ingest/datahub): Create Structured property templates in advance and batch processing #13355
Conversation
treff7es
commented
Apr 29, 2025
•
treff7es
commented
- Structured properties ingested first to make sure template created in advance.
- Add process database data in batches to make sure it won't run out of memory on db side
github-actions
bot
added
the
ingestion
Codecov ReportAttention: Patch coverage is ✅ All tests successful. No failed tests found.
📢 Thoughts on this report? Let us know! |
datahub-cyborg
bot
added
the
needs-review
| ) | ||
|
|
||
| time.sleep( | ||
| self.config.structured_properties_template_cache_invalidation_interval |
There was a problem hiding this comment.
This time should be based on on how many structured properties exist.
There was a problem hiding this comment.
LGTM, assuming green CI.
Ideally we could use something like: https://docs.datahub.com/docs/advanced/api-tracing to know when structured properties have been processed.
datahub-cyborg
bot
added
pending-submitter-merge
and removed
needs-review
treff7es
changed the title
| ) | ||
|
|
||
| query_timeout: Optional[int] = Field( | ||
| hidden_from_docs=True, |
There was a problem hiding this comment.
This should not be hidden from docs
| logger.debug("Fetching soft-deleted URNs") | ||
|
|
||
| # Use server-side cursor implementation | ||
| rows = self.execute_server_cursor(self.soft_deleted_urns_query, params) |
There was a problem hiding this comment.
@treff7es this query should also take into account the execution dates + limits & offsets. We can have ingestion sources emitting a lot of soft deletes so the disk issue on the server is also applicable here.
| def _get_rows( | ||
| self, from_createdon: datetime, stop_time: datetime | ||
| self, | ||
| from_createdon: datetime, | ||
| stop_time: datetime, | ||
| set_structured_properties_filter: bool = False, | ||
| ) -> Iterable[Dict[str, Any]]: | ||
| params = { | ||
| "exclude_aspects": list(self.config.exclude_aspects), | ||
| "since_createdon": from_createdon.strftime(DATETIME_FORMAT), | ||
| } | ||
| yield from self.execute_server_cursor(self.query, params) | ||
| # Query and yield rows for this date batch | ||
| yield from self._get_rows_for_date_range( | ||
| from_createdon, | ||
| stop_time, | ||
| set_structured_properties_filter, | ||
| limit=self.config.database_query_batch_size, | ||
| ) |
There was a problem hiding this comment.
Doesn't seem like we need both these functions anymore
| ) | ||
|
|
||
| query_timeout: Optional[int] = Field( | ||
| hidden_from_docs=True, |
| {"" if self.config.include_soft_deleted_entities else " AND (removed = false or removed is NULL)"} | ||
| {structured_prop_filter} |
There was a problem hiding this comment.
Nit: feels like either both expressions should be factored into a variable or neither
| if start_date != last_createdon: | ||
| start_date = last_createdon | ||
| offset = 0 |
There was a problem hiding this comment.
If start_date == last_createdon then we need to update offset, for sure, no?
|
🔴 Meticulous spotted visual differences in 36 of 612 screens tested: view and approve differences detected. Meticulous evaluated ~10 hours of user flows against your PR. Last updated for commit 8afa715. This comment will update as new commits are pushed. |
| duplicate_found = True | ||
| break | ||
| if first_iteration: | ||
| start_date = row.get("createdon", start_date) |
treff7es
force-pushed
the
datahub_structured_property_fix
branch
from
8afa715 to
3f01d68
Compare
