ethan-oro opened a new issue, #41160:
URL: https://github.com/apache/airflow/issues/41160
### 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 to
fail with following error: `AssertionError: unexpected expand_input:
DictOfListsExpandInput(value={})` raises
[here](https://github.com/apache/airflow/blob/main/airflow/decorators/base.py#L558);
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](https://github.com/apache/airflow/blob/main/airflow/models/dag.py#L2581)
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
```python
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
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()
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()
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](https://github.com/Recidiviz/pulse-data/blob/main/recidiviz/tools/postgres/local_postgres_helpers.py)
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?
- [X] Yes I am willing to submit a PR!
### Code of Conduct
- [X] I agree to follow this project's [Code of
Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
