From 3d3806a574e10f2c03f6fbca4730c342467c7ed6 Mon Sep 17 00:00:00 2001 From: Ramachandra Nalam Date: Wed, 14 Jan 2026 10:54:06 -0800 Subject: [PATCH 1/4] docs: resolve low-priority cleanups --- .../rbac/user-roles-permissions.md | 10 ++---- .../dagster-plus/hybrid/kubernetes/setup.md | 4 +-- .../etl-pipeline/6-create-a-sensor.md | 6 ++-- .../sensors/testing-run-status-sensors.md | 3 +- .../assets/asset-versioning-and-caching.md | 3 +- .../assets/passing-data-between-assets.md | 5 ++- .../troubleshooting-components.md | 16 +++++++++- .../aws/aws-lambda-pipeline.md | 3 +- .../external-pipelines/databricks-pipeline.md | 3 +- .../external-pipelines/kubernetes-pipeline.md | 3 +- .../guides/build/external-resources/index.md | 4 +-- .../using-bare-python-objects-as-resources.md | 6 ++-- ...dependencies-between-partitioned-assets.md | 22 ------------- .../partitioning-assets.md | 4 +-- .../projects/dagster-plus-project-history.md | 2 -- .../workspaces/creating-workspaces.md | 1 - .../workspaces/migrating-workspace-yaml.md | 32 +++++++++++++++++-- .../test/running-a-subset-of-asset-checks.md | 3 +- .../testing-partitioned-config-and-jobs.md | 4 +-- .../libraries/duckdb/reference.md | 9 +++--- .../duckdb/using-duckdb-with-dagster.md | 8 ++--- .../libraries/gcp/bigquery/reference.md | 5 ++- .../bigquery/using-bigquery-with-dagster.md | 4 +-- .../libraries/snowflake/reference.md | 12 +++---- docs/scripts/build-kinds-tags.sh | 2 +- docs/sphinx/sections/api/dagster/assets.rst | 7 ++++ .../libraries/omni/dagster-omni.rst | 2 +- docs/src/theme/DocCard/index.tsx | 3 -- docs/src/theme/DocCard/styles.module.css | 3 +- .../test_sensors.py | 19 +++++++---- .../guides/components/test_components_docs.py | 20 +++++------- .../test_integration_files_load.py | 8 ++--- .../VirtualizedAssetCheckTable.tsx | 3 +- .../libraries/dagster-cloud-cli/setup.py | 2 +- .../libraries/dagster-dg-cli/setup.py | 2 +- .../libraries/dagster-dg-cli/tox.ini | 2 +- .../libraries/dagster-dg-core/setup.py | 2 +- python_modules/libraries/dagster-omni/tox.ini | 4 +-- .../dagstermill/examples/repository.py | 12 +++---- .../api-docs.rst.tmpl | 8 ++--- .../setup.py.tmpl | 6 ++-- .../tox.ini.tmpl | 4 +-- 42 files changed, 138 insertions(+), 143 deletions(-) diff --git a/docs/docs/deployment/dagster-plus/authentication-and-access-control/rbac/user-roles-permissions.md b/docs/docs/deployment/dagster-plus/authentication-and-access-control/rbac/user-roles-permissions.md index 381f3c43423e6..d2bb4ecedb318 100644 --- a/docs/docs/deployment/dagster-plus/authentication-and-access-control/rbac/user-roles-permissions.md +++ b/docs/docs/deployment/dagster-plus/authentication-and-access-control/rbac/user-roles-permissions.md @@ -99,7 +99,7 @@ To override a code location role for an individual user: 2. Click **Edit**. 3. Click the toggle to the left of the deployment to open a list of code locations. 4. Next to a code location, click **Edit user role**. -5. Select the user role for the code location: {/* TODO: add picture previously at "/images/dagster-cloud/user-token-management/code-location-override.png" */} +5. Select the user role for the code location. 6. Click **Save**. #### Team members @@ -118,13 +118,9 @@ The above also applies to code locations and Branch Deployment roles. #### Viewing overrides -To view deployment-level overrides for a specific user, locate the user on the **Users** page and hover over a deployment: +To view deployment-level overrides for a specific user, locate the user on the **Users** page and hover over a deployment. A tooltip will display any overrides that apply at the deployment level. -{/* TODO: add picture previously at "/images/dagster-cloud/user-token-management/user-overrides-popup.png" */} - -If there are code location-level overrides, a small **N override(s)** link will display beneath the user's deployment role. Hover over it to display the list of overrides: - -{/* TODO: add picture previously at "/images/dagster-cloud/user-token-management/code-location-override-popup.png" */} +If there are code location-level overrides, a small **N override(s)** link will display beneath the user's deployment role. Hover over it to display the list of overrides. #### Removing overrides diff --git a/docs/docs/deployment/dagster-plus/hybrid/kubernetes/setup.md b/docs/docs/deployment/dagster-plus/hybrid/kubernetes/setup.md index f00aeedf9af0f..005db4e628c49 100644 --- a/docs/docs/deployment/dagster-plus/hybrid/kubernetes/setup.md +++ b/docs/docs/deployment/dagster-plus/hybrid/kubernetes/setup.md @@ -69,9 +69,7 @@ helm --namespace dagster-cloud upgrade agent \ ## Troubleshooting tips -You can see basic health information about your agent in the Dagster+ UI: - -[comment]: <> (TODO: Screenshot of Dagster+ Deployments agents tab) +You can see basic health information about your agent in the Dagster+ UI. ### View logs diff --git a/docs/docs/examples/full-pipelines/etl-pipeline/6-create-a-sensor.md b/docs/docs/examples/full-pipelines/etl-pipeline/6-create-a-sensor.md index 4de4e36dfd480..35a0a6631e571 100644 --- a/docs/docs/examples/full-pipelines/etl-pipeline/6-create-a-sensor.md +++ b/docs/docs/examples/full-pipelines/etl-pipeline/6-create-a-sensor.md @@ -61,8 +61,7 @@ Now copy the following sensor code in the `sensors.py` file: 2. Navigate to the Automation page. 3. Turn on the `adhoc_request_sensor`. 4. Click on the `adhoc_request_sensor` details. - -{/* TODO: Screenshot */} + You should see the sensor details view with recent ticks and the run history panel. 5. Create a `data/requests` directory in `dagster_tutorial`. Then include a `request.json` file: @@ -76,8 +75,7 @@ Now copy the following sensor code in the `sensors.py` file: ``` 6. Click on the green tick to see the run for this request. - -{/* TODO: Screenshot */} + The run should appear in the runs list, and the tick details will show the run ID. ## Summary diff --git a/docs/docs/guides/automate/sensors/testing-run-status-sensors.md b/docs/docs/guides/automate/sensors/testing-run-status-sensors.md index f65299ca68df7..f5ded24f21f86 100644 --- a/docs/docs/guides/automate/sensors/testing-run-status-sensors.md +++ b/docs/docs/guides/automate/sensors/testing-run-status-sensors.md @@ -33,8 +33,7 @@ Then we can execute this job and pull the attributes we need to build the `conte title="src//defs/sensors.py" /> -{/* TODO the methods and statuses below do not exist in API docs */} -{/* We have provided convenience functions and for retrieving `DagsterRunStatus.SUCCESS` and `DagsterRunStatus.FAILURE` events, respectively. If you have a run status sensor triggered on another status, you can retrieve all events from `result` and filter based on your event type. */} +To inspect the resulting status in tests, use the returned by `execute_in_process`. Its `dagster_run` and `all_events` properties let you filter for the status or event type you care about. We can use the same pattern to build the context for . If we wanted to test this run failure sensor: diff --git a/docs/docs/guides/build/assets/asset-versioning-and-caching.md b/docs/docs/guides/build/assets/asset-versioning-and-caching.md index b1f14858ad03c..6cca3787d7ae2 100644 --- a/docs/docs/guides/build/assets/asset-versioning-and-caching.md +++ b/docs/docs/guides/build/assets/asset-versioning-and-caching.md @@ -18,8 +18,7 @@ Dagster's versioning system helps you determine ahead of time whether materializ Dagster has two versioning concepts to represent the code and input data used for each materialization: - **Code version.** A string that represents the version of the code that computes an asset. This is the `code_version` argument of . -- **Data version.** A string that represents the version of the data represented by the asset. This is represented as a `DataVersion` object. -{/* TODO link `DataVersion` to API docs */} +- **Data version.** A string that represents the version of the data represented by the asset. This is represented as a object. By keeping track of code and data versions, Dagster can predict whether a materialization will change the underlying value. This allows Dagster to skip redundant materializations and instead return the previously computed value. In more technical terms, Dagster offers a limited form of [memoization](https://en.wikipedia.org/wiki/Memoization) for assets: the last-computed asset value is always cached. diff --git a/docs/docs/guides/build/assets/passing-data-between-assets.md b/docs/docs/guides/build/assets/passing-data-between-assets.md index d8234185ff40f..384648562c7cc 100644 --- a/docs/docs/guides/build/assets/passing-data-between-assets.md +++ b/docs/docs/guides/build/assets/passing-data-between-assets.md @@ -117,4 +117,7 @@ The downsides of this approach are: ## Related resources -{/* TODO: add links to relevant API documentation here. */} +- +- +- +- diff --git a/docs/docs/guides/build/components/building-pipelines-with-components/troubleshooting-components.md b/docs/docs/guides/build/components/building-pipelines-with-components/troubleshooting-components.md index 5f5aee80f2ce4..607b11c89f787 100644 --- a/docs/docs/guides/build/components/building-pipelines-with-components/troubleshooting-components.md +++ b/docs/docs/guides/build/components/building-pipelines-with-components/troubleshooting-components.md @@ -5,4 +5,18 @@ sidebar_position: 500 unlisted: true --- -TODO +This page lists common checks when a component fails to load or behaves unexpectedly. + +## Quick checks + +- **Validate the component type string**. Ensure the type name matches the registered component class and that the package providing it is installed in your environment. +- **Confirm configuration fields**. Component configuration is validated on load. If a field is missing or mis-typed, the error message will include the field path to update. +- **Verify workspace paths**. If you are using a workspace, confirm that `dg.toml` points at the correct project directories and that the project loads on its own. + +## Inspecting load errors + +If a component fails to load, the code location will show a load error in the UI and in CLI output. Use the stack trace there to locate the file and configuration entry that caused the failure. + +## Next steps + +For a refresher on how components are structured, see the [components overview](/guides/build/components). diff --git a/docs/docs/guides/build/external-pipelines/aws/aws-lambda-pipeline.md b/docs/docs/guides/build/external-pipelines/aws/aws-lambda-pipeline.md index 92b4dd7e05a33..018fa345e3942 100644 --- a/docs/docs/guides/build/external-pipelines/aws/aws-lambda-pipeline.md +++ b/docs/docs/guides/build/external-pipelines/aws/aws-lambda-pipeline.md @@ -158,8 +158,7 @@ Here's what we did in this example: This argument is passed to the `run` method of , which submits the provided information to the [boto `invoke` API](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/lambda/client/invoke.html) and then invokes the specified function (`function_name`). -- Returned a object representing the result of execution. This is obtained by calling `get_materialize_result` on the `PipesClientCompletedInvocation` object returned by `run` after the execution in AWS Lambda has completed. -{/* TODO replace `PipesClientCompletedInvocation` with */} +- Returned a object representing the result of execution. This is obtained by calling `get_materialize_result` on the object returned by `run` after the execution in AWS Lambda has completed. ### Step 2.2: Create Dagster Definitions diff --git a/docs/docs/guides/build/external-pipelines/databricks-pipeline.md b/docs/docs/guides/build/external-pipelines/databricks-pipeline.md index e76f95bff4f0f..6b943e885b4c4 100644 --- a/docs/docs/guides/build/external-pipelines/databricks-pipeline.md +++ b/docs/docs/guides/build/external-pipelines/databricks-pipeline.md @@ -95,8 +95,7 @@ Let's review what's happening in this code: - **Passes the `SubmitTask` object, `AssetExecutionContext`, and `extras` dictionary to the `run` method of **. This method synchronously executes the Databricks job specified by the `SubmitTask` object. It slightly modifies the object by injecting some environment variables under `new_cluster.spark_env_vars` before submitting the object to the Databricks API. -- **Returns a object representing the result of execution**. This is obtained by calling `get_materialize_result` on the `PipesClientCompletedInvocation` object returned by `run` after the Databricks job has finished. **Note**: Execution can take several minutes even for trivial scripts due to Databricks cluster provisioning times. -{/* TODO replace `PipesClientCompletedInvocation` with */} +- **Returns a object representing the result of execution**. This is obtained by calling `get_materialize_result` on the object returned by `run` after the Databricks job has finished. **Note**: Execution can take several minutes even for trivial scripts due to Databricks cluster provisioning times. ### Step 1.2: Define the Databricks Pipes client and Definitions diff --git a/docs/docs/guides/build/external-pipelines/kubernetes-pipeline.md b/docs/docs/guides/build/external-pipelines/kubernetes-pipeline.md index 8776b577e35bf..9a211e957fd2a 100644 --- a/docs/docs/guides/build/external-pipelines/kubernetes-pipeline.md +++ b/docs/docs/guides/build/external-pipelines/kubernetes-pipeline.md @@ -164,8 +164,7 @@ Here's what we did in this example: These arguments are passed to the `run` method of , which submits the provided cluster information to the Kubernetes API and then runs the specified `image`. -- Returned a object representing the result of execution. This is obtained by calling `get_materialize_result` on the `PipesClientCompletedInvocation` object returned by `run` after the execution in Kubernetes has completed. -{/* TODO replace `PipesClientCompletedInvocation` with */} +- Returned a object representing the result of execution. This is obtained by calling `get_materialize_result` on the object returned by `run` after the execution in Kubernetes has completed. :::note diff --git a/docs/docs/guides/build/external-resources/index.md b/docs/docs/guides/build/external-resources/index.md index e73975223343c..751ce1933d81c 100644 --- a/docs/docs/guides/build/external-resources/index.md +++ b/docs/docs/guides/build/external-resources/index.md @@ -24,12 +24,10 @@ Resources allow you to: ## Relevant APIs -{/* TODO replace `ResourceParam` with in table below */} - | Name | Description | | ------------------------------------------------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | | The base class extended to define resources. Under the hood, implements . | -| `ResourceParam` | An annotation used to specify that a plain Python object parameter for an asset or op is a resource. | +| | An annotation used to specify that a plain Python object parameter for an asset or op is a resource. | | | Class for resource definitions. You almost never want to use initialize this class directly. Instead, you should extend the class which implements . | | | The context object provided to a resource during initialization. This object contains required resources, config, and other run information. | | | Function for building an outside of execution, intended to be used when testing a resource. | diff --git a/docs/docs/guides/build/external-resources/using-bare-python-objects-as-resources.md b/docs/docs/guides/build/external-resources/using-bare-python-objects-as-resources.md index f524b1ec17b63..15d984042e0a0 100644 --- a/docs/docs/guides/build/external-resources/using-bare-python-objects-as-resources.md +++ b/docs/docs/guides/build/external-resources/using-bare-python-objects-as-resources.md @@ -10,10 +10,8 @@ import ScaffoldResource from '@site/docs/partials/\_ScaffoldResource.md'; When starting to build a set of assets or jobs, you may want to use a bare Python object without configuration as a resource, such as a third-party API client. -Dagster supports passing bare Python objects as resources. This follows a similar pattern to using a subclass; however, assets that use these resources must [annotate](https://docs.python.org/3/library/typing.html#typing.Annotated) them with `ResourceParam`. This annotation lets Dagster know that the parameter is a resource and not an upstream input. - -{/* TODO replace `ResourceParam` with */} +Dagster supports passing bare Python objects as resources. This follows a similar pattern to using a subclass; however, assets that use these resources must [annotate](https://docs.python.org/3/library/typing.html#typing.Annotated) them with . This annotation lets Dagster know that the parameter is a resource and not an upstream input. - \ No newline at end of file + diff --git a/docs/docs/guides/build/partitions-and-backfills/defining-dependencies-between-partitioned-assets.md b/docs/docs/guides/build/partitions-and-backfills/defining-dependencies-between-partitioned-assets.md index 65b3552bb105c..c0a2763aa1130 100644 --- a/docs/docs/guides/build/partitions-and-backfills/defining-dependencies-between-partitioned-assets.md +++ b/docs/docs/guides/build/partitions-and-backfills/defining-dependencies-between-partitioned-assets.md @@ -83,25 +83,3 @@ We recommend using [automation conditions](/guides/automate/declarative-automati ::: - -{/* ## Dependencies between time-based partitions and un-partitioned assets */} - -{/* TODO */} - -{/* ## Dependencies between time-based and static partitions */} - -{/* Combining time-based and static partitions allows you to analyze data across both temporal and categorical dimensions. This is particularly useful for scenarios like regional time series analysis. */} - -{/* TODO */} - -{/* ## Dependencies between time-based and dynamic partitions */} - -{/* TODO */} - -{/* ## Dependencies between time-based partitions and un-partitioned assets */} - -{/* TODO */} - -{/* ## Integrating Dagster partitions with external systems: incremental models and dbt */} - -{/* TODO */} diff --git a/docs/docs/guides/build/partitions-and-backfills/partitioning-assets.md b/docs/docs/guides/build/partitions-and-backfills/partitioning-assets.md index 1db82dcd67a7e..cb50736e7fe73 100644 --- a/docs/docs/guides/build/partitions-and-backfills/partitioning-assets.md +++ b/docs/docs/guides/build/partitions-and-backfills/partitioning-assets.md @@ -35,7 +35,7 @@ Sometimes you have a set of predefined categories for your data. For instance, y -{/* TODO: Link to Backfill page to explain how to backfill regional sales data */} +To backfill historical partitions for regional sales data, see [Backfilling data](/guides/build/partitions-and-backfills/backfilling-data). ## Two-dimensional partitions \{#two-dimensional-partitions} @@ -97,4 +97,4 @@ After a partition has been successfully materialized, it will display as green i To view materializations by partition for a specific asset, navigate to the **Activity** tab of the asset's **Details** page: -![Asset activity section of asset details page](/images/guides/build/partitions-and-backfills/materialized-partitioned-asset-activity.png) \ No newline at end of file +![Asset activity section of asset details page](/images/guides/build/partitions-and-backfills/materialized-partitioned-asset-activity.png) diff --git a/docs/docs/guides/build/projects/dagster-plus-project-history.md b/docs/docs/guides/build/projects/dagster-plus-project-history.md index 541f9a98ac2d0..0bf99b44d79bf 100644 --- a/docs/docs/guides/build/projects/dagster-plus-project-history.md +++ b/docs/docs/guides/build/projects/dagster-plus-project-history.md @@ -16,7 +16,6 @@ Dagster+ automatically tracks metadata every time a project (code location) is l 1. In the Dagster+ UI, navigate to the **Deployment** tab. 2. In the row associated with the project you're interested in, click **View history** in the **Updated** column. -{/* TODO - update screenshot when code location renamed in UI */} ![Screenshot highlighting the "Updated" column for a project/code location](/images/dagster-plus/deployment/code-locations/view-code-location-history.png) This will bring up a modal showing a history of every time that project has been loaded, and metadata associated with that load. If you have connected Dagster+ to a GitHub or GitLab repository, each row will have a link to the commit that was deployed at that point in time. @@ -45,7 +44,6 @@ If you notice an issue with newly deployed code, or your code fails to deploy su 2. In the row associated with the project you're interested in, click **View history** in the **Updated** column. 3. In the **Actions** column click the dropdown menu to the right of **View metadata**, select **Rollback to this version**. -{/* TODO - update screenshot when code location renamed in UI */} ![Screenshot highlighting the "Updated" column for a project/code location](/images/dagster-plus/deployment/code-locations/rollback-code-location.png) :::tip diff --git a/docs/docs/guides/build/projects/workspaces/creating-workspaces.md b/docs/docs/guides/build/projects/workspaces/creating-workspaces.md index 2b28fc15a5fd0..b48aad40bc980 100644 --- a/docs/docs/guides/build/projects/workspaces/creating-workspaces.md +++ b/docs/docs/guides/build/projects/workspaces/creating-workspaces.md @@ -99,5 +99,4 @@ Finally, let's load our two projects with `dg dev`. When you run `dg dev` from t -{/* TODO - replace screenshot */} ![](/images/guides/build/projects-and-components/setting-up-a-workspace/two-projects.png) diff --git a/docs/docs/guides/build/projects/workspaces/migrating-workspace-yaml.md b/docs/docs/guides/build/projects/workspaces/migrating-workspace-yaml.md index ac2e8a1f5bef4..407680f98bad1 100644 --- a/docs/docs/guides/build/projects/workspaces/migrating-workspace-yaml.md +++ b/docs/docs/guides/build/projects/workspaces/migrating-workspace-yaml.md @@ -1,8 +1,36 @@ --- title: Migrating from workspace.yaml to dg.toml sidebar_position: 2000 -description: TODO +description: Migrate legacy workspace.yaml configurations to a dg.toml workspace. unlisted: true --- -TODO \ No newline at end of file +If you have an existing `workspace.yaml` file, you can migrate to a `dg.toml`-based workspace to align with the current Dagster OSS project and workspace structure. + +## Overview + +`workspace.yaml` directly lists code locations. A `dg.toml` workspace points to one or more Dagster projects, each with its own project configuration. The migration is therefore a move from "direct code locations" to "workspace + projects." + +## Step 1: Create a workspace root + +If you don't already have a workspace directory, create one by following [Creating workspaces](/guides/build/projects/workspaces/creating-workspaces). This will create a workspace root with a `dg.toml` file. + +## Step 2: Convert code locations into projects + +Each entry in `workspace.yaml` should map to a Dagster project in your workspace: + +1. For each code location, create a Dagster project using [Creating projects](/guides/build/projects/creating-projects). +2. Move or reference your existing definitions in that project. +3. Add each project path under `[[workspace.projects]]` in `dg.toml`. + +For the full `dg.toml` format, see the [dg.toml reference](/guides/build/projects/workspaces/dg-toml). + +## Step 3: Validate and remove workspace.yaml + +Run the workspace from the root directory: + +```shell +dg dev +``` + +If everything loads correctly, you can remove `workspace.yaml` and update any scripts to point to `dg.toml` (for example, `dg dev -w path/to/dg.toml`). diff --git a/docs/docs/guides/test/running-a-subset-of-asset-checks.md b/docs/docs/guides/test/running-a-subset-of-asset-checks.md index 839b417804442..861a4187018d9 100644 --- a/docs/docs/guides/test/running-a-subset-of-asset-checks.md +++ b/docs/docs/guides/test/running-a-subset-of-asset-checks.md @@ -18,8 +18,7 @@ This article assumes familiarity with [asset checks](/guides/test/asset-checks) Using the decorator's `specs` and `can_subset` arguments, you can execute a subset of checks in a single op. -{/* TODO link this to proper API doc */} -Inside the body of the function, we can use `AssetCheckExecutionContext.selected_asset_check_keys` to identify which computations to run. We can also set the decorator's `can_subset` parameter to `True` to execute a subset of the asset checks that the computation contains. +Inside the body of the function, we can use and its `selected_asset_check_keys` property to identify which computations to run. We can also set the decorator's `can_subset` parameter to `True` to execute a subset of the asset checks that the computation contains. As we don't know in advance which checks will be executed, we explicitly `yield` each asset check result that we're expected to create: diff --git a/docs/docs/guides/test/testing-partitioned-config-and-jobs.md b/docs/docs/guides/test/testing-partitioned-config-and-jobs.md index c6651f5f986cf..ba53874c26dcc 100644 --- a/docs/docs/guides/test/testing-partitioned-config-and-jobs.md +++ b/docs/docs/guides/test/testing-partitioned-config-and-jobs.md @@ -50,9 +50,7 @@ If you want to test that a : You can also specify the schema as part of the asset's key: diff --git a/docs/docs/integrations/libraries/duckdb/using-duckdb-with-dagster.md b/docs/docs/integrations/libraries/duckdb/using-duckdb-with-dagster.md index 070f04dc37c6c..59092e8b1ec8e 100644 --- a/docs/docs/integrations/libraries/duckdb/using-duckdb-with-dagster.md +++ b/docs/docs/integrations/libraries/duckdb/using-duckdb-with-dagster.md @@ -11,7 +11,7 @@ The `dagster-duckdb` library provides two ways to interact with DuckDB tables: - [Resource](/guides/build/external-resources): The resource allows you to directly run SQL queries against tables within an asset's compute function. Available resources: . - [I/O manager](/guides/build/io-managers): The I/O manager transfers the responsibility of storing and loading DataFrames as DuckdB tables to Dagster. Available I/O managers: , , . -This tutorial is divided into two sections to demonstrate the differences between the DuckDB resource and the DuckDB I/O manager. Each section will create the same assets, but the first section will use the DuckDB resource to store data in DuckDB, whereas the second section will use the DuckDB I/O manager. When writing your own assets, you may choose one or the other (or both) approaches depending on your storage requirements. {/* TODO fix link See [When to use I/O managers](/guides/build/io-managers#when-to-use-io-managers) to learn more about when to use I/O managers and when to use resources. */} +This tutorial is divided into two sections to demonstrate the differences between the DuckDB resource and the DuckDB I/O manager. Each section will create the same assets, but the first section will use the DuckDB resource to store data in DuckDB, whereas the second section will use the DuckDB I/O manager. When writing your own assets, you may choose one or the other (or both) approaches depending on your storage requirements. See [When to use I/O managers](/guides/build/io-managers#when-to-use-io-managers) to learn more about when to use I/O managers and when to use resources. In [Option 1](#option-1-using-the-duckdb-resource) you will: @@ -25,8 +25,6 @@ In [Option 2](#option-2-using-the-duckdb-io-manager) you will: - Use Pandas to create a DataFrame, then delegate responsibility creating a table to the DuckDB I/O manager. - Use the DuckDB I/O manager to load the table into memory so that you can interact with it using the Pandas library. -When writing your own assets, you may choose one or the other (or both) approaches depending on your storage requirements. {/* See [When to use I/O managers](/guides/build/io-managers#when-to-use-io-managers) to learn more. */} - By the end of the tutorial, you will: - Understand how to interact with a DuckDB database using the DuckDB resource. @@ -109,13 +107,11 @@ When finished, your code should look like the following: ## Option 2: Using the DuckDB I/O manager -You may want to use an I/O manager to handle storing DataFrames as tables in DuckDB and loading DuckDB tables as DataFrames in downstream assets. You may want to use an I/O manager if: +You may want to use an I/O manager to handle storing DataFrames as tables in DuckDB and loading DuckDB tables as DataFrames in downstream assets. See [When to use I/O managers](/guides/build/io-managers#when-to-use-io-managers) for more guidance. You may want to use an I/O manager if: - You want your data to be loaded in memory so that you can interact with it using Python. - You'd like to have Dagster manage how you store the data and load it as an input in downstream assets. -{/* Using an I/O manager is not required, and you can reference [When to use I/O managers](/guides/build/io-managers#when-to-use-io-managers) to learn more. */} - This section of the guide focuses on storing and loading Pandas DataFrames in DuckDB, but Dagster also supports using PySpark and Polars DataFrames with DuckDB. The concepts from this guide apply to working with PySpark and Polars DataFrames, and you can learn more about setting up and using the DuckDB I/O manager with PySpark and Polars DataFrames in the [reference guide](/integrations/libraries/duckdb/reference). ### Step 1: Configure the DuckDB I/O manager diff --git a/docs/docs/integrations/libraries/gcp/bigquery/reference.md b/docs/docs/integrations/libraries/gcp/bigquery/reference.md index f479d7c640b59..f011add77b0e8 100644 --- a/docs/docs/integrations/libraries/gcp/bigquery/reference.md +++ b/docs/docs/integrations/libraries/gcp/bigquery/reference.md @@ -132,7 +132,7 @@ When materializing the above assets, a partition must be selected, as described SELECT * WHERE SPECIES in ('Iris-setosa') AND TIMESTAMP_SECONDS(TIME) >= '2023-01-02 00:00:00' - AND TIMESTAMP_SECONDS(TIME) < '2023-01-03 00:00:00'` + AND TIMESTAMP_SECONDS(TIME) < '2023-01-03 00:00:00' ``` @@ -154,12 +154,11 @@ If you want to store assets in different datasets, you can specify the dataset a You can also specify the dataset as part of the asset's asset key: -{/* TODO add dedent=4 to CodeExample below */} - The dataset will be the last prefix before the asset's name. In this example, the `iris_data` asset will be stored in the `IRIS` dataset, and the `daffodil_data` asset will be found in the `DAFFODIL` dataset. diff --git a/docs/docs/integrations/libraries/gcp/bigquery/using-bigquery-with-dagster.md b/docs/docs/integrations/libraries/gcp/bigquery/using-bigquery-with-dagster.md index 9c5c0ee9443d0..625ef7376d8f2 100644 --- a/docs/docs/integrations/libraries/gcp/bigquery/using-bigquery-with-dagster.md +++ b/docs/docs/integrations/libraries/gcp/bigquery/using-bigquery-with-dagster.md @@ -124,13 +124,11 @@ When finished, your code should look like the following: ## Option 2: Using the BigQuery I/O manager -While using an I/O manager is not required, you may want to use an I/O manager to handle storing DataFrames as tables in BigQuery and loading BigQuery tables as DataFrames in downstream assets. You may want to use an I/O manager if: +While using an I/O manager is not required, you may want to use an I/O manager to handle storing DataFrames as tables in BigQuery and loading BigQuery tables as DataFrames in downstream assets. See [When to use I/O managers](/guides/build/io-managers#when-to-use-io-managers) for guidance. You may want to use an I/O manager if: - You want your data to be loaded in memory so that you can interact with it using Python. - You'd like to have Dagster manage how you store the data and load it as an input in downstream assets. -{/* TODO fix link: Using an I/O manager is not required, and you can reference [When to use I/O managers](/guides/build/io-managers#when-to-use-io-managers) to learn more. */} - This section of the guide focuses on storing and loading Pandas DataFrames in BigQuery, but Dagster also supports using PySpark DataFrames with BigQuery. The concepts from this guide apply to working with PySpark DataFrames, and you can learn more about setting up and using the BigQuery I/O manager with PySpark DataFrames in the [reference guide](/integrations/libraries/gcp/bigquery/reference). ### Step 1: Configure the BigQuery I/O manager diff --git a/docs/docs/integrations/libraries/snowflake/reference.md b/docs/docs/integrations/libraries/snowflake/reference.md index a1ae5f30e8085..bc611314ae21e 100644 --- a/docs/docs/integrations/libraries/snowflake/reference.md +++ b/docs/docs/integrations/libraries/snowflake/reference.md @@ -101,7 +101,7 @@ SELECT * When the `partition_expr` value is injected into this statement, the resulting SQL query must follow Snowflake's SQL syntax. Refer to the [Snowflake documentation](https://docs.snowflake.com/en/sql-reference/constructs) for more information. -{/* TODO fix link: When materializing the above assets, a partition must be selected, as described in [Materializing partitioned assets](/concepts/partitions-schedules-sensors/partitioning-assets#materializing-partitioned-assets). */} When materializing the above assets, a partition must be selected. In this example, the query used when materializing the `Iris-setosa` partition of the above assets would be: +When materializing the above assets, a partition must be selected, as described in [Materializing partitioned assets](/guides/build/partitions-and-backfills/partitioning-assets#materializing-partitioned-assets). In this example, the query used when materializing the `Iris-setosa` partition of the above assets would be: ```sql SELECT * @@ -129,7 +129,7 @@ SELECT * When the `partition_expr` value is injected into this statement, the resulting SQL query must follow Snowflake's SQL syntax. Refer to the [Snowflake documentation](https://docs.snowflake.com/en/sql-reference/constructs) for more information. -{/* TODO fix link: When materializing the above assets, a partition must be selected, as described in [Materializing partitioned assets](/concepts/partitions-schedules-sensors/partitioning-assets#materializing-partitioned-assets). */} When materializing the above assets, a partition must be selected. The `[partition_start]` and `[partition_end]` bounds are of the form `YYYY-MM-DD HH:MM:SS`. In this example, the query when materializing the `2023-01-02` partition of the above assets would be: +When materializing the above assets, a partition must be selected, as described in [Materializing partitioned assets](/guides/build/partitions-and-backfills/partitioning-assets#materializing-partitioned-assets). The `[partition_start]` and `[partition_end]` bounds are of the form `YYYY-MM-DD HH:MM:SS`. In this example, the query when materializing the `2023-01-02` partition of the above assets would be: ```sql SELECT * @@ -152,7 +152,7 @@ The Snowflake I/O manager can also store data partitioned on multiple dimensions Dagster uses the `partition_expr` metadata to craft the `SELECT` statement when loading the correct partition in a downstream asset. For multi-partitions, Dagster concatenates the `WHERE` statements described in the above sections to craft the correct `SELECT` statement. -{/* TODO fix link: When materializing the above assets, a partition must be selected, as described in [Materializing partitioned assets](/concepts/partitions-schedules-sensors/partitioning-assets#materializing-partitioned-assets). */} When materializing the above assets, a partition must be selected. For example, when materializing the `2023-01-02|Iris-setosa` partition of the above assets, the following query will be used: +When materializing the above assets, a partition must be selected, as described in [Materializing partitioned assets](/guides/build/partitions-and-backfills/partitioning-assets#materializing-partitioned-assets). For example, when materializing the `2023-01-02|Iris-setosa` partition of the above assets, the following query will be used: ```sql SELECT * @@ -172,22 +172,20 @@ You can specify the default schema where data will be stored as configuration to To store assets in different schemas, specify the schema as metadata: -{/* TODO add dedent=4 prop to CodeExample below */} - You can also specify the schema as part of the asset's asset key: -{/* TODO add dedent=4 prop to CodeExample below */} - In this example, the `iris_dataset` asset will be stored in the `IRIS` schema, and the `daffodil_dataset` asset will be found in the `DAFFODIL` schema. diff --git a/docs/scripts/build-kinds-tags.sh b/docs/scripts/build-kinds-tags.sh index ee7b6c74d1552..fde4903932901 100755 --- a/docs/scripts/build-kinds-tags.sh +++ b/docs/scripts/build-kinds-tags.sh @@ -18,7 +18,7 @@ uv_activate_venv() { uv pip install tox } -# TODO - refactor Vercel logic shared between `build-kinds-tags` and `build-api-docs` into single script +# NOTE: Vercel logic is shared with `build-api-docs` and could be consolidated later. if [ "$VERCEL" = "1" ]; then echo "Detected Vercel environment. Running Vercel-specific commands and configurations." export LC_ALL=C.UTF-8 diff --git a/docs/sphinx/sections/api/dagster/assets.rst b/docs/sphinx/sections/api/dagster/assets.rst index fffeac15bc34e..ffc6ad16cb9aa 100644 --- a/docs/sphinx/sections/api/dagster/assets.rst +++ b/docs/sphinx/sections/api/dagster/assets.rst @@ -90,6 +90,13 @@ Refer to the `Asset observation `_. diff --git a/docs/src/theme/DocCard/index.tsx b/docs/src/theme/DocCard/index.tsx index 52d2d2126ec13..36871f8bb550e 100644 --- a/docs/src/theme/DocCard/index.tsx +++ b/docs/src/theme/DocCard/index.tsx @@ -13,9 +13,6 @@ import type {PropSidebarItemCategory, PropSidebarItemLink} from '@docusaurus/plu import styles from './styles.module.css'; import useBaseUrl from '@docusaurus/useBaseUrl'; -// TODO - text for folders -// TODO - indicator for "community supported" integration - function useCategoryItemsPlural() { const {selectMessage} = usePluralForm(); return (count: number) => diff --git a/docs/src/theme/DocCard/styles.module.css b/docs/src/theme/DocCard/styles.module.css index be17c0c5d29d7..cbb1bd2c45031 100644 --- a/docs/src/theme/DocCard/styles.module.css +++ b/docs/src/theme/DocCard/styles.module.css @@ -1,5 +1,4 @@ -/* TODO: Refactor with scss to simplify/match other docs styling, merge with card styling in custom.scss */ -/* TODO: This duplicates/conflicts with .card styles set in custom.scss. Try to create a more cohesive card styling system */ +/* Doc card layout overrides the global .card styles from custom.scss to match docs-specific layout. */ .cardContainer { --ifm-link-color: var(--ifm-color-emphasis-800); diff --git a/examples/docs_snippets/docs_snippets_tests/concepts_tests/partitions_schedules_sensors_tests/test_sensors.py b/examples/docs_snippets/docs_snippets_tests/concepts_tests/partitions_schedules_sensors_tests/test_sensors.py index 6aeb086f4b5b9..7d367d5d97162 100644 --- a/examples/docs_snippets/docs_snippets_tests/concepts_tests/partitions_schedules_sensors_tests/test_sensors.py +++ b/examples/docs_snippets/docs_snippets_tests/concepts_tests/partitions_schedules_sensors_tests/test_sensors.py @@ -7,6 +7,7 @@ my_slack_on_run_success, slack_on_run_failure, ) +from docs_snippets.concepts.partitions_schedules_sensors.sensors import sensors as sensors_module from docs_snippets.concepts.partitions_schedules_sensors.sensors.sensors import ( log_file_job, my_directory_sensor, @@ -36,15 +37,21 @@ def test_log_file_job(): assert result.success -def test_my_directory_sensor(): - # TODO: Actually test - assert my_directory_sensor +def test_my_directory_sensor(tmp_path, monkeypatch): + test_file = tmp_path / "example.txt" + test_file.write_text("data", encoding="utf8") + + monkeypatch.setattr(sensors_module, "MY_DIRECTORY", str(tmp_path)) + run_requests = list(my_directory_sensor()) + + assert len(run_requests) == 1 + assert run_requests[0].run_key == "example.txt" + assert run_requests[0].run_config is not None def test_interval_sensors(): - # TODO: Actually test - assert sensor_A - assert sensor_B + assert sensor_A.minimum_interval_seconds == 30 + assert sensor_B.minimum_interval_seconds == 45 def test_run_failure_sensor_def(): diff --git a/examples/docs_snippets/docs_snippets_tests/snippet_checks/guides/components/test_components_docs.py b/examples/docs_snippets/docs_snippets_tests/snippet_checks/guides/components/test_components_docs.py index 4f5197756481b..0beaf4663ecf2 100644 --- a/examples/docs_snippets/docs_snippets_tests/snippet_checks/guides/components/test_components_docs.py +++ b/examples/docs_snippets/docs_snippets_tests/snippet_checks/guides/components/test_components_docs.py @@ -184,9 +184,8 @@ def next_snip_no() -> int: context.run_command_and_snippet_output( cmd="dg scaffold defs 'dagster_sling.SlingReplicationCollectionComponent' ingest_files", snippet_path=f"{next_snip_no()}-dg-scaffold-sling-replication.txt", - # TODO turn output back on when we figure out how to handle multiple - # "Using ..." messages from multiple dagster-components calls under the hood (when - # cache disabled for pip) + # NOTE: Output stays off while multiple "Using ..." messages are emitted from nested + # dagster-components calls (when cache is disabled for pip). ignore_output=True, ) @@ -310,9 +309,8 @@ def next_snip_no() -> int: context.run_command_and_snippet_output( cmd="dg scaffold defs dagster_dbt.DbtProjectComponent jdbt --project-path dbt/jdbt", snippet_path=f"{next_snip_no()}-dg-scaffold-jdbt.txt", - # TODO turn output back on when we figure out how to handle multiple - # "Using ..." messages from multiple dagster-components calls under the hood - # (when cache disabled for pip) + # NOTE: Output stays off while multiple "Using ..." messages are emitted from nested + # dagster-components calls (when cache is disabled for pip). ignore_output=True, ) context.check_file( @@ -386,9 +384,8 @@ def next_snip_no() -> int: context.run_command_and_snippet_output( cmd="dg scaffold defs dagster_evidence.EvidenceProject jaffle_dashboard", snippet_path=f"{next_snip_no()}-scaffold-jaffle-dashboard.txt", - # TODO turn output back on when we figure out how to handle multiple - # "Using ..." messages from multiple dagster-components calls under the hood - # (when cache disabled for pip) + # NOTE: Output stays off while multiple "Using ..." messages are emitted from nested + # dagster-components calls (when cache is disabled for pip). ignore_output=True, ) @@ -440,9 +437,8 @@ def next_snip_no() -> int: context.run_command_and_snippet_output( cmd="dg scaffold defs dagster.schedule daily_jaffle.py", snippet_path=f"{next_snip_no()}-scaffold-daily-jaffle.txt", - # TODO turn output back on when we figure out how to handle multiple - # "Using ..." messages from multiple dagster-components calls under the hood (when - # cache disabled for pip) + # NOTE: Output stays off while multiple "Using ..." messages are emitted from nested + # dagster-components calls (when cache is disabled for pip). ignore_output=True, ) diff --git a/examples/docs_snippets/docs_snippets_tests/test_integration_files_load.py b/examples/docs_snippets/docs_snippets_tests/test_integration_files_load.py index df71b06033325..90e9ceeb670ae 100644 --- a/examples/docs_snippets/docs_snippets_tests/test_integration_files_load.py +++ b/examples/docs_snippets/docs_snippets_tests/test_integration_files_load.py @@ -17,7 +17,7 @@ f"{snippets_folder}/salesforce.py", f"{snippets_folder}/sftp.py", f"{snippets_folder}/sharepoint.py", - # FIXME: need to enable the following once we have a way to run their init/compile script in CI + # NOTE: Enable these once we have a way to run their init/compile script in CI. f"{snippets_folder}/dbt.py", f"{snippets_folder}/sdf.py", f"{snippets_folder}/airbyte.py", @@ -45,9 +45,9 @@ f"{snippets_folder}/airbyte_cloud/create_airbyte_cloud_all_assets_job.py", f"{snippets_folder}/airbyte_cloud/create_airbyte_cloud_selection_job.py", f"{snippets_folder}/airbyte_cloud/schedule_airbyte_cloud_jobs.py", - # FIXME: this breaks on py3.8 and seems related to the non-dagster dependencies + # NOTE: This breaks on py3.8 and seems related to the non-dagster dependencies. f"{snippets_folder}/pandera.py", - # FIXME: include tests + # NOTE: Include tests when the optional deps are available in CI. f"{snippets_folder}/anthropic.py", f"{snippets_folder}/chroma.py", f"{snippets_folder}/gemini.py", @@ -134,7 +134,7 @@ f"{snippets_folder}/tableau/refresh-and-materialize-tableau-assets.py", f"{snippets_folder}/tableau/representing-tableau-cloud-assets.py", f"{snippets_folder}/tableau/representing-tableau-server-assets.py", - # FIXME: serverless pipes files require Databricks environment/credentials + # NOTE: Serverless pipes files require Databricks environment/credentials. f"{snippets_folder}/databricks/dagster_code_serverless.py", f"{snippets_folder}/databricks/databricks_code_serverless.py", f"{snippets_folder}/iceberg/io_manager_daft.py", diff --git a/js_modules/dagster-ui/packages/ui-core/src/assets/asset-checks/VirtualizedAssetCheckTable.tsx b/js_modules/dagster-ui/packages/ui-core/src/assets/asset-checks/VirtualizedAssetCheckTable.tsx index 4ba1ec33c4b58..0a85c008441c0 100644 --- a/js_modules/dagster-ui/packages/ui-core/src/assets/asset-checks/VirtualizedAssetCheckTable.tsx +++ b/js_modules/dagster-ui/packages/ui-core/src/assets/asset-checks/VirtualizedAssetCheckTable.tsx @@ -71,9 +71,10 @@ interface AssetCheckRowProps { export const VirtualizedAssetCheckRow = ({assetNode, height, start, row}: AssetCheckRowProps) => { const execution = row.executionForLatestMaterialization; const timestamp = execution?.evaluation?.timestamp; + const rowTestId = `row-${assetNode.assetKey.path.join('-')}-${row.name}`; return ( - + diff --git a/python_modules/libraries/dagster-cloud-cli/setup.py b/python_modules/libraries/dagster-cloud-cli/setup.py index bda09dd61181f..77c310cf19994 100644 --- a/python_modules/libraries/dagster-cloud-cli/setup.py +++ b/python_modules/libraries/dagster-cloud-cli/setup.py @@ -20,7 +20,7 @@ def get_version() -> str: author="Dagster Labs", author_email="hello@dagsterlabs.com", license="Apache-2.0", - description="", # TODO - fill out description + description="Dagster Cloud command line interface.", url=( "https://github.com/dagster-io/dagster/tree/master/python_modules/libraries/dagster-cloud-cli" ), diff --git a/python_modules/libraries/dagster-dg-cli/setup.py b/python_modules/libraries/dagster-dg-cli/setup.py index f709c9cba9132..61bf3749c0740 100644 --- a/python_modules/libraries/dagster-dg-cli/setup.py +++ b/python_modules/libraries/dagster-dg-cli/setup.py @@ -20,7 +20,7 @@ def get_version() -> str: author="Dagster Labs", author_email="hello@dagsterlabs.com", license="Apache-2.0", - description="", # TODO - fill out description + description="Dagster dg command line interface.", url=( "https://github.com/dagster-io/dagster/tree/master/python_modules/libraries/dagster-dg-cli" ), diff --git a/python_modules/libraries/dagster-dg-cli/tox.ini b/python_modules/libraries/dagster-dg-cli/tox.ini index 368a1e6b57ac0..69636d73e9813 100644 --- a/python_modules/libraries/dagster-dg-cli/tox.ini +++ b/python_modules/libraries/dagster-dg-cli/tox.ini @@ -8,7 +8,7 @@ passenv = BUILDKITE* PYTEST_ADDOPTS PYTEST_PLUGINS - # TODO add additional env var requirements and add them to .buildkite/dagster-buildkite/dagster_buildkite/steps/packages.py + # NOTE: Add additional env var requirements and add them to .buildkite/dagster-buildkite/dagster_buildkite/steps/packages.py. install_command = uv pip install {opts} {packages} diff --git a/python_modules/libraries/dagster-dg-core/setup.py b/python_modules/libraries/dagster-dg-core/setup.py index 0743355ebffa1..7e421cf76ee61 100644 --- a/python_modules/libraries/dagster-dg-core/setup.py +++ b/python_modules/libraries/dagster-dg-core/setup.py @@ -20,7 +20,7 @@ def get_version() -> str: author="Dagster Labs", author_email="hello@dagsterlabs.com", license="Apache-2.0", - description="", # TODO - fill out description + description="Core utilities for Dagster dg.", url=( "https://github.com/dagster-io/dagster/tree/master/python_modules/libraries/dagster-dg-core" ), diff --git a/python_modules/libraries/dagster-omni/tox.ini b/python_modules/libraries/dagster-omni/tox.ini index 21fcc24c7ece2..77008c15fe3f7 100644 --- a/python_modules/libraries/dagster-omni/tox.ini +++ b/python_modules/libraries/dagster-omni/tox.ini @@ -8,7 +8,7 @@ passenv = BUILDKITE* PYTEST_ADDOPTS PYTEST_PLUGINS - # TODO add additional env var requirements and add them to .buildkite/dagster-buildkite/dagster_buildkite/steps/packages.py + # NOTE: Add additional env var requirements and add them to .buildkite/dagster-buildkite/dagster_buildkite/steps/packages.py. install_command = uv pip install {opts} {packages} deps = -e ../../dagster[test] @@ -21,4 +21,4 @@ allowlist_externals = uv commands = !windows: /bin/bash -c '! pip list --exclude-editable | grep -e dagster -e dagit' - pytest -vv ./dagster_omni_tests --durations 10 {posargs} \ No newline at end of file + pytest -vv ./dagster_omni_tests --durations 10 {posargs} diff --git a/python_modules/libraries/dagstermill/dagstermill/examples/repository.py b/python_modules/libraries/dagstermill/dagstermill/examples/repository.py index f37fed8c6dd95..80787a42b5338 100644 --- a/python_modules/libraries/dagstermill/dagstermill/examples/repository.py +++ b/python_modules/libraries/dagstermill/dagstermill/examples/repository.py @@ -83,7 +83,7 @@ def test_nb_op(name, **kwargs): common_resource_defs = { "output_notebook_io_manager": local_output_notebook_io_manager, - "io_manager": FilesystemIOManager(), # TODO check if necessary + "io_manager": FilesystemIOManager(), # NOTE: Used for default outputs in these examples. } hello_world = test_nb_op("hello_world", outs={}) @@ -305,17 +305,15 @@ def error_job(): # We need type-ignores here because type checkers don't understand the `*_PRESENT` kwargs. clean_data = test_nb_op("clean_data", outs={DEFAULT_OUTPUT: Out(DataFrame)}) # pyright: ignore[reportPossiblyUnboundVariable] - # FIXME add an output to this - tutorial_LR = test_nb_op("tutorial_LR", ins={"df": In(DataFrame)}) # pyright: ignore[reportPossiblyUnboundVariable] + tutorial_lr = test_nb_op("tutorial_LR", ins={"df": In(DataFrame)}) # pyright: ignore[reportPossiblyUnboundVariable] - tutorial_RF = test_nb_op("tutorial_RF", ins={"df": In(DataFrame)}) # pyright: ignore[reportPossiblyUnboundVariable] + tutorial_rf = test_nb_op("tutorial_RF", ins={"df": In(DataFrame)}) # pyright: ignore[reportPossiblyUnboundVariable] @job(resource_defs=common_resource_defs) def tutorial_job(): dfr, _ = clean_data() - # FIXME get better names for these - tutorial_LR(dfr) - tutorial_RF(dfr) + tutorial_lr.alias("train_linear_regression")(dfr) + tutorial_rf.alias("train_random_forest")(dfr) @op(required_resource_keys={"list"}) diff --git a/scripts/templates_create_dagster_package/api-docs.rst.tmpl b/scripts/templates_create_dagster_package/api-docs.rst.tmpl index a7d010f73e431..6ba94c780a5f1 100644 --- a/scripts/templates_create_dagster_package/api-docs.rst.tmpl +++ b/scripts/templates_create_dagster_package/api-docs.rst.tmpl @@ -1,15 +1,15 @@ {{ formal_name }} ({{ hyphen_name }}) ----------------------------------------------------- -.. TODO description +.. Description goes here. .. currentmodule:: {{ underscore_name }} -.. TODO fill out remaining docs +.. Fill out remaining docs. -.. TODO add your package name to the following files: +.. Add your package name to the following files: .. docs/content/_navigation.json .. docs/content/_apidocs.mdx .. docs/sphinx/conf.py -.. docs/sphinx/index.rst \ No newline at end of file +.. docs/sphinx/index.rst diff --git a/scripts/templates_create_dagster_package/setup.py.tmpl b/scripts/templates_create_dagster_package/setup.py.tmpl index 32a374ace96bc..015cf4f045eb2 100644 --- a/scripts/templates_create_dagster_package/setup.py.tmpl +++ b/scripts/templates_create_dagster_package/setup.py.tmpl @@ -10,7 +10,7 @@ def get_version() -> str: return version["__version__"] -# TODO - add your package to scripts/install_dev_python_modules.py +# NOTE: Add your package to scripts/install_dev_python_modules.py when ready. ver = get_version() # dont pin dev installs to avoid pip dep resolver issues @@ -21,7 +21,7 @@ setup( author="Dagster Labs", author_email="hello@dagsterlabs.com", license="Apache-2.0", - description="", # TODO - fill out description + description="", # NOTE: Fill out description. url=( "https://github.com/dagster-io/dagster/tree/master/python_modules/libraries/" "{{ hyphen_name }}" @@ -39,7 +39,7 @@ setup( include_package_data=True, install_requires=[ f"dagster{pin}", - # TODO - fill in remaining dependencies + # NOTE: Fill in remaining dependencies. ], zip_safe=False, ) diff --git a/scripts/templates_create_dagster_package/tox.ini.tmpl b/scripts/templates_create_dagster_package/tox.ini.tmpl index 5e78bd7cd87c4..50742b872feb2 100644 --- a/scripts/templates_create_dagster_package/tox.ini.tmpl +++ b/scripts/templates_create_dagster_package/tox.ini.tmpl @@ -9,12 +9,12 @@ passenv = BUILDKITE* PYTEST_ADDOPTS PYTEST_PLUGINS - # TODO add additional env var requirements and add them to .buildkite/dagster-buildkite/dagster_buildkite/steps/packages.py + # NOTE: Add additional env var requirements and add them to .buildkite/dagster-buildkite/dagster_buildkite/steps/packages.py. install_command = uv pip install {opts} {packages} deps = -e ../../dagster[test] -e . - # TODO fill in remaining dependencies + # NOTE: Fill in remaining dependencies. allowlist_externals = /bin/bash uv From 47566f3bb7fbe7b3d720d2294fde495966f0680a Mon Sep 17 00:00:00 2001 From: Ramachandra Nalam Date: Wed, 14 Jan 2026 11:24:03 -0800 Subject: [PATCH 2/4] Improve scaling and diagnostics --- .../configeditor/codemirror-yaml/mode.tsx | 18 ++ .../automation/parse_spark_configs.py | 5 +- .../implementation/fetch_assets.py | 78 ++++++-- .../asset_graph_view/asset_graph_view.py | 33 +++- .../assets/graph/base_asset_graph.py | 8 + .../dagster/dagster/_core/events/utils.py | 59 +++--- .../dagster/_core/execution/asset_backfill.py | 47 ++++- .../_core/executor/child_process_executor.py | 171 ++++++++++++++++-- .../dagster/_core/instance/instance.py | 30 +++ .../dagster/_core/storage/event_log/base.py | 17 ++ .../_core/storage/event_log/sql_event_log.py | 34 ++++ .../_daemon/monitoring/run_monitoring.py | 105 +++++++---- .../dagster/dagster/_daemon/sensor.py | 28 ++- .../_utils/caching_instance_queryer.py | 19 ++ .../dagster_gcp/bigquery/configs.py | 57 +++--- .../dagster-gcp/dagster_gcp/bigquery/ops.py | 21 +++ .../dagster_spark/configs_spark.py | 33 ++-- .../dagster-spark/dagster_spark/types.py | 36 ++++ 18 files changed, 637 insertions(+), 162 deletions(-) diff --git a/js_modules/dagster-ui/packages/ui-components/src/components/configeditor/codemirror-yaml/mode.tsx b/js_modules/dagster-ui/packages/ui-components/src/components/configeditor/codemirror-yaml/mode.tsx index 07b4a8cd648c1..edd04d5139411 100644 --- a/js_modules/dagster-ui/packages/ui-components/src/components/configeditor/codemirror-yaml/mode.tsx +++ b/js_modules/dagster-ui/packages/ui-components/src/components/configeditor/codemirror-yaml/mode.tsx @@ -110,6 +110,9 @@ export const RegExps = { VARIABLE: /^\s*(\&|\*)[a-z0-9\._-]+$/i, }; +const MAX_YAML_LINT_CHARS = 200_000; +const MAX_YAML_LINT_LINES = 5_000; + const defineYamlMode = () => { CodeMirror.defineMode('yaml', () => { return { @@ -775,6 +778,21 @@ const registerYamlLint = () => { ): Promise> => { const codeMirrorDoc = editor.getDoc(); + const lineCount = text.split('\n').length; + if (text.length > MAX_YAML_LINT_CHARS || lineCount > MAX_YAML_LINT_LINES) { + return [ + { + message: + `YAML lint skipped for ${lineCount} lines / ${text.length} chars ` + + `to keep the editor responsive.`, + severity: 'warning', + type: 'syntax', + from: {line: 0, ch: 0}, + to: {line: 0, ch: 0}, + }, + ]; + } + // TODO: In some scenarios where every line yields an error `parseDocument` can take 1s+ // and returns 20,000+ errors. The library does not have a "bail out" option but we need one. // However we can't switch libraries because we need the structured document model this returns. diff --git a/python_modules/automation/automation/parse_spark_configs.py b/python_modules/automation/automation/parse_spark_configs.py index 0165acf5b1018..c06eb294112c5 100644 --- a/python_modules/automation/automation/parse_spark_configs.py +++ b/python_modules/automation/automation/parse_spark_configs.py @@ -24,8 +24,8 @@ class ConfigType(Enum): INT = "IntSource" FLOAT = "Float" BOOL = "Bool" - MEMORY = "StringSource" # TODO: We should handle memory field types - TIME = "StringSource" # TODO: We should handle time field types + MEMORY = "SparkMemory" + TIME = "SparkTime" CONFIG_TYPES = { @@ -260,6 +260,7 @@ def serialize(result: SparkConfigNode) -> bytes: with IndentingBufferPrinter() as printer: printer.write_header() printer.line("from dagster import Bool, Field, Float, IntSource, Permissive, StringSource") + printer.line("from dagster_spark.types import SparkMemory, SparkTime") printer.blank_line() printer.blank_line() printer.line("def spark_config():") diff --git a/python_modules/dagster-graphql/dagster_graphql/implementation/fetch_assets.py b/python_modules/dagster-graphql/dagster_graphql/implementation/fetch_assets.py index a8eae3c388a3f..dec5599fb5086 100644 --- a/python_modules/dagster-graphql/dagster_graphql/implementation/fetch_assets.py +++ b/python_modules/dagster-graphql/dagster_graphql/implementation/fetch_assets.py @@ -1,6 +1,8 @@ +import bisect import datetime +import heapq from collections import defaultdict -from collections.abc import Sequence +from collections.abc import Iterator, Sequence from typing import TYPE_CHECKING, AbstractSet, Optional, Union, cast # noqa: UP035 import dagster_shared.seven as seven @@ -49,6 +51,9 @@ from dagster_graphql.schema.util import ResolveInfo +ASSET_KEY_PAGE_SIZE = 1000 + + def _normalize_asset_cursor_str(cursor_string: Optional[str]) -> Optional[str]: # the cursor for assets is derived from a json serialized string of the path. Because there are # json serialization differences between JS and Python in its treatment of whitespace, we should @@ -78,7 +83,7 @@ def get_asset_records( materialized_assets = sorted( # TODO(salazarm): Replace this with `get_asset_records` once that supports pagination arguments. instance.get_asset_keys(prefix=prefix, limit=limit, cursor=normalized_cursor_str), - key=str, + key=lambda asset_key: asset_key.to_string(), ) return GrapheneAssetRecordConnection( @@ -93,6 +98,42 @@ def get_asset_records( ) +def _iter_asset_graph_keys( + asset_graph, prefix: Optional[Sequence[str]], cursor: Optional[str] +) -> Iterator[tuple[str, int, AssetKey]]: + prefix_list = list(prefix) if prefix else None + key_strings = asset_graph.sorted_asset_key_strings + sorted_keys = asset_graph.sorted_asset_keys + start_idx = bisect.bisect_right(key_strings, cursor) if cursor else 0 + + for key_str, key in zip(key_strings[start_idx:], sorted_keys[start_idx:]): + if not prefix_list or key.path[: len(prefix_list)] == prefix_list: + yield (key_str, 1, key) + + +def _iter_materialized_keys( + instance, + prefix: Optional[Sequence[str]], + cursor: Optional[str], + limit: Optional[int], +) -> Iterator[tuple[str, int, AssetKey]]: + if limit and limit > 0: + page_size = min(limit, ASSET_KEY_PAGE_SIZE) + else: + page_size = ASSET_KEY_PAGE_SIZE + page_cursor = cursor + + while True: + page = instance.get_asset_keys(prefix=prefix, limit=page_size, cursor=page_cursor) + if not page: + break + for key in page: + yield (key.to_string(), 0, key) + if len(page) < page_size: + break + page_cursor = page[-1].to_string() + + def get_assets( graphene_info: "ResolveInfo", prefix: Optional[Sequence[str]] = None, @@ -115,26 +156,25 @@ def get_assets( normalized_cursor_str = _normalize_asset_cursor_str(cursor) if asset_keys is None: - materialized_keys = instance.get_asset_keys( - prefix=prefix, limit=limit, cursor=normalized_cursor_str + asset_graph_iter = _iter_asset_graph_keys( + graphene_info.context.asset_graph, prefix, normalized_cursor_str ) - - asset_graph_keys = { - asset_key - for asset_key in graphene_info.context.asset_graph.get_all_asset_keys() - if ( - (not prefix or asset_key.path[: len(prefix)] == prefix) - and (not normalized_cursor_str or asset_key.to_string() > normalized_cursor_str) - and (not asset_keys or asset_key in asset_keys) - ) - } - - merged_asset_keys = sorted(set(materialized_keys).union(asset_graph_keys), key=str) + materialized_iter = _iter_materialized_keys( + instance, prefix, normalized_cursor_str, limit + ) + merged_asset_keys = [] + last_key = None + for key_str, _, asset_key in heapq.merge(materialized_iter, asset_graph_iter): + if key_str == last_key: + continue + merged_asset_keys.append(asset_key) + last_key = key_str + if limit and len(merged_asset_keys) >= limit: + break else: merged_asset_keys = asset_keys - - if limit: - merged_asset_keys = merged_asset_keys[:limit] + if limit: + merged_asset_keys = merged_asset_keys[:limit] return GrapheneAssetConnection( nodes=[GrapheneAsset(key=asset_key) for asset_key in merged_asset_keys], diff --git a/python_modules/dagster/dagster/_core/asset_graph_view/asset_graph_view.py b/python_modules/dagster/dagster/_core/asset_graph_view/asset_graph_view.py index 0cbdeae878f75..1786e3e08076a 100644 --- a/python_modules/dagster/dagster/_core/asset_graph_view/asset_graph_view.py +++ b/python_modules/dagster/dagster/_core/asset_graph_view/asset_graph_view.py @@ -724,15 +724,30 @@ async def _compute_missing_asset_subset( ) return from_subset.compute_difference(materialized_subset) else: - # more expensive call - missing_asset_partitions = { - ap - for ap in from_subset.expensively_compute_asset_partitions() - if not self._queryer.asset_partition_has_materialization_or_observation(ap) - } - return self.get_asset_subset_from_asset_partitions( - key=key, asset_partitions=missing_asset_partitions - ) + partitions_def = self._get_partitions_def(key) + if partitions_def: + observed_partition_keys = self._queryer.get_materialized_or_observed_partition_keys( + asset_key=key + ) + with partition_loading_context(new_ctx=self.partition_loading_context): + observed_partition_keys = self._validate_partition_keys( + key, observed_partition_keys + ) + observed_subset = EntitySubset( + self, + key=key, + value=_ValidatedEntitySubsetValue( + partitions_def.subset_with_partition_keys(observed_partition_keys) + ), + ) + return from_subset.compute_difference(observed_subset) + else: + has_materialization_or_observation = ( + self._queryer.asset_partition_has_materialization_or_observation( + AssetKeyPartitionKey(key) + ) + ) + return self.get_empty_subset(key=key) if has_materialization_or_observation else from_subset @cached_method async def compute_run_in_progress_subset(self, *, key: EntityKey) -> EntitySubset: diff --git a/python_modules/dagster/dagster/_core/definitions/assets/graph/base_asset_graph.py b/python_modules/dagster/dagster/_core/definitions/assets/graph/base_asset_graph.py index 0816851d23c28..7f42f200f31af 100644 --- a/python_modules/dagster/dagster/_core/definitions/assets/graph/base_asset_graph.py +++ b/python_modules/dagster/dagster/_core/definitions/assets/graph/base_asset_graph.py @@ -305,6 +305,14 @@ def entity_dep_graph(self) -> DependencyGraph[EntityKey]: def get_all_asset_keys(self) -> AbstractSet[AssetKey]: return set(self._asset_nodes_by_key) + @cached_property + def sorted_asset_keys(self) -> Sequence[AssetKey]: + return sorted(self._asset_nodes_by_key, key=lambda key: key.to_string()) + + @cached_property + def sorted_asset_key_strings(self) -> Sequence[str]: + return [key.to_string() for key in self.sorted_asset_keys] + # since this is an ABC and @cached_property has class level locking on py < 3.12 # use property & cached_method instead diff --git a/python_modules/dagster/dagster/_core/events/utils.py b/python_modules/dagster/dagster/_core/events/utils.py index 1b8d0385ffd59..9102eba758a9f 100644 --- a/python_modules/dagster/dagster/_core/events/utils.py +++ b/python_modules/dagster/dagster/_core/events/utils.py @@ -1,3 +1,4 @@ +import json from json import JSONDecodeError from dagster_shared.serdes import deserialize_value @@ -19,33 +20,37 @@ def filter_dagster_events_from_cli_logs(log_lines): """ check.list_param(log_lines, "log_lines", str) - coalesced_lines = [] - buffer = [] - in_split_line = False - for raw_line in log_lines: - line = raw_line.strip() - if not in_split_line and line.startswith("{"): - if line.endswith("}"): - coalesced_lines.append(line) - else: - buffer.append(line) - in_split_line = True - elif in_split_line: - buffer.append(line) - if line.endswith("}"): # Note: hack, this may not have been the end of the full object - coalesced_lines.append("".join(buffer)) - buffer = [] - in_split_line = False - events = [] - for line in coalesced_lines: - try: - events.append(deserialize_value(line, DagsterEvent)) - except JSONDecodeError: - pass - except check.CheckError: - pass - except DeserializationError: - pass + decoder = json.JSONDecoder() + buffer = "" + for raw_line in log_lines: + buffer += raw_line.rstrip("\r\n") + + while buffer: + trimmed = buffer.lstrip() + if not trimmed: + buffer = "" + break + + start_idx = trimmed.find("{") + if start_idx == -1: + buffer = "" + break + + if start_idx: + trimmed = trimmed[start_idx:] + + try: + _, end_idx = decoder.raw_decode(trimmed) + except JSONDecodeError: + buffer = trimmed + break + + candidate = trimmed[:end_idx] + buffer = trimmed[end_idx:] + try: + events.append(deserialize_value(candidate, DagsterEvent)) + except (JSONDecodeError, check.CheckError, DeserializationError): + pass return events diff --git a/python_modules/dagster/dagster/_core/execution/asset_backfill.py b/python_modules/dagster/dagster/_core/execution/asset_backfill.py index 473a40fb458cb..1d372e5bf9aa4 100644 --- a/python_modules/dagster/dagster/_core/execution/asset_backfill.py +++ b/python_modules/dagster/dagster/_core/execution/asset_backfill.py @@ -80,6 +80,40 @@ def get_asset_backfill_run_chunk_size(): ) +def _get_first_partition_keys_in_subset( + asset_graph_view: AssetGraphView, + partitions_def: PartitionsDefinition, + subset: PartitionsSubset, + limit: int, +) -> Sequence[str]: + if limit <= 0: + return [] + + result: list[str] = [] + cursor = None + page_limit = max(limit, 100) + with partition_loading_context(new_ctx=asset_graph_view.partition_loading_context): + while len(result) < limit: + page = partitions_def.get_paginated_partition_keys( + context=asset_graph_view.partition_loading_context, + limit=page_limit, + ascending=True, + cursor=cursor, + ) + if not page.results: + break + for partition_key in page.results: + if partition_key in subset: + result.append(partition_key) + if len(result) >= limit: + break + if not page.has_more: + break + cursor = page.cursor + + return result + + class AssetBackfillStatus(Enum): IN_PROGRESS = "IN_PROGRESS" MATERIALIZED = "MATERIALIZED" @@ -1844,16 +1878,17 @@ def _should_backfill_atomic_asset_subset_unit( num_allowed_partitions = ( self_dependent_node.backfill_policy.max_partitions_per_run ) - # TODO add a method for paginating through the keys in order - # and returning the first N instead of listing all of them - # (can't use expensively_compute_asset_partitions because it returns - # an unordered set) internal_value = entity_subset_to_filter.get_internal_value() partition_keys_to_include = ( - list(internal_value.get_partition_keys()) + _get_first_partition_keys_in_subset( + asset_graph_view, + check.not_none(self_dependent_node.partitions_def), + internal_value, + num_allowed_partitions, + ) if isinstance(internal_value, PartitionsSubset) else [None] - )[:num_allowed_partitions] + ) partition_subset_to_include = AssetGraphSubset.from_asset_partition_set( { AssetKeyPartitionKey(self_dependent_node.key, partition_key) diff --git a/python_modules/dagster/dagster/_core/executor/child_process_executor.py b/python_modules/dagster/dagster/_core/executor/child_process_executor.py index 7d8606f279947..5186b6d7a63b0 100644 --- a/python_modules/dagster/dagster/_core/executor/child_process_executor.py +++ b/python_modules/dagster/dagster/_core/executor/child_process_executor.py @@ -1,10 +1,13 @@ """Facilities for running arbitrary commands in child processes.""" +import io import os import queue import sys +import tempfile from abc import ABC, abstractmethod from collections.abc import Iterator +from contextlib import contextmanager, redirect_stderr, redirect_stdout from multiprocessing import Queue from multiprocessing.context import BaseContext as MultiprocessingBaseContext from multiprocessing.process import BaseProcess @@ -35,11 +38,24 @@ class ChildProcessDoneEvent(NamedTuple("ChildProcessDoneEvent", [("pid", int)]), class ChildProcessSystemErrorEvent( NamedTuple( - "ChildProcessSystemErrorEvent", [("pid", int), ("error_info", SerializableErrorInfo)] + "ChildProcessSystemErrorEvent", + [ + ("pid", int), + ("error_info", SerializableErrorInfo), + ("stdout", Optional[str]), + ("stderr", Optional[str]), + ], ), ChildProcessEvent, ): - pass + def __new__( + cls, + pid: int, + error_info: SerializableErrorInfo, + stdout: Optional[str] = None, + stderr: Optional[str] = None, + ): + return super().__new__(cls, pid, error_info, stdout, stderr) class ChildProcessCommand(ABC): @@ -59,13 +75,125 @@ def execute(self) -> Iterator[Union[ChildProcessEvent, "DagsterEvent"]]: class ChildProcessCrashException(Exception): """Thrown when the child process crashes.""" - def __init__(self, pid, exit_code=None): + def __init__( + self, + pid: int, + exit_code: Optional[int] = None, + stdout: Optional[str] = None, + stderr: Optional[str] = None, + ): self.pid = pid self.exit_code = exit_code - super().__init__() + self.stdout = stdout + self.stderr = stderr + super().__init__(self._build_message()) + + def _build_message(self) -> str: + base_message = f"Child process {self.pid} crashed with exit code {self.exit_code}." + output = _format_captured_output(self.stdout, self.stderr) + if output: + return f"{base_message}\n\n{output}" + return base_message + + def __str__(self) -> str: + return self._build_message() + + +CAPTURED_CHILD_PROCESS_LOG_BYTES = int( + os.getenv("DAGSTER_CHILD_PROCESS_LOG_CAPTURE_BYTES", "65536") +) + + +class _TeeStream(io.TextIOBase): + def __init__(self, *streams: io.TextIOBase): + self._streams = streams + + def write(self, s: str) -> int: + for stream in self._streams: + stream.write(s) + stream.flush() + return len(s) + + def flush(self) -> None: + for stream in self._streams: + stream.flush() -def _execute_command_in_child_process(event_queue: Queue, command: ChildProcessCommand): +def _format_captured_output(stdout: Optional[str], stderr: Optional[str]) -> str: + sections = [] + if stdout: + sections.append(f"Captured stdout:\n{stdout}") + if stderr: + sections.append(f"Captured stderr:\n{stderr}") + return "\n\n".join(sections) + + +def _read_log_tail(path: Optional[str]) -> Optional[str]: + if not path: + return None + try: + with open(path, "rb") as log_file: + log_file.seek(0, os.SEEK_END) + size = log_file.tell() + if size > CAPTURED_CHILD_PROCESS_LOG_BYTES: + log_file.seek(-CAPTURED_CHILD_PROCESS_LOG_BYTES, os.SEEK_END) + else: + log_file.seek(0) + data = log_file.read() + return data.decode("utf-8", errors="replace") + except OSError: + return None + + +def _with_output_snippet( + error_info: SerializableErrorInfo, stdout: Optional[str], stderr: Optional[str] +) -> SerializableErrorInfo: + output = _format_captured_output(stdout, stderr) + if not output: + return error_info + return SerializableErrorInfo( + message=f"{error_info.message.rstrip()}\n\n{output}", + stack=error_info.stack, + cls_name=error_info.cls_name, + cause=error_info.cause, + context=error_info.context, + ) + + +def _make_temp_log_file(prefix: str) -> str: + fd, path = tempfile.mkstemp(prefix=prefix, suffix=".log") + os.close(fd) + return path + + +@contextmanager +def _redirect_output(stdout_path: Optional[str], stderr_path: Optional[str]): + if not stdout_path and not stderr_path: + yield + return + + stdout_file = open(stdout_path, "w", encoding="utf-8") if stdout_path else None + stderr_file = open(stderr_path, "w", encoding="utf-8") if stderr_path else None + try: + stdout_stream = _TeeStream(sys.stdout, stdout_file) if stdout_file else sys.stdout + stderr_stream = _TeeStream(sys.stderr, stderr_file) if stderr_file else sys.stderr + with redirect_stdout(stdout_stream), redirect_stderr(stderr_stream): + yield + finally: + if stdout_file: + stdout_file.flush() + stdout_file.close() + if stderr_file: + stderr_file.flush() + stderr_file.close() + + +def _execute_command_in_child_process( + event_queue: Queue, + command: ChildProcessCommand, + stdout_path: Optional[str], + stderr_path: Optional[str], +): """Wraps the execution of a ChildProcessCommand. Handles errors and communicates across a queue with the parent process. @@ -76,18 +204,24 @@ def _execute_command_in_child_process(event_queue: Queue, command: ChildProcessC pid = os.getpid() event_queue.put(ChildProcessStartEvent(pid=pid)) try: - for step_event in command.execute(): - event_queue.put(step_event) - event_queue.put(ChildProcessDoneEvent(pid=pid)) + with _redirect_output(stdout_path, stderr_path): + for step_event in command.execute(): + event_queue.put(step_event) + event_queue.put(ChildProcessDoneEvent(pid=pid)) except ( Exception, KeyboardInterrupt, DagsterExecutionInterruptedError, ): + stdout = _read_log_tail(stdout_path) + stderr = _read_log_tail(stderr_path) + error_info = _with_output_snippet( + serializable_error_info_from_exc_info(sys.exc_info()), stdout, stderr + ) event_queue.put( ChildProcessSystemErrorEvent( - pid=pid, error_info=serializable_error_info_from_exc_info(sys.exc_info()) + pid=pid, error_info=error_info, stdout=stdout, stderr=stderr ) ) @@ -147,9 +281,12 @@ def execute_child_process_command( check.inst_param(command, "command", ChildProcessCommand) event_queue = multiprocessing_ctx.Queue() + stdout_path = _make_temp_log_file("dagster-child-stdout-") + stderr_path = _make_temp_log_file("dagster-child-stderr-") try: process = multiprocessing_ctx.Process( # type: ignore - target=_execute_command_in_child_process, args=(event_queue, command) + target=_execute_command_in_child_process, + args=(event_queue, command, stdout_path, stderr_path), ) process.start() yield process @@ -168,9 +305,19 @@ def execute_child_process_command( completed_properly = True if not completed_properly: - # TODO Figure out what to do about stderr/stdout - raise ChildProcessCrashException(pid=process.pid, exit_code=process.exitcode) + process.join() + stdout = _read_log_tail(stdout_path) + stderr = _read_log_tail(stderr_path) + raise ChildProcessCrashException( + pid=process.pid, exit_code=process.exitcode, stdout=stdout, stderr=stderr + ) process.join() finally: event_queue.close() + for path in (stdout_path, stderr_path): + if path: + try: + os.remove(path) + except OSError: + pass diff --git a/python_modules/dagster/dagster/_core/instance/instance.py b/python_modules/dagster/dagster/_core/instance/instance.py index 031885aa95ee0..456512c599c45 100644 --- a/python_modules/dagster/dagster/_core/instance/instance.py +++ b/python_modules/dagster/dagster/_core/instance/instance.py @@ -609,6 +609,20 @@ def add_dynamic_partitions( """ return self._event_storage.add_dynamic_partitions(partitions_def_name, partition_keys) + @public + @traced + def delete_dynamic_partitions( + self, partitions_def_name: str, partition_keys: Sequence[str] + ) -> None: + """Delete partitions for the specified :py:class:`DynamicPartitionsDefinition`. + If a partition does not exist, it is ignored. + + Args: + partitions_def_name (str): The name of the `DynamicPartitionsDefinition`. + partition_keys (Sequence[str]): Partition keys to delete. + """ + return self._event_storage.delete_dynamic_partitions(partitions_def_name, partition_keys) + @public @traced def delete_dynamic_partition(self, partitions_def_name: str, partition_key: str) -> None: @@ -631,6 +645,22 @@ def get_dynamic_partitions(self, partitions_def_name: str) -> Sequence[str]: """ return self._event_storage.get_dynamic_partitions(partitions_def_name) + @public + @traced + def get_dynamic_partitions_by_keys( + self, partitions_def_name: str, partition_keys: Sequence[str] + ) -> Sequence[str]: + """Return the subset of provided partition keys that exist for the specified + :py:class:`DynamicPartitionsDefinition`. + + Args: + partitions_def_name (str): The name of the `DynamicPartitionsDefinition`. + partition_keys (Sequence[str]): Partition keys to check. + """ + return self._event_storage.get_dynamic_partitions_by_keys( + partitions_def_name, partition_keys + ) + def get_dynamic_partitions_definition_id(self, partitions_def_name: str) -> str: from dagster._core.definitions.partitions.context import partition_loading_context from dagster._core.definitions.partitions.utils import ( diff --git a/python_modules/dagster/dagster/_core/storage/event_log/base.py b/python_modules/dagster/dagster/_core/storage/event_log/base.py index e9b3dbcbded80..809d63a43cdae 100644 --- a/python_modules/dagster/dagster/_core/storage/event_log/base.py +++ b/python_modules/dagster/dagster/_core/storage/event_log/base.py @@ -533,6 +533,23 @@ def delete_dynamic_partition(self, partitions_def_name: str, partition_key: str) """Delete a partition for the specified dynamic partitions definition.""" raise NotImplementedError() + def delete_dynamic_partitions( + self, partitions_def_name: str, partition_keys: Sequence[str] + ) -> None: + """Delete partitions for the specified dynamic partitions definition.""" + for partition_key in partition_keys: + self.delete_dynamic_partition(partitions_def_name, partition_key) + + def get_dynamic_partitions_by_keys( + self, partitions_def_name: str, partition_keys: Sequence[str] + ) -> Sequence[str]: + """Return the subset of partition keys that exist for the dynamic partitions definition.""" + return [ + partition_key + for partition_key in partition_keys + if self.has_dynamic_partition(partitions_def_name, partition_key) + ] + def alembic_version(self) -> Optional[AlembicVersion]: return None diff --git a/python_modules/dagster/dagster/_core/storage/event_log/sql_event_log.py b/python_modules/dagster/dagster/_core/storage/event_log/sql_event_log.py index 498ea3e0e45ba..ceaa8b64b4714 100644 --- a/python_modules/dagster/dagster/_core/storage/event_log/sql_event_log.py +++ b/python_modules/dagster/dagster/_core/storage/event_log/sql_event_log.py @@ -2131,6 +2131,24 @@ def has_dynamic_partition(self, partitions_def_name: str, partition_key: str) -> return len(results) > 0 + def get_dynamic_partitions_by_keys( + self, partitions_def_name: str, partition_keys: Sequence[str] + ) -> Sequence[str]: + self._check_partitions_table() + if not partition_keys: + return [] + + query = db_select([DynamicPartitionsTable.c.partition]).where( + db.and_( + DynamicPartitionsTable.c.partitions_def_name == partitions_def_name, + DynamicPartitionsTable.c.partition.in_(partition_keys), + ) + ) + with self.index_connection() as conn: + results = conn.execute(query).fetchall() + + return [cast("str", row[0]) for row in results] + def add_dynamic_partitions( self, partitions_def_name: str, partition_keys: Sequence[str] ) -> None: @@ -2172,6 +2190,22 @@ def delete_dynamic_partition(self, partitions_def_name: str, partition_key: str) ) ) + def delete_dynamic_partitions( + self, partitions_def_name: str, partition_keys: Sequence[str] + ) -> None: + self._check_partitions_table() + if not partition_keys: + return + with self.index_connection() as conn: + conn.execute( + DynamicPartitionsTable.delete().where( + db.and_( + DynamicPartitionsTable.c.partitions_def_name == partitions_def_name, + DynamicPartitionsTable.c.partition.in_(partition_keys), + ) + ) + ) + @cached_property def supports_global_concurrency_limits(self) -> bool: # pyright: ignore[reportIncompatibleMethodOverride] return self.has_table(ConcurrencySlotsTable.name) diff --git a/python_modules/dagster/dagster/_daemon/monitoring/run_monitoring.py b/python_modules/dagster/dagster/_daemon/monitoring/run_monitoring.py index de039fe7a37ed..089c9cf162bff 100644 --- a/python_modules/dagster/dagster/_daemon/monitoring/run_monitoring.py +++ b/python_modules/dagster/dagster/_daemon/monitoring/run_monitoring.py @@ -1,7 +1,8 @@ import logging +import os import sys import time -from collections.abc import Iterator +from collections.abc import Iterator, Sequence from typing import Optional from dagster import ( @@ -24,6 +25,10 @@ from dagster._utils.error import SerializableErrorInfo, serializable_error_info_from_exc_info RESUME_RUN_LOG_MESSAGE = "Launching a new run worker to resume run" +RUN_MONITORING_RUN_FETCH_CHUNK_SIZE = max( + 1, + int(os.getenv("DAGSTER_RUN_MONITORING_RUN_FETCH_CHUNK_SIZE", "1000")), +) def monitor_starting_run( @@ -179,49 +184,71 @@ def execute_run_monitoring_iteration( ) -> Iterator[Optional[SerializableErrorInfo]]: instance = workspace_process_context.instance - # TODO: consider limiting number of runs to fetch - run_records = list( - instance.get_run_records( - filters=RunsFilter( - statuses=IN_PROGRESS_RUN_STATUSES - + [DagsterRunStatus.CANCELING, DagsterRunStatus.NOT_STARTED] + def _get_run_record_batches() -> Iterator[Sequence[RunRecord]]: + cursor = None + while True: + run_records = instance.get_run_records( + filters=RunsFilter( + statuses=IN_PROGRESS_RUN_STATUSES + + [DagsterRunStatus.CANCELING, DagsterRunStatus.NOT_STARTED] + ), + limit=RUN_MONITORING_RUN_FETCH_CHUNK_SIZE, + cursor=cursor, ) + if not run_records: + break + yield run_records + if len(run_records) < RUN_MONITORING_RUN_FETCH_CHUNK_SIZE: + break + cursor = run_records[-1].dagster_run.run_id + + run_record_batches = _get_run_record_batches() + total_runs = 0 + batch_count = 0 + workspace = workspace_process_context.create_request_context() + for run_records in run_record_batches: + if not run_records: + continue + batch_count += 1 + total_runs += len(run_records) + logger.info( + f"Collected {len(run_records)} runs for monitoring (batch {batch_count})" ) - ) + for run_record in run_records: + try: + logger.info(f"Checking run {run_record.dagster_run.run_id}") + + if ( + instance.run_monitoring_start_timeout_seconds > 0 + and run_record.dagster_run.status + in {DagsterRunStatus.STARTING, DagsterRunStatus.NOT_STARTED} + ): + monitor_starting_run(instance, run_record, logger) + elif run_record.dagster_run.status == DagsterRunStatus.STARTED: + monitor_started_run(instance, workspace, run_record, logger) + elif ( + instance.run_monitoring_cancel_timeout_seconds > 0 + and run_record.dagster_run.status == DagsterRunStatus.CANCELING + ): + monitor_canceling_run(instance, run_record, logger) + pass + else: + check.invariant( + False, f"Unexpected run status: {run_record.dagster_run.status}" + ) + except Exception: + yield DaemonErrorCapture.process_exception( + exc_info=sys.exc_info(), + logger=logger, + log_message=f"Hit error while monitoring run {run_record.dagster_run.run_id}", + ) + else: + yield - if not run_records: + if batch_count == 0: return - logger.info(f"Collected {len(run_records)} runs for monitoring") - workspace = workspace_process_context.create_request_context() - for run_record in run_records: - try: - logger.info(f"Checking run {run_record.dagster_run.run_id}") - - if ( - instance.run_monitoring_start_timeout_seconds > 0 - and run_record.dagster_run.status - in {DagsterRunStatus.STARTING, DagsterRunStatus.NOT_STARTED} - ): - monitor_starting_run(instance, run_record, logger) - elif run_record.dagster_run.status == DagsterRunStatus.STARTED: - monitor_started_run(instance, workspace, run_record, logger) - elif ( - instance.run_monitoring_cancel_timeout_seconds > 0 - and run_record.dagster_run.status == DagsterRunStatus.CANCELING - ): - monitor_canceling_run(instance, run_record, logger) - pass - else: - check.invariant(False, f"Unexpected run status: {run_record.dagster_run.status}") - except Exception: - yield DaemonErrorCapture.process_exception( - exc_info=sys.exc_info(), - logger=logger, - log_message=f"Hit error while monitoring run {run_record.dagster_run.run_id}", - ) - else: - yield + logger.info(f"Collected {total_runs} runs for monitoring across {batch_count} batches") def check_run_timeout( diff --git a/python_modules/dagster/dagster/_daemon/sensor.py b/python_modules/dagster/dagster/_daemon/sensor.py index 6293bea43d149..ed43c1b655ebb 100644 --- a/python_modules/dagster/dagster/_daemon/sensor.py +++ b/python_modules/dagster/dagster/_daemon/sensor.py @@ -922,13 +922,21 @@ def _handle_dynamic_partitions_requests( context: SensorLaunchContext, ) -> None: for request in dynamic_partitions_requests: - existent_partitions = [] - nonexistent_partitions = [] - for partition_key in request.partition_keys: - if instance.has_dynamic_partition(request.partitions_def_name, partition_key): - existent_partitions.append(partition_key) - else: - nonexistent_partitions.append(partition_key) + existing_partitions = set( + instance.get_dynamic_partitions_by_keys( + request.partitions_def_name, request.partition_keys + ) + ) + existent_partitions = [ + partition_key + for partition_key in request.partition_keys + if partition_key in existing_partitions + ] + nonexistent_partitions = [ + partition_key + for partition_key in request.partition_keys + if partition_key not in existing_partitions + ] if isinstance(request, AddDynamicPartitionsRequest): if nonexistent_partitions: @@ -958,9 +966,9 @@ def _handle_dynamic_partitions_requests( ) elif isinstance(request, DeleteDynamicPartitionsRequest): if existent_partitions: - # TODO add a bulk delete method to the instance - for partition in existent_partitions: - instance.delete_dynamic_partition(request.partitions_def_name, partition) + instance.delete_dynamic_partitions( + request.partitions_def_name, existent_partitions + ) context.logger.info( "Deleted partition keys from dynamic partitions definition" diff --git a/python_modules/dagster/dagster/_utils/caching_instance_queryer.py b/python_modules/dagster/dagster/_utils/caching_instance_queryer.py index c07f9e3c1a12f..f56d40aed8407 100644 --- a/python_modules/dagster/dagster/_utils/caching_instance_queryer.py +++ b/python_modules/dagster/dagster/_utils/caching_instance_queryer.py @@ -386,6 +386,25 @@ def asset_partition_has_materialization_or_observation( after_cursor or 0 ) + @cached_method + def get_materialized_or_observed_partition_keys( + self, *, asset_key: AssetKey + ) -> AbstractSet[str]: + if not self.asset_graph.get(asset_key).is_partitioned: + return set() + + partition_keys: set[str] = set() + for event_type in ( + DagsterEventType.ASSET_MATERIALIZATION, + DagsterEventType.ASSET_OBSERVATION, + ): + partition_keys.update( + self.instance.get_latest_storage_id_by_partition( + asset_key, event_type=event_type + ).keys() + ) + return partition_keys + def get_latest_materialization_or_observation_record( self, asset_partition: AssetKeyPartitionKey, diff --git a/python_modules/libraries/dagster-gcp/dagster_gcp/bigquery/configs.py b/python_modules/libraries/dagster-gcp/dagster_gcp/bigquery/configs.py index 5ad2473559a9a..3f64a95c3552d 100644 --- a/python_modules/libraries/dagster-gcp/dagster_gcp/bigquery/configs.py +++ b/python_modules/libraries/dagster-gcp/dagster_gcp/bigquery/configs.py @@ -4,7 +4,7 @@ https://googleapis.github.io/google-cloud-python/latest/bigquery/reference.html """ -from dagster import Array, Bool, Field, IntSource, String, StringSource +from dagster import Array, Bool, Field, IntSource, Map, Permissive, Selector, String, StringSource from dagster_gcp.bigquery.types import ( BQCreateDisposition, @@ -186,25 +186,38 @@ def define_bigquery_query_config(): is_required=False, ) - # TODO: - # Type: Shape[str, google.cloud.bigquery.external_config.ExternalConfig] - # table_definitions = Field( - # PermissiveShape(), - # description='''Definitions for external tables or None if not set. - # See https://g.co/cloud/bigquery/docs/reference/rest/v2/jobs#configuration.query.tableDefinitions - # ''', - # is_required=False, - # ) - - # TODO: Need to add this - # Type: [google.cloud.bigquery.query.UDFResource] - # udf_resources = Field( - # String, - # description='''user defined function resources (empty by default) - # See: https://g.co/cloud/bigquery/docs/reference/rest/v2/jobs#configuration.query.userDefinedFunctionResources - # ''', - # is_required=False - # ) + table_definitions = Field( + Map( + str, + Permissive( + fields={ + "source_uris": Field([StringSource], is_required=False), + "source_format": Field(BQSourceFormat, is_required=False), + "autodetect": Field(Bool, is_required=False), + "schema": Field(Array(inner_type=dict), is_required=False), + } + ), + ), + description="""Definitions for external tables or None if not set. Values are passed to + `ExternalConfig.from_api_repr` when available. + See https://g.co/cloud/bigquery/docs/reference/rest/v2/jobs#configuration.query.tableDefinitions + """, + is_required=False, + ) + + udf_resource = Selector( + { + "inline_code": StringSource, + "resource_uri": StringSource, + } + ) + udf_resources = Field( + [udf_resource], + description="""User defined function resources (empty by default). + See https://g.co/cloud/bigquery/docs/reference/rest/v2/jobs#configuration.query.userDefinedFunctionResources + """, + is_required=False, + ) use_legacy_sql = Field( Bool, @@ -237,10 +250,10 @@ def define_bigquery_query_config(): "maximum_bytes_billed": maximum_bytes_billed, "priority": priority, "query_parameters": query_parameters, - # TODO: table_definitions + "table_definitions": table_definitions, "schema_update_options": sf["schema_update_options"], "time_partitioning": sf["time_partitioning"], - # TODO: udf_resources + "udf_resources": udf_resources, "use_legacy_sql": use_legacy_sql, "use_query_cache": use_query_cache, "write_disposition": sf["write_disposition"], diff --git a/python_modules/libraries/dagster-gcp/dagster_gcp/bigquery/ops.py b/python_modules/libraries/dagster-gcp/dagster_gcp/bigquery/ops.py index 4e13bd8fb1138..ed2d0e65a9242 100644 --- a/python_modules/libraries/dagster-gcp/dagster_gcp/bigquery/ops.py +++ b/python_modules/libraries/dagster-gcp/dagster_gcp/bigquery/ops.py @@ -28,6 +28,8 @@ def _preprocess_config(cfg): destination_encryption_configuration = cfg.get("destination_encryption_configuration") time_partitioning = cfg.get("time_partitioning") + table_definitions = cfg.get("table_definitions") + udf_resources = cfg.get("udf_resources") if destination_encryption_configuration is not None: cfg["destination_encryption_configuration"] = EncryptionConfiguration( @@ -37,6 +39,25 @@ def _preprocess_config(cfg): if time_partitioning is not None: cfg["time_partitioning"] = TimePartitioning(**time_partitioning) + if table_definitions is not None: + try: + from google.cloud.bigquery.external_config import ExternalConfig + except ImportError: + pass + else: + cfg["table_definitions"] = { + name: ExternalConfig.from_api_repr(defn) if isinstance(defn, dict) else defn + for name, defn in table_definitions.items() + } + + if udf_resources is not None: + from google.cloud.bigquery.query import UDFResource + + cfg["udf_resources"] = [ + UDFResource(**resource) if isinstance(resource, dict) else resource + for resource in udf_resources + ] + return cfg diff --git a/python_modules/libraries/dagster-spark/dagster_spark/configs_spark.py b/python_modules/libraries/dagster-spark/dagster_spark/configs_spark.py index 96e31bf71aaca..a1d24ca6e0ab3 100644 --- a/python_modules/libraries/dagster-spark/dagster_spark/configs_spark.py +++ b/python_modules/libraries/dagster-spark/dagster_spark/configs_spark.py @@ -8,6 +8,7 @@ """ from dagster import Bool, Field, Float, IntSource, Permissive, StringSource +from dagster_spark.types import SparkMemory, SparkTime def spark_config(): @@ -37,17 +38,17 @@ def spark_config(): is_required=False, ), "maxResultSize": Field( - StringSource, + SparkMemory, description="""Application Properties: Limit of total size of serialized results of all partitions for each Spark action (e.g. collect) in bytes. Should be at least 1M, or 0 for unlimited. Jobs will be aborted if the total size is above this limit. Having a high limit may cause out-of-memory errors in driver (depends on spark.driver.memory and memory overhead of objects in JVM). Setting a proper limit can protect the driver from out-of-memory errors.""", is_required=False, ), "memory": Field( - StringSource, + SparkMemory, description="""Application Properties: Amount of memory to use for the driver process, i.e. where SparkContext is initialized, in the same format as JVM memory strings with a size unit suffix ("k", "m", "g" or "t") (e.g. 512m, 2g). Note: In client mode, this config must not be set through the SparkConf directly in your application, because the driver JVM has already started at that point. Instead, please set this through the --driver-memory command line option or in your default properties file.""", is_required=False, ), "memoryOverhead": Field( - StringSource, + SparkMemory, description="""Application Properties: Amount of non-heap memory to be allocated per driver process in cluster mode, in MiB unless otherwise specified. This is memory that accounts for things like VM overheads, interned strings, other native overheads, etc. This tends to grow with the container size (typically 6-10%). This option is currently supported on YARN, Mesos and Kubernetes. Note: Non-heap memory includes off-heap memory (when spark.memory.offHeap.enabled=true) and memory used by other driver processes (e.g. python process that goes with a PySpark driver) and memory used by other non-driver processes running in the same container. The maximum memory size of container to running driver is determined by the sum of spark.driver.memoryOverhead and spark.driver.memory.""", is_required=False, ), @@ -189,7 +190,7 @@ def spark_config(): Permissive( fields={ "memory": Field( - StringSource, + SparkMemory, description="""Application Properties: Amount of memory to use per executor process, in the same format as JVM memory strings with a size unit suffix ("k", "m", "g" or "t") (e.g. 512m, 2g).""", is_required=False, ), @@ -197,7 +198,7 @@ def spark_config(): Permissive( fields={ "memory": Field( - StringSource, + SparkMemory, description="""Application Properties: The amount of memory to be allocated to PySpark in each executor, in MiB unless otherwise specified. If set, PySpark memory for an executor will be limited to this amount. If not set, Spark will not limit Python's memory use and it is up to the application to avoid exceeding the overhead memory space shared with other non-JVM processes. When PySpark is run in YARN or Kubernetes, this memory is added to executor resource requests. Note: This feature is dependent on Python's `resource` module; therefore, the behaviors and limitations are inherited. For instance, Windows does not support resource limiting and actual resource is not limited on MacOS.""", is_required=False, ), @@ -205,7 +206,7 @@ def spark_config(): ) ), "memoryOverhead": Field( - StringSource, + SparkMemory, description="""Application Properties: Amount of additional memory to be allocated per executor process, in MiB unless otherwise specified. This is memory that accounts for things like VM overheads, interned strings, other native overheads, etc. This tends to grow with the executor size (typically 6-10%). This option is currently supported on YARN and Kubernetes. Note: Additional memory includes PySpark executor memory (when spark.executor.pyspark.memory is not configured) and memory used by other non-executor processes running in the same container. The maximum memory size of container to running executor is determined by the sum of spark.executor.memoryOverhead, spark.executor.memory, spark.memory.offHeap.size and spark.executor.pyspark.memory.""", is_required=False, ), @@ -346,7 +347,7 @@ def spark_config(): is_required=False, ), "heartbeatInterval": Field( - StringSource, + SparkTime, description="""Execution Behavior: Interval between each executor's heartbeats to the driver. Heartbeats let the driver know that the executor is still alive and update it with metrics for in-progress tasks. spark.executor.heartbeatInterval should be significantly less than spark.network.timeout""", is_required=False, ), @@ -494,7 +495,7 @@ def spark_config(): Permissive( fields={ "memory": Field( - StringSource, + SparkMemory, description="""Runtime Environment: Amount of memory to use per python worker process during aggregation, in the same format as JVM memory strings with a size unit suffix ("k", "m", "g" or "t") (e.g. 512m, 2g). If the memory used during aggregation goes above this amount, it will spill the data into disks.""", is_required=False, ), @@ -534,7 +535,7 @@ def spark_config(): ) ), "fetchTimeout": Field( - StringSource, + SparkTime, description="""Execution Behavior: Communication timeout to use when fetching files added through SparkContext.addFile() from the driver.""", is_required=False, ), @@ -638,7 +639,7 @@ def spark_config(): Permissive( fields={ "maxSizeInFlight": Field( - StringSource, + SparkMemory, description="""Shuffle Behavior: Maximum size of map outputs to fetch simultaneously from each reduce task, in MiB unless otherwise specified. Since each output requires us to create a buffer to receive it, this represents a fixed memory overhead per reduce task, so keep it small unless you have a large amount of memory.""", is_required=False, ), @@ -667,7 +668,7 @@ def spark_config(): Permissive( fields={ "buffer": Field( - StringSource, + SparkMemory, description="""Shuffle Behavior: Size of the in-memory buffer for each shuffle file output stream, in KiB unless otherwise specified. These buffers reduce the number of disk seeks and system calls made in creating intermediate shuffle files.""", is_required=False, ), @@ -732,7 +733,7 @@ def spark_config(): is_required=False, ), "retryWait": Field( - StringSource, + SparkTime, description="""Shuffle Behavior: (Netty only) How long to wait between retries of fetches. The maximum delay caused by retrying is 15 seconds by default, calculated as maxRetries * retryWait.""", is_required=False, ), @@ -779,7 +780,7 @@ def spark_config(): Permissive( fields={ "size": Field( - StringSource, + SparkMemory, description="""Shuffle Behavior: Cache entries limited to the specified memory footprint, in bytes unless otherwise specified.""", is_required=False, ), @@ -1423,7 +1424,7 @@ def spark_config(): is_required=False, ), "blockSize": Field( - StringSource, + SparkMemory, description="""Execution Behavior: Size of each piece of a block for TorrentBroadcastFactory, in KiB unless otherwise specified. Too large a value decreases parallelism during broadcast (makes it slower); however, if it is too small, BlockManager might take a performance hit.""", is_required=False, ), @@ -1660,7 +1661,7 @@ def spark_config(): ) ), "memoryMapThreshold": Field( - StringSource, + SparkMemory, description="""Execution Behavior: Size of a block above which Spark memory maps when reading a block from disk. Default unit is bytes, unless specified otherwise. This prevents Spark from memory mapping very small blocks. In general, memory mapping has high overhead for blocks close to or below the page size of the operating system.""", is_required=False, ), @@ -1733,7 +1734,7 @@ def spark_config(): Permissive( fields={ "interval": Field( - StringSource, + SparkTime, description="""Memory Management: Controls how often to trigger a garbage collection. This context cleaner triggers cleanups only when weak references are garbage collected. In long-running applications with large driver JVMs, where there is little memory pressure on the driver, this may happen very occasionally or not at all. Not cleaning at all may lead to executors running out of disk space after a while.""", is_required=False, ), diff --git a/python_modules/libraries/dagster-spark/dagster_spark/types.py b/python_modules/libraries/dagster-spark/dagster_spark/types.py index 921477ea8df38..b429975ff657e 100644 --- a/python_modules/libraries/dagster-spark/dagster_spark/types.py +++ b/python_modules/libraries/dagster-spark/dagster_spark/types.py @@ -1,4 +1,8 @@ +import re + from dagster import Enum, EnumValue +from dagster._config.errors import PostProcessingError +from dagster._config.source import StringSourceType SparkDeployModeCluster = EnumValue("cluster") SparkDeployModeClient = EnumValue("client") @@ -7,5 +11,37 @@ ) +_SPARK_MEMORY_PATTERN = re.compile( + r"^\d+(?:\.\d+)?(?:b|k|kb|m|mb|g|gb|t|tb|p|pb)?$", re.IGNORECASE +) +_SPARK_TIME_PATTERN = re.compile(r"^\d+(?:\.\d+)?(?:us|ms|s|m|min|h|d)?$", re.IGNORECASE) + + +class SparkMemorySourceType(StringSourceType): + def post_process(self, value): + value = super().post_process(value) + if not _SPARK_MEMORY_PATTERN.match(value): + raise PostProcessingError( + f"Invalid Spark memory value '{value}'. Expected a number with an optional unit " + "(b, k, m, g, t, p)." + ) + return value + + +class SparkTimeSourceType(StringSourceType): + def post_process(self, value): + value = super().post_process(value) + if not _SPARK_TIME_PATTERN.match(value): + raise PostProcessingError( + f"Invalid Spark time value '{value}'. Expected a number with an optional unit " + "(us, ms, s, m, min, h, d)." + ) + return value + + +SparkMemory: SparkMemorySourceType = SparkMemorySourceType() +SparkTime: SparkTimeSourceType = SparkTimeSourceType() + + class SparkOpError(Exception): pass From f38d2bf9b5de9ad42180fc32cd7dd7fbe715a056 Mon Sep 17 00:00:00 2001 From: Ramachandra Nalam Date: Mon, 19 Jan 2026 22:54:21 -0800 Subject: [PATCH 3/4] Restore TODO placeholders --- .../rbac/user-roles-permissions.md | 3 +++ docs/docs/deployment/dagster-plus/hybrid/kubernetes/setup.md | 1 + .../full-pipelines/etl-pipeline/6-create-a-sensor.md | 2 ++ .../guides/automate/sensors/testing-run-status-sensors.md | 1 + .../docs/guides/build/assets/asset-versioning-and-caching.md | 1 + docs/docs/guides/build/assets/passing-data-between-assets.md | 1 + .../troubleshooting-components.md | 1 + .../build/external-pipelines/aws/aws-lambda-pipeline.md | 1 + .../guides/build/external-pipelines/databricks-pipeline.md | 1 + .../guides/build/external-pipelines/kubernetes-pipeline.md | 1 + docs/docs/guides/build/external-resources/index.md | 1 + .../using-bare-python-objects-as-resources.md | 1 + .../defining-dependencies-between-partitioned-assets.md | 5 +++++ .../build/partitions-and-backfills/partitioning-assets.md | 1 + .../guides/build/projects/dagster-plus-project-history.md | 2 ++ .../guides/build/projects/workspaces/creating-workspaces.md | 1 + .../build/projects/workspaces/migrating-workspace-yaml.md | 3 +++ docs/docs/guides/test/running-a-subset-of-asset-checks.md | 1 + docs/docs/guides/test/testing-partitioned-config-and-jobs.md | 1 + docs/docs/integrations/libraries/duckdb/reference.md | 4 ++++ .../libraries/duckdb/using-duckdb-with-dagster.md | 1 + docs/docs/integrations/libraries/gcp/bigquery/reference.md | 1 + .../libraries/gcp/bigquery/using-bigquery-with-dagster.md | 1 + docs/docs/integrations/libraries/snowflake/reference.md | 5 +++++ docs/scripts/build-kinds-tags.sh | 1 + .../sections/integrations/libraries/omni/dagster-omni.rst | 2 +- docs/src/theme/DocCard/index.tsx | 2 ++ docs/src/theme/DocCard/styles.module.css | 2 ++ .../partitions_schedules_sensors_tests/test_sensors.py | 2 ++ .../snippet_checks/guides/components/test_components_docs.py | 4 ++++ .../src/assets/asset-checks/VirtualizedAssetCheckTable.tsx | 1 + python_modules/libraries/dagster-cloud-cli/setup.py | 1 + python_modules/libraries/dagster-dg-cli/setup.py | 1 + python_modules/libraries/dagster-dg-cli/tox.ini | 1 + python_modules/libraries/dagster-dg-core/setup.py | 1 + python_modules/libraries/dagster-omni/tox.ini | 1 + .../libraries/dagstermill/dagstermill/examples/repository.py | 1 + scripts/templates_create_dagster_package/api-docs.rst.tmpl | 3 +++ scripts/templates_create_dagster_package/setup.py.tmpl | 3 +++ scripts/templates_create_dagster_package/tox.ini.tmpl | 2 ++ 40 files changed, 68 insertions(+), 1 deletion(-) diff --git a/docs/docs/deployment/dagster-plus/authentication-and-access-control/rbac/user-roles-permissions.md b/docs/docs/deployment/dagster-plus/authentication-and-access-control/rbac/user-roles-permissions.md index d2bb4ecedb318..6e362a3c57531 100644 --- a/docs/docs/deployment/dagster-plus/authentication-and-access-control/rbac/user-roles-permissions.md +++ b/docs/docs/deployment/dagster-plus/authentication-and-access-control/rbac/user-roles-permissions.md @@ -99,6 +99,7 @@ To override a code location role for an individual user: 2. Click **Edit**. 3. Click the toggle to the left of the deployment to open a list of code locations. 4. Next to a code location, click **Edit user role**. +5. Select the user role for the code location: {/* TODO: add picture previously at "/images/dagster-cloud/user-token-management/code-location-override.png" */} 5. Select the user role for the code location. 6. Click **Save**. @@ -118,6 +119,8 @@ The above also applies to code locations and Branch Deployment roles. #### Viewing overrides +{/* TODO: add picture previously at "/images/dagster-cloud/user-token-management/user-overrides-popup.png" */} +{/* TODO: add picture previously at "/images/dagster-cloud/user-token-management/code-location-override-popup.png" */} To view deployment-level overrides for a specific user, locate the user on the **Users** page and hover over a deployment. A tooltip will display any overrides that apply at the deployment level. If there are code location-level overrides, a small **N override(s)** link will display beneath the user's deployment role. Hover over it to display the list of overrides. diff --git a/docs/docs/deployment/dagster-plus/hybrid/kubernetes/setup.md b/docs/docs/deployment/dagster-plus/hybrid/kubernetes/setup.md index 005db4e628c49..b3b41331df149 100644 --- a/docs/docs/deployment/dagster-plus/hybrid/kubernetes/setup.md +++ b/docs/docs/deployment/dagster-plus/hybrid/kubernetes/setup.md @@ -69,6 +69,7 @@ helm --namespace dagster-cloud upgrade agent \ ## Troubleshooting tips +[comment]: <> (TODO: Screenshot of Dagster+ Deployments agents tab) You can see basic health information about your agent in the Dagster+ UI. ### View logs diff --git a/docs/docs/examples/full-pipelines/etl-pipeline/6-create-a-sensor.md b/docs/docs/examples/full-pipelines/etl-pipeline/6-create-a-sensor.md index 35a0a6631e571..2995eeef282e3 100644 --- a/docs/docs/examples/full-pipelines/etl-pipeline/6-create-a-sensor.md +++ b/docs/docs/examples/full-pipelines/etl-pipeline/6-create-a-sensor.md @@ -61,6 +61,7 @@ Now copy the following sensor code in the `sensors.py` file: 2. Navigate to the Automation page. 3. Turn on the `adhoc_request_sensor`. 4. Click on the `adhoc_request_sensor` details. +{/* TODO: Screenshot */} You should see the sensor details view with recent ticks and the run history panel. 5. Create a `data/requests` directory in `dagster_tutorial`. Then include a `request.json` file: @@ -75,6 +76,7 @@ Now copy the following sensor code in the `sensors.py` file: ``` 6. Click on the green tick to see the run for this request. +{/* TODO: Screenshot */} The run should appear in the runs list, and the tick details will show the run ID. ## Summary diff --git a/docs/docs/guides/automate/sensors/testing-run-status-sensors.md b/docs/docs/guides/automate/sensors/testing-run-status-sensors.md index f5ded24f21f86..86f57f867e69b 100644 --- a/docs/docs/guides/automate/sensors/testing-run-status-sensors.md +++ b/docs/docs/guides/automate/sensors/testing-run-status-sensors.md @@ -33,6 +33,7 @@ Then we can execute this job and pull the attributes we need to build the `conte title="src//defs/sensors.py" /> +{/* TODO the methods and statuses below do not exist in API docs */} To inspect the resulting status in tests, use the returned by `execute_in_process`. Its `dagster_run` and `all_events` properties let you filter for the status or event type you care about. We can use the same pattern to build the context for . If we wanted to test this run failure sensor: diff --git a/docs/docs/guides/build/assets/asset-versioning-and-caching.md b/docs/docs/guides/build/assets/asset-versioning-and-caching.md index 6cca3787d7ae2..691e705cc73ac 100644 --- a/docs/docs/guides/build/assets/asset-versioning-and-caching.md +++ b/docs/docs/guides/build/assets/asset-versioning-and-caching.md @@ -18,6 +18,7 @@ Dagster's versioning system helps you determine ahead of time whether materializ Dagster has two versioning concepts to represent the code and input data used for each materialization: - **Code version.** A string that represents the version of the code that computes an asset. This is the `code_version` argument of . +{/* TODO link `DataVersion` to API docs */} - **Data version.** A string that represents the version of the data represented by the asset. This is represented as a object. By keeping track of code and data versions, Dagster can predict whether a materialization will change the underlying value. This allows Dagster to skip redundant materializations and instead return the previously computed value. In more technical terms, Dagster offers a limited form of [memoization](https://en.wikipedia.org/wiki/Memoization) for assets: the last-computed asset value is always cached. diff --git a/docs/docs/guides/build/assets/passing-data-between-assets.md b/docs/docs/guides/build/assets/passing-data-between-assets.md index 384648562c7cc..f1e41dfe0192f 100644 --- a/docs/docs/guides/build/assets/passing-data-between-assets.md +++ b/docs/docs/guides/build/assets/passing-data-between-assets.md @@ -117,6 +117,7 @@ The downsides of this approach are: ## Related resources +{/* TODO: add links to relevant API documentation here. */} - - - diff --git a/docs/docs/guides/build/components/building-pipelines-with-components/troubleshooting-components.md b/docs/docs/guides/build/components/building-pipelines-with-components/troubleshooting-components.md index 607b11c89f787..4d8f51a1ed2b7 100644 --- a/docs/docs/guides/build/components/building-pipelines-with-components/troubleshooting-components.md +++ b/docs/docs/guides/build/components/building-pipelines-with-components/troubleshooting-components.md @@ -5,6 +5,7 @@ sidebar_position: 500 unlisted: true --- +TODO This page lists common checks when a component fails to load or behaves unexpectedly. ## Quick checks diff --git a/docs/docs/guides/build/external-pipelines/aws/aws-lambda-pipeline.md b/docs/docs/guides/build/external-pipelines/aws/aws-lambda-pipeline.md index 018fa345e3942..ace8500dd22de 100644 --- a/docs/docs/guides/build/external-pipelines/aws/aws-lambda-pipeline.md +++ b/docs/docs/guides/build/external-pipelines/aws/aws-lambda-pipeline.md @@ -158,6 +158,7 @@ Here's what we did in this example: This argument is passed to the `run` method of , which submits the provided information to the [boto `invoke` API](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/lambda/client/invoke.html) and then invokes the specified function (`function_name`). +{/* TODO replace `PipesClientCompletedInvocation` with */} - Returned a object representing the result of execution. This is obtained by calling `get_materialize_result` on the object returned by `run` after the execution in AWS Lambda has completed. ### Step 2.2: Create Dagster Definitions diff --git a/docs/docs/guides/build/external-pipelines/databricks-pipeline.md b/docs/docs/guides/build/external-pipelines/databricks-pipeline.md index 6b943e885b4c4..7b33f1a2a6c4b 100644 --- a/docs/docs/guides/build/external-pipelines/databricks-pipeline.md +++ b/docs/docs/guides/build/external-pipelines/databricks-pipeline.md @@ -95,6 +95,7 @@ Let's review what's happening in this code: - **Passes the `SubmitTask` object, `AssetExecutionContext`, and `extras` dictionary to the `run` method of **. This method synchronously executes the Databricks job specified by the `SubmitTask` object. It slightly modifies the object by injecting some environment variables under `new_cluster.spark_env_vars` before submitting the object to the Databricks API. +{/* TODO replace `PipesClientCompletedInvocation` with */} - **Returns a object representing the result of execution**. This is obtained by calling `get_materialize_result` on the object returned by `run` after the Databricks job has finished. **Note**: Execution can take several minutes even for trivial scripts due to Databricks cluster provisioning times. ### Step 1.2: Define the Databricks Pipes client and Definitions diff --git a/docs/docs/guides/build/external-pipelines/kubernetes-pipeline.md b/docs/docs/guides/build/external-pipelines/kubernetes-pipeline.md index 9a211e957fd2a..040e0e631168b 100644 --- a/docs/docs/guides/build/external-pipelines/kubernetes-pipeline.md +++ b/docs/docs/guides/build/external-pipelines/kubernetes-pipeline.md @@ -164,6 +164,7 @@ Here's what we did in this example: These arguments are passed to the `run` method of , which submits the provided cluster information to the Kubernetes API and then runs the specified `image`. +{/* TODO replace `PipesClientCompletedInvocation` with */} - Returned a object representing the result of execution. This is obtained by calling `get_materialize_result` on the object returned by `run` after the execution in Kubernetes has completed. :::note diff --git a/docs/docs/guides/build/external-resources/index.md b/docs/docs/guides/build/external-resources/index.md index 751ce1933d81c..a47ec5f8a2c2f 100644 --- a/docs/docs/guides/build/external-resources/index.md +++ b/docs/docs/guides/build/external-resources/index.md @@ -24,6 +24,7 @@ Resources allow you to: ## Relevant APIs +{/* TODO replace `ResourceParam` with in table below */} | Name | Description | | ------------------------------------------------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | | The base class extended to define resources. Under the hood, implements . | diff --git a/docs/docs/guides/build/external-resources/using-bare-python-objects-as-resources.md b/docs/docs/guides/build/external-resources/using-bare-python-objects-as-resources.md index 15d984042e0a0..102d8ea6398a0 100644 --- a/docs/docs/guides/build/external-resources/using-bare-python-objects-as-resources.md +++ b/docs/docs/guides/build/external-resources/using-bare-python-objects-as-resources.md @@ -10,6 +10,7 @@ import ScaffoldResource from '@site/docs/partials/\_ScaffoldResource.md'; When starting to build a set of assets or jobs, you may want to use a bare Python object without configuration as a resource, such as a third-party API client. +{/* TODO replace `ResourceParam` with */} Dagster supports passing bare Python objects as resources. This follows a similar pattern to using a subclass; however, assets that use these resources must [annotate](https://docs.python.org/3/library/typing.html#typing.Annotated) them with . This annotation lets Dagster know that the parameter is a resource and not an upstream input. diff --git a/docs/docs/guides/build/partitions-and-backfills/defining-dependencies-between-partitioned-assets.md b/docs/docs/guides/build/partitions-and-backfills/defining-dependencies-between-partitioned-assets.md index c0a2763aa1130..dcadd9c6d65dd 100644 --- a/docs/docs/guides/build/partitions-and-backfills/defining-dependencies-between-partitioned-assets.md +++ b/docs/docs/guides/build/partitions-and-backfills/defining-dependencies-between-partitioned-assets.md @@ -83,3 +83,8 @@ We recommend using [automation conditions](/guides/automate/declarative-automati ::: +{/* TODO */} +{/* TODO */} +{/* TODO */} +{/* TODO */} +{/* TODO */} diff --git a/docs/docs/guides/build/partitions-and-backfills/partitioning-assets.md b/docs/docs/guides/build/partitions-and-backfills/partitioning-assets.md index cb50736e7fe73..ee8bbe4cc3f3d 100644 --- a/docs/docs/guides/build/partitions-and-backfills/partitioning-assets.md +++ b/docs/docs/guides/build/partitions-and-backfills/partitioning-assets.md @@ -35,6 +35,7 @@ Sometimes you have a set of predefined categories for your data. For instance, y +{/* TODO: Link to Backfill page to explain how to backfill regional sales data */} To backfill historical partitions for regional sales data, see [Backfilling data](/guides/build/partitions-and-backfills/backfilling-data). ## Two-dimensional partitions \{#two-dimensional-partitions} diff --git a/docs/docs/guides/build/projects/dagster-plus-project-history.md b/docs/docs/guides/build/projects/dagster-plus-project-history.md index 0bf99b44d79bf..541f9a98ac2d0 100644 --- a/docs/docs/guides/build/projects/dagster-plus-project-history.md +++ b/docs/docs/guides/build/projects/dagster-plus-project-history.md @@ -16,6 +16,7 @@ Dagster+ automatically tracks metadata every time a project (code location) is l 1. In the Dagster+ UI, navigate to the **Deployment** tab. 2. In the row associated with the project you're interested in, click **View history** in the **Updated** column. +{/* TODO - update screenshot when code location renamed in UI */} ![Screenshot highlighting the "Updated" column for a project/code location](/images/dagster-plus/deployment/code-locations/view-code-location-history.png) This will bring up a modal showing a history of every time that project has been loaded, and metadata associated with that load. If you have connected Dagster+ to a GitHub or GitLab repository, each row will have a link to the commit that was deployed at that point in time. @@ -44,6 +45,7 @@ If you notice an issue with newly deployed code, or your code fails to deploy su 2. In the row associated with the project you're interested in, click **View history** in the **Updated** column. 3. In the **Actions** column click the dropdown menu to the right of **View metadata**, select **Rollback to this version**. +{/* TODO - update screenshot when code location renamed in UI */} ![Screenshot highlighting the "Updated" column for a project/code location](/images/dagster-plus/deployment/code-locations/rollback-code-location.png) :::tip diff --git a/docs/docs/guides/build/projects/workspaces/creating-workspaces.md b/docs/docs/guides/build/projects/workspaces/creating-workspaces.md index b48aad40bc980..2b28fc15a5fd0 100644 --- a/docs/docs/guides/build/projects/workspaces/creating-workspaces.md +++ b/docs/docs/guides/build/projects/workspaces/creating-workspaces.md @@ -99,4 +99,5 @@ Finally, let's load our two projects with `dg dev`. When you run `dg dev` from t +{/* TODO - replace screenshot */} ![](/images/guides/build/projects-and-components/setting-up-a-workspace/two-projects.png) diff --git a/docs/docs/guides/build/projects/workspaces/migrating-workspace-yaml.md b/docs/docs/guides/build/projects/workspaces/migrating-workspace-yaml.md index 407680f98bad1..2bf92823e125d 100644 --- a/docs/docs/guides/build/projects/workspaces/migrating-workspace-yaml.md +++ b/docs/docs/guides/build/projects/workspaces/migrating-workspace-yaml.md @@ -1,10 +1,13 @@ --- title: Migrating from workspace.yaml to dg.toml sidebar_position: 2000 +# description: TODO description: Migrate legacy workspace.yaml configurations to a dg.toml workspace. unlisted: true --- +TODO + If you have an existing `workspace.yaml` file, you can migrate to a `dg.toml`-based workspace to align with the current Dagster OSS project and workspace structure. ## Overview diff --git a/docs/docs/guides/test/running-a-subset-of-asset-checks.md b/docs/docs/guides/test/running-a-subset-of-asset-checks.md index 861a4187018d9..e425b58be6b9d 100644 --- a/docs/docs/guides/test/running-a-subset-of-asset-checks.md +++ b/docs/docs/guides/test/running-a-subset-of-asset-checks.md @@ -18,6 +18,7 @@ This article assumes familiarity with [asset checks](/guides/test/asset-checks) Using the decorator's `specs` and `can_subset` arguments, you can execute a subset of checks in a single op. +{/* TODO link this to proper API doc */} Inside the body of the function, we can use and its `selected_asset_check_keys` property to identify which computations to run. We can also set the decorator's `can_subset` parameter to `True` to execute a subset of the asset checks that the computation contains. As we don't know in advance which checks will be executed, we explicitly `yield` each asset check result that we're expected to create: diff --git a/docs/docs/guides/test/testing-partitioned-config-and-jobs.md b/docs/docs/guides/test/testing-partitioned-config-and-jobs.md index ba53874c26dcc..7367aa584ea00 100644 --- a/docs/docs/guides/test/testing-partitioned-config-and-jobs.md +++ b/docs/docs/guides/test/testing-partitioned-config-and-jobs.md @@ -50,6 +50,7 @@ If you want to test that a : , , . This tutorial is divided into two sections to demonstrate the differences between the DuckDB resource and the DuckDB I/O manager. Each section will create the same assets, but the first section will use the DuckDB resource to store data in DuckDB, whereas the second section will use the DuckDB I/O manager. When writing your own assets, you may choose one or the other (or both) approaches depending on your storage requirements. See [When to use I/O managers](/guides/build/io-managers#when-to-use-io-managers) to learn more about when to use I/O managers and when to use resources. +{/* TODO fix link See [When to use I/O managers](/guides/build/io-managers#when-to-use-io-managers) to learn more about when to use I/O managers and when to use resources. */} In [Option 1](#option-1-using-the-duckdb-resource) you will: diff --git a/docs/docs/integrations/libraries/gcp/bigquery/reference.md b/docs/docs/integrations/libraries/gcp/bigquery/reference.md index f011add77b0e8..bd6442df4dc63 100644 --- a/docs/docs/integrations/libraries/gcp/bigquery/reference.md +++ b/docs/docs/integrations/libraries/gcp/bigquery/reference.md @@ -154,6 +154,7 @@ If you want to store assets in different datasets, you can specify the dataset a You can also specify the dataset as part of the asset's asset key: +{/* TODO add dedent=4 to CodeExample below */} You can also specify the schema as part of the asset's asset key: +{/* TODO add dedent=4 prop to CodeExample below */} `_. +.. TODO add link to state-backed component docs once they existFor more on state-backed components, see the `state-backed components guide `_. diff --git a/docs/src/theme/DocCard/index.tsx b/docs/src/theme/DocCard/index.tsx index 36871f8bb550e..883fa70634136 100644 --- a/docs/src/theme/DocCard/index.tsx +++ b/docs/src/theme/DocCard/index.tsx @@ -12,6 +12,8 @@ import type {PropSidebarItemCategory, PropSidebarItemLink} from '@docusaurus/plu import styles from './styles.module.css'; import useBaseUrl from '@docusaurus/useBaseUrl'; +// TODO - text for folders +// TODO - indicator for "community supported" integration function useCategoryItemsPlural() { const {selectMessage} = usePluralForm(); diff --git a/docs/src/theme/DocCard/styles.module.css b/docs/src/theme/DocCard/styles.module.css index cbb1bd2c45031..b452325c5aa53 100644 --- a/docs/src/theme/DocCard/styles.module.css +++ b/docs/src/theme/DocCard/styles.module.css @@ -1,3 +1,5 @@ +/* TODO: Refactor with scss to simplify/match other docs styling, merge with card styling in custom.scss */ +/* TODO: This duplicates/conflicts with .card styles set in custom.scss. Try to create a more cohesive card styling system */ /* Doc card layout overrides the global .card styles from custom.scss to match docs-specific layout. */ .cardContainer { diff --git a/examples/docs_snippets/docs_snippets_tests/concepts_tests/partitions_schedules_sensors_tests/test_sensors.py b/examples/docs_snippets/docs_snippets_tests/concepts_tests/partitions_schedules_sensors_tests/test_sensors.py index 7d367d5d97162..5f02b16f2e6e3 100644 --- a/examples/docs_snippets/docs_snippets_tests/concepts_tests/partitions_schedules_sensors_tests/test_sensors.py +++ b/examples/docs_snippets/docs_snippets_tests/concepts_tests/partitions_schedules_sensors_tests/test_sensors.py @@ -37,6 +37,7 @@ def test_log_file_job(): assert result.success + # TODO: Actually test def test_my_directory_sensor(tmp_path, monkeypatch): test_file = tmp_path / "example.txt" test_file.write_text("data", encoding="utf8") @@ -50,6 +51,7 @@ def test_my_directory_sensor(tmp_path, monkeypatch): def test_interval_sensors(): + # TODO: Actually test assert sensor_A.minimum_interval_seconds == 30 assert sensor_B.minimum_interval_seconds == 45 diff --git a/examples/docs_snippets/docs_snippets_tests/snippet_checks/guides/components/test_components_docs.py b/examples/docs_snippets/docs_snippets_tests/snippet_checks/guides/components/test_components_docs.py index 0beaf4663ecf2..8083ad236ea8f 100644 --- a/examples/docs_snippets/docs_snippets_tests/snippet_checks/guides/components/test_components_docs.py +++ b/examples/docs_snippets/docs_snippets_tests/snippet_checks/guides/components/test_components_docs.py @@ -184,6 +184,7 @@ def next_snip_no() -> int: context.run_command_and_snippet_output( cmd="dg scaffold defs 'dagster_sling.SlingReplicationCollectionComponent' ingest_files", snippet_path=f"{next_snip_no()}-dg-scaffold-sling-replication.txt", + # TODO turn output back on when we figure out how to handle multiple # NOTE: Output stays off while multiple "Using ..." messages are emitted from nested # dagster-components calls (when cache is disabled for pip). ignore_output=True, @@ -309,6 +310,7 @@ def next_snip_no() -> int: context.run_command_and_snippet_output( cmd="dg scaffold defs dagster_dbt.DbtProjectComponent jdbt --project-path dbt/jdbt", snippet_path=f"{next_snip_no()}-dg-scaffold-jdbt.txt", + # TODO turn output back on when we figure out how to handle multiple # NOTE: Output stays off while multiple "Using ..." messages are emitted from nested # dagster-components calls (when cache is disabled for pip). ignore_output=True, @@ -384,6 +386,7 @@ def next_snip_no() -> int: context.run_command_and_snippet_output( cmd="dg scaffold defs dagster_evidence.EvidenceProject jaffle_dashboard", snippet_path=f"{next_snip_no()}-scaffold-jaffle-dashboard.txt", + # TODO turn output back on when we figure out how to handle multiple # NOTE: Output stays off while multiple "Using ..." messages are emitted from nested # dagster-components calls (when cache is disabled for pip). ignore_output=True, @@ -437,6 +440,7 @@ def next_snip_no() -> int: context.run_command_and_snippet_output( cmd="dg scaffold defs dagster.schedule daily_jaffle.py", snippet_path=f"{next_snip_no()}-scaffold-daily-jaffle.txt", + # TODO turn output back on when we figure out how to handle multiple # NOTE: Output stays off while multiple "Using ..." messages are emitted from nested # dagster-components calls (when cache is disabled for pip). ignore_output=True, diff --git a/js_modules/dagster-ui/packages/ui-core/src/assets/asset-checks/VirtualizedAssetCheckTable.tsx b/js_modules/dagster-ui/packages/ui-core/src/assets/asset-checks/VirtualizedAssetCheckTable.tsx index 0a85c008441c0..84fc0014a96ec 100644 --- a/js_modules/dagster-ui/packages/ui-core/src/assets/asset-checks/VirtualizedAssetCheckTable.tsx +++ b/js_modules/dagster-ui/packages/ui-core/src/assets/asset-checks/VirtualizedAssetCheckTable.tsx @@ -71,6 +71,7 @@ interface AssetCheckRowProps { export const VirtualizedAssetCheckRow = ({assetNode, height, start, row}: AssetCheckRowProps) => { const execution = row.executionForLatestMaterialization; const timestamp = execution?.evaluation?.timestamp; + // TODO_USE_CHECK_ID const rowTestId = `row-${assetNode.assetKey.path.join('-')}-${row.name}`; return ( diff --git a/python_modules/libraries/dagster-cloud-cli/setup.py b/python_modules/libraries/dagster-cloud-cli/setup.py index 77c310cf19994..3986356ae0873 100644 --- a/python_modules/libraries/dagster-cloud-cli/setup.py +++ b/python_modules/libraries/dagster-cloud-cli/setup.py @@ -20,6 +20,7 @@ def get_version() -> str: author="Dagster Labs", author_email="hello@dagsterlabs.com", license="Apache-2.0", + description="", # TODO - fill out description description="Dagster Cloud command line interface.", url=( "https://github.com/dagster-io/dagster/tree/master/python_modules/libraries/dagster-cloud-cli" diff --git a/python_modules/libraries/dagster-dg-cli/setup.py b/python_modules/libraries/dagster-dg-cli/setup.py index 61bf3749c0740..a736c8bf152f5 100644 --- a/python_modules/libraries/dagster-dg-cli/setup.py +++ b/python_modules/libraries/dagster-dg-cli/setup.py @@ -20,6 +20,7 @@ def get_version() -> str: author="Dagster Labs", author_email="hello@dagsterlabs.com", license="Apache-2.0", + description="", # TODO - fill out description description="Dagster dg command line interface.", url=( "https://github.com/dagster-io/dagster/tree/master/python_modules/libraries/dagster-dg-cli" diff --git a/python_modules/libraries/dagster-dg-cli/tox.ini b/python_modules/libraries/dagster-dg-cli/tox.ini index 69636d73e9813..7bd4befa38153 100644 --- a/python_modules/libraries/dagster-dg-cli/tox.ini +++ b/python_modules/libraries/dagster-dg-cli/tox.ini @@ -8,6 +8,7 @@ passenv = BUILDKITE* PYTEST_ADDOPTS PYTEST_PLUGINS + # TODO add additional env var requirements and add them to .buildkite/dagster-buildkite/dagster_buildkite/steps/packages.py # NOTE: Add additional env var requirements and add them to .buildkite/dagster-buildkite/dagster_buildkite/steps/packages.py. install_command = uv pip install {opts} {packages} diff --git a/python_modules/libraries/dagster-dg-core/setup.py b/python_modules/libraries/dagster-dg-core/setup.py index 7e421cf76ee61..7ed29b3781926 100644 --- a/python_modules/libraries/dagster-dg-core/setup.py +++ b/python_modules/libraries/dagster-dg-core/setup.py @@ -20,6 +20,7 @@ def get_version() -> str: author="Dagster Labs", author_email="hello@dagsterlabs.com", license="Apache-2.0", + description="", # TODO - fill out description description="Core utilities for Dagster dg.", url=( "https://github.com/dagster-io/dagster/tree/master/python_modules/libraries/dagster-dg-core" diff --git a/python_modules/libraries/dagster-omni/tox.ini b/python_modules/libraries/dagster-omni/tox.ini index 77008c15fe3f7..8e1690e100e75 100644 --- a/python_modules/libraries/dagster-omni/tox.ini +++ b/python_modules/libraries/dagster-omni/tox.ini @@ -8,6 +8,7 @@ passenv = BUILDKITE* PYTEST_ADDOPTS PYTEST_PLUGINS + # TODO add additional env var requirements and add them to .buildkite/dagster-buildkite/dagster_buildkite/steps/packages.py # NOTE: Add additional env var requirements and add them to .buildkite/dagster-buildkite/dagster_buildkite/steps/packages.py. install_command = uv pip install {opts} {packages} deps = diff --git a/python_modules/libraries/dagstermill/dagstermill/examples/repository.py b/python_modules/libraries/dagstermill/dagstermill/examples/repository.py index 80787a42b5338..0e03087f1a8c3 100644 --- a/python_modules/libraries/dagstermill/dagstermill/examples/repository.py +++ b/python_modules/libraries/dagstermill/dagstermill/examples/repository.py @@ -83,6 +83,7 @@ def test_nb_op(name, **kwargs): common_resource_defs = { "output_notebook_io_manager": local_output_notebook_io_manager, + # TODO check if necessary "io_manager": FilesystemIOManager(), # NOTE: Used for default outputs in these examples. } diff --git a/scripts/templates_create_dagster_package/api-docs.rst.tmpl b/scripts/templates_create_dagster_package/api-docs.rst.tmpl index 6ba94c780a5f1..8f2831525b2c7 100644 --- a/scripts/templates_create_dagster_package/api-docs.rst.tmpl +++ b/scripts/templates_create_dagster_package/api-docs.rst.tmpl @@ -1,13 +1,16 @@ {{ formal_name }} ({{ hyphen_name }}) ----------------------------------------------------- +.. TODO description .. Description goes here. .. currentmodule:: {{ underscore_name }} +.. TODO fill out remaining docs .. Fill out remaining docs. +.. TODO add your package name to the following files: .. Add your package name to the following files: .. docs/content/_navigation.json .. docs/content/_apidocs.mdx diff --git a/scripts/templates_create_dagster_package/setup.py.tmpl b/scripts/templates_create_dagster_package/setup.py.tmpl index 015cf4f045eb2..1c9a3f7feb1cf 100644 --- a/scripts/templates_create_dagster_package/setup.py.tmpl +++ b/scripts/templates_create_dagster_package/setup.py.tmpl @@ -10,6 +10,7 @@ def get_version() -> str: return version["__version__"] +# TODO - add your package to scripts/install_dev_python_modules.py # NOTE: Add your package to scripts/install_dev_python_modules.py when ready. ver = get_version() @@ -21,6 +22,7 @@ setup( author="Dagster Labs", author_email="hello@dagsterlabs.com", license="Apache-2.0", + # TODO - fill out description description="", # NOTE: Fill out description. url=( "https://github.com/dagster-io/dagster/tree/master/python_modules/libraries/" @@ -39,6 +41,7 @@ setup( include_package_data=True, install_requires=[ f"dagster{pin}", + # TODO - fill in remaining dependencies # NOTE: Fill in remaining dependencies. ], zip_safe=False, diff --git a/scripts/templates_create_dagster_package/tox.ini.tmpl b/scripts/templates_create_dagster_package/tox.ini.tmpl index 50742b872feb2..5b92e8e9a642d 100644 --- a/scripts/templates_create_dagster_package/tox.ini.tmpl +++ b/scripts/templates_create_dagster_package/tox.ini.tmpl @@ -9,11 +9,13 @@ passenv = BUILDKITE* PYTEST_ADDOPTS PYTEST_PLUGINS + # TODO add additional env var requirements and add them to .buildkite/dagster-buildkite/dagster_buildkite/steps/packages.py # NOTE: Add additional env var requirements and add them to .buildkite/dagster-buildkite/dagster_buildkite/steps/packages.py. install_command = uv pip install {opts} {packages} deps = -e ../../dagster[test] -e . + # TODO fill in remaining dependencies # NOTE: Fill in remaining dependencies. allowlist_externals = /bin/bash From 26deec09305fbc7173b76f2538468dc5ccc003cb Mon Sep 17 00:00:00 2001 From: deepinsight coder Date: Fri, 19 Jun 2026 23:44:23 +0000 Subject: [PATCH 4/4] docs: fix omni state-backed link; restore dagstermill FIXME Addresses @cmpadden review feedback on #33263: - dagster-omni.rst: the restored TODO was concatenated onto the link sentence with no newline, turning the whole line into a hidden RST comment. The state-backed components guide now exists, so this replaces the TODO with the proper visible link (valid-content replacement, per the review request). - dagstermill examples/repository.py: restore the dropped "# FIXME add an output to this" marker (the op still has no output). Co-Authored-By: Claude Opus 4.8 (1M context) --- .../sections/integrations/libraries/omni/dagster-omni.rst | 2 +- .../libraries/dagstermill/dagstermill/examples/repository.py | 1 + 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/docs/sphinx/sections/integrations/libraries/omni/dagster-omni.rst b/docs/sphinx/sections/integrations/libraries/omni/dagster-omni.rst index fb8ebfeaf1691..d8c7068035d3d 100644 --- a/docs/sphinx/sections/integrations/libraries/omni/dagster-omni.rst +++ b/docs/sphinx/sections/integrations/libraries/omni/dagster-omni.rst @@ -23,4 +23,4 @@ it to, and you will need to redeploy your code location after updating your meta The simplest way to update the stored state of your ``OmniComponent`` is to use the ``dg utils refresh-component-state`` command. When deploying your code location, this command should be executed in your CI/CD workflow (e.g. github actions). -.. TODO add link to state-backed component docs once they existFor more on state-backed components, see the `state-backed components guide `_. +For more on state-backed components, see the `state-backed components guide `_. diff --git a/python_modules/libraries/dagstermill/dagstermill/examples/repository.py b/python_modules/libraries/dagstermill/dagstermill/examples/repository.py index 0e03087f1a8c3..6fe805945c21e 100644 --- a/python_modules/libraries/dagstermill/dagstermill/examples/repository.py +++ b/python_modules/libraries/dagstermill/dagstermill/examples/repository.py @@ -306,6 +306,7 @@ def error_job(): # We need type-ignores here because type checkers don't understand the `*_PRESENT` kwargs. clean_data = test_nb_op("clean_data", outs={DEFAULT_OUTPUT: Out(DataFrame)}) # pyright: ignore[reportPossiblyUnboundVariable] + # FIXME add an output to this tutorial_lr = test_nb_op("tutorial_LR", ins={"df": In(DataFrame)}) # pyright: ignore[reportPossiblyUnboundVariable] tutorial_rf = test_nb_op("tutorial_RF", ins={"df": In(DataFrame)}) # pyright: ignore[reportPossiblyUnboundVariable]