dataframe_operators.py

from business_rules.operators import BaseType, type_operator
from typing import Union, Any, List, Tuple, Sequence
from business_rules.fields import FIELD_DATAFRAME
from cdisc_rules_engine.check_operators.helpers import (
    flatten_list,
    is_valid_date,
    vectorized_is_valid,
    vectorized_is_valid_duration,
    vectorized_is_complete_date,
    vectorized_get_dict_key,
    vectorized_is_in,
    vectorized_case_insensitive_is_in,
    apply_regex,
    vectorized_compare_dates,
    apply_rounding,
    is_in,
)
from cdisc_rules_engine.enums.dataset_title_case import DatasetTitleCase
from cdisc_rules_engine.constants import NULL_FLAVORS
from cdisc_rules_engine.utilities.utils import (
    dates_overlap,
    parse_date,
    custom_str_conversion,
)
import numpy as np
import dask.dataframe as dd
import pandas as pd
import re
import operator
from titlecase import titlecase
from uuid import uuid4
from cdisc_rules_engine.models.dataset.dask_dataset import DaskDataset
from cdisc_rules_engine.models.dataset.dataset_interface import DatasetInterface
from pandas.api.types import is_integer_dtype
from cdisc_rules_engine.services import logger
from functools import wraps
import traceback


def log_operator_execution(func):
    @wraps(func)
    def wrapper(self, other_value, *args, **kwargs):
        try:
            logger.info(f"Starting check operator: {func.__name__}")
            result = func(self, other_value)
            logger.info(f"Completed check operator: {func.__name__}")
            return result
        except Exception as e:
            logger.error(
                f"Error in {func.__name__}: {str(e)}, "
                f"traceback: {traceback.format_exc()}"
            )
            error_message = str(e)
            if isinstance(e, TypeError) and (
                "NoneType" in error_message
                or "None" in error_message
                or any(
                    phrase in error_message
                    for phrase in [
                        "NoneType",
                        "object is None",
                        "'NoneType'",
                        "None has no attribute",
                        "unsupported operand type",
                        "bad operand type",
                        "object is not",
                        "cannot be None",
                    ]
                )
            ):
                return None
            else:
                raise

    return wrapper


