diff --git a/airflow-core/src/airflow/ui/public/i18n/locales/en/components.json b/airflow-core/src/airflow/ui/public/i18n/locales/en/components.json index 7de3c1683a2be..19790cb7bc242 100644 --- a/airflow-core/src/airflow/ui/public/i18n/locales/en/components.json +++ b/airflow-core/src/airflow/ui/public/i18n/locales/en/components.json @@ -5,6 +5,7 @@ "affectedNone": "No runs matching selected criteria.", "allRuns": "All Runs", "backwards": "Run Backwards", + "completed": "Completed", "dateRange": "Date Range", "errorStartDateBeforeEndDate": "Start Date must be before the End Date", "exceptionReason": { @@ -24,10 +25,12 @@ "permissionDenied": "Dry Run Failed: User does not have permission to create backfills.", "reprocessBehavior": "Reprocess Behavior", "run": "Run Backfill", + "running": "Running", "scheduleNotBackfillable": "This Dag's schedule does not support backfills", "schedulerPriorityHint": "Backfill Dag runs are ordered after non-backfill Dag runs in each scheduler cycle. Backfill runs may remain queued longer if other non-backfill runs are present.", "selectDescription": "Run this Dag for a range of dates", "selectLabel": "Backfill", + "status": "Status", "title": "Run Backfill", "toaster": { "success": { diff --git a/airflow-core/src/airflow/ui/public/i18n/locales/en/dags.json b/airflow-core/src/airflow/ui/public/i18n/locales/en/dags.json index 5bd68179bc3e8..16af49b5d0f7e 100644 --- a/airflow-core/src/airflow/ui/public/i18n/locales/en/dags.json +++ b/airflow-core/src/airflow/ui/public/i18n/locales/en/dags.json @@ -22,6 +22,7 @@ }, "favoriteState": "Favorite state", "favoriteStatePlaceholder": "Select favorite state…", + "hasImportErrors": "Has import errors", "noTimetableTypesFound": "No timetable types found", "paused": { "active": "Not paused", diff --git a/airflow-core/src/airflow/ui/src/components/Assets/AssetEventsFilter.tsx b/airflow-core/src/airflow/ui/src/components/Assets/AssetEventsFilter.tsx index 50fa277a59490..7f0575c4f5fe3 100644 --- a/airflow-core/src/airflow/ui/src/components/Assets/AssetEventsFilter.tsx +++ b/airflow-core/src/airflow/ui/src/components/Assets/AssetEventsFilter.tsx @@ -25,6 +25,9 @@ const FILTER_KEYS: Array = [ SearchParamsKeys.ASSET_EVENT_DATE_RANGE, SearchParamsKeys.DAG_ID, SearchParamsKeys.TASK_ID, + SearchParamsKeys.SOURCE_RUN_ID, + SearchParamsKeys.SOURCE_MAP_INDEX, + SearchParamsKeys.PARTITION_KEY, ]; export const AssetEventsFilter = () => { diff --git a/airflow-core/src/airflow/ui/src/constants/filterConfigs.tsx b/airflow-core/src/airflow/ui/src/constants/filterConfigs.tsx index 05a04f4b4f5d1..3b47e30c74aff 100644 --- a/airflow-core/src/airflow/ui/src/constants/filterConfigs.tsx +++ b/airflow-core/src/airflow/ui/src/constants/filterConfigs.tsx @@ -28,6 +28,7 @@ import { MdCode, MdComputer, MdDateRange, + MdErrorOutline, MdHistory, MdHourglassEmpty, MdLoop, @@ -114,6 +115,15 @@ export const useFilterConfigs = () => { })); const filterConfigMap = { + [SearchParamsKeys.ACTIVE]: { + icon: , + label: translate("components:backfill.status"), + options: [ + { label: translate("components:backfill.running"), value: "true" }, + { label: translate("components:backfill.completed"), value: "false" }, + ], + type: FilterTypes.SELECT, + }, [SearchParamsKeys.ASSET_EVENT_DATE_RANGE]: { endKey: SearchParamsKeys.END_DATE, icon: , @@ -266,6 +276,15 @@ export const useFilterConfigs = () => { ], type: FilterTypes.SELECT, }, + [SearchParamsKeys.HAS_IMPORT_ERRORS]: { + icon: , + label: translate("dags:filters.hasImportErrors"), + options: [ + { label: translate("common:yes"), value: "true" }, + { label: translate("common:no"), value: "false" }, + ], + type: FilterTypes.SELECT, + }, [SearchParamsKeys.HOSTNAME]: { hotkeyDisabled: true, icon: , @@ -373,6 +392,12 @@ export const useFilterConfigs = () => { placeholder: translate("common:table.ownerPlaceholder"), type: FilterTypes.MULTISELECT, }, + [SearchParamsKeys.PARTITION_KEY]: { + hotkeyDisabled: true, + icon: , + label: translate("common:dagRun.partitionKey"), + type: FilterTypes.TEXT, + }, [SearchParamsKeys.PARTITION_KEY_PATTERN]: { hotkeyDisabled: true, icon: , @@ -490,6 +515,18 @@ export const useFilterConfigs = () => { })), type: FilterTypes.SELECT, }, + [SearchParamsKeys.SOURCE_MAP_INDEX]: { + icon: , + label: translate("common:mapIndex"), + min: -1, + type: FilterTypes.NUMBER, + }, + [SearchParamsKeys.SOURCE_RUN_ID]: { + hotkeyDisabled: true, + icon: , + label: translate("common:runId"), + type: FilterTypes.TEXT, + }, [SearchParamsKeys.START_DATE_RANGE]: { endKey: SearchParamsKeys.START_DATE_LTE, icon: , diff --git a/airflow-core/src/airflow/ui/src/constants/searchParams.ts b/airflow-core/src/airflow/ui/src/constants/searchParams.ts index 66f6f788ec37e..ca604eefc14a6 100644 --- a/airflow-core/src/airflow/ui/src/constants/searchParams.ts +++ b/airflow-core/src/airflow/ui/src/constants/searchParams.ts @@ -17,6 +17,7 @@ * under the License. */ export enum SearchParamsKeys { + ACTIVE = "active", ADVANCED_SEARCH = "advanced_search", AFTER = "after", ASSET_EVENT_DATE_RANGE = "asset_event_date_range", @@ -71,6 +72,7 @@ export enum SearchParamsKeys { GRAPH_TASK_STATE = "graph-task_state", GROUP_PATTERN = "group_pattern", HAS_EVENTS = "has_events", + HAS_IMPORT_ERRORS = "has_import_errors", HOSTNAME = "hostname", INCLUDED_EVENTS = "included_events", ITERATION = "iteration", @@ -100,6 +102,7 @@ export enum SearchParamsKeys { OPERATOR = "operator", OPERATOR_NAME_PATTERN = "operator_name_pattern", OWNERS = "owners", + PARTITION_KEY = "partition_key", PARTITION_KEY_PATTERN = "partition_key_pattern", PAUSED = "paused", POOL = "pool", @@ -122,6 +125,8 @@ export enum SearchParamsKeys { SCHEDULING_STATE = "scheduling_state", SORT = "sort", SOURCE = "log_source", + SOURCE_MAP_INDEX = "source_map_index", + SOURCE_RUN_ID = "source_run_id", START_DATE = "start_date", START_DATE_GTE = "start_date_gte", START_DATE_LTE = "start_date_lte", diff --git a/airflow-core/src/airflow/ui/src/mocks/handlers/assets.ts b/airflow-core/src/airflow/ui/src/mocks/handlers/assets.ts index 4bbae3e92cf3f..e8a1217f4dba5 100644 --- a/airflow-core/src/airflow/ui/src/mocks/handlers/assets.ts +++ b/airflow-core/src/airflow/ui/src/mocks/handlers/assets.ts @@ -42,8 +42,53 @@ const plainAsset = { watchers: [], }; +const assetEvent = { + asset_id: 1, + created_dagruns: [], + extra: {}, + group: "asset_group", + id: 100, + name: "asset_with_dependencies", + partition_key: null, + source_dag_id: "producer_dag", + source_map_index: -1, + source_run_id: "manual__2025-01-10", + source_task_id: "producer_task", + timestamp: "2025-01-10T00:00:00Z", + uri: "s3://bucket/asset", +}; + +// Same asset, but a mapped and partitioned event, so each of the new filters has two rows to +// choose between rather than one row that always matches. +const mappedPartitionedEvent = { + ...assetEvent, + id: 101, + partition_key: "2025-01-11", + source_dag_id: "scheduled_producer_dag", + source_map_index: 3, + source_run_id: "scheduled__2025-01-11", + timestamp: "2025-01-11T00:00:00Z", +}; + +export const assetEvents = [assetEvent, mappedPartitionedEvent]; + export const handlers: Array = [ http.get("/ui/assets", () => HttpResponse.json({ assets: [assetWithDependencies], total_entries: 1 })), + http.get("/api/v2/assets/events", ({ request }) => { + const url = new URL(request.url); + const sourceRunId = url.searchParams.get("source_run_id"); + const sourceMapIndex = url.searchParams.get("source_map_index"); + const partitionKey = url.searchParams.get("partition_key"); + + const filtered = assetEvents.filter( + (event) => + (sourceRunId === null || event.source_run_id === sourceRunId) && + (partitionKey === null || event.partition_key === partitionKey) && + (sourceMapIndex === null || event.source_map_index === Number(sourceMapIndex)), + ); + + return HttpResponse.json({ asset_events: filtered, total_entries: filtered.length }); + }), http.get("/api/v2/assets/:assetId", ({ params }) => HttpResponse.json(params.assetId === "2" ? plainAsset : assetWithDependencies), ), diff --git a/airflow-core/src/airflow/ui/src/mocks/handlers/backfills.ts b/airflow-core/src/airflow/ui/src/mocks/handlers/backfills.ts index ee895e16987e0..c0d821ee7c6cf 100644 --- a/airflow-core/src/airflow/ui/src/mocks/handlers/backfills.ts +++ b/airflow-core/src/airflow/ui/src/mocks/handlers/backfills.ts @@ -48,6 +48,22 @@ const backfillInRange = { updated_at: "2025-01-16T01:00:00Z", }; +// Still running, so ``completed_at`` is null — the one row the ``active`` filter can single out. +const backfillRunning = { + completed_at: null, + created_at: "2025-02-01T00:00:00Z", + dag_display_name: "tutorial_taskflow_api", + dag_id: "tutorial_taskflow_api", + dag_run_conf: {}, + from_date: "2025-02-01T00:00:00Z", + id: 3, + is_paused: false, + max_active_runs: 10, + reprocess_behavior: "none", + to_date: "2025-02-15T00:00:00Z", + updated_at: "2025-02-01T00:00:00Z", +}; + export const handlers: Array = [ http.get("/ui/backfills", ({ request }) => { const url = new URL(request.url); @@ -55,13 +71,19 @@ export const handlers: Array = [ const fromDateLte = url.searchParams.get("from_date_lte"); const toDateGte = url.searchParams.get("to_date_gte"); const toDateLte = url.searchParams.get("to_date_lte"); + const active = url.searchParams.get("active"); - const allBackfills = [backfillBeforeFilter, backfillInRange]; + const allBackfills = [backfillBeforeFilter, backfillInRange, backfillRunning]; const filtered = allBackfills.filter((backfill) => { const fromDate = new Date(backfill.from_date); const toDate = new Date(backfill.to_date); + // The backend models this as "completed_at is null". + if (active !== null && (backfill.completed_at === null) !== (active === "true")) { + return false; + } + if (fromDateGte !== null && fromDate < new Date(fromDateGte)) { return false; } diff --git a/airflow-core/src/airflow/ui/src/mocks/handlers/dags.ts b/airflow-core/src/airflow/ui/src/mocks/handlers/dags.ts index 41b4144c54ec0..d3b30d91a5adf 100644 --- a/airflow-core/src/airflow/ui/src/mocks/handlers/dags.ts +++ b/airflow-core/src/airflow/ui/src/mocks/handlers/dags.ts @@ -137,21 +137,37 @@ export const drainingDag = { timetable_type: "NullTimetable", }; +// Failed to import, so the Dags list has something for the import-errors filter to single out. +export const brokenDag = { + ...pausedDag, + dag_display_name: "broken_dag", + dag_id: "broken_dag", + has_import_errors: true, + is_paused: false, + tags: [{ dag_id: "broken_dag", name: "example" }], +}; + export const defaultDags = [successDag, failedDag, pausedDag]; const filterDags = ({ dags, + hasImportErrors, paused, schedulingState, timetableTypes, }: { - dags: Array<(typeof defaultDags)[number] | typeof drainingDag>; + dags: Array<(typeof defaultDags)[number] | typeof brokenDag | typeof drainingDag>; + hasImportErrors: string | null; paused: string | null; schedulingState: string | null; timetableTypes: Array; }) => { let allDags = dags; + if (hasImportErrors !== null) { + allDags = allDags.filter((dag) => dag.has_import_errors === (hasImportErrors === "true")); + } + if (timetableTypes.length > 0) { allDags = allDags.filter((dag) => timetableTypes.includes(dag.timetable_type)); } @@ -177,13 +193,14 @@ const filterDags = ({ * enlarging the set every other suite renders. */ export const dagsListHandler = ( - dags: Array<(typeof defaultDags)[number] | typeof drainingDag> = defaultDags, + dags: Array<(typeof defaultDags)[number] | typeof brokenDag | typeof drainingDag> = defaultDags, ) => http.get("/ui/dags", ({ request }) => { const url = new URL(request.url); const lastDagRunState = url.searchParams.get("last_dag_run_state"); const anyDagRunState = url.searchParams.get("dag_run_state"); const orderBy = url.searchParams.get("order_by"); + const hasImportErrors = url.searchParams.get("has_import_errors"); const paused = url.searchParams.get("paused"); const schedulingState = url.searchParams.get("scheduling_state"); const timetableTypes = url.searchParams.getAll("timetable_type"); @@ -209,7 +226,7 @@ export const dagsListHandler = ( }); } - let filtered = filterDags({ dags, paused, schedulingState, timetableTypes }); + let filtered = filterDags({ dags, hasImportErrors, paused, schedulingState, timetableTypes }); if (orderBy === "last_run_run_after" || orderBy === "-last_run_run_after") { filtered = [failedDag, successDag, pausedDag, drainingDag].filter((dag) => filtered.includes(dag)); diff --git a/airflow-core/src/airflow/ui/src/pages/Asset/AssetEvents.tsx b/airflow-core/src/airflow/ui/src/pages/Asset/AssetEvents.tsx index 8b8b98d34cdb4..2d00112c282bd 100644 --- a/airflow-core/src/airflow/ui/src/pages/Asset/AssetEvents.tsx +++ b/airflow-core/src/airflow/ui/src/pages/Asset/AssetEvents.tsx @@ -35,8 +35,10 @@ export const AssetEvents = () => { const [sort] = sorting; const orderBy = sort ? [`${sort.desc ? "-" : ""}${sort.id}`] : ["-timestamp"]; - const { DAG_ID, END_DATE, START_DATE, TASK_ID } = SearchParamsKeys; + const { DAG_ID, END_DATE, PARTITION_KEY, SOURCE_MAP_INDEX, SOURCE_RUN_ID, START_DATE, TASK_ID } = + SearchParamsKeys; const [searchParams] = useSearchParams(); + const sourceMapIndex = searchParams.get(SOURCE_MAP_INDEX); const { data, isLoading } = useAssetServiceGetAssetEvents( { @@ -44,7 +46,10 @@ export const AssetEvents = () => { limit: pagination.pageSize, offset: pagination.pageIndex * pagination.pageSize, orderBy, + partitionKey: searchParams.get(PARTITION_KEY) ?? undefined, sourceDagId: searchParams.get(DAG_ID) ?? undefined, + sourceMapIndex: sourceMapIndex === null || sourceMapIndex === "" ? undefined : Number(sourceMapIndex), + sourceRunId: searchParams.get(SOURCE_RUN_ID) ?? undefined, sourceTaskId: searchParams.get(TASK_ID) ?? undefined, timestampGte: searchParams.get(START_DATE) ?? undefined, timestampLte: searchParams.get(END_DATE) ?? undefined, diff --git a/airflow-core/src/airflow/ui/src/pages/Asset/AssetEventsFilters.test.tsx b/airflow-core/src/airflow/ui/src/pages/Asset/AssetEventsFilters.test.tsx new file mode 100644 index 0000000000000..905e11c336626 --- /dev/null +++ b/airflow-core/src/airflow/ui/src/pages/Asset/AssetEventsFilters.test.tsx @@ -0,0 +1,48 @@ +/*! + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +import "@testing-library/jest-dom/vitest"; +import { render, screen, waitFor } from "@testing-library/react"; +import { describe, expect, it } from "vitest"; + +import { AppWrapper } from "src/utils/AppWrapper"; + +// The assets mock handler serves two events for asset 1: +// id 100 — producer_dag, run manual__2025-01-10, map index -1, no partition key +// id 101 — scheduled_producer_dag, run scheduled__2025-01-11, map index 3, partition key 2025-01-11 +// The card shows the source Dag id rather than the run id, so assertions use the Dag id. +const route = (query: string) => `/assets/1?${query}`; + +describe("Asset event filters", () => { + it.each([ + { kept: "producer_dag", query: "source_run_id=manual__2025-01-10" }, + { kept: "scheduled_producer_dag", query: "source_map_index=3" }, + { kept: "scheduled_producer_dag", query: "partition_key=2025-01-11" }, + ])("narrows the events to one row for $query", async ({ kept, query }) => { + render(); + + await waitFor(() => expect(screen.getByText(kept)).toBeInTheDocument()); + }); + + it("shows both events when no filter is applied", async () => { + render(); + + await waitFor(() => expect(screen.getByText("producer_dag")).toBeInTheDocument()); + expect(screen.getByText("scheduled_producer_dag")).toBeInTheDocument(); + }); +}); diff --git a/airflow-core/src/airflow/ui/src/pages/Dag/Backfills/Backfills.tsx b/airflow-core/src/airflow/ui/src/pages/Dag/Backfills/Backfills.tsx index 0cd6772d70544..6d91e0b82c0ae 100644 --- a/airflow-core/src/airflow/ui/src/pages/Dag/Backfills/Backfills.tsx +++ b/airflow-core/src/airflow/ui/src/pages/Dag/Backfills/Backfills.tsx @@ -49,6 +49,7 @@ import { BackfillDagRunsModal } from "./BackfillDagRunsModal"; import { BackfillsFilters } from "./BackfillsFilters"; const { + ACTIVE: ACTIVE_PARAM, COMPLETED_AT_GTE: COMPLETED_AT_GTE_PARAM, COMPLETED_AT_LTE: COMPLETED_AT_LTE_PARAM, CREATED_AT_GTE: CREATED_AT_GTE_PARAM, @@ -228,9 +229,12 @@ export const Backfills = () => { const durationGte = searchParams.get(DURATION_GTE_PARAM); const durationLte = searchParams.get(DURATION_LTE_PARAM); const reprocessBehaviorParam = searchParams.get(REPROCESS_BEHAVIOR_PARAM); + const activeParam = searchParams.get(ACTIVE_PARAM); const reprocessBehavior = isReprocessBehavior(reprocessBehaviorParam) ? reprocessBehaviorParam : undefined; const { data, error, isLoading } = useBackfillServiceListBackfillsUi({ + // The backend reads this as "completed_at is null", so only an explicit choice sends it. + active: activeParam === null || activeParam === "" ? undefined : activeParam === "true", completedAtGte: completedAtGte ?? undefined, completedAtLte: completedAtLte ?? undefined, createdAtGte: createdAtGte ?? undefined, diff --git a/airflow-core/src/airflow/ui/src/pages/Dag/Backfills/BackfillsDateFilter.test.tsx b/airflow-core/src/airflow/ui/src/pages/Dag/Backfills/BackfillsDateFilter.test.tsx index 2431d6823ae39..228b423d3d87a 100644 --- a/airflow-core/src/airflow/ui/src/pages/Dag/Backfills/BackfillsDateFilter.test.tsx +++ b/airflow-core/src/airflow/ui/src/pages/Dag/Backfills/BackfillsDateFilter.test.tsx @@ -17,7 +17,7 @@ * under the License. */ import "@testing-library/jest-dom/vitest"; -import { render, screen, waitFor } from "@testing-library/react"; +import { render, screen, waitFor, within } from "@testing-library/react"; import { describe, expect, it } from "vitest"; import { AppWrapper } from "src/utils/AppWrapper"; @@ -50,3 +50,33 @@ describe("Backfills start/end date filter", () => { expect(getFromDateCells()).not.toContain("2024-12-01T00:00:00Z"); }); }); + +const getTableDateCells = () => + within(screen.getByRole("table")) + .getAllByTestId("time-display") + .map((element) => element.getAttribute("datetime")); + +describe("Backfills status filter", () => { + it.each([ + { expected: "2025-02-01T00:00:00Z", param: "active=true", unexpected: "2025-01-01T00:00:00Z" }, + { expected: "2025-01-01T00:00:00Z", param: "active=false", unexpected: "2025-02-01T00:00:00Z" }, + ])("filters backfills by $param", async ({ expected, param, unexpected }) => { + render(); + + // Both conditions in one waitFor: the list keeps the previous rows on screen while the new + // query resolves, so the absence only holds once that has landed. + // Scoped to the table: a running backfill also drives a banner elsewhere on the Dag page, + // whose dates would otherwise count as rows. + await waitFor(() => { + expect(getTableDateCells()).toContain(expected); + expect(getTableDateCells()).not.toContain(unexpected); + }); + }); + + it("shows running and completed backfills together when the filter is unset", async () => { + render(); + + await waitFor(() => expect(getTableDateCells()).toContain("2025-02-01T00:00:00Z")); + expect(getTableDateCells()).toContain("2025-01-01T00:00:00Z"); + }); +}); diff --git a/airflow-core/src/airflow/ui/src/pages/Dag/Backfills/BackfillsFilters.tsx b/airflow-core/src/airflow/ui/src/pages/Dag/Backfills/BackfillsFilters.tsx index 48b9694d6ec73..3f88e9dadd0ca 100644 --- a/airflow-core/src/airflow/ui/src/pages/Dag/Backfills/BackfillsFilters.tsx +++ b/airflow-core/src/airflow/ui/src/pages/Dag/Backfills/BackfillsFilters.tsx @@ -34,6 +34,7 @@ export const BackfillsFilters = () => { SearchParamsKeys.DURATION_GTE, SearchParamsKeys.DURATION_LTE, SearchParamsKeys.REPROCESS_BEHAVIOR, + SearchParamsKeys.ACTIVE, ]; const { filterConfigs, handleFiltersChange, initialValues } = useFiltersHandler(searchParamKeys); diff --git a/airflow-core/src/airflow/ui/src/pages/DagsList/DagsFilters/DagsFilters.test.tsx b/airflow-core/src/airflow/ui/src/pages/DagsList/DagsFilters/DagsFilters.test.tsx index 380fd374c678e..b6fa3491adc23 100644 --- a/airflow-core/src/airflow/ui/src/pages/DagsList/DagsFilters/DagsFilters.test.tsx +++ b/airflow-core/src/airflow/ui/src/pages/DagsList/DagsFilters/DagsFilters.test.tsx @@ -20,7 +20,7 @@ import "@testing-library/jest-dom/vitest"; import { fireEvent, render, screen, waitFor } from "@testing-library/react"; import { afterEach, describe, it, expect, vi } from "vitest"; -import { dagsListHandler, defaultDags, drainingDag } from "src/mocks/handlers/dags"; +import { brokenDag, dagsListHandler, defaultDags, drainingDag } from "src/mocks/handlers/dags"; import { server } from "src/mocks/server"; import { AppWrapper } from "src/utils/AppWrapper"; @@ -252,3 +252,26 @@ describe("Links written before draining moved into the pill", () => { expect(screen.getByText("draining_dag")).toBeInTheDocument(); }); }); + +const withBrokenDag = () => server.use(dagsListHandler([...defaultDags, brokenDag])); + +describe("Import errors filter", () => { + it.each([ + { expected: "broken_dag", param: "true", unexpected: "tutorial_taskflow_api_success" }, + { expected: "tutorial_taskflow_api_success", param: "false", unexpected: "broken_dag" }, + ])("filters dags by has_import_errors=$param", async ({ expected, param, unexpected }) => { + withBrokenDag(); + render(); + + await waitFor(() => expect(screen.getByText(expected)).toBeInTheDocument()); + expect(screen.queryByText(unexpected)).not.toBeInTheDocument(); + }); + + it("shows both when the filter is unset", async () => { + withBrokenDag(); + render(); + + await waitFor(() => expect(screen.getByText("broken_dag")).toBeInTheDocument()); + expect(screen.getByText("tutorial_taskflow_api_success")).toBeInTheDocument(); + }); +}); diff --git a/airflow-core/src/airflow/ui/src/pages/DagsList/DagsFilters/DagsFilters.tsx b/airflow-core/src/airflow/ui/src/pages/DagsList/DagsFilters/DagsFilters.tsx index d84ecc14fb609..0a5d92560baa6 100644 --- a/airflow-core/src/airflow/ui/src/pages/DagsList/DagsFilters/DagsFilters.tsx +++ b/airflow-core/src/airflow/ui/src/pages/DagsList/DagsFilters/DagsFilters.tsx @@ -37,6 +37,7 @@ export const DagsFilters = () => { SearchParamsKeys.OWNERS, SearchParamsKeys.TIMETABLE_TYPE, SearchParamsKeys.FAVORITE, + SearchParamsKeys.HAS_IMPORT_ERRORS, ]; if (multiTeamEnabled) { diff --git a/airflow-core/src/airflow/ui/src/pages/DagsList/DagsList.tsx b/airflow-core/src/airflow/ui/src/pages/DagsList/DagsList.tsx index e7e37ff9f4416..7b209354ed649 100644 --- a/airflow-core/src/airflow/ui/src/pages/DagsList/DagsList.tsx +++ b/airflow-core/src/airflow/ui/src/pages/DagsList/DagsList.tsx @@ -67,6 +67,7 @@ const { DAG_RUN_STATE, DAG_RUN_STATE_WITHIN_HOURS, FAVORITE, + HAS_IMPORT_ERRORS, LAST_DAG_RUN_STATE, NAME_PATTERN, NEEDS_REVIEW, @@ -149,6 +150,7 @@ export const DagsList = () => { // Derived through the pill's own mapping so the query can never filter on more than the pill shows. const { paused, schedulingState } = pausedStateToQuery(pausedStateFromSearchParams(searchParams)); const showFavorites = searchParams.get(FAVORITE); + const importErrors = searchParams.get(HAS_IMPORT_ERRORS); const lastDagRunState = searchParams.get(LAST_DAG_RUN_STATE) as DagRunState; const dagRunState = searchParams.get(DAG_RUN_STATE) as DagRunState; @@ -216,6 +218,7 @@ export const DagsList = () => { }; let isFavorite = undefined; + let hasImportErrors = undefined; let pendingHitl = undefined; if (showFavorites === "true") { @@ -224,6 +227,12 @@ export const DagsList = () => { isFavorite = false; } + if (importErrors === "true") { + hasImportErrors = true; + } else if (importErrors === "false") { + hasImportErrors = false; + } + if (pendingReviews === "true") { pendingHitl = true; } else if (pendingReviews === "false") { @@ -237,6 +246,8 @@ export const DagsList = () => { dagRunsLimit, dagRunState, dagRunStateWithinHours: Boolean(dagRunState) ? withinHours : undefined, + excludeStale: hasImportErrors !== true, + hasImportErrors, isFavorite, lastDagRunState, limit: pagination.pageSize, diff --git a/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstances.test.tsx b/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstances.test.tsx index 217213439fdc5..d4a60bf3ecda0 100644 --- a/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstances.test.tsx +++ b/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstances.test.tsx @@ -326,3 +326,39 @@ describe("TaskInstances", () => { ); }); }); + +describe("TaskInstances date filters", () => { + beforeEach(() => { + mockParams = { dagId: "example_dag", runId: "manual__2026-06-07T00:00:00+00:00" }; + vi.mocked(useTaskInstanceServiceGetTaskInstances).mockReturnValue(getTaskInstancesResponse([])); + }); + + it("sends the run_after range from the URL", () => { + mockSearchParams = new URLSearchParams( + "run_after_gte=2026-06-01T00:00:00Z&run_after_lte=2026-06-30T00:00:00Z", + ); + + render(, { wrapper: Wrapper }); + + expect(useTaskInstanceServiceGetTaskInstances).toHaveBeenCalledWith( + expect.objectContaining({ + runAfterGte: "2026-06-01T00:00:00Z", + runAfterLte: "2026-06-30T00:00:00Z", + }), + undefined, + expect.anything(), + ); + }); + + it("still honours logical_date links even though the pill is gone", () => { + mockSearchParams = new URLSearchParams("logical_date_gte=2026-06-01T00:00:00Z"); + + render(, { wrapper: Wrapper }); + + expect(useTaskInstanceServiceGetTaskInstances).toHaveBeenCalledWith( + expect.objectContaining({ logicalDateGte: "2026-06-01T00:00:00Z" }), + undefined, + expect.anything(), + ); + }); +}); diff --git a/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstances.tsx b/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstances.tsx index b7b845cadde21..479ceaf0fa56c 100644 --- a/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstances.tsx +++ b/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstances.tsx @@ -85,6 +85,8 @@ const { POOL_NAME_PATTERN: POOL_NAME_PATTERN_PARAM, QUEUE_NAME_PATTERN: QUEUE_NAME_PATTERN_PARAM, RENDERED_MAP_INDEX: RENDERED_MAP_INDEX_PARAM, + RUN_AFTER_GTE: RUN_AFTER_GTE_PARAM, + RUN_AFTER_LTE: RUN_AFTER_LTE_PARAM, RUN_ID_PATTERN: RUN_ID_PATTERN_PARAM, START_DATE: START_DATE_PARAM, TASK_STATE: STATE_PARAM, @@ -339,6 +341,8 @@ export const TaskInstances = () => { const durationLte = searchParams.get(DURATION_LTE_PARAM); const logicalDateGte = searchParams.get(LOGICAL_DATE_GTE_PARAM); const logicalDateLte = searchParams.get(LOGICAL_DATE_LTE_PARAM); + const runAfterGte = searchParams.get(RUN_AFTER_GTE_PARAM); + const runAfterLte = searchParams.get(RUN_AFTER_LTE_PARAM); const tryNumberFilter = searchParams.get(TRY_NUMBER_PARAM); const mapIndexFilter = searchParams.get(MAP_INDEX_PARAM); const loopFilter = searchParams.get(SearchParamsKeys.LOOP_ID); @@ -423,6 +427,8 @@ export const TaskInstances = () => { logicalDateLte: logicalDateLte ?? undefined, loopId: loopInScope ? loopFilter : undefined, mapIndex: mapIndexFilter !== null && mapIndexFilter !== "" ? [Number(mapIndexFilter)] : undefined, + runAfterGte: runAfterGte ?? undefined, + runAfterLte: runAfterLte ?? undefined, ...operatorNameArg, orderBy, ...poolNameArg, diff --git a/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstancesFilter.test.tsx b/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstancesFilter.test.tsx new file mode 100644 index 0000000000000..4a3da0d5bd4c3 --- /dev/null +++ b/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstancesFilter.test.tsx @@ -0,0 +1,46 @@ +/*! + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +import "@testing-library/jest-dom/vitest"; +import { fireEvent, render, screen } from "@testing-library/react"; +import { describe, expect, it } from "vitest"; + +import { AppWrapper } from "src/utils/AppWrapper"; + +describe("Task Instances date filters", () => { + it("shows a pill for a logical_date link so the filter is visible and removable", async () => { + render(); + + // A registered filter renders its pill from the URL, so the param can be seen, changed and + // cleared. Honouring it without a pill would narrow the list with nothing explaining it, and + // Reset only clears filters it knows about. + const pill = await screen.findByTestId("logical_date_range-pill"); + + expect(pill).toBeInTheDocument(); + expect(screen.getByRole("button", { name: /Remove .* filter/u })).toBeInTheDocument(); + }); + + it("offers both run after and logical date in the add-filter menu", async () => { + render(); + + fireEvent.click(await screen.findByTestId("add-filter-button")); + + expect(await screen.findByTestId("add-filter-run_after_range")).toBeInTheDocument(); + expect(screen.getByTestId("add-filter-logical_date_range")).toBeInTheDocument(); + }); +}); diff --git a/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstancesFilter.tsx b/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstancesFilter.tsx index 7abfba112b16d..4015aeefe8647 100644 --- a/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstancesFilter.tsx +++ b/airflow-core/src/airflow/ui/src/pages/TaskInstances/TaskInstancesFilter.tsx @@ -38,6 +38,7 @@ const { POOL_NAME_PATTERN: POOL_NAME_PATTERN_PARAM, QUEUE_NAME_PATTERN: QUEUE_NAME_PATTERN_PARAM, RENDERED_MAP_INDEX: RENDERED_MAP_INDEX_PARAM, + RUN_AFTER_RANGE: RUN_AFTER_RANGE_PARAM, RUN_ID_PATTERN: RUN_ID_PATTERN_PARAM, TASK_STATE: STATE_PARAM, TEAMS: TEAMS_PARAM, @@ -49,6 +50,7 @@ export const TaskInstancesFilter = () => { const multiTeamEnabled = Boolean(useConfig("multi_team")); const paramKeys: Array = [ NAME_PATTERN_PARAM as FilterableSearchParamsKeys, + RUN_AFTER_RANGE_PARAM as FilterableSearchParamsKeys, LOGICAL_DATE_RANGE_PARAM as FilterableSearchParamsKeys, ASSET_EVENT_DATE_RANGE_PARAM as FilterableSearchParamsKeys, DURATION_GTE_PARAM as FilterableSearchParamsKeys, diff --git a/airflow-core/src/airflow/ui/src/queries/useDags.tsx b/airflow-core/src/airflow/ui/src/queries/useDags.tsx index 80c740f9ae0dc..31354ce54d1a1 100644 --- a/airflow-core/src/airflow/ui/src/queries/useDags.tsx +++ b/airflow-core/src/airflow/ui/src/queries/useDags.tsx @@ -30,6 +30,7 @@ export const useDags = ({ dagRunState, dagRunStateWithinHours, excludeStale = true, + hasImportErrors, isFavorite, lastDagRunState, limit, @@ -53,6 +54,7 @@ export const useDags = ({ dagRunState?: DagRunState; dagRunStateWithinHours?: number; excludeStale?: boolean; + hasImportErrors?: boolean; isFavorite?: boolean; lastDagRunState?: DagRunState; limit?: number; @@ -80,6 +82,7 @@ export const useDags = ({ dagRunState, dagRunStateWithinHours, excludeStale, + hasImportErrors, hasPendingActions: pendingHitl, isFavorite, lastDagRunState, diff --git a/airflow-core/src/airflow/ui/src/utils/useFiltersHandler.ts b/airflow-core/src/airflow/ui/src/utils/useFiltersHandler.ts index 18d434321786f..99786b38a9151 100644 --- a/airflow-core/src/airflow/ui/src/utils/useFiltersHandler.ts +++ b/airflow-core/src/airflow/ui/src/utils/useFiltersHandler.ts @@ -57,6 +57,7 @@ const handleDateRangeChange = ( }; export type FilterableSearchParamsKeys = + | SearchParamsKeys.ACTIVE | SearchParamsKeys.ASSET_EVENT_DATE_RANGE | SearchParamsKeys.BODY_SEARCH | SearchParamsKeys.BUNDLE_VERSION @@ -79,6 +80,7 @@ export type FilterableSearchParamsKeys = | SearchParamsKeys.FROM_RANGE | SearchParamsKeys.GROUP_PATTERN | SearchParamsKeys.HAS_EVENTS + | SearchParamsKeys.HAS_IMPORT_ERRORS | SearchParamsKeys.HOSTNAME | SearchParamsKeys.JOB_STATE | SearchParamsKeys.JOB_TYPE @@ -94,6 +96,7 @@ export type FilterableSearchParamsKeys = | SearchParamsKeys.NEEDS_REVIEW | SearchParamsKeys.OPERATOR_NAME_PATTERN | SearchParamsKeys.OWNERS + | SearchParamsKeys.PARTITION_KEY | SearchParamsKeys.PARTITION_KEY_PATTERN | SearchParamsKeys.PAUSED | SearchParamsKeys.POOL_NAME_PATTERN @@ -107,6 +110,8 @@ export type FilterableSearchParamsKeys = | SearchParamsKeys.RUN_ID_PATTERN | SearchParamsKeys.RUN_STATE | SearchParamsKeys.RUN_TYPE + | SearchParamsKeys.SOURCE_MAP_INDEX + | SearchParamsKeys.SOURCE_RUN_ID | SearchParamsKeys.START_DATE_RANGE | SearchParamsKeys.STATE | SearchParamsKeys.SUBJECT_SEARCH