diff --git a/airflow-core/newsfragments/69205.bugfix.rst b/airflow-core/newsfragments/69205.bugfix.rst new file mode 100644 index 0000000000000..f0a7bca7c2cba --- /dev/null +++ b/airflow-core/newsfragments/69205.bugfix.rst @@ -0,0 +1 @@ +Fix the mapped XCom sequence-slice API returning incorrect rows for crossed-bound slices (where ``stop`` comes before ``start``). Such slices now return an empty result, matching Python slicing, instead of emitting a negative SQL ``LIMIT`` -- which silently returned wrong rows on SQLite and failed the request on PostgreSQL/MySQL. diff --git a/airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py b/airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py index 9ab8eb200712a..f6ca14d177385 100644 --- a/airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py +++ b/airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py @@ -178,6 +178,22 @@ class GetXComSliceFilterParams(BaseModel): include_prior_dates: bool = False +def _sliced_or_empty(query: Select, low: int, high: int) -> Select: + """ + Apply ``.slice(low, high)`` but return no rows when the bounds are crossed. + + ``Select.slice(low, high)`` compiles to ``OFFSET low LIMIT (high - low)`` and does not + clamp a negative limit. A crossed slice (``high <= low``), which Python evaluates to an + empty sequence, would otherwise emit a negative SQL ``LIMIT`` -- silently returning rows + from ``OFFSET`` onward on SQLite (negative ``LIMIT`` means "no limit") and raising on + PostgreSQL/MySQL. By this point the query is already ordered and step handling is applied + separately, so each slice is an ascending window and ``high <= low`` is exactly the empty case. + """ + if high <= low: + return query.limit(0) + return query.slice(low, high) + + @router.get( "/{dag_id}/{run_id}/{task_id}/{key:path}/slice", description="Get XCom values from a mapped task by sequence slice", @@ -235,9 +251,9 @@ def get_mapped_xcom_by_slice( if stop < 0: stop += get_query_count(query, session=session) if step >= 0: - query = query.slice(start, stop) + query = _sliced_or_empty(query, start, stop) else: - query = query.slice(stop + 1, start + 1) + query = _sliced_or_empty(query, stop + 1, start + 1) else: query = query.order_by(XComModel.map_index.desc()) step = -step @@ -250,9 +266,9 @@ def get_mapped_xcom_by_slice( if stop >= 0: stop -= get_query_count(query, session=session) if step > 0: - query = query.slice(-1 - start, -1 - stop) + query = _sliced_or_empty(query, -1 - start, -1 - stop) else: - query = query.slice(-stop, -start) + query = _sliced_or_empty(query, -stop, -start) values = [row.value for row in session.execute(query.with_only_columns(XComModel.value)).all()] if step != 1: diff --git a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py index 277e2a78fd8dc..6dff173ba58a5 100644 --- a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py +++ b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py @@ -237,6 +237,12 @@ def __init__(self, *, x, **kwargs): pytest.param(slice(-1, None, -1), id="-1::-1"), pytest.param(slice(-2, -1, None), id="-2:-1"), pytest.param(slice(-1, -3, -1), id="-1:-3:-1"), + # Crossed-bound slices (stop before start) must return [] rather than + # emitting a negative SQL LIMIT; one case per .slice() branch. + pytest.param(slice(2, 1, None), id="2:1"), + pytest.param(slice(0, 1, -1), id="0:1:-1"), + pytest.param(slice(-2, -1, -1), id="-2:-1:-1"), + pytest.param(slice(-1, -2, None), id="-1:-2"), ], ) def test_xcom_get_with_slice(self, client, dag_maker, session, key):