search

apache / airflow

Apache Airflow - A platform to programmatically author, schedule, and monitor workflows
https://airflow.apache.org/
Apache License 2.0
36.27k stars 14.08k forks source link

`id` of sentinel values not preserved by `partial_subset` #41160

Open ethan-oro opened 1 month ago

ethan-oro commented 1 month ago

Apache Airflow version

Other Airflow 2 version (please specify below)

If "Other Airflow 2 version" selected, which one?

2.7.3

What happened?

when running a dag created with partial_subset, the id of sentinel values are not properly maintained for decorated python mapped tasks, specifically EXPAND_INPUT_EMPTY that is set as the default value for expand_input for any mapped @task. this causes all calls to the mapped, decorated tasks to fail with following error: AssertionError: unexpected expand_input: DictOfListsExpandInput(value={}) raises here as the comparison is done using is not; i have added a minimally reproducible example below.

in debugging, adding EXPAND_INPUT_EMPTY to deepcopy's memo arg fixed the issue within partial_subset like

memo = {id(self.task_dict): None, id(self._task_group): None, id(EXPAND_INPUT_EMPTY): EXPAND_INPUT_EMPTY}

as I imagine the comparison to a sentinel value by id instead of by value will want to continue. If that is the desired path forward, I can open a PR to that effect.

What you think should happen instead?

the mapped task should run without failure.

How to reproduce

import datetime
import os

from airflow import settings
from airflow.utils.state import DagRunState
from airflow.utils.db import initdb, resetdb
from airflow.decorators import dag
from airflow.decorators import task

@task
def generate_input() -> list[int]:
    return [1, 2, 3, 4, 5, 6, 7]

@task
def mapped(multiplier: int, value: int) -> int:
    return multiplier * value

@task
def validate_output(multiplier: int, results: list[int]) -> None:
    for result in results:
        if result % multiplier != 0:
            raise ValueError("Oops!")

@dag(
    dag_id="reproducible_dag",
    default_args={
        "start_date": datetime.date.today().strftime("%Y-%m-%d"),
        "retries": 0,
    },
    schedule=None,
    catchup=False,
    render_template_as_native_obj=True,
)
def build_reproducible_dag() -> None:
    mapped_input = generate_input()
    mapped_output = mapped.partial(multiplier=3).expand(value=mapped_input)
    validate_output(multiplier=3, results=mapped_output)

reproducible_dag = build_reproducible_dag()

if __name__ == "__main__":
    from recidiviz.tools.postgres import local_postgres_helpers

    # set up
    temp_dir = local_postgres_helpers.start_on_disk_postgresql_database()
    os.environ["AIRFLOW__DATABASE__SQL_ALCHEMY_CONN"] = (
        local_postgres_helpers.on_disk_postgres_db_url().render_as_string()
    )
    settings.initialize()
    initdb(load_connections=False)

    dag_run = reproducible_dag.test()  # this run will succeeed

    assert dag_run.get_state() == DagRunState.SUCCESS

    # reset for test
    resetdb()

    reproducible_subset = reproducible_dag.partial_subset(
        task_ids_or_regex=[
            "generate_input",
            "mapped",
            "validate_output",
        ]
    )
    dag_run = reproducible_subset.test()  # this run will fail

    assert dag_run.get_state() == DagRunState.SUCCESS

    # tear down
    resetdb(skip_init=True)
    local_postgres_helpers.stop_and_clear_on_disk_postgresql_database(temp_dir)

for local_postgres_helpers code see, pulse-data

task failure trace

[2024-07-31 11:42:49,263] {taskinstance.py:1937} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1518, in _run_raw_task
    self._execute_task_with_callbacks(context, test_mode, session=session)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1647, in _execute_task_with_callbacks
    task_orig = self.render_templates(context=context)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 2285, in render_templates
    original_task.render_template_fields(context)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/mappedoperator.py", line 725, in render_template_fields
    mapped_kwargs, seen_oids = self._expand_mapped_kwargs(context, session)
                               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/decorators/base.py", line 527, in _expand_mapped_kwargs
    raise AssertionError(f"unexpected expand_input: {self.expand_input}")
AssertionError: unexpected expand_input: DictOfListsExpandInput(value={})
ERROR [airflow.task] Task failed with exception
Traceback (most recent call last):
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1518, in _run_raw_task
    self._execute_task_with_callbacks(context, test_mode, session=session)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1647, in _execute_task_with_callbacks
    task_orig = self.render_templates(context=context)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 2285, in render_templates
    original_task.render_template_fields(context)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/mappedoperator.py", line 725, in render_template_fields
    mapped_kwargs, seen_oids = self._expand_mapped_kwargs(context, session)
                               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/decorators/base.py", line 527, in _expand_mapped_kwargs
    raise AssertionError(f"unexpected expand_input: {self.expand_input}")
AssertionError: unexpected expand_input: DictOfListsExpandInput(value={})

Operating System

macOS::14.1

Versions of Apache Airflow Providers

apache-airflow==2.7.3
apache-airflow-providers-apache-beam==5.6.3
apache-airflow-providers-celery==3.6.0
apache-airflow-providers-cncf-kubernetes==8.1.0
apache-airflow-providers-common-sql==1.12.0
apache-airflow-providers-dbt-cloud==3.7.1
apache-airflow-providers-ftp==3.8.0
apache-airflow-providers-google==10.17.0
apache-airflow-providers-hashicorp==3.6.4
apache-airflow-providers-http==4.10.1
apache-airflow-providers-imap==3.5.0
apache-airflow-providers-mysql==5.5.4
apache-airflow-providers-postgres==5.10.2
apache-airflow-providers-sendgrid==3.4.0
apache-airflow-providers-sftp==4.9.1
apache-airflow-providers-sqlite==3.7.1
apache-airflow-providers-ssh==3.10.1
google-cloud-orchestration-airflow==1.12.1

