[FEATURE]: Support for Streaming in Dataset level checks

Is there an existing issue for this?

• I have searched the existing issues

Problem statement

The existing aggregation aggregation checks like is_aggr_greater_than, is_aggr_less_than, and is_unique are using Window.partitionBy. Structured Streaming only supports time-window aggregation using the WINDOW function.

Proposed Solution

For streaming use time-based windowed aggregation (groupBy(window(), ...)) and watermark.

Additional Context

Error from DLT:
Append mode error: [NON_TIME_WINDOW_NOT_SUPPORTED_IN_STREAMING] Window function is not supported in COUNT(VENDOR_ID#7677) (as column __metric_vendor_id_count_greater_than_f238ab7cee3742f598d56c0e7a34b26d) on streaming DataFrames/Datasets.
Structured Streaming only supports time-window aggregation using the WINDOW function. (window specification: (ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING)) SQLSTATE: 42KDE
Complete mode error: [STREAMING_OUTPUT_MODE.UNSUPPORTED_OPERATION] Invalid streaming output mode: complete. This output mode is not supported for no streaming aggregations on streaming DataFrames/DataSets. SQLSTATE: 42KDE

Code snipped for changes required in _is_aggr_compare for streaming:

    def apply(df: DataFrame) -> DataFrame:
        df_with_watermark = df.withWatermark(ts_column, "10 minutes")
        df_with_window = df_with_watermark.withColumn("window", F.window(ts_column, "10 minutes"))

        group_exprs = [F.col(c) for c in group_by] if group_by else []
        if not group_by or (isinstance(column, str) and column not in group_by):
            group_exprs.append(F.col(column))

        group_cols = [F.col("window")] + group_exprs

        filter_col = F.expr(row_filter) if row_filter else F.lit(True)
        filtered_expr = F.when(filter_col, aggr_col_expr)
        aggr_expr = getattr(F, aggr_type)(filtered_expr).alias(metric_col)

        agg_df = (
            df_with_window.groupBy(*group_cols)
            .agg(aggr_expr)
            .withColumn(condition_col, compare_op(F.col(metric_col), limit_expr))
            .withColumnRenamed("window", "agg_window")
        )

        # Build safe join conditions
        join_conditions = [df_with_window["window"] == agg_df["agg_window"]]
        if group_by:
            join_conditions += [df_with_window[c] == agg_df[c] for c in group_by if c != "window"]

        # Select relevant columns from agg_df
        select_cols = ["agg_window", metric_col, condition_col]
        if group_by:
            select_cols += [c for c in group_by if c != "window"]

        joined_df = (
            df_with_window
            .join(agg_df.select(*select_cols), on=join_conditions, how="left")
            .drop("agg_window")
        )

        return joined_df

Start by examining _is_aggr_compare and the existing aggregation checks is_aggr_greater_than, is_aggr_less_than, and is_unique. Run the relevant streaming validation or reproduction described in the issue, then determine how time windows and watermarks should integrate with grouping and joins. Done means these checks work on Structured Streaming DataFrames without the reported append- or complete-mode errors.