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: validation queries for oracle db #270

Open
wants to merge 8 commits into
base: main
Choose a base branch
from
Open

fix: validation queries for oracle db #270

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
7d6a53b
fix: numeric, completeness and uniqueness validation for oracle db
ashank2603 Nov 6, 2024
e40a365
fix: ssn, permID and string validation for oracle db
ashank2603 Nov 6, 2024
c55e76f
fix: timestamp metrics for oracle db
ashank2603 Nov 8, 2024
73083c8
fix: freshness metric for oracle db
ashank2603 Nov 11, 2024
10b6b51
fix: all space metric for oracle db
ashank2603 Nov 12, 2024
0cd9365
fix: filter for metrics in oracle db
ashank2603 Nov 14, 2024
fd9c320
fix: min validation filter
ashank2603 Nov 14, 2024
14452e6
fix: timestamp metric test case
ashank2603 Nov 14, 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
32 changes: 5 additions & 27 deletions dcs_core/core/datasource/sql_datasource.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -670,7 +670,7 @@ def query_negative_metric(
total_count_query += f" WHERE {filters}"

if operation == "percent":
query = f"SELECT (CAST(({negative_query}) AS float) / CAST(({total_count_query}) AS float)) * 100"
query = f"SELECT (CAST(({negative_query}) AS float) / CAST(({total_count_query}) AS float)) * 100 FROM {qualified_table_name}"
else:
query = negative_query

Expand Down Expand Up @@ -718,7 +718,7 @@ def query_get_null_keyword_count(
FROM {qualified_table_name}"""

if filters:
query += f" AND {filters}"
query += f" WHERE {filters}"

result = self.fetchone(query)

Expand All @@ -731,14 +731,12 @@ def query_timestamp_metric(
self,
table: str,
field: str,
operation: str,
predefined_regex: str,
filters: str = None,
) -> Union[float, int]:
"""
:param table: Table name
:param field: Column name
:param operation: Metric operation ("count" or "percent")
:param predefined_regex: regex pattern
:param filters: filter condition
:return: Tuple containing valid count and total count (or percentage)
Expand Down Expand Up @@ -811,12 +809,7 @@ def query_timestamp_metric(
)
total_count = self.fetchone(total_count_query)[0]

if operation == "count":
return valid_count, total_count
elif operation == "percent":
return valid_count, total_count
else:
raise ValueError(f"Unknown operation: {operation}")
return valid_count, total_count

except Exception as e:
print(f"Error occurred: {e}")
Expand All @@ -826,14 +819,12 @@ def query_timestamp_not_in_future_metric(
self,
table: str,
field: str,
operation: str,
predefined_regex: str,
filters: str = None,
) -> Union[float, int]:
"""
:param table: Table name
:param field: Column name
:param operation: Metric operation ("count" or "percent")
:param predefined_regex: regex pattern
:param filters: filter condition
:return: Tuple containing count of valid timestamps not in the future and total count
Expand Down Expand Up @@ -906,12 +897,7 @@ def query_timestamp_not_in_future_metric(
)
total_count = self.fetchone(total_count_query)[0]

if operation == "count":
return valid_count, total_count
elif operation == "percent":
return valid_count, total_count
else:
raise ValueError(f"Unknown operation: {operation}")
return valid_count, total_count

except Exception as e:
print(f"Error occurred: {e}")
Expand All @@ -921,14 +907,12 @@ def query_timestamp_date_not_in_future_metric(
self,
table: str,
field: str,
operation: str,
predefined_regex: str,
filters: str = None,
) -> Union[float, int]:
"""
:param table: Table name
:param field: Column name
:param operation: Metric operation ("count" or "percent")
:param predefined_regex: The regex pattern to use (e.g., "timestamp_iso")
:param filters: Optional filter condition
:return: Tuple containing count of valid dates not in the future and total count
Expand Down Expand Up @@ -996,13 +980,7 @@ def query_timestamp_date_not_in_future_metric(
)
total_count = self.fetchone(total_count_query)[0]

if operation == "count":
return valid_count, total_count
elif operation == "percent":
return valid_count, total_count
else:
raise ValueError(f"Unknown operation: {operation}")

return valid_count, total_count
except Exception as e:
logger.error(f"Error occurred: {e}")
return 0, 0
34 changes: 30 additions & 4 deletions dcs_core/core/validation/completeness_validation.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,19 +12,27 @@
# See the License for the specific language governing permissions and
# limitations under the License.

import re
from typing import Union

from dcs_core.core.datasource.search_datasource import SearchIndexDataSource
from dcs_core.core.datasource.sql_datasource import SQLDataSource
from dcs_core.core.validation.base import Validation
from dcs_core.integrations.databases.oracle import OracleDataSource


class CountNullValidation(Validation):
def _generate_metric_value(self, **kwargs) -> Union[float, int]:
if isinstance(self.data_source, SQLDataSource):
if isinstance(self.data_source, OracleDataSource) and self.where_filter:
self.where_filter = re.sub(
r"(\b[a-zA-Z_]+\b)(?=\s*[=<>])", r'"\1"', self.where_filter
)
return self.data_source.query_get_null_count(
table=self.dataset_name,
field=self.field_name,
field=f'"{self.field_name}"'
if isinstance(self.data_source, OracleDataSource)
else self.field_name,
filters=self.where_filter if self.where_filter is not None else None,
)
elif isinstance(self.data_source, SearchIndexDataSource):
Expand All @@ -40,9 +48,15 @@ def _generate_metric_value(self, **kwargs) -> Union[float, int]:
class PercentageNullValidation(Validation):
def _generate_metric_value(self, **kwargs) -> Union[float, int]:
if isinstance(self.data_source, SQLDataSource):
if isinstance(self.data_source, OracleDataSource) and self.where_filter:
self.where_filter = re.sub(
r"(\b[a-zA-Z_]+\b)(?=\s*[=<>])", r'"\1"', self.where_filter
)
return self.data_source.query_get_null_percentage(
table=self.dataset_name,
field=self.field_name,
field=f'"{self.field_name}"'
if isinstance(self.data_source, OracleDataSource)
else self.field_name,
filters=self.where_filter if self.where_filter is not None else None,
)
elif isinstance(self.data_source, SearchIndexDataSource):
Expand All @@ -58,9 +72,15 @@ def _generate_metric_value(self, **kwargs) -> Union[float, int]:
class CountEmptyStringValidation(Validation):
def _generate_metric_value(self, **kwargs) -> Union[float, int]:
if isinstance(self.data_source, SQLDataSource):
if isinstance(self.data_source, OracleDataSource) and self.where_filter:
self.where_filter = re.sub(
r"(\b[a-zA-Z_]+\b)(?=\s*[=<>])", r'"\1"', self.where_filter
)
return self.data_source.query_get_empty_string_count(
table=self.dataset_name,
field=self.field_name,
field=f'"{self.field_name}"'
if isinstance(self.data_source, OracleDataSource)
else self.field_name,
filters=self.where_filter if self.where_filter is not None else None,
)
elif isinstance(self.data_source, SearchIndexDataSource):
Expand All @@ -76,9 +96,15 @@ def _generate_metric_value(self, **kwargs) -> Union[float, int]:
class PercentageEmptyStringValidation(Validation):
def _generate_metric_value(self, **kwargs) -> Union[float, int]:
if isinstance(self.data_source, SQLDataSource):
if isinstance(self.data_source, OracleDataSource) and self.where_filter:
self.where_filter = re.sub(
r"(\b[a-zA-Z_]+\b)(?=\s*[=<>])", r'"\1"', self.where_filter
)
return self.data_source.query_get_empty_string_percentage(
table=self.dataset_name,
field=self.field_name,
field=f'"{self.field_name}"'
if isinstance(self.data_source, OracleDataSource)
else self.field_name,
filters=self.where_filter if self.where_filter is not None else None,
)
elif isinstance(self.data_source, SearchIndexDataSource):
Expand Down
Loading
Loading