Deployment

Google Cloud Composer

Deployment details

pipfile for env, drawn from https://cloud.google.com/composer/docs/concepts/versioning/composer-versions for composer-2.7.1-airflow-2.7.3

[[source]]
url = "https://pypi.org/simple"
verify_ssl = true
name = "pypi"

[dev-packages]
pytest = "*"
pylint = "*"
more-itertools = "*"
mypy = "*"
black = "*"
freezegun = "*"

[packages]
# Packages specified in cloud-composer.tf
us = "==3.1.1"
apache-airflow-providers-sftp = "==4.9.1"
python-levenshtein = "==0.25.1"
dateparser = "==1.2.0"

# We pin all dependencies to make sure they remain on versions supported by our current 
# Cloud Composer environment. We must manually update the dependencies when we update 
# our Cloud Composer version. https://cloud.google.com/composer/docs/concepts/versioning/composer-versions
absl-py = "==2.1.0"
agate = "==1.6.3"
aiodebug = "==2.3.0"
aiofiles = "==23.2.1"
aiohttp = "==3.9.4"
aiosignal = "==1.3.1"
alembic = "==1.13.1"
amqp = "==5.2.0"
annotated-types = "==0.6.0"
anyio = "==4.3.0"
apache-airflow = "==2.7.3"
apache-airflow-providers-apache-beam = "==5.6.3"
apache-airflow-providers-celery = "==3.6.0"
apache-airflow-providers-cncf-kubernetes = "==8.1.0"
apache-airflow-providers-common-sql = "==1.12.0"
apache-airflow-providers-dbt-cloud = "==3.7.1"
apache-airflow-providers-ftp = "==3.8.0"
apache-airflow-providers-google = "==10.17.0"
apache-airflow-providers-hashicorp = "==3.6.4"
apache-airflow-providers-http = "==4.10.1"
apache-airflow-providers-imap = "==3.5.0"
apache-airflow-providers-mysql = "==5.5.4"
apache-airflow-providers-postgres = "==5.10.2"
apache-airflow-providers-sendgrid = "==3.4.0"
apache-airflow-providers-sqlite = "==3.7.1"
apache-airflow-providers-ssh = "==3.10.1"
apache-beam = "==2.55.1"
apispec = "==6.6.0"
appdirs = "==1.4.4"
argcomplete = "==3.3.0"
asgiref = "==3.8.1"
astunparse = "==1.6.3"
attrs = "==23.2.0"
Babel = "==2.14.0"
backoff = "==2.2.1"
backports-tarfile = "==1.0.0"
bcrypt = "==4.1.2"
billiard = "==4.2.0"
blinker = "==1.7.0"
CacheControl = "==0.14.0"
cachelib = "==0.9.0"
cachetools = "==5.3.3"
cattrs = "==23.2.3"
celery = "==5.3.6"
certifi = "==2024.2.2"
cffi = "==1.16.0"
chardet = "==5.2.0"
charset-normalizer = "==3.3.2"
click = "==8.1.3"
click-didyoumean = "==0.3.1"
click-plugins = "==1.1.1"
click-repl = "==0.3.0"
clickclick = "==20.10.2"
cloudpickle = "==2.2.1"
colorama = "==0.4.6"
colorlog = "==4.8.0"
ConfigUpdater = "==3.2"
connexion = "==2.14.2"
crcmod = "==1.7"
cron-descriptor = "==1.4.3"
croniter = "==2.0.3"
cryptography = "==41.0.7"
db-dtypes = "==1.2.0"
dbt-bigquery = "==1.5.4"
dbt-core = "==1.5.4"
dbt-extractor = "==0.4.1"
decorator = "==5.1.1"
Deprecated = "==1.2.14"
diff_cover = "==9.0.0"
dill = "==0.3.1.1"
distlib = "==0.3.8"
dnspython = "==2.6.1"
docopt = "==0.6.2"
docstring_parser = "==0.16"
docutils = "==0.21.1"
email-validator = "==1.3.1"
fastavro = "==1.9.4"
fasteners = "==0.19"
filelock = "==3.13.3"
firebase-admin = "==6.5.0"
Flask = "==2.2.5"
Flask-AppBuilder = "==4.3.6"
Flask-Babel = "==2.0.0"
Flask-Bcrypt = "==1.0.1"
Flask-Caching = "==2.1.0"
Flask-JWT-Extended = "==4.6.0"
Flask-Limiter = "==3.5.1"
Flask-Login = "==0.6.3"
Flask-Session = "==0.5.0"
Flask-SQLAlchemy = "==2.5.1"
Flask-WTF = "==1.2.1"
flatbuffers = "==24.3.25"
flower = "==2.0.1"
frozenlist = "==1.4.1"
fsspec = "==2024.3.1"
future = "==1.0.0"
gast = "==0.5.4"
gcloud-aio-auth = "==4.2.3"
gcloud-aio-bigquery = "==7.1.0"
gcloud-aio-storage = "==9.2.0"
gcsfs = "==2024.3.1"
google-ads = "==23.1.0"
google-analytics-admin = "==0.22.7"
google-api-core = "==2.18.0"
google-api-python-client = "==2.125.0"
google-apitools = "==0.5.32"
google-auth = "==2.29.0"
google-auth-httplib2 = "==0.2.0"
google-auth-oauthlib = "==1.2.0"
google-cloud-access-context-manager = "==0.2.0"
google-cloud-aiplatform = "==1.47.0"
google-cloud-appengine-logging = "==1.4.3"
google-cloud-asset = "==3.26.0"
google-cloud-audit-log = "==0.2.5"
google-cloud-automl = "==2.13.3"
google-cloud-batch = "==0.17.17"
google-cloud-bigquery = "==3.20.1"
google-cloud-bigquery-datatransfer = "==3.15.1"
google-cloud-bigtable = "==2.23.0"
google-cloud-build = "==3.24.0"
google-cloud-common = "==1.3.3"
google-cloud-compute = "==1.18.0"
google-cloud-container = "==2.45.0"
google-cloud-core = "==2.4.1"
google-cloud-datacatalog = "==3.19.0"
google-cloud-datacatalog-lineage = "==0.3.1"
# google-cloud-datacatalog-lineage-producer-client = "==0.1.0"
google-cloud-dataflow-client = "==0.8.10"
google-cloud-dataform = "==0.5.9"
google-cloud-dataplex = "==1.13.0"
google-cloud-dataproc = "==5.9.3"
google-cloud-dataproc-metastore = "==1.15.3"
google-cloud-datastore = "==2.19.0"
google-cloud-dlp = "==3.16.0"
google-cloud-documentai = "==2.25.0"
google-cloud-filestore = "==1.9.3"
google-cloud-firestore = "==2.16.0"
google-cloud-kms = "==2.21.3"
google-cloud-language = "==2.13.3"
google-cloud-logging = "==3.10.0"
google-cloud-memcache = "==1.9.3"
google-cloud-monitoring = "==2.19.3"
google-cloud-orchestration-airflow = "==1.12.1"
google-cloud-org-policy = "==1.11.0"
google-cloud-os-config = "==1.17.3"
google-cloud-os-login = "==2.14.3"
google-cloud-pubsub = "==2.21.1"
google-cloud-pubsublite = "==0.6.1"
google-cloud-redis = "==2.15.3"
google-cloud-resource-manager = "==1.12.3"
google-cloud-run = "==0.10.5"
google-cloud-secret-manager = "==2.19.0"
google-cloud-spanner = "==3.44.0"
google-cloud-speech = "==2.26.0"
google-cloud-storage = "==2.16.0"
google-cloud-storage-transfer = "==1.11.3"
google-cloud-tasks = "==2.16.3"
google-cloud-texttospeech = "==2.16.3"
google-cloud-translate = "==3.15.3"
google-cloud-videointelligence = "==2.13.3"
google-cloud-vision = "==3.7.2"
google-cloud-workflows = "==1.14.3"
google-crc32c = "==1.5.0"
google-pasta = "==0.2.0"
google-re2 = "==1.1"
google-resumable-media = "==2.7.0"
googleapis-common-protos = "==1.63.0"
graphviz = "==0.20.3"
greenlet = "==3.0.3"
grpc-google-iam-v1 = "==0.13.0"
grpc-interceptor = "==0.15.4"
grpcio = "==1.62.1"
grpcio-gcp = "==0.2.2"
grpcio-status = "==1.62.1"
gunicorn = "==21.2.0"
h11 = "==0.14.0"
h5py = "==3.11.0"
hdfs = "==2.7.3"
hologram = "==0.0.16"
httpcore = "==1.0.5"
httplib2 = "==0.22.0"
httpx = "==0.27.0"
humanize = "==4.9.0"
hvac = "==2.1.0"
idna = "==3.7"
importlib-metadata = "==7.0.0"
importlib_resources = "==6.4.0"
inflection = "==0.5.1"
iniconfig = "==2.0.0"
isodate = "==0.6.1"
itsdangerous = "==2.1.2"
jaraco-classes = "==3.4.0"
jaraco-context = "==5.3.0"
jaraco-functools = "==4.0.0"
jeepney = "==0.8.0"
Jinja2 = "==3.1.2"
Js2Py = "==0.74"
json-merge-patch = "==0.2"
jsonpickle = "==3.0.4"
jsonschema = "==4.21.1"
jsonschema-specifications = "==2023.12.1"
keras = "==3.2.1"
keyring = "==25.1.0"
keyrings-google-artifactregistry-auth = "==1.1.2"
kombu = "==5.3.7"
kubernetes = "==29.0.0"
kubernetes_asyncio = "==29.0.0"
lazy-object-proxy = "==1.10.0"
leather = "==0.4.0"
libclang = "==18.1.1"
limits = "==3.10.1"
linkify-it-py = "==2.0.3"
lockfile = "==0.12.2"
Logbook = "==1.5.3"
looker-sdk = "==24.4.0"
Mako = "==1.3.3"
Markdown = "==3.6"
markdown-it-py = "==3.0.0"
MarkupSafe = "==2.1.5"
marshmallow = "==3.21.1"
marshmallow-oneofschema = "==3.1.1"
marshmallow-sqlalchemy = "==0.26.1"
mashumaro = "==3.6"
mdit-py-plugins = "==0.4.0"
mdurl = "==0.1.2"
minimal-snowplow-tracker = "==0.0.2"
ml-dtypes = "==0.3.2"
more-itertools = "==10.2.0"
msgpack = "==1.0.8"
multidict = "==6.0.5"
mysql-connector-python = "==8.3.0"
mysqlclient = "==2.2.4"
namex = "==0.0.8"
networkx = "==2.8.8"
numpy = "==1.26.4"
oauth2client = "==4.1.3"
oauthlib = "==3.2.2"
objsize = "==0.7.0"
opentelemetry-api = "==1.24.0"
opentelemetry-exporter-otlp = "==1.24.0"
opentelemetry-exporter-otlp-proto-common = "==1.24.0"
opentelemetry-exporter-otlp-proto-grpc = "==1.24.0"
opentelemetry-exporter-otlp-proto-http = "==1.24.0"
opentelemetry-proto = "==1.24.0"
opentelemetry-sdk = "==1.24.0"
opentelemetry-semantic-conventions = "==0.45b0"
opt-einsum = "==3.3.0"
optree = "==0.11.0"
ordered-set = "==4.1.0"
orjson = "==3.10.0"
overrides = "==6.5.0"
packaging = "==24.0"
pandas = "==2.1.4"
pandas-gbq = "==0.22.0"
paramiko = "==3.4.0"
parsedatetime = "==2.4"
pathspec = "==0.11.2"
pendulum = "==2.1.2"
pip = "==23.2.1"
pipdeptree = "==2.18.1"
platformdirs = "==4.2.0"
pluggy = "==1.4.0"
prison = "==0.2.1"
prometheus_client = "==0.20.0"
prompt-toolkit = "==3.0.43"
proto-plus = "==1.23.0"
protobuf = "==4.25.3"
psutil = "==5.9.8"
psycopg2-binary = "==2.9.9"
pyarrow = "==14.0.2"
pyarrow-hotfix = "==0.6"
pyasn1 = "==0.5.1"
pyasn1-modules = "==0.3.0"
pycparser = "==2.22"
pydantic = "==2.7.0"
pydantic_core = "==2.18.1"
pydata-google-auth = "==1.8.2"
pydot = "==1.4.2"
Pygments = "==2.17.2"
pyjsparser = "==2.7.1"
PyJWT = "==2.8.0"
pymongo = "==4.6.3"
PyNaCl = "==1.5.0"
pyOpenSSL = "==24.1.0"
pyparsing = "==3.1.2"
pytest = "==7.4.4"
python-daemon = "==3.0.1"
python-dateutil = "==2.8.2"
python-http-client = "==3.3.7"
python-nvd3 = "==0.15.0"
python-slugify = "==8.0.4"
pytimeparse = "==1.1.8"
pytz = "==2024.1"
pytzdata = "==2020.1"
PyYAML = "==6.0.1"
redis = "==4.6.0"
referencing = "==0.34.0"
regex = "==2023.12.25"
requests = "==2.31.0"
requests-oauthlib = "==2.0.0"
requests-toolbelt = "==1.0.0"
rfc3339-validator = "==0.1.4"
rich = "==13.7.1"
rich-argparse = "==1.4.0"
rpds-py = "==0.18.0"
rsa = "==4.9"
SecretStorage = "==3.3.3"
sendgrid = "==6.11.0"
setproctitle = "==1.3.3"
setuptools = "==66.1.1"
shapely = "==2.0.3"
six = "==1.16.0"
sniffio = "==1.3.1"
SQLAlchemy = "==1.4.52"
sqlalchemy-bigquery = "==1.10.0"
SQLAlchemy-JSONField = "==1.0.2"
sqlalchemy-spanner = "==1.6.2"
SQLAlchemy-Utils = "==0.41.2"
sqlfluff = "==2.3.5"
sqllineage = "==1.4.9"
sqlparse = "==0.4.4"
sshtunnel = "==0.4.0"
starkbank-ecdsa = "==2.2.0"
statsd = "==4.0.1"
tabulate = "==0.9.0"
tblib = "==3.0.0"
tenacity = "==8.2.3"
tensorboard = "==2.16.2"
tensorboard-data-server = "==0.7.2"
tensorflow = "==2.16.1"
tensorflow-io-gcs-filesystem = "==0.36.0"
termcolor = "==2.4.0"
text-unidecode = "==1.3"
tornado = "==6.4"
tqdm = "==4.66.2"
typing_extensions = "==4.11.0"
tzdata = "==2024.1"
tzlocal = "==5.2"
uc-micro-py = "==1.0.3"
unicodecsv = "==0.14.1"
uritemplate = "==4.1.1"
urllib3 = "==2.2.1"
vine = "==5.1.0"
virtualenv = "==20.25.1"
wcwidth = "==0.2.13"
websocket-client = "==1.7.0"
Werkzeug = "==2.2.3"
wheel = "==0.43.0"
wrapt = "==1.16.0"
WTForms = "==3.0.1"
yarl = "==1.9.4"
zipp = "==3.18.1"
zstandard = "==0.22.0"