class DataframeType(BaseType):

    name = "dataframe"

    def __init__(self, data):
        self.value: DatasetInterface = data["value"]
        self.column_prefix_map = data.get("column_prefix_map", {})
        self.value_level_metadata = data.get("value_level_metadata", [])
        self.column_codelist_map = data.get("column_codelist_map", {})
        self.codelist_term_maps = data.get("codelist_term_maps", [])

    def _assert_valid_value_and_cast(self, value):
        if isinstance(value, dict):
            value = self._resolve_prefixes(value)
        return value

    def _regex_str_conversion(self, x):
        """Convert value to string for regex operations.
        Only converts non-null values. Returns NaN/None as-is.
        """
        if pd.notna(x):
            if isinstance(x, int):
                return str(x).strip()
            elif isinstance(x, float):
                return f"{x:.0f}" if x.is_integer() else str(x).strip()
        return x

    def convert_string_data_to_lower(self, data):
        if self.value.is_series(data):
            data = data.str.lower()
        else:
            data = data.lower()
        return data

    def _is_null_or_empty(self, value):
        try:
            result = pd.isna(value) | (value == "") | (value is None)
            if hasattr(result, "all"):
                return result.all()  # True only if ALL elements are null/empty
            return result
        except (ValueError, TypeError):
            return pd.isna(value) or value is None or value == ""

    def replace_prefix(self, value: str) -> Union[str, Any]:
        if isinstance(value, str):
            for prefix, replacement in self.column_prefix_map.items():
                if value.startswith(prefix) and replacement is not None:
                    return value.replace(prefix, replacement, 1)
        return value

    def replace_all_prefixes(self, values: List[str]) -> List[str]:
        for i in range(len(values)):
            values[i] = self.replace_prefix(values[i])
        return values

    def _resolve_prefixes(self, other_value: dict) -> dict:
        other_value = other_value.copy()
        for key, value in other_value.items():
            if isinstance(value, str):
                other_value[key] = self.replace_prefix(value)
            elif isinstance(value, list):
                other_value[key] = self.replace_all_prefixes(value)
        return other_value

    def _normalize_grouping_columns(
        self, within: Union[str, Sequence[str]]
    ) -> List[str]:
        if within is None:
            raise ValueError("within parameter is required")
        columns = list(within) if isinstance(within, (list, tuple)) else [within]
        if not columns or any(
            not isinstance(column, str) or not column for column in columns
        ):
            raise ValueError("within must contain valid column names")
        return list(dict.fromkeys(columns))

    def get_comparator_data(self, comparator, value_is_literal: bool = False):
        if value_is_literal:
            return comparator
        else:
            return self.value.get(comparator, comparator)

    @log_operator_execution
    def is_column_of_iterables(self, column):
        if not self.value.is_series(column):
            return False
        non_null_values = column[column.notna()]
        return len(non_null_values) > 0 and all(
            isinstance(val, (list, set)) for val in non_null_values
        )

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def exists(self, other_value):
        target_column = other_value.get("target")

        def check_row(row):
            return any(target_column in item for item in row if isinstance(item, list))

        column_exists = target_column in self.value.columns
        if column_exists:
            return self.value.convert_to_series([True] * len(self.value))
        else:
            exists_in_nested = self.value.apply(check_row, axis=1).any()
            return self.value.convert_to_series([exists_in_nested] * len(self.value))

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def not_exists(self, other_value):
        return ~self.exists(other_value)

    def _check_equality(
        self,
        row,
        target,
        comparator,
        value_is_literal: bool = False,
        value_is_reference: bool = False,
        case_insensitive: bool = False,
        type_insensitive: bool = False,
        round_values: bool = False,
    ) -> bool:
        """
        Equality checks work slightly differently for clinical datasets.
        See truth table below:
        Operator       --A         --B         Outcome
        equal_to       "" or null  "" or null  False
        equal_to       "" or null  Populated   False
        equal_to       Populated   "" or null  False
        equal_to       Populated   Populated   A == B
        """
        if value_is_reference:
            dynamic_column_name = row[comparator]
            comparison_data = row[dynamic_column_name]
        else:
            comparison_data = (
                comparator
                if comparator not in row or value_is_literal
                else row[comparator]
            )
        both_null = self._is_null_or_empty(comparison_data) & self._is_null_or_empty(
            row[target]
        )
        if both_null:
            return False
        target_val = row[target]
        comparison_val = comparison_data
        if round_values:
            target_val, comparison_val = apply_rounding(target_val, comparison_val)
        if type_insensitive:
            target_val = custom_str_conversion(target_val)
            comparison_val = custom_str_conversion(comparison_val)
        if case_insensitive:
            target_val = target_val.lower() if isinstance(target_val, str) and target_val else None
            comparison_val = comparison_val.lower() if isinstance(comparison_val, str) and comparison_val else None
            return target_val == comparison_val
        return target_val == comparison_val

    def _check_inequality(
        self,
        row,
        target,
        comparator,
        value_is_literal: bool = False,
        value_is_reference: bool = False,
        case_insensitive: bool = False,
        type_insensitive: bool = False,
        round_values: bool = False,
    ) -> bool:
        """
        Equality checks work slightly differently for clinical datasets.
        See truth table below:
        Operator       --A         --B         Outcome
        not_equal_to   "" or null  "" or null  False
        not_equal_to   "" or null  Populated   True
        not_equal_to   Populated   "" or null  True
        not_equal_to   Populated   Populated   A != B
        """
        if value_is_reference:
            dynamic_column_name = row[comparator]
            comparison_data = row[dynamic_column_name]
        else:
            comparison_data = (
                comparator
                if comparator not in row or value_is_literal
                else row[comparator]
            )
        both_null = self._is_null_or_empty(comparison_data) & self._is_null_or_empty(
            row[target]
        )
        if both_null:
            return False
        target_val = row[target]
        comparison_val = comparison_data
        if round_values:
            target_val, comparison_val = apply_rounding(target_val, comparison_val)
        if type_insensitive:
            target_val = custom_str_conversion(target_val)
            comparison_val = custom_str_conversion(comparison_val)
        if case_insensitive:
            target_val = target_val.lower() if isinstance(target_val, str) and target_val else None
            comparison_val = comparison_val.lower() if isinstance(comparison_val, str) and comparison_val else None
            return target_val != comparison_val
        return target_val != comparison_val

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def equal_to(self, other_value):
        target = other_value.get("target")
        value_is_literal = other_value.get("value_is_literal", False)
        value_is_reference = other_value.get("value_is_reference", False)
        type_insensitive = other_value.get("type_insensitive", False)
        round_values = other_value.get("round_values", False)
        comparator = other_value.get("comparator")
        return self.value.apply(
            lambda row: self._check_equality(
                row,
                target,
                comparator,
                value_is_literal,
                value_is_reference,
                type_insensitive=type_insensitive,
                round_values=round_values,
            ),
            axis=1,
            meta=(None, "bool"),
        ).reset_index(drop=True)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def equal_to_case_insensitive(self, other_value):
        target = other_value.get("target")
        value_is_literal = other_value.get("value_is_literal", False)
        value_is_reference = other_value.get("value_is_reference", False)
        type_insensitive = other_value.get("type_insensitive", False)
        round_values = other_value.get("round_values", False)
        comparator = other_value.get("comparator")

        return self.value.apply(
            lambda row: self._check_equality(
                row,
                target,
                comparator,
                value_is_literal,
                value_is_reference,
                case_insensitive=True,
                type_insensitive=type_insensitive,
                round_values=round_values,
            ),
            axis=1,
            meta=(None, "bool"),
        )

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def not_equal_to_case_insensitive(self, other_value):
        target = other_value.get("target")
        value_is_literal = other_value.get("value_is_literal", False)
        value_is_reference = other_value.get("value_is_reference", False)
        type_insensitive = other_value.get("type_insensitive", False)
        round_values = other_value.get("round_values", False)
        comparator = other_value.get("comparator")

        return self.value.apply(
            lambda row: self._check_inequality(
                row,
                target,
                comparator,
                value_is_literal,
                value_is_reference,
                case_insensitive=True,
                type_insensitive=type_insensitive,
                round_values=round_values,
            ),
            axis=1,
        )

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def not_equal_to(self, other_value):
        target = other_value.get("target")
        value_is_literal = other_value.get("value_is_literal", False)
        value_is_reference = other_value.get("value_is_reference", False)
        type_insensitive = other_value.get("type_insensitive", False)
        round_values = other_value.get("round_values", False)
        comparator = other_value.get("comparator")

        return self.value.apply(
            lambda row: self._check_inequality(
                row,
                target,
                comparator,
                value_is_literal,
                value_is_reference,
                type_insensitive=type_insensitive,
                round_values=round_values,
            ),
            axis=1,
            meta=(None, "bool"),
        ).reset_index(drop=True)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def suffix_equal_to(self, other_value: dict):
        """
        Checks if target suffix is equal to comparator.
        """
        target: str = other_value.get("target")
        value_is_literal: bool = other_value.get("value_is_literal", False)
        comparator: Union[str, Any] = other_value.get("comparator")
        comparison_data = self.get_comparator_data(comparator, value_is_literal)
        suffix: int = other_value.get("suffix")
        return self._check_equality_of_string_part(
            target, comparison_data, "suffix", suffix
        )

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def suffix_not_equal_to(self, other_value: dict):
        """
        Checks if target suffix is not equal to comparator.
        """
        return ~self.suffix_equal_to(other_value)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def prefix_equal_to(self, other_value: dict):
        """
        Checks if target prefix is equal to comparator.
        """
        target: str = other_value.get("target")
        value_is_literal: bool = other_value.get("value_is_literal", False)
        comparator: Union[str, Any] = other_value.get("comparator")
        if comparator == "DOMAIN":
            comparison_data = self.column_prefix_map["--"]
        else:
            comparison_data = self.get_comparator_data(comparator, value_is_literal)
        prefix: int = other_value.get("prefix")
        return self._check_equality_of_string_part(
            target, comparison_data, "prefix", prefix
        )

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def prefix_not_equal_to(self, other_value: dict):
        """
        Checks if target prefix is not equal to comparator.
        """
        return ~self.prefix_equal_to(other_value)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def prefix_is_contained_by(self, other_value: dict):
        """
        Checks if target prefix is contained by the comparator.
        """
        target: str = other_value.get("target")
        value_is_literal: bool = other_value.get("value_is_literal", False)
        comparator: Union[str, Any] = other_value.get("comparator")
        comparison_data = self.get_comparator_data(comparator, value_is_literal)
        prefix_length: int = other_value.get("prefix")
        series_to_validate = self._get_string_part_series(
            "prefix", prefix_length, target
        )
        return self._value_is_contained_by(series_to_validate, comparison_data)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def prefix_is_not_contained_by(self, other_value: dict):
        return ~self.prefix_is_contained_by(other_value)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def suffix_is_contained_by(self, other_value: dict):
        """
        Checks if target prefix is equal to comparator.
        """
        target: str = other_value.get("target")
        value_is_literal: bool = other_value.get("value_is_literal", False)
        comparator: Union[str, Any] = other_value.get("comparator")
        comparison_data = self.get_comparator_data(comparator, value_is_literal)
        suffix_length: int = other_value.get("suffix")
        series_to_validate = self._get_string_part_series(
            "suffix", suffix_length, target
        )
        return self._value_is_contained_by(series_to_validate, comparison_data)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def suffix_is_not_contained_by(self, other_value: dict):
        return ~self.suffix_is_contained_by(other_value)

    def _get_string_part_series(self, part_to_validate: str, length: int, target: str):
        if not self.value[target].apply(type).eq(str).all():
            raise ValueError("The operator can't be used with non-string values")

        if part_to_validate == "suffix":
            series_to_validate = self.value[target].str.slice(-length)
        elif part_to_validate == "prefix":
            series_to_validate = self.value[target].str.slice(stop=length)
        else:
            raise ValueError(f"Invalid part to validate: {part_to_validate}. \
                    Valid values are: suffix, prefix")
        series_to_validate = series_to_validate.mask(pd.isna(self.value[target]))
        return series_to_validate

    def _value_is_contained_by(self, series, comparison_data):
        if self.is_column_of_iterables(comparison_data):
            results = vectorized_is_in(series, comparison_data)
        else:
            results = series.isin(comparison_data)
        return self.value.convert_to_series(results)

    def _check_equality_of_string_part(
        self,
        target: str,
        comparison_data,
        part_to_validate: str,
        length: int,
    ):
        """
        Checks if the given string part is equal to comparison data.
        """
        series_to_validate = self._get_string_part_series(
            part_to_validate, length, target
        )
        return series_to_validate.eq(comparison_data).astype(bool)

    def _where_less_than(self, target, comparison):
        return np.where(target < comparison, True, False)

    def _where_greater_than(self, target, comparison):
        return np.where(target > comparison, True, False)

    def _where_less_than_or_equal_to(self, target, comparison):
        return np.where(target <= comparison, True, False)

    def _where_greater_than_or_equal_to(self, target, comparison):
        return np.where(target >= comparison, True, False)

    def _to_numeric(self, target, **kwargs):
        return pd.to_numeric(target, **kwargs)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def less_than(self, other_value):
        target = other_value.get("target")
        value_is_literal = other_value.get("value_is_literal", False)
        comparator = other_value.get("comparator")
        comparison_data = self.get_comparator_data(comparator, value_is_literal)
        target_column = self._to_numeric(self.value[target], errors="coerce")
        if self.value.is_series(comparison_data):
            comparison_data = self._to_numeric(comparison_data, errors="coerce")
        results = self._where_less_than(target_column, comparison_data)
        return self.value.convert_to_series(results)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def less_than_or_equal_to(self, other_value):
        target = other_value.get("target")
        value_is_literal = other_value.get("value_is_literal", False)
        comparator = other_value.get("comparator")
        comparison_data = self.get_comparator_data(comparator, value_is_literal)
        target_column = self._to_numeric(self.value[target], errors="coerce")
        if self.value.is_series(comparison_data):
            comparison_data = self._to_numeric(comparison_data, errors="coerce")
        results = self._where_less_than_or_equal_to(target_column, comparison_data)
        return self.value.convert_to_series(results)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def greater_than_or_equal_to(self, other_value):
        target = other_value.get("target")
        value_is_literal = other_value.get("value_is_literal", False)
        comparator = other_value.get("comparator")
        comparison_data = self.get_comparator_data(comparator, value_is_literal)
        target_column = self._to_numeric(self.value[target], errors="coerce")
        if self.value.is_series(comparison_data):
            comparison_data = self._to_numeric(comparison_data, errors="coerce")
        results = self._where_greater_than_or_equal_to(target_column, comparison_data)
        return self.value.convert_to_series(results)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def greater_than(self, other_value):
        target = other_value.get("target")
        value_is_literal = other_value.get("value_is_literal", False)
        comparator = other_value.get("comparator")
        comparison_data = self.get_comparator_data(comparator, value_is_literal)
        target_column = self._to_numeric(self.value[target], errors="coerce")
        if self.value.is_series(comparison_data):
            comparison_data = self._to_numeric(comparison_data, errors="coerce")
        results = self._where_greater_than(target_column, comparison_data)
        return self.value.convert_to_series(results)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def contains(self, other_value):
        target = other_value.get("target")
        value_is_literal = other_value.get("value_is_literal", False)
        comparator = other_value.get("comparator")
        comparison_data = self.get_comparator_data(comparator, value_is_literal)
        if self.is_column_of_iterables(self.value[target]) or isinstance(
            comparison_data, str
        ):
            results = vectorized_is_in(comparison_data, self.value[target])
        elif self.value.is_series(comparison_data):
            results = self._series_is_in(self.value[target], comparison_data)
        else:
            # Handles numeric case. This case should never occur
            results = np.where(self.value[target] == comparison_data, True, False)
        return self.value.convert_to_series(results)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def does_not_contain(self, other_value):
        return ~self.contains(other_value)

    def _series_is_in(self, target, comparison_data):
        return np.where(comparison_data.isin(target), True, False)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def contains_case_insensitive(self, other_value):
        target = other_value.get("target")
        value_is_literal = other_value.get("value_is_literal", False)
        comparator = other_value.get("comparator")
        comparison_data = self.get_comparator_data(comparator, value_is_literal)
        comparison_data = self.convert_string_data_to_lower(comparison_data)
        if self.is_column_of_iterables(self.value[target]):
            results = vectorized_case_insensitive_is_in(
                comparison_data, self.value[target]
            )
        elif self.value.is_series(comparison_data):
            # column vs column case: perform element-wise case-insensitive substring check
            target_series = self.convert_string_data_to_lower(self.value[target])
            comparison_series = self.convert_string_data_to_lower(comparison_data)
            results = target_series.combine(
                comparison_series,
                lambda t, c: (
                    vectorized_case_insensitive_is_in(c, [t])[0]
                    if pd.notna(t) and pd.notna(c)
                    else False
                ),
            )
        else:
            results = vectorized_case_insensitive_is_in(
                comparison_data.lower(), self.value[target]
            )
        return self.value.convert_to_series(results)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def does_not_contain_case_insensitive(self, other_value):
        return ~self.contains_case_insensitive(other_value)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def is_contained_by(self, other_value):
        target = other_value.get("target")
        value_is_literal = other_value.get("value_is_literal", False)
        comparator = other_value.get("comparator")
        comparison_data = self.get_comparator_data(comparator, value_is_literal)
        target_data = self.value[target]
        if self.is_column_of_iterables(target_data):
            results = []
            for i in range(len(target_data)):
                target_val = target_data.iloc[i]
                comp_val = (
                    comparison_data.iloc[i]
                    if hasattr(comparison_data, "iloc")
                    else comparison_data
                )
                if isinstance(target_val, list):
                    result = any(is_in(item, comp_val) for item in target_val)
                else:
                    result = is_in(target_val, comp_val)
                results.append(result)
            results = pd.Series(results)
        elif self.is_column_of_iterables(comparison_data):
            results = vectorized_is_in(target_data, comparison_data)
        else:
            if isinstance(comparison_data, pd.Series):
                comparison_data = comparison_data.apply(
                    lambda x: list(x) if isinstance(x, set) else x
                )
            elif isinstance(comparison_data, set):
                comparison_data = list(comparison_data)
            results = target_data.isin(comparison_data)
        return self.value.convert_to_series(results)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def is_not_contained_by(self, other_value):
        return ~self.is_contained_by(other_value)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def is_contained_by_case_insensitive(self, other_value):
        target = other_value.get("target")
        comparator = other_value.get("comparator", [])
        value_is_literal = other_value.get("value_is_literal", False)
        if isinstance(comparator, list):
            comparator = [val.lower() for val in comparator]
        comparison_data = self.get_comparator_data(comparator, value_is_literal)
        if self.is_column_of_iterables(comparison_data):
            results = vectorized_case_insensitive_is_in(
                self.value[target].str.lower(), comparison_data
            )
            return self.value.convert_to_series(results)
        elif self.value.is_series(comparison_data):
            results = self.value[target].str.lower().isin(comparison_data.str.lower())
        else:
            results = self.value[target].str.lower().isin(comparison_data)
        return results

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def is_not_contained_by_case_insensitive(self, other_value):
        return ~self.is_contained_by_case_insensitive(other_value)

    @staticmethod
    def _map_regex(series, func):
        # pandas 3 returns nullable BooleanDtype from .map(); normalize to numpy
        # bool so ~ and & behave identically for both positive and negated callers.
        return series.map(func, na_action="ignore").fillna(False).astype(bool)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def prefix_matches_regex(self, other_value):
        target = other_value.get("target")
        comparator = other_value.get("comparator")
        prefix = other_value.get("prefix")
        converted_strings = self.value[target].map(
            lambda x: self._regex_str_conversion(x)
        )
        return converted_strings.notna() & self._map_regex(
            converted_strings.astype(str),
            lambda x: re.search(comparator, x[:prefix]) is not None,
        )

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def not_prefix_matches_regex(self, other_value):
        target = other_value.get("target")
        comparator = other_value.get("comparator")
        prefix = other_value.get("prefix")
        converted_strings = self.value[target].map(
            lambda x: self._regex_str_conversion(x)
        )
        return converted_strings.notna() & ~self._map_regex(
            converted_strings.astype(str),
            lambda x: re.search(comparator, x[:prefix]) is not None,
        )

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def suffix_matches_regex(self, other_value):
        target = other_value.get("target")
        comparator = other_value.get("comparator")
        suffix = other_value.get("suffix")
        converted_strings = self.value[target].map(
            lambda x: self._regex_str_conversion(x)
        )
        return converted_strings.notna() & self._map_regex(
            converted_strings.astype(str),
            lambda x: re.search(comparator, x[-suffix:]) is not None,
        )

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def not_suffix_matches_regex(self, other_value):
        target = other_value.get("target")
        comparator = other_value.get("comparator")
        suffix = other_value.get("suffix")
        converted_strings = self.value[target].map(
            lambda x: self._regex_str_conversion(x)
        )
        return converted_strings.notna() & ~self._map_regex(
            converted_strings.astype(str),
            lambda x: re.search(comparator, x[-suffix:]) is not None,
        )

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def matches_regex(self, other_value):
        target = other_value.get("target")
        comparator = other_value.get("comparator")
        converted_strings = self.value[target].map(
            lambda x: self._regex_str_conversion(x)
        )
        return converted_strings.notna() & self._map_regex(
            converted_strings.astype(str),
            lambda x: re.match(comparator, x) is not None,
        )

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def not_matches_regex(self, other_value):
        target = other_value.get("target")
        comparator = other_value.get("comparator")
        converted_strings = self.value[target].map(
            lambda x: self._regex_str_conversion(x)
        )
        return converted_strings.notna() & ~self._map_regex(
            converted_strings.astype(str),
            lambda x: re.match(comparator, x) is not None,
        )

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def equals_string_part(self, other_value):
        """
        Checks that the values in the target column
        equal the result of parsing the value in the comparison
        column with a regex
        """
        target = other_value.get("target")
        comparator = other_value.get("comparator")
        regex = other_value.get("regex")
        value_is_literal: bool = other_value.get("value_is_literal", False)
        comparison_data = self.get_comparator_data(comparator, value_is_literal)
        if isinstance(comparison_data, str):
            parsed_data = apply_regex(regex, comparison_data)
        else:
            parsed_data = comparison_data.str.findall(regex).str[0]
        parsed_id = str(uuid4())
        self.value[parsed_id] = parsed_data
        return self.value.apply(
            lambda row: self._check_equality(
                row, target, parsed_id, value_is_literal=False
            ),
            axis=1,
        )

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def does_not_equal_string_part(self, other_value):
        return ~self.equals_string_part(other_value)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def starts_with(self, other_value):
        target = other_value.get("target")
        comparator = other_value.get("comparator")
        value_is_literal: bool = other_value.get("value_is_literal", False)
        comparison_data = self.get_comparator_data(comparator, value_is_literal)
        if self.value.is_series(comparison_data):
            # need to convert series to tuple to make startswith operator work correctly
            comparison_data: Tuple[str] = tuple(comparison_data)
        results = self.value[target].str.startswith(comparison_data)
        return results

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def ends_with(self, other_value):
        target = other_value.get("target")
        comparator = other_value.get("comparator")
        value_is_literal: bool = other_value.get("value_is_literal", False)
        comparison_data = self.get_comparator_data(comparator, value_is_literal)
        if self.value.is_series(comparison_data):
            # need to convert series to tuple to make endswith operator work correctly
            comparison_data: Tuple[str] = tuple(comparison_data)
        results = self.value[target].str.endswith(comparison_data)
        return results

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def has_equal_length(self, other_value: dict):
        """
        Checks that the target length is the same as comparator.
        If comparing two columns (value_is_literal is False), the operator
        compares lengths of values in these columns.
        """
        target = other_value.get("target")
        comparator = other_value.get("comparator")
        value_is_literal: bool = other_value.get("value_is_literal", False)
        comparison_data = self.get_comparator_data(comparator, value_is_literal)
        if self.value.is_series(comparison_data):
            if is_integer_dtype(comparison_data):
                results = self.value[target].str.len().eq(comparison_data).astype(bool)
            else:
                results = (
                    self.value[target]
                    .str.len()
                    .eq(comparison_data.str.len())
                    .astype(bool)
                )
        else:
            results = self.value[target].str.len().eq(comparator).astype(bool)
        return results

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def has_not_equal_length(self, other_value: dict):
        return ~self.has_equal_length(other_value)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def longer_than(self, other_value: dict):
        """
        Checks if the target is longer than the comparator.
        If comparing two columns (value_is_literal is False), the operator
        compares lengths of values in these columns.
        """
        target = other_value.get("target")
        comparator = other_value.get("comparator")
        value_is_literal: bool = other_value.get("value_is_literal", False)
        comparison_data = self.get_comparator_data(comparator, value_is_literal)
        if self.value.is_series(comparison_data):
            if is_integer_dtype(comparison_data):
                results = self.value[target].str.len().gt(comparison_data)
            else:
                results = self.value[target].str.len().gt(comparison_data.str.len())
        else:
            results = self.value[target].str.len().gt(comparison_data)
        return results

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def longer_than_or_equal_to(self, other_value: dict):
        target = other_value.get("target")
        comparator = other_value.get("comparator")
        value_is_literal: bool = other_value.get("value_is_literal", False)
        comparison_data = self.get_comparator_data(comparator, value_is_literal)
        if self.value.is_series(comparison_data):
            if is_integer_dtype(comparison_data):
                results = self.value[target].str.len().ge(comparison_data)
            else:
                results = self.value[target].str.len().ge(comparison_data.str.len())
        else:
            results = self.value[target].str.len().ge(comparator)
        return results

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def shorter_than(self, other_value: dict):
        return ~self.longer_than_or_equal_to(other_value)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def shorter_than_or_equal_to(self, other_value: dict):
        return ~self.longer_than(other_value)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def split_parts_have_equal_length(self, other_value: dict):
        """
        Splits string values by a separator and checks if both parts have equal length.
        """
        target = other_value.get("target")
        separator = other_value.get("separator", "/")

        target_series = self.value[target]
        is_null_or_empty = target_series.isna() | (target_series == "")
        target_str = target_series.astype(str)
        split_series = target_str.str.split(separator, expand=False)

        def validate_split(parts):
            if not isinstance(parts, list):
                return True
            if len(parts) == 1:
                return True
            if len(parts) == 2:
                return len(parts[0]) == len(parts[1])
            return False

        results = split_series.apply(validate_split)
        results = results | is_null_or_empty

        return results

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def split_parts_have_unequal_length(self, other_value: dict):
        """
        Complement of split_parts_have_equal_length.
        """
        return ~self.split_parts_have_equal_length(other_value)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def empty(self, other_value: dict):
        target = other_value.get("target")
        series = self.value[target]

        def check_empty(x):
            return isinstance(x, (set, list, dict)) and len(x) == 0

        if hasattr(series, "map_partitions"):
            is_empty_collection = series.map(check_empty, meta=("x", "bool"))
        else:
            is_empty_collection = series.map(check_empty)
        results = np.where(
            self.value[target].isin(NULL_FLAVORS)
            | pd.isna(self.value[target])
            | is_empty_collection,
            True,
            False,
        )
        return self.value.convert_to_series(results)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def empty_within_except_last_row(self, other_value: dict):
        target = other_value.get("target")
        comparator = other_value.get("comparator")
        order_by_column: str = other_value.get("ordering")
        # group all targets by comparator
        if order_by_column:
            ordered_df = self.value.sort_values(by=[comparator, order_by_column])
        else:
            ordered_df = self.value.sort_values(by=[comparator])
        grouped_target = ordered_df.groupby(comparator)[target]
        # validate all targets except the last one
        results = grouped_target.apply(lambda x: x[:-1]).apply(
            lambda x: (
                pd.isna(x).all()
                if isinstance(x, (pd.Series, list))
                else (x in NULL_FLAVORS or pd.isna(x))
            )
        )
        if isinstance(self.value, DaskDataset) and self.value.is_series(results):
            results = results.compute()
        # return values with corresponding indexes from results
        return pd.Series(results.reset_index(level=0, drop=True))

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def non_empty(self, other_value: dict):
        return ~self.empty(other_value)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def non_empty_within_except_last_row(self, other_value: dict):
        target = other_value.get("target")
        comparator = other_value.get("comparator")
        order_by_column: str = other_value.get("ordering")
        # group all targets by comparator
        if order_by_column:
            ordered_df = self.value.sort_values(by=[comparator, order_by_column])
        else:
            ordered_df = self.value.sort_values(by=[comparator])
        grouped_target = ordered_df.groupby(comparator)[target]
        # validate all targets except the last one
        results = ~grouped_target.apply(lambda x: x[:-1]).apply(
            lambda x: (
                pd.isna(x).all()
                if isinstance(x, (pd.Series, list))
                else (x in NULL_FLAVORS or pd.isna(x))
            )
        )
        if isinstance(self.value, DaskDataset) and self.value.is_series(results):
            computed_results = results.compute()
            return computed_results.reset_index(level=0, drop=True)

        # return values with corresponding indexes from results
        return pd.Series(results.reset_index(level=0, drop=True))

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def contains_all(self, other_value: dict):
        target = other_value.get("target")
        value_is_literal: bool = other_value.get("value_is_literal", False)
        comparator = other_value.get("comparator")
        if self.is_column_of_iterables(
            self.value[target]
        ) and self.is_column_of_iterables(self.value[comparator]):
            comparison_data = self.get_comparator_data(comparator, value_is_literal)
            results = []
            for i in range(len(self.value[target])):
                target_val = self.value[target].iloc[i]
                comp_val = comparison_data.iloc[i]
                results.append(all(is_in(item, target_val) for item in comp_val))
        else:
            if isinstance(comparator, list):
                # get column as array of values
                values = flatten_list(self.value, comparator)
            else:
                values = self.value[comparator].unique()
            results = set(values).issubset(set(self.value[target].unique()))
        return self.value.convert_to_series(results)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def not_contains_all(self, other_value: dict):
        return ~self.contains_all(other_value)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def invalid_date(self, other_value):
        target = other_value.get("target")
        results = ~vectorized_is_valid(self.value[target])
        return self.value.convert_to_series(results)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def invalid_duration(self, other_value):
        target = other_value.get("target")
        if other_value.get("negative") is False:
            results = ~vectorized_is_valid_duration(self.value[target], False)
        else:
            results = ~vectorized_is_valid_duration(self.value[target], True)
        return self.value.convert_to_series(results)

    def date_comparison(self, other_value, operator):
        target = other_value.get("target")
        comparator = other_value.get("comparator")
        value_is_literal: bool = other_value.get("value_is_literal", False)
        comparison_data = self.get_comparator_data(comparator, value_is_literal)
        component = other_value.get("date_component")
        results = np.where(
            vectorized_compare_dates(
                component, self.value[target], comparison_data, operator
            ),
            True,
            False,
        )
        return self.value.convert_to_series(results)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def date_equal_to(self, other_value):
        return self.date_comparison(other_value, operator.eq)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def date_not_equal_to(self, other_value):
        return self.date_comparison(other_value, operator.ne)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def date_less_than(self, other_value):
        return self.date_comparison(other_value, operator.lt)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def date_less_than_or_equal_to(self, other_value):
        return self.date_comparison(other_value, operator.le)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def date_greater_than_or_equal_to(self, other_value):
        return self.date_comparison(other_value, operator.ge)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def date_greater_than(self, other_value):
        return self.date_comparison(other_value, operator.gt)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def is_incomplete_date(self, other_value):
        return ~self.is_complete_date(other_value)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def is_complete_date(self, other_value):
        target = other_value.get("target")
        results = vectorized_is_complete_date(self.value[target])
        return self.value.convert_to_series(results)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def is_inconsistent_across_dataset(self, other_value):
        target = other_value.get("target")
        comparator = other_value.get("comparator")
        regex = other_value.get("regex")
        if isinstance(regex, list) and regex:
            regex = regex[0]

        grouping_cols = []
        if isinstance(comparator, str):
            if comparator in self.value.columns:
                grouping_cols.append(comparator)
        else:
            for col in comparator:
                if col in self.value.columns:
                    grouping_cols.append(col)
        df_check = self.value[grouping_cols + [target]].copy()
        df_check = df_check.fillna("_NaN_")
        if regex:
            try:
                pattern = re.compile(regex)
            except re.error:
                raise ValueError(
                    f"Invalid regex: {regex}. Remove parameter or fix the regex."
                )
            if pattern.groups == 0:
                regex = f"({regex})"
            extracted = df_check[target].astype(str).str.extract(regex, expand=True)[0]
            df_check[target] = extracted.fillna(df_check[target])
        results = self._check_inconsistency(df_check, grouping_cols, target)
        return results

    @staticmethod
    def _check_inconsistency(df_check, grouping_cols: list[Any], target):
        results = pd.Series(False, index=df_check.index)
        for name, group in df_check.groupby(grouping_cols, dropna=False):
            if group[target].nunique() > 1:
                value_counts = group[target].value_counts()
                max_count = value_counts.max()
                # if same amount of inconsistency values, flag all
                most_common_values = value_counts[value_counts == max_count]
                if len(most_common_values) > 1:
                    results[group.index] = True
                else:
                    most_common_value = most_common_values.index[0]
                    minority_rows = group[group[target] != most_common_value]
                    results[minority_rows.index] = True
        return results

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def is_unique_set(self, other_value):
        target = other_value.get("target")
        comparator = other_value.get("comparator")
        regex_pattern = other_value.get("regex")
        values = [target, comparator]
        target_data = flatten_list(self.value, values)
        target_names = []
        for target_name in target_data:
            if target_name in self.value.columns:
                target_names.append(target_name)
        target_names = list(set(target_names))
        df_group = self.value[target_names].copy()
        if regex_pattern:
            for col in df_group.columns:
                sample_value = (
                    df_group[col].dropna().iloc[0]
                    if not df_group[col].dropna().empty
                    else None
                )
                if (
                    sample_value
                    and isinstance(sample_value, str)
                    and re.match(regex_pattern, sample_value)
                ):
                    df_group[col] = df_group[col].apply(
                        lambda x: (
                            apply_regex(regex_pattern, x)
                            if isinstance(x, str) and x
                            else x
                        )
                    )
        df_group = df_group.fillna("_NaN_")
        group_sizes = df_group.groupby(target_names).size()
        counts = df_group.apply(tuple, axis=1).map(group_sizes)
        results = np.where(counts <= 1, True, False)
        return self.value.convert_to_series(results)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def is_not_unique_set(self, other_value):
        return ~self.is_unique_set(other_value)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def is_not_unique_relationship(self, other_value):
        """
        Validates one-to-one relationship between two columns (target and comparator)
        within a dataset. One-to-one means that a columns values can be duplicated
        but it must always corresponds to one value of comparator and vice versa.
        A violation occurs when a NON-NULL value in either column maps to multiple
        different values in the other column.
        """
        target = other_value.get("target")
        comparator = other_value.get("comparator")
        if isinstance(comparator, list):
            columns = [target] + comparator
        else:
            columns = [target, comparator]

        df_subset = self.value[columns].dropna(how="all")
        df_without_duplicates = df_subset.drop_duplicates()
        violated_targets, violated_comparators = self._find_relationship_violations(
            df_without_duplicates, target, comparator
        )

        # flag violations from target and comparator
        result = self.value.convert_to_series([False] * len(self.value))
        if violated_targets:
            clean_targets = {
                v for v in violated_targets if pd.notna(v) and v != "" and v is not None
            }
            if clean_targets:
                result = result | self.value[target].isin(clean_targets)
        if violated_comparators:
            clean_comparators = {
                v
                for v in violated_comparators
                if pd.notna(v) and v != "" and v is not None
            }
            if clean_comparators:
                if isinstance(comparator, list):
                    # For multi-column comparators, match on tuple combinations
                    for comp_tuple in clean_comparators:
                        mask = self.value.convert_to_series([True] * len(self.value))
                        for i, col in enumerate(comparator):
                            mask = mask & (self.value[col] == comp_tuple[i])
                        result = result | mask
                else:
                    result = result | self.value[comparator].isin(clean_comparators)

        return result

    def _find_relationship_violations(self, df_without_duplicates, target, comparator):
        """
        Find all values that violate one-to-one relationship constraints.
        Returns two sets:
        - violated_targets: non-null target values that map to multiple comparators
        - violated_comparators: non-null comparator values that map to multiple targets
        """
        violated_targets = self._check_column_violations(
            df_without_duplicates, target, comparator
        )
        violated_comparators = self._check_column_violations(
            df_without_duplicates, comparator, target
        )
        return violated_targets, violated_comparators

    def _check_column_violations(self, df_without_duplicates, key_column, value_column):
        violated_keys = set()
        if isinstance(key_column, list):
            key_data = df_without_duplicates[key_column]
            unique_keys = [tuple(row) for row in key_data.drop_duplicates().values]
        else:
            unique_keys = df_without_duplicates[key_column].dropna().unique()
        for key_val in unique_keys:
            if isinstance(key_column, list):
                if any(v == "" or pd.isna(v) or v is None for v in key_val):
                    continue
                mask = pd.Series(
                    [True] * len(df_without_duplicates),
                    index=df_without_duplicates.index,
                )
                for i, col in enumerate(key_column):
                    mask = mask & (df_without_duplicates[col] == key_val[i])
                key_rows = df_without_duplicates[mask]
            else:
                if key_val == "":
                    continue
                key_rows = df_without_duplicates[
                    df_without_duplicates[key_column] == key_val
                ]

            if isinstance(value_column, list):
                value_tuples = [tuple(row) for row in key_rows[value_column].values]
                if self._has_multiple_mappings_for_tuples(value_tuples):
                    violated_keys.add(key_val)
            else:
                if self._has_multiple_mappings(key_rows[value_column]):
                    violated_keys.add(key_val)
        return violated_keys

    def _has_multiple_mappings_for_tuples(self, value_tuples):
        """
        Check if a list of tuples contains multiple different non-null tuples.
        Returns True if there are multiple non-null tuples, or at least one
        non-null tuple plus a null tuple.
        """
        unique_tuples = set()
        has_null = False
        for val_tuple in value_tuples:
            if any(pd.isna(v) or v == "" or v is None for v in val_tuple):
                has_null = True
            else:
                unique_tuples.add(val_tuple)
        return len(unique_tuples) > 1 or (len(unique_tuples) >= 1 and has_null)

    def _has_multiple_mappings(self, values):
        """
        Check if a series of values contains multiple different values.
        Returns True if there are multiple non-null values, or at least one
        non-null value plus null.
        """
        unique_values = set()
        has_null = False
        for val in values:
            if pd.isna(val) or val == "" or val is None:
                has_null = True
            else:
                unique_values.add(val)
        return len(unique_values) > 1 or (len(unique_values) >= 1 and has_null)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def is_unique_relationship(self, other_value):
        return ~self.is_not_unique_relationship(other_value)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def is_ordered_set(self, other_value):
        target = other_value.get("target")
        value = other_value.get("comparator")
        if not isinstance(value, (str, list)):
            raise Exception("Comparator must be a String or list of Strings")
        if isinstance(value, list) and not all(isinstance(v, str) for v in value):
            raise Exception("All comparator values must be Strings")
        return self.value.is_column_sorted_within(value, target)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def is_not_ordered_set(self, other_value):
        return ~self.is_ordered_set(other_value)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def has_next_corresponding_record(self, other_value: dict):
        """
        The operator ensures that value of target in current row
        is the same as value of comparator in the next row.
        In order to achieve this, we just remove last row from target
        and first row from comparator and compare the resulting contents.
        The result is reported for target.
        """
        target = other_value.get("target")
        comparator = other_value.get("comparator")
        group_by_column: str = other_value.get("within")
        order_by_column: str = other_value.get("ordering")
        target_columns = [target, comparator, group_by_column, order_by_column]
        ordered_df = self.value[target_columns].sort_values(by=[order_by_column])
        grouped_df = ordered_df.groupby(group_by_column)
        results = grouped_df.apply(
            lambda x: self.compare_target_with_comparator_next_row(
                x, target, comparator
            )
        )
        return self.value.convert_to_series(results.explode().tolist())

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def does_not_have_next_corresponding_record(self, other_value: dict):
        return ~self.has_next_corresponding_record(other_value)

    def compare_target_with_comparator_next_row(
        self, df: DatasetInterface, target: str, comparator: str
    ):
        """
        Compares current row of a target with the next row of comparator.
        We can't
        compare last row of target with the next row of comparator
        because there is no row after the last one.
        """
        target_without_last_row = df[target].drop(df[target].tail(1).index)
        comparator_without_first_row = df[comparator].drop(df[comparator].head(1).index)
        results = np.where(
            target_without_last_row.values == comparator_without_first_row.values,
            True,
            False,
        )
        # we add True at the end as the last row of target has nothing to compare
        # so as to not raise errors or incorrect issues in the report with False or NaN
        return self.value.convert_to_series(
            [
                *results,
                True,
            ]
        ).tolist()

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def present_on_multiple_rows_within(self, other_value: dict):
        """
        The operator ensures that the target is present on multiple rows
        within a group_by column. The dataframe is grouped by a certain column
        and the check is applied to each group.
        """
        target = other_value.get("target")
        min_count: int = other_value.get("comparator") or 1
        group_by_column = other_value.get("within")
        grouped = self.value.groupby([group_by_column, target])
        meta = (target, bool)
        results = grouped.apply(
            lambda x: self.validate_series_length(x, target, min_count), meta=meta
        )
        uuid = str(uuid4())
        return self.value.merge(
            results.rename(uuid).reset_index(), on=[group_by_column, target]
        )[uuid]

    def validate_series_length(
        self, data: DatasetInterface, target: str, min_length: int
    ):
        return len(data) > min_length

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def not_present_on_multiple_rows_within(self, other_value: dict):
        return ~self.present_on_multiple_rows_within(other_value)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def inconsistent_enumerated_columns(self, other_value: dict):
        """
        Check for inconsistencies in enumerated columns of a DataFrame.

        Starting with the smallest/largest enumeration of the given variable,
        return an error if VARIABLE(N+1) is populated but VARIABLE(N) is not populated.
        Repeat for all variables belonging to the enumeration.
        Note that the initial variable will not have an index (VARIABLE) and
        the next enumerated variable has index 1 (VARIABLE1).
        """
        variable_name: str = other_value.get("target")
        df = self.value
        pattern = rf"^{re.escape(variable_name)}(\d*)$"
        matching_columns = [col for col in df.columns if re.match(pattern, col)]
        if not matching_columns:
            return pd.Series(
                [False] * len(df)
            )  # Return a series of False values if no matching columns
        sorted_columns = sorted(matching_columns, key=lambda x: (len(x), x))

        def check_inconsistency(row):
            prev_populated = (
                pd.notna(row[sorted_columns[0]]) and row[sorted_columns[0]] != ""
            )
            for i in range(1, len(sorted_columns)):
                curr_col = sorted_columns[i]
                curr_value = row[curr_col]
                if pd.notna(curr_value) and curr_value != "" and not prev_populated:
                    return True
                prev_populated = pd.notna(curr_value) and curr_value != ""
            return False

        return df.apply(check_inconsistency, axis=1)

    def next_column_exists_and_previous_is_null(self, row) -> bool:
        row = row.reset_index(drop=True)
        for index in row[
            row.isin(NULL_FLAVORS) | pd.isna(row)
        ].index:  # leaving null values only
            next_position: int = index + 1
            if next_position < len(row) and not (
                pd.isna(row[next_position]) or row[next_position] in NULL_FLAVORS
            ):
                return True
        return False

    def valid_codelist_reference(self, column_name, codelist):
        if column_name in self.column_codelist_map:
            return codelist in self.column_codelist_map[column_name]
        elif self.column_prefix_map:
            # Check for generic versions of variables (i.e --DECOD)
            for key in self.column_prefix_map:
                if column_name.startswith(self.column_prefix_map[key]):
                    generic_column_name = column_name.replace(
                        self.column_prefix_map[key], key, 1
                    )
                    if generic_column_name in self.column_codelist_map:
                        return codelist in self.column_codelist_map.get(
                            generic_column_name
                        )
        return True

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def has_different_values(self, other_value: dict):
        """
        The operator ensures that the target column has different values.
        """
        target = other_value.get("target")
        is_valid: bool = len(self.value[target].unique()) > 1
        return self.value.convert_to_series([is_valid] * len(self.value[target]))

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def has_same_values(self, other_value: dict):
        return ~self.has_different_values(other_value)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def is_ordered_by(self, other_value: dict):
        """
        Checking validity based on target order.
        """
        target = other_value.get("target")
        sort_order: str = other_value.get("order", "asc")
        if sort_order not in ["asc", "dsc"]:
            raise ValueError("invalid sorting order")
        sort_order_bool: bool = sort_order == "asc"
        return (
            self.value[target]
            .eq(
                self.value[target].sort_values(
                    ascending=sort_order_bool, ignore_index=True
                )
            )
            .astype(bool)
        )

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def is_not_ordered_by(self, other_value: dict):
        return ~self.is_ordered_by(other_value)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def value_has_multiple_references(self, other_value: dict):
        """
        Requires a target column and a reference count column whose values
        are a dictionary containing the number of times that value appears.
        """
        target = other_value.get("target")
        reference_count_column: str = other_value.get("comparator")
        result = np.where(
            vectorized_get_dict_key(
                self.value[reference_count_column], self.value[target]
            )
            > 1,
            True,
            False,
        )
        return self.value.convert_to_series(result)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def value_does_not_have_multiple_references(self, other_value: dict):
        return ~self.value_has_multiple_references(other_value)

    def _mark_invalid_null_positions(self, is_valid, group, null_mask, na_pos):
        null_indices = group[null_mask].index.tolist()
        non_null_indices = group[~null_mask].index.tolist()
        index_order = group.index.tolist()

        if not null_indices or not non_null_indices:
            return is_valid

        if na_pos == "last":
            last_non_null = max(index_order.index(i) for i in non_null_indices)
            first_null = min(index_order.index(i) for i in null_indices)
            if first_null < last_non_null:
                is_valid[null_mask] = False
        else:
            last_null = max(index_order.index(i) for i in null_indices)
            first_non_null = min(index_order.index(i) for i in non_null_indices)
            if last_null > first_non_null:
                is_valid[null_mask] = False

        return is_valid

    def _verify_neighbor_consistency(
        self,
        is_valid,
        non_null_rows,
        target,
        comparator,
        ascending,
        is_numeric_comparator,
    ):
        indices = non_null_rows.index.tolist()
        comparator_vals = non_null_rows[comparator].tolist()

        for i, idx in enumerate(indices):
            if not is_valid.loc[idx]:
                continue

            curr = comparator_vals[i]
            if self._is_null_or_empty(curr):
                continue

            prev = next(
                (
                    comparator_vals[j]
                    for j in range(i - 1, -1, -1)
                    if not self._is_null_or_empty(comparator_vals[j])
                ),
                None,
            )
            nxt = next(
                (
                    comparator_vals[j]
                    for j in range(i + 1, len(comparator_vals))
                    if not self._is_null_or_empty(comparator_vals[j])
                ),
                None,
            )
            if not is_numeric_comparator and not is_valid_date(str(curr)):
                continue

            if ascending:
                if prev is not None and curr < prev:
                    is_valid.loc[idx] = False
                elif nxt is not None and curr > nxt:
                    is_valid.loc[idx] = False
            else:
                if prev is not None and curr > prev:
                    is_valid.loc[idx] = False
                elif nxt is not None and curr < nxt:
                    is_valid.loc[idx] = False

        return is_valid

    def check_target_ascending_in_sorted_group_with_regex(
        self, group, target, comparator, ascending, na_pos
    ):
        """
        Check if target values are in ascending order within a group
        already sorted by comparator. Supports regex extraction.
        """
        is_valid = pd.Series(True, index=group.index)
        is_numeric_comparator = pd.api.types.is_numeric_dtype(group[comparator])
        is_numeric_target = pd.api.types.is_numeric_dtype(group[target])

        null_mask = group[comparator].isna() | (
            group[comparator].astype(str).str.strip() == ""
        )
        non_null_rows = group[~null_mask]

        is_valid = self._mark_invalid_null_positions(is_valid, group, null_mask, na_pos)
        overlap_check = self.check_date_overlaps(group, target, comparator)
        overlap_invalid_mask = ~overlap_check
        non_null_rows_for_order_check = non_null_rows[
            ~overlap_invalid_mask[non_null_rows.index]
        ]

        non_null_sorted = non_null_rows_for_order_check.sort_values(
            by=comparator, ascending=ascending
        )
        actual_target = non_null_rows_for_order_check[target].tolist()
        expected_target = non_null_sorted[target].tolist()
        non_null_indices = non_null_rows_for_order_check.index.tolist()

        for i in range(len(actual_target)):
            actual = actual_target[i]
            expected_val = expected_target[i]

            if pd.isna(actual) and pd.isna(expected_val):
                continue
            elif pd.isna(actual) or pd.isna(expected_val):
                is_valid.loc[non_null_indices[i]] = False
            elif (
                not is_numeric_target
                and is_valid_date(actual)
                and is_valid_date(expected_val)
            ):
                date1, _ = parse_date(actual)
                date2, _ = parse_date(expected_val)
                if date1 != date2:
                    is_valid.loc[non_null_indices[i]] = False
            else:
                if actual != expected_val:
                    is_valid.loc[non_null_indices[i]] = False

        non_null_target_sorted = non_null_rows.sort_values(by=target, ascending=True)
        is_valid = self._verify_neighbor_consistency(
            is_valid,
            non_null_target_sorted,
            target,
            comparator,
            ascending,
            is_numeric_comparator,
        )
        return is_valid

    def check_date_overlaps(self, group, target, comparator):
        comparator_values = group[comparator].tolist()
        is_valid = pd.Series(True, index=group.index)
        is_numeric = pd.api.types.is_numeric_dtype(group[comparator])

        if is_numeric:
            return is_valid

        valid_positions = [
            i
            for i in range(len(comparator_values))
            if not (
                pd.isna(comparator_values[i]) or str(comparator_values[i]).strip() == ""
            )
        ]

        for i in range(len(valid_positions)):
            curr_pos = valid_positions[i]
            current = comparator_values[curr_pos]
            if not is_valid_date(current):
                continue
            _, prec1 = parse_date(current)
            for j in range(len(valid_positions)):
                if i == j:
                    continue
                other_pos = valid_positions[j]
                other = comparator_values[other_pos]
                if not is_valid_date(other):
                    continue
                _, prec2 = parse_date(other)
                if prec1 != prec2:
                    overlaps, _ = dates_overlap(current, prec1, other, prec2)
                    if overlaps:
                        is_valid.iloc[curr_pos] = False
                        is_valid.iloc[other_pos] = False

        return is_valid

    def _process_grouped_result(
        self,
        grouped_result,
        grouped_df,
        within_columns,
        sorted_df,
        check_func=None,
    ):
        if isinstance(grouped_result, pd.DataFrame):
            grouped_result = grouped_result.stack()
        if isinstance(grouped_result.index, pd.MultiIndex):
            if len(within_columns) < grouped_result.index.nlevels:
                grouped_result = grouped_result.droplevel(
                    list(range(len(within_columns)))
                )
            elif len(within_columns) == grouped_result.index.nlevels:
                grouped_result = grouped_result.reset_index(drop=True)
            else:
                result_list = []
                index_list = []
                for name, group in grouped_df:
                    if check_func is not None:
                        group_result = check_func(group)
                    else:
                        group_result = (
                            grouped_result.loc[name]
                            if hasattr(grouped_result, "loc")
                            else grouped_result.iloc[0]
                        )
                    result_list.extend(group_result.tolist())
                    index_list.extend(group.index.tolist())
                grouped_result = pd.Series(result_list, index=index_list)
        return grouped_result.reindex(sorted_df.index, fill_value=True)

    def _extract_regex_group(self, series: pd.Series, regex_pattern: str) -> pd.Series:
        """
        Extract the first capturing group from a regex pattern and convert to numeric if possible.
        Handles zero-padded numbers by converting to numeric.

        Args:
            series: Pandas series with string values
            regex_pattern: Regex pattern with capturing group(s)

        Returns:
            Series with extracted and converted values
        """

        def extract_and_convert(value):
            if pd.isna(value) or value == "":
                return np.nan

            # YAML escapes backslashes, so we receive ".*\\d+$" which Python interprets as raw \
            # We need to convert this to the actual regex pattern by replacing \\ with \
            # However, since strings from YAML come already unescaped, we just use as-is
            match = re.search(regex_pattern, str(value))
            if match and match.groups():
                extracted = match.group(1)  # First capturing group
                # Try to convert to numeric to handle both padded and non-padded numbers
                try:
                    return pd.to_numeric(extracted)
                except (ValueError, TypeError):
                    return extracted
            return np.nan

        return series.apply(extract_and_convert)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def target_is_sorted_by(self, other_value: dict):
        target = other_value.get("target")
        within_columns = self._normalize_grouping_columns(other_value.get("within"))
        columns = other_value["comparator"]
        target_regex = other_value.get("regex")  # parameter for regex extraction

        result = pd.Series([True] * len(self.value), index=self.value.index)

        for col in columns:
            comparator: str = self.replace_prefix(col["name"])
            ascending: bool = col["sort_order"].lower() != "desc"
            na_pos: str = col.get("null_position", "last")

            selected_columns = list(
                dict.fromkeys([target, comparator, *within_columns])
            )

            # If regex is provided, extract and convert target values
            if target_regex:
                working_df = self.value[selected_columns].copy()
                # Create a temporary column with extracted regex values
                working_df[f"{target}_extracted"] = self._extract_regex_group(
                    working_df[target], target_regex
                )
                target_for_sorting = f"{target}_extracted"
                # Sort by within columns AND extracted target
                sorted_df = working_df.sort_values(
                    by=[*within_columns, target_for_sorting],
                    ascending=[True] * (len(within_columns) + 1),
                )
            else:
                working_df = self.value[selected_columns]
                target_for_sorting = target
                sorted_df = working_df.sort_values(
                    by=[*within_columns, target],
                    ascending=[True] * (len(within_columns) + 1),
                )

            grouped_df = sorted_df.groupby(within_columns, sort=False)

            target_check = grouped_df.apply(
                lambda x: self.check_target_ascending_in_sorted_group_with_regex(
                    x, target_for_sorting, comparator, ascending, na_pos
                )
            )
            target_check = self._process_grouped_result(
                target_check,
                grouped_df,
                within_columns,
                sorted_df,
                lambda group: self.check_target_ascending_in_sorted_group_with_regex(
                    group, target_for_sorting, comparator, ascending, na_pos
                ),
            )

            date_overlap_check = grouped_df.apply(
                lambda x: self.check_date_overlaps(x, target_for_sorting, comparator)
            )
            date_overlap_check = self._process_grouped_result(
                date_overlap_check,
                grouped_df,
                within_columns,
                sorted_df,
                lambda group: self.check_date_overlaps(
                    group, target_for_sorting, comparator
                ),
            )

            combined_check = target_check & date_overlap_check
            result = result & combined_check.reindex(self.value.index, fill_value=True)

            if isinstance(result, (pd.DataFrame, dd.DataFrame)):
                if isinstance(result, dd.DataFrame):
                    result = result.compute()
                result = result.squeeze()

        return result

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def target_is_not_sorted_by(self, other_value: dict):
        return ~self.target_is_sorted_by(other_value)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def shares_at_least_one_element_with(self, other_value: dict):
        target: str = other_value.get("target")
        comparator: str = other_value.get("comparator")

        def check_shared_elements(row):
            target_set = (
                set(row[target])
                if isinstance(row[target], (list, set))
                else {row[target]}
            )
            comparator_set = (
                set(row[comparator])
                if isinstance(row[comparator], (list, set))
                else {row[comparator]}
            )
            return bool(target_set.intersection(comparator_set))

        return self.value.apply(check_shared_elements, axis=1)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def shares_exactly_one_element_with(self, other_value: dict):
        target: str = other_value.get("target")
        comparator: str = other_value.get("comparator")

        def check_exactly_one_shared_element(row):
            target_set = (
                set(row[target])
                if isinstance(row[target], (list, set))
                else {row[target]}
            )
            comparator_set = (
                set(row[comparator])
                if isinstance(row[comparator], (list, set))
                else {row[comparator]}
            )
            return len(target_set.intersection(comparator_set)) == 1

        return self.value.apply(check_exactly_one_shared_element, axis=1)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def shares_no_elements_with(self, other_value: dict):
        target: str = other_value.get("target")
        comparator: str = other_value.get("comparator")

        def check_no_shared_elements(row):
            target_set = (
                set(row[target])
                if isinstance(row[target], (list, set))
                else {row[target]}
            )
            comparator_set = (
                set(row[comparator])
                if isinstance(row[comparator], (list, set))
                else {row[comparator]}
            )
            return len(target_set.intersection(comparator_set)) == 0

        return self.value.apply(check_no_shared_elements, axis=1)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def is_ordered_subset_of(self, other_value: dict):
        target: str = other_value.get("target")
        comparator: str = other_value.get("comparator")
        missing_columns = set()

        def check_order(row):
            target_list = row[target]
            comparator_list = row[comparator]
            comparator_positions = {col: idx for idx, col in enumerate(comparator_list)}
            positions = []
            for col in target_list:
                if col in comparator_positions:
                    positions.append(comparator_positions[col])
                else:
                    missing_columns.add(col)
                    return False
            return positions == sorted(positions)

        if isinstance(self.value, DaskDataset):
            results = self.value.apply(check_order, axis=1, meta=("check_order", bool))
            results = self.value.convert_to_series(results)
        else:
            results = self.value.apply(check_order, axis=1)
        if missing_columns:
            logger.info(
                f"Columns not found in comparator list {comparator}: {', '.join(sorted(missing_columns))}"
            )
        return results

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def is_not_ordered_subset_of(self, other_value: dict):
        return ~self.is_ordered_subset_of(other_value)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def is_title_case(self, other_value: dict):
        """
        Checks if target column values are in proper title case.
        """
        target = other_value.get("target")
        acronyms = DatasetTitleCase.Acronyms.value
        lowercase_exceptions = DatasetTitleCase.Lowercase_Exceptions.value

        def acronym_callback(word, **kwargs):
            if word.lower() in lowercase_exceptions:
                return word.lower()
            if any(word.upper() == acr.upper() for acr in acronyms):
                return word.upper()
            return None

        def check_title_case(value):
            if pd.isna(value) or value == "" or value in NULL_FLAVORS:
                return True
            str_value = str(value).strip()
            expected = titlecase(str_value, callback=acronym_callback)
            expected = expected[0].upper() + expected[1:]
            return str_value == expected

        results = self.value[target].apply(check_title_case)
        return self.value.convert_to_series(results)

    @log_operator_execution
    @type_operator(FIELD_DATAFRAME)
    def is_not_title_case(self, other_value: dict):
        """
        Checks if target column values are NOT in proper title case.
        """
        return ~self.is_title_case(other_value)