Method _apply

python/pathway/stdlib/temporal/_window.py:272–356

(
        self,
        table: pw.Table,
        key: pw.ColumnExpression,
        behavior: Behavior | None,
        instance: pw.ColumnExpression | None,
    )

    @check_arg_types
    def _apply(
        self,
        table: pw.Table,
        key: pw.ColumnExpression,
        behavior: Behavior | None,
        instance: pw.ColumnExpression | None,
    ) -> pw.GroupedTable:
        check_joint_types(
            {
                "time_expr": (key, TimeEventType),
                "window.hop": (self.hop, IntervalType),
                "window.duration": (self.duration, IntervalType),
                "window.origin": (self.origin, TimeEventType),
            }
        )

        target = _assign_sliding_windows(
            table,
            key=key,
            instance=instance,
            window=self,
            with_original_id=True,
        )

        if behavior is not None:
            if isinstance(behavior, ExactlyOnceBehavior):
                duration: IntervalType
                # that is split in two if-s, as it helps mypy figure out proper types
                # one if impl left either self.ratio or self.duration as optionals
                # which won't fit into the duration variable of type IntervalType
                if self.duration is not None:
                    duration = self.duration
                elif self.ratio is not None:
                    duration = self.ratio * self.hop
                shift = (
                    behavior.shift
                    if behavior.shift is not None
                    else zero_length_interval(type(duration))
                )
                behavior = common_behavior(
                    duration + shift, shift, True  # type:ignore
                )
            elif not isinstance(behavior, CommonBehavior):
                raise ValueError(
                    f"behavior {behavior} unsupported in sliding/tumbling window"
                )

            if behavior.cutoff is not None:
                cutoff_threshold = pw.this._pw_window_end + behavior.cutoff
                target = target._freeze(cutoff_threshold, pw.this._pw_key)
            if behavior.delay is not None:
                target = target._buffer(
                    target._pw_window_start + behavior.delay, target._pw_key
                )
                target = target.with_columns(
                    _pw_key=pw.if_else(
                        target._pw_key > target._pw_window_start + behavior.delay,
                        target._pw_key,