[requires]
python_version = "3.11.8"

Anything else?

here are full logs from failed dag run

[2024-07-31T12:01:54.270-0400] {environment.py:213} INFO - Test environment, proceeding.
[2024-07-31T12:01:54.274-0400] {environment.py:213} INFO - Test environment, proceeding.
Created database `recidiviz_test_db0` on postgres instance bound to port 54300
[2024-07-31T12:01:55.179-0400] {environment.py:213} INFO - Test environment, proceeding.
To query data, connect with `psql postgresql://recidiviz_test_usr@localhost:54300/recidiviz_test_db0`
[2024-07-31T12:01:55.180-0400] {environment.py:213} INFO - Test environment, proceeding.
[2024-07-31T12:01:55.309-0400] {migration.py:216} INFO - Context impl PostgresqlImpl.
[2024-07-31T12:01:55.309-0400] {migration.py:219} INFO - Will assume transactional DDL.
INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
INFO  [alembic.runtime.migration] Running stamp_revision  -> 405de8318b3a
[2024-07-31 12:01:55,790] {taskinstance.py:1662} INFO - Exporting env vars: AIRFLOW_CTX_DAG_OWNER='airflow' AIRFLOW_CTX_DAG_ID='reproducible_dag' AIRFLOW_CTX_TASK_ID='generate_input' AIRFLOW_CTX_EXECUTION_DATE='2024-07-31T16:01:55.743454+00:00' AIRFLOW_CTX_TRY_NUMBER='1' AIRFLOW_CTX_DAG_RUN_ID='manual__2024-07-31T16:01:55.743454+00:00'
INFO  [airflow.task] Exporting env vars: AIRFLOW_CTX_DAG_OWNER='airflow' AIRFLOW_CTX_DAG_ID='reproducible_dag' AIRFLOW_CTX_TASK_ID='generate_input' AIRFLOW_CTX_EXECUTION_DATE='2024-07-31T16:01:55.743454+00:00' AIRFLOW_CTX_TRY_NUMBER='1' AIRFLOW_CTX_DAG_RUN_ID='manual__2024-07-31T16:01:55.743454+00:00'
[2024-07-31 12:01:55,791] {python.py:194} INFO - Done. Returned value was: [1, 2, 3, 4, 5, 6, 7]
INFO  [airflow.task.operators] Done. Returned value was: [1, 2, 3, 4, 5, 6, 7]
[2024-07-31 12:01:55,795] {taskinstance.py:1400} INFO - Marking task as SUCCESS. dag_id=reproducible_dag, task_id=generate_input, execution_date=20240731T160155, start_date=, end_date=20240731T160155
INFO  [airflow.task] Marking task as SUCCESS. dag_id=reproducible_dag, task_id=generate_input, execution_date=20240731T160155, start_date=, end_date=20240731T160155
[2024-07-31 12:01:55,816] {taskinstance.py:1937} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1518, in _run_raw_task
    self._execute_task_with_callbacks(context, test_mode, session=session)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1647, in _execute_task_with_callbacks
    task_orig = self.render_templates(context=context)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 2285, in render_templates
    original_task.render_template_fields(context)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/mappedoperator.py", line 725, in render_template_fields
    mapped_kwargs, seen_oids = self._expand_mapped_kwargs(context, session)
                               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/decorators/base.py", line 527, in _expand_mapped_kwargs
    raise AssertionError(f"unexpected expand_input: {self.expand_input}")
