Skip to content
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

fix(ingest): profiling (bigquery) - Address biquery profiling query error due to timestamp vs data mismatch #6874

Merged
merged 5 commits into from
Dec 30, 2022
Merged

fix(ingest): profiling (bigquery) - Address biquery profiling query error due to timestamp vs data mismatch #6874

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
602f0ba
Address biquery profiling query error due to timestamp vs data mismatch
treff7es Dec 28, 2022
6806fc8
Adding comment about default type
treff7es Dec 28, 2022
6432cbf
lint
treff7es Dec 28, 2022
a3cbebe
Setting prifle if updated default to None
treff7es Dec 29, 2022
4e6be62
Merge branch 'master' into bq_profiling_query_fix
treff7es Dec 29, 2022
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
9 changes: 4 additions & 5 deletions metadata-ingestion/src/datahub/ingestion/source/bigquery_v2/profiler.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -121,15 +121,14 @@ def generate_partition_profiler_query(
] = partition
return None, None

partition_column_type: str = "DATE"
# ingestion time partitoned tables partition column is not in the schema, so we default to TIMESTAMP type
partition_column_type: str = "TIMESTAMP"
for c in table.columns:
if c.is_partition_column:
partition_column_type = c.data_type

if table.time_partitioning.type_ in ("DAY", "MONTH", "YEAR"):
partition_where_clause = f"`{table.time_partitioning.field}` BETWEEN {partition_column_type}('{partition_datetime}') AND {partition_column_type}('{upper_bound_partition_datetime}')"
elif table.time_partitioning.type_ in ("HOUR"):
partition_where_clause = f"`{table.time_partitioning.field}` BETWEEN '{partition_datetime}' AND '{upper_bound_partition_datetime}'"
if table.time_partitioning.type_ in ("HOUR", "DAY", "MONTH", "YEAR"):
partition_where_clause = f"{partition_column_type}(`{table.time_partitioning.field}`) BETWEEN {partition_column_type}('{partition_datetime}') AND {partition_column_type}('{upper_bound_partition_datetime}')"
else:
logger.warning(
f"Not supported partition type {table.time_partitioning.type_}"
Expand Down
1 change: 0 additions & 1 deletion metadata-ingestion/src/datahub/ingestion/source/kafka.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -330,7 +330,6 @@ def build_custom_properties(
topic_detail: Optional[TopicMetadata],
extra_topic_config: Optional[Dict[str, ConfigEntry]],
) -> Dict[str, str]:

custom_props: Dict[str, str] = {}
self.update_custom_props_with_topic_details(topic, topic_detail, custom_props)
self.update_custom_props_with_topic_config(
Expand Down
5 changes: 4 additions & 1 deletion metadata-ingestion/src/datahub/ingestion/source/sql/sql_generic_profiler.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -147,7 +147,10 @@ def is_dataset_eligible_for_profiling(
if last_profiled
else None
)
if self.config.profiling.profile_if_updated_since_days is not None:
if (
Copy link
Collaborator

Choose a reason for hiding this comment

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

Should we disable "self.config.profiling.profile_if_updated_since_days" (i.e. set to None) if store_last_profiling_timestamps is set to True.
The defaults for self.config.profiling.profile_if_updated_since_days config are very conservative i.e. 1 day and it will interfere with the profiling state handler, which smartly determines the whether to profile a table.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

thanks, you are absolutely right, I will address this

not threshold_time
and self.config.profiling.profile_if_updated_since_days is not None
):
threshold_time = datetime.now(timezone.utc) - timedelta(
self.config.profiling.profile_if_updated_since_days
)
Expand Down