Implement enough interface for MappedOperator to be baggable

[uranusjr]

The tests are changed to actually bag mapped tasks so we can test they are actually compatible.

[ashb]

Anyone else find it odd that Operator is "lower level" than BaseOperator 😁

[ashb]

Why change from a collection to a sequence? (This might cause knock-on impacts for typing of all other operators)

[uranusjr]

It’s the other way around, from Sequence to a Collection; Collection is looser than Sequence and should work in subclasses (I think; we can always change this back).

[uranusjr]

Hmm, #20931 still does not correctly establish task dependencies, we’re missing something.

Oh that’s a simple yet difficult-to-spot oversight. Everything should be working now!

[uranusjr]

As an unrelated discovery, we can sort of trace what we need to check/change to account for MappedOperator by adding annotation to DAG.add_task() and DAG.add_tasks(). It cascades into a nice chain of errors that alerts us what originally can only contain BaseOperator now may also contain MappedOperator and may need to be modified.

[ashb]

We could choose to make them writable properties that affect the slot in partial_kwargs couldn't we?

    @weight_rule.setter
    def _set_weigh_rule(self, value):
        self.partial_kwargs['weight_rule'] = value

[uranusjr]

Hmm yeah theoratically (and maybe also a deleter for Mypy). I’ll investigate.

[uranusjr]

Uh, Mypy is not smart enough unfortunately python/mypy#4125.

We can’t use the Protocol workaround mentioned in the issue either because BaseOperator has a custom metaclass, while Protocol is also implemented with metaclass, and the two would conflict.

[ashb]

Type ignore it is the other option?

[github-actions]

The PR most likely needs to run full matrix of tests because it modifies parts of the core of Airflow. However, committers might decide to merge it quickly and take the risk. If they don't merge it quickly - please rebase it to the latest main at your convenience, or amend the last commit of the PR, and push it with --force-with-lease.

[uranusjr]

(Documenting for myself on Monday) This is currently failing with

   tests/models/test_taskinstance.py:2288: 
  _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  airflow/utils/session.py:71: in wrapper
      return func(*args, session=session, **kwargs)
  airflow/models/taskinstance.py:1621: in run
      session=session,
  airflow/utils/session.py:68: in wrapper
      return func(*args, **kwargs)
  airflow/models/taskinstance.py:1194: in check_and_change_state_before_execution
      dep_context=non_requeueable_dep_context, session=session, verbose=True
  airflow/utils/session.py:68: in wrapper
      return func(*args, **kwargs)
  airflow/models/taskinstance.py:1031: in are_dependencies_met
      for dep_status in self.get_failed_dep_statuses(dep_context=dep_context, session=session):
  airflow/models/taskinstance.py:1052: in get_failed_dep_statuses
      for dep_status in dep.get_dep_statuses(self, session, dep_context):
  airflow/ti_deps/deps/base_ti_dep.py:95: in get_dep_statuses
      yield from self._get_dep_statuses(ti, session, dep_context)
  _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
  
  self = <TIDep(Previous Dagrun State)>
  ti = <[DetachedInstanceError('Instance <TaskInstance at 0x7fefa43f9850> is not bound to a Session; attribute refresh operation cannot proceed') raised in repr()] TaskInstance object at 0x7fefa43f9850>
  session = <sqlalchemy.orm.session.Session object at 0x7fefd5d5f050>
  dep_context = <airflow.ti_deps.dep_context.DepContext object at 0x7fefa4430050>
  
      @provide_session
      def _get_dep_statuses(self, ti, session, dep_context):
          if dep_context.ignore_depends_on_past:
              reason = "The context specified that the state of past DAGs could be ignored."
              yield self._passing_status(reason=reason)
              return
      
  >       if not ti.task.depends_on_past:
  E       AttributeError: 'MappedOperator' object has no attribute 'depends_on_past'
  
  airflow/ti_deps/deps/prev_dagrun_dep.py:41: AttributeError

[ashb]

We also need wait_for_downstream in a similar vein for the scheduler - but that can be added in a separate pr

[uranusjr]

depends_on_past depends on wait_for_downstream, so I implemented them both.

[ashb]

Only failure is an actual timing error:

test_heartbeat_failed_fast: assert 0.5497589999999999 < 0.5

Good to go now!