AssertionError: unexpected expand_input: DictOfListsExpandInput(value={})
ERROR [airflow.task] Task failed with exception
Traceback (most recent call last):
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1518, in _run_raw_task
    self._execute_task_with_callbacks(context, test_mode, session=session)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1647, in _execute_task_with_callbacks
    task_orig = self.render_templates(context=context)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 2285, in render_templates
    original_task.render_template_fields(context)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/mappedoperator.py", line 725, in render_template_fields
    mapped_kwargs, seen_oids = self._expand_mapped_kwargs(context, session)
                               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/decorators/base.py", line 527, in _expand_mapped_kwargs
    raise AssertionError(f"unexpected expand_input: {self.expand_input}")
AssertionError: unexpected expand_input: DictOfListsExpandInput(value={})
[2024-07-31 12:01:55,818] {taskinstance.py:1983} ERROR - Unable to unmap task to determine if we need to send an alert email
ERROR [airflow.task] Unable to unmap task to determine if we need to send an alert email
[2024-07-31 12:01:55,818] {taskinstance.py:1400} INFO - Marking task as FAILED. dag_id=reproducible_dag, task_id=mapped, map_index=0, execution_date=20240731T160155, start_date=, end_date=20240731T160155
INFO  [airflow.task] Marking task as FAILED. dag_id=reproducible_dag, task_id=mapped, map_index=0, execution_date=20240731T160155, start_date=, end_date=20240731T160155
ERROR [airflow.models.dag.DAG] Task failed; ti=<TaskInstance: reproducible_dag.mapped manual__2024-07-31T16:01:55.743454+00:00 map_index=0 [failed]>
Traceback (most recent call last):
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/dag.py", line 2776, in test
    ret = _run_task(ti, session=session)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/dag.py", line 3938, in _run_task
    ret = ti._run_raw_task(session=session)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/utils/session.py", line 76, in wrapper
    return func(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1518, in _run_raw_task
    self._execute_task_with_callbacks(context, test_mode, session=session)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1647, in _execute_task_with_callbacks
    task_orig = self.render_templates(context=context)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 2285, in render_templates
    original_task.render_template_fields(context)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/mappedoperator.py", line 725, in render_template_fields
    mapped_kwargs, seen_oids = self._expand_mapped_kwargs(context, session)
                               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/decorators/base.py", line 527, in _expand_mapped_kwargs
    raise AssertionError(f"unexpected expand_input: {self.expand_input}")
AssertionError: unexpected expand_input: DictOfListsExpandInput(value={})
[2024-07-31 12:01:55,825] {taskinstance.py:1937} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1518, in _run_raw_task
    self._execute_task_with_callbacks(context, test_mode, session=session)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1647, in _execute_task_with_callbacks
    task_orig = self.render_templates(context=context)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 2285, in render_templates
    original_task.render_template_fields(context)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/mappedoperator.py", line 725, in render_template_fields
    mapped_kwargs, seen_oids = self._expand_mapped_kwargs(context, session)
                               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/decorators/base.py", line 527, in _expand_mapped_kwargs
    raise AssertionError(f"unexpected expand_input: {self.expand_input}")
AssertionError: unexpected expand_input: DictOfListsExpandInput(value={})
ERROR [airflow.task] Task failed with exception
Traceback (most recent call last):
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1518, in _run_raw_task
    self._execute_task_with_callbacks(context, test_mode, session=session)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1647, in _execute_task_with_callbacks
    task_orig = self.render_templates(context=context)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 2285, in render_templates
    original_task.render_template_fields(context)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/mappedoperator.py", line 725, in render_template_fields
    mapped_kwargs, seen_oids = self._expand_mapped_kwargs(context, session)
                               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/decorators/base.py", line 527, in _expand_mapped_kwargs
    raise AssertionError(f"unexpected expand_input: {self.expand_input}")
AssertionError: unexpected expand_input: DictOfListsExpandInput(value={})
[2024-07-31 12:01:55,825] {taskinstance.py:1983} ERROR - Unable to unmap task to determine if we need to send an alert email
ERROR [airflow.task] Unable to unmap task to determine if we need to send an alert email
[2024-07-31 12:01:55,826] {taskinstance.py:1400} INFO - Marking task as FAILED. dag_id=reproducible_dag, task_id=mapped, map_index=1, execution_date=20240731T160155, start_date=, end_date=20240731T160155
INFO  [airflow.task] Marking task as FAILED. dag_id=reproducible_dag, task_id=mapped, map_index=1, execution_date=20240731T160155, start_date=, end_date=20240731T160155
ERROR [airflow.models.dag.DAG] Task failed; ti=<TaskInstance: reproducible_dag.mapped manual__2024-07-31T16:01:55.743454+00:00 map_index=1 [failed]>
Traceback (most recent call last):
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/dag.py", line 2776, in test
    ret = _run_task(ti, session=session)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/dag.py", line 3938, in _run_task
    ret = ti._run_raw_task(session=session)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/utils/session.py", line 76, in wrapper
    return func(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1518, in _run_raw_task
    self._execute_task_with_callbacks(context, test_mode, session=session)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1647, in _execute_task_with_callbacks
    task_orig = self.render_templates(context=context)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 2285, in render_templates
    original_task.render_template_fields(context)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/mappedoperator.py", line 725, in render_template_fields
    mapped_kwargs, seen_oids = self._expand_mapped_kwargs(context, session)
                               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/decorators/base.py", line 527, in _expand_mapped_kwargs
    raise AssertionError(f"unexpected expand_input: {self.expand_input}")
AssertionError: unexpected expand_input: DictOfListsExpandInput(value={})
[2024-07-31 12:01:55,831] {taskinstance.py:1937} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1518, in _run_raw_task
    self._execute_task_with_callbacks(context, test_mode, session=session)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1647, in _execute_task_with_callbacks
    task_orig = self.render_templates(context=context)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 2285, in render_templates
    original_task.render_template_fields(context)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/mappedoperator.py", line 725, in render_template_fields
    mapped_kwargs, seen_oids = self._expand_mapped_kwargs(context, session)
                               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/decorators/base.py", line 527, in _expand_mapped_kwargs
    raise AssertionError(f"unexpected expand_input: {self.expand_input}")
AssertionError: unexpected expand_input: DictOfListsExpandInput(value={})
ERROR [airflow.task] Task failed with exception
Traceback (most recent call last):
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1518, in _run_raw_task
    self._execute_task_with_callbacks(context, test_mode, session=session)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1647, in _execute_task_with_callbacks
    task_orig = self.render_templates(context=context)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 2285, in render_templates
    original_task.render_template_fields(context)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/mappedoperator.py", line 725, in render_template_fields
    mapped_kwargs, seen_oids = self._expand_mapped_kwargs(context, session)
                               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/decorators/base.py", line 527, in _expand_mapped_kwargs
    raise AssertionError(f"unexpected expand_input: {self.expand_input}")
AssertionError: unexpected expand_input: DictOfListsExpandInput(value={})
[2024-07-31 12:01:55,832] {taskinstance.py:1983} ERROR - Unable to unmap task to determine if we need to send an alert email
ERROR [airflow.task] Unable to unmap task to determine if we need to send an alert email
[2024-07-31 12:01:55,832] {taskinstance.py:1400} INFO - Marking task as FAILED. dag_id=reproducible_dag, task_id=mapped, map_index=2, execution_date=20240731T160155, start_date=, end_date=20240731T160155
INFO  [airflow.task] Marking task as FAILED. dag_id=reproducible_dag, task_id=mapped, map_index=2, execution_date=20240731T160155, start_date=, end_date=20240731T160155
ERROR [airflow.models.dag.DAG] Task failed; ti=<TaskInstance: reproducible_dag.mapped manual__2024-07-31T16:01:55.743454+00:00 map_index=2 [failed]>
Traceback (most recent call last):
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/dag.py", line 2776, in test
    ret = _run_task(ti, session=session)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/dag.py", line 3938, in _run_task
    ret = ti._run_raw_task(session=session)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/utils/session.py", line 76, in wrapper
    return func(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1518, in _run_raw_task
    self._execute_task_with_callbacks(context, test_mode, session=session)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1647, in _execute_task_with_callbacks
    task_orig = self.render_templates(context=context)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 2285, in render_templates
    original_task.render_template_fields(context)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/mappedoperator.py", line 725, in render_template_fields
    mapped_kwargs, seen_oids = self._expand_mapped_kwargs(context, session)
                               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/decorators/base.py", line 527, in _expand_mapped_kwargs
    raise AssertionError(f"unexpected expand_input: {self.expand_input}")
AssertionError: unexpected expand_input: DictOfListsExpandInput(value={})
[2024-07-31 12:01:55,838] {taskinstance.py:1937} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1518, in _run_raw_task
    self._execute_task_with_callbacks(context, test_mode, session=session)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1647, in _execute_task_with_callbacks
    task_orig = self.render_templates(context=context)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 2285, in render_templates
    original_task.render_template_fields(context)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/mappedoperator.py", line 725, in render_template_fields
    mapped_kwargs, seen_oids = self._expand_mapped_kwargs(context, session)
                               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/decorators/base.py", line 527, in _expand_mapped_kwargs
    raise AssertionError(f"unexpected expand_input: {self.expand_input}")
AssertionError: unexpected expand_input: DictOfListsExpandInput(value={})
ERROR [airflow.task] Task failed with exception
Traceback (most recent call last):
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1518, in _run_raw_task
    self._execute_task_with_callbacks(context, test_mode, session=session)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1647, in _execute_task_with_callbacks
    task_orig = self.render_templates(context=context)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 2285, in render_templates
    original_task.render_template_fields(context)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/mappedoperator.py", line 725, in render_template_fields
    mapped_kwargs, seen_oids = self._expand_mapped_kwargs(context, session)
                               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/decorators/base.py", line 527, in _expand_mapped_kwargs
    raise AssertionError(f"unexpected expand_input: {self.expand_input}")
AssertionError: unexpected expand_input: DictOfListsExpandInput(value={})
[2024-07-31 12:01:55,840] {taskinstance.py:1983} ERROR - Unable to unmap task to determine if we need to send an alert email
ERROR [airflow.task] Unable to unmap task to determine if we need to send an alert email
[2024-07-31 12:01:55,840] {taskinstance.py:1400} INFO - Marking task as FAILED. dag_id=reproducible_dag, task_id=mapped, map_index=3, execution_date=20240731T160155, start_date=, end_date=20240731T160155
INFO  [airflow.task] Marking task as FAILED. dag_id=reproducible_dag, task_id=mapped, map_index=3, execution_date=20240731T160155, start_date=, end_date=20240731T160155
ERROR [airflow.models.dag.DAG] Task failed; ti=<TaskInstance: reproducible_dag.mapped manual__2024-07-31T16:01:55.743454+00:00 map_index=3 [failed]>
Traceback (most recent call last):
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/dag.py", line 2776, in test
    ret = _run_task(ti, session=session)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/dag.py", line 3938, in _run_task
    ret = ti._run_raw_task(session=session)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/utils/session.py", line 76, in wrapper
    return func(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1518, in _run_raw_task
    self._execute_task_with_callbacks(context, test_mode, session=session)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1647, in _execute_task_with_callbacks
    task_orig = self.render_templates(context=context)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 2285, in render_templates
    original_task.render_template_fields(context)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/mappedoperator.py", line 725, in render_template_fields
    mapped_kwargs, seen_oids = self._expand_mapped_kwargs(context, session)
                               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/decorators/base.py", line 527, in _expand_mapped_kwargs
    raise AssertionError(f"unexpected expand_input: {self.expand_input}")
AssertionError: unexpected expand_input: DictOfListsExpandInput(value={})
[2024-07-31 12:01:55,847] {taskinstance.py:1937} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1518, in _run_raw_task
    self._execute_task_with_callbacks(context, test_mode, session=session)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1647, in _execute_task_with_callbacks
    task_orig = self.render_templates(context=context)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 2285, in render_templates
    original_task.render_template_fields(context)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/mappedoperator.py", line 725, in render_template_fields
    mapped_kwargs, seen_oids = self._expand_mapped_kwargs(context, session)
                               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/decorators/base.py", line 527, in _expand_mapped_kwargs
    raise AssertionError(f"unexpected expand_input: {self.expand_input}")
AssertionError: unexpected expand_input: DictOfListsExpandInput(value={})
ERROR [airflow.task] Task failed with exception
Traceback (most recent call last):
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1518, in _run_raw_task
    self._execute_task_with_callbacks(context, test_mode, session=session)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1647, in _execute_task_with_callbacks
    task_orig = self.render_templates(context=context)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 2285, in render_templates
    original_task.render_template_fields(context)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/mappedoperator.py", line 725, in render_template_fields
    mapped_kwargs, seen_oids = self._expand_mapped_kwargs(context, session)
                               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/decorators/base.py", line 527, in _expand_mapped_kwargs
    raise AssertionError(f"unexpected expand_input: {self.expand_input}")
AssertionError: unexpected expand_input: DictOfListsExpandInput(value={})
[2024-07-31 12:01:55,848] {taskinstance.py:1983} ERROR - Unable to unmap task to determine if we need to send an alert email
ERROR [airflow.task] Unable to unmap task to determine if we need to send an alert email
[2024-07-31 12:01:55,848] {taskinstance.py:1400} INFO - Marking task as FAILED. dag_id=reproducible_dag, task_id=mapped, map_index=4, execution_date=20240731T160155, start_date=, end_date=20240731T160155
INFO  [airflow.task] Marking task as FAILED. dag_id=reproducible_dag, task_id=mapped, map_index=4, execution_date=20240731T160155, start_date=, end_date=20240731T160155
ERROR [airflow.models.dag.DAG] Task failed; ti=<TaskInstance: reproducible_dag.mapped manual__2024-07-31T16:01:55.743454+00:00 map_index=4 [failed]>
Traceback (most recent call last):
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/dag.py", line 2776, in test
    ret = _run_task(ti, session=session)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/dag.py", line 3938, in _run_task
    ret = ti._run_raw_task(session=session)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/utils/session.py", line 76, in wrapper
    return func(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1518, in _run_raw_task
    self._execute_task_with_callbacks(context, test_mode, session=session)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1647, in _execute_task_with_callbacks
    task_orig = self.render_templates(context=context)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 2285, in render_templates
    original_task.render_template_fields(context)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/mappedoperator.py", line 725, in render_template_fields
    mapped_kwargs, seen_oids = self._expand_mapped_kwargs(context, session)
                               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/decorators/base.py", line 527, in _expand_mapped_kwargs
    raise AssertionError(f"unexpected expand_input: {self.expand_input}")
AssertionError: unexpected expand_input: DictOfListsExpandInput(value={})
[2024-07-31 12:01:55,853] {taskinstance.py:1937} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1518, in _run_raw_task
    self._execute_task_with_callbacks(context, test_mode, session=session)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1647, in _execute_task_with_callbacks
    task_orig = self.render_templates(context=context)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 2285, in render_templates
    original_task.render_template_fields(context)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/mappedoperator.py", line 725, in render_template_fields
    mapped_kwargs, seen_oids = self._expand_mapped_kwargs(context, session)
                               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/decorators/base.py", line 527, in _expand_mapped_kwargs
    raise AssertionError(f"unexpected expand_input: {self.expand_input}")
AssertionError: unexpected expand_input: DictOfListsExpandInput(value={})
ERROR [airflow.task] Task failed with exception
Traceback (most recent call last):
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1518, in _run_raw_task
    self._execute_task_with_callbacks(context, test_mode, session=session)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1647, in _execute_task_with_callbacks
    task_orig = self.render_templates(context=context)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 2285, in render_templates
    original_task.render_template_fields(context)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/mappedoperator.py", line 725, in render_template_fields
    mapped_kwargs, seen_oids = self._expand_mapped_kwargs(context, session)
                               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/decorators/base.py", line 527, in _expand_mapped_kwargs
    raise AssertionError(f"unexpected expand_input: {self.expand_input}")
AssertionError: unexpected expand_input: DictOfListsExpandInput(value={})
[2024-07-31 12:01:55,854] {taskinstance.py:1983} ERROR - Unable to unmap task to determine if we need to send an alert email
ERROR [airflow.task] Unable to unmap task to determine if we need to send an alert email
[2024-07-31 12:01:55,854] {taskinstance.py:1400} INFO - Marking task as FAILED. dag_id=reproducible_dag, task_id=mapped, map_index=5, execution_date=20240731T160155, start_date=, end_date=20240731T160155
INFO  [airflow.task] Marking task as FAILED. dag_id=reproducible_dag, task_id=mapped, map_index=5, execution_date=20240731T160155, start_date=, end_date=20240731T160155
ERROR [airflow.models.dag.DAG] Task failed; ti=<TaskInstance: reproducible_dag.mapped manual__2024-07-31T16:01:55.743454+00:00 map_index=5 [failed]>
Traceback (most recent call last):
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/dag.py", line 2776, in test
    ret = _run_task(ti, session=session)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/dag.py", line 3938, in _run_task
    ret = ti._run_raw_task(session=session)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/utils/session.py", line 76, in wrapper
    return func(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1518, in _run_raw_task
    self._execute_task_with_callbacks(context, test_mode, session=session)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1647, in _execute_task_with_callbacks
    task_orig = self.render_templates(context=context)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 2285, in render_templates
    original_task.render_template_fields(context)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/mappedoperator.py", line 725, in render_template_fields
    mapped_kwargs, seen_oids = self._expand_mapped_kwargs(context, session)
                               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/decorators/base.py", line 527, in _expand_mapped_kwargs
    raise AssertionError(f"unexpected expand_input: {self.expand_input}")
AssertionError: unexpected expand_input: DictOfListsExpandInput(value={})
[2024-07-31 12:01:55,860] {taskinstance.py:1937} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1518, in _run_raw_task
    self._execute_task_with_callbacks(context, test_mode, session=session)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1647, in _execute_task_with_callbacks
    task_orig = self.render_templates(context=context)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 2285, in render_templates
    original_task.render_template_fields(context)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/mappedoperator.py", line 725, in render_template_fields
    mapped_kwargs, seen_oids = self._expand_mapped_kwargs(context, session)
                               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/decorators/base.py", line 527, in _expand_mapped_kwargs
    raise AssertionError(f"unexpected expand_input: {self.expand_input}")
AssertionError: unexpected expand_input: DictOfListsExpandInput(value={})
ERROR [airflow.task] Task failed with exception
Traceback (most recent call last):
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1518, in _run_raw_task
    self._execute_task_with_callbacks(context, test_mode, session=session)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1647, in _execute_task_with_callbacks
    task_orig = self.render_templates(context=context)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 2285, in render_templates
    original_task.render_template_fields(context)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/mappedoperator.py", line 725, in render_template_fields
    mapped_kwargs, seen_oids = self._expand_mapped_kwargs(context, session)
                               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/decorators/base.py", line 527, in _expand_mapped_kwargs
    raise AssertionError(f"unexpected expand_input: {self.expand_input}")
AssertionError: unexpected expand_input: DictOfListsExpandInput(value={})
[2024-07-31 12:01:55,862] {taskinstance.py:1983} ERROR - Unable to unmap task to determine if we need to send an alert email
ERROR [airflow.task] Unable to unmap task to determine if we need to send an alert email
[2024-07-31 12:01:55,862] {taskinstance.py:1400} INFO - Marking task as FAILED. dag_id=reproducible_dag, task_id=mapped, map_index=6, execution_date=20240731T160155, start_date=, end_date=20240731T160155
INFO  [airflow.task] Marking task as FAILED. dag_id=reproducible_dag, task_id=mapped, map_index=6, execution_date=20240731T160155, start_date=, end_date=20240731T160155
ERROR [airflow.models.dag.DAG] Task failed; ti=<TaskInstance: reproducible_dag.mapped manual__2024-07-31T16:01:55.743454+00:00 map_index=6 [failed]>
Traceback (most recent call last):
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/dag.py", line 2776, in test
    ret = _run_task(ti, session=session)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/dag.py", line 3938, in _run_task
    ret = ti._run_raw_task(session=session)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/utils/session.py", line 76, in wrapper
    return func(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1518, in _run_raw_task
    self._execute_task_with_callbacks(context, test_mode, session=session)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 1647, in _execute_task_with_callbacks
    task_orig = self.render_templates(context=context)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 2285, in render_templates
    original_task.render_template_fields(context)
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/models/mappedoperator.py", line 725, in render_template_fields
    mapped_kwargs, seen_oids = self._expand_mapped_kwargs(context, session)
                               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/oro/.local/share/virtualenvs/airflow-JBf5o6AR/lib/python3.11/site-packages/airflow/decorators/base.py", line 527, in _expand_mapped_kwargs
    raise AssertionError(f"unexpected expand_input: {self.expand_input}")
AssertionError: unexpected expand_input: DictOfListsExpandInput(value={})
ERROR [airflow.models.dagrun.DagRun] Marking run <DagRun reproducible_dag @ 2024-07-31T16:01:55.743454+00:00: manual__2024-07-31T16:01:55.743454+00:00, state:running, queued_at: None. externally triggered: False> failed
Traceback (most recent call last):
  File "/Users/oro/Library/Application Support/Code/User/globalStorage/buenon.scratchpads/scratchpads/1a4f2b73c916c9af3926773a674c6453/scratch107.py", line 74, in <module>
    assert dag_run.get_state() == DagRunState.SUCCESS
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
AssertionError

Are you willing to submit PR?

Code of Conduct

boring-cyborg[bot] commented 1 month ago

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

harjeevanmaan commented 4 weeks ago

I would like to work on it