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..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 @@ -100,6 +100,7 @@ To override a code location role for an individual user: 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 +119,11 @@ 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: - {/* 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" */} +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. #### 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 91216ea891cde..b3b41331df149 100644 --- a/docs/docs/deployment/dagster-plus/hybrid/kubernetes/setup.md +++ b/docs/docs/deployment/dagster-plus/hybrid/kubernetes/setup.md @@ -50,43 +50,27 @@ You can use Helm to do rolling upgrades of your Dagster+ agent. The version of t :::tip -We recommend upgrading your Dagster+ agent every six months. The version of your agent is visible on the **Deployments** > **Agents** tab at `https://.dagster.plus/deployment/health`. - -Agent version numbering follows [Dagster release](https://github.com/dagster-io/dagster/releases) version numbering. +We recommend upgrading your Dagster+ agent every 6 months. The version of your agent is visible on the "Deployments", "Agents" tab https://your-org.dagster.plus/deployment/health. The current version of the agent matches the most [recent Dagster release](https://github.com/dagster-io/dagster/releases). ::: - - - -To upgrade to the latest version of the Dagster agent, run the following `helm upgrade` command: - -```shell -helm --namespace dagster-cloud upgrade agent \ - dagster-cloud/dagster-cloud-agent \ - --values ./values.yaml +```yaml +# values.yaml +dagsterCloudAgent: + image: + tag: latest ``` - - - -To upgrade to a specific version of the Dagster agent, run the `helm upgrade` command with the `--version` flag: - ```shell helm --namespace dagster-cloud upgrade agent \ dagster-cloud/dagster-cloud-agent \ --values ./values.yaml - --version ``` - - - ## 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 @@ -114,36 +98,12 @@ If you have an older Dagster+ deployment, you may have a `dagster_cloud.yaml` fi ### Configure your agents to serve branch deployments -[Branch deployments](/deployment/dagster-plus/deploying-code/branch-deployments) are lightweight staging environments created for each code change. While you can use your existing production agent for branch deployments on Dagster+ Hybrid, we recommend creating a dedicated branch deployment agent. This ensures that your production instance isn't negatively impacted by the workload associated with branch deployments. - - - - -If you have a dedicated branch deployment agent, you do not need to specify your full deployment in the branch deployment agent configuration: +[Branch deployments](/deployment/dagster-plus/deploying-code/branch-deployments) are lightweight staging environments created for each code change. To configure your Dagster+ agent to manage them: ```yaml # values.yaml dagsterCloud: - branchDeployments: true -``` - -```shell -helm --namespace dagster-cloud upgrade agent \ - dagster-cloud/dagster-cloud-agent \ - --values ./values.yaml -``` - -::: - - - - -If you have a single agent that serves both your [full deployment](/deployment/dagster-plus/deploying-code/full-deployments) and branch deployments, you will need to specify the full deployment in the agent configuration: - -```yaml -# values.yaml -dagsterCloud: - deployments: + deployments: # can omit full deployments to serve only branch deployments in a new agent - prod branchDeployments: true ``` @@ -154,9 +114,6 @@ helm --namespace dagster-cloud upgrade agent \ --values ./values.yaml ``` - - - ### Deploy a high availability architecture You can configure your Dagster+ agent to run with multiple replicas. Work will be load balanced across all replicas. 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..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,8 +61,8 @@ 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 +76,8 @@ 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..86f57f867e69b 100644 --- a/docs/docs/guides/automate/sensors/testing-run-status-sensors.md +++ b/docs/docs/guides/automate/sensors/testing-run-status-sensors.md @@ -34,7 +34,7 @@ Then we can execute this job and pull the attributes we need to build the `conte /> {/* 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..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,8 +18,8 @@ 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..f1e41dfe0192f 100644 --- a/docs/docs/guides/build/assets/passing-data-between-assets.md +++ b/docs/docs/guides/build/assets/passing-data-between-assets.md @@ -118,3 +118,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..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 @@ -6,3 +6,18 @@ 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..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,8 +158,8 @@ 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..7b33f1a2a6c4b 100644 --- a/docs/docs/guides/build/external-pipelines/databricks-pipeline.md +++ b/docs/docs/guides/build/external-pipelines/databricks-pipeline.md @@ -95,8 +95,8 @@ 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..040e0e631168b 100644 --- a/docs/docs/guides/build/external-pipelines/kubernetes-pipeline.md +++ b/docs/docs/guides/build/external-pipelines/kubernetes-pipeline.md @@ -164,8 +164,8 @@ 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..a47ec5f8a2c2f 100644 --- a/docs/docs/guides/build/external-resources/index.md +++ b/docs/docs/guides/build/external-resources/index.md @@ -25,11 +25,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..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,10 +10,9 @@ 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..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,25 +83,8 @@ 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..ee8bbe4cc3f3d 100644 --- a/docs/docs/guides/build/partitions-and-backfills/partitioning-assets.md +++ b/docs/docs/guides/build/partitions-and-backfills/partitioning-assets.md @@ -36,6 +36,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 +98,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/workspaces/migrating-workspace-yaml.md b/docs/docs/guides/build/projects/workspaces/migrating-workspace-yaml.md index ac2e8a1f5bef4..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,8 +1,39 @@ --- title: Migrating from workspace.yaml to dg.toml sidebar_position: 2000 -description: TODO +# description: TODO +description: Migrate legacy workspace.yaml configurations to a dg.toml workspace. unlisted: true --- -TODO \ No newline at end of file +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 + +`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..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 @@ -19,7 +19,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..7367aa584ea00 100644 --- a/docs/docs/guides/test/testing-partitioned-config-and-jobs.md +++ b/docs/docs/guides/test/testing-partitioned-config-and-jobs.md @@ -51,8 +51,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..8e26e9781db34 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,8 @@ 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. +{/* 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: @@ -25,8 +26,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 +108,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..bd6442df4dc63 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' ``` @@ -155,11 +155,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..bc7969211e14b 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,12 @@ 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..12ba1b1967cd6 100644 --- a/docs/docs/integrations/libraries/snowflake/reference.md +++ b/docs/docs/integrations/libraries/snowflake/reference.md @@ -102,6 +102,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 * @@ -130,6 +131,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 * @@ -153,6 +155,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 * @@ -173,21 +176,21 @@ 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 fc1fd7e6628af..fba7809f23902 100755 --- a/docs/scripts/build-kinds-tags.sh +++ b/docs/scripts/build-kinds-tags.sh @@ -12,14 +12,14 @@ uv_install() { } uv_activate_venv() { - export UV_PYTHON_DOWNLOADS=automatic uv python install 3.11 - uv venv --python 3.11 + uv venv source .venv/bin/activate 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..883fa70634136 100644 --- a/docs/src/theme/DocCard/index.tsx +++ b/docs/src/theme/DocCard/index.tsx @@ -12,7 +12,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 diff --git a/docs/src/theme/DocCard/styles.module.css b/docs/src/theme/DocCard/styles.module.css index be17c0c5d29d7..b452325c5aa53 100644 --- a/docs/src/theme/DocCard/styles.module.css +++ b/docs/src/theme/DocCard/styles.module.css @@ -1,5 +1,6 @@ /* 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..bbea64752581d 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 @@ -1,6 +1,9 @@ from unittest import mock from dagster import Definitions, OpExecutionContext, build_sensor_context, job, op +from docs_snippets.concepts.partitions_schedules_sensors.sensors import ( + sensors as sensors_module, +) from docs_snippets.concepts.partitions_schedules_sensors.sensors.sensor_alert import ( email_on_run_failure, my_slack_on_run_failure, @@ -35,16 +38,25 @@ 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 d01db2c18a600..a7d4f7c920b7f 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 @@ -49,9 +49,6 @@ # Mask this until we figure out how to get rid of it. _MASK_EMPTY_WARNINGS = (r"\n +warnings.warn\(message\)\n", "") -# Mask BetaWarning messages from dagster-evidence -_MASK_BETA_WARNING = (r"/[^\n]*BetaWarning:[^\n]*\n[^\n]*\n", "") - @pytest.mark.parametrize("package_manager", ["pip", "uv"]) @pytest.mark.flaky(max_runs=2) @@ -80,7 +77,6 @@ def next_snip_no() -> int: MASK_JAFFLE_PLATFORM, _MASK_USING_ENVIRONMENT_LOG_MESSAGE, _MASK_EMPTY_WARNINGS, - _MASK_BETA_WARNING, MASK_PLUGIN_CACHE_REBUILD, ], # For multi-parameter tests which share snippets, we don't want to clear the @@ -189,8 +185,8 @@ def next_snip_no() -> int: 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, ) @@ -315,8 +311,8 @@ def next_snip_no() -> int: 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( @@ -391,8 +387,8 @@ def next_snip_no() -> int: 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, ) @@ -445,8 +441,8 @@ def next_snip_no() -> int: 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-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/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..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,9 +71,11 @@ 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/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 ea83cf6e3a48e..6d848c663a979 100644 --- a/python_modules/dagster-graphql/dagster_graphql/implementation/fetch_assets.py +++ b/python_modules/dagster-graphql/dagster_graphql/implementation/fetch_assets.py @@ -1,7 +1,9 @@ +import bisect import datetime +import heapq from collections import defaultdict -from collections.abc import Sequence -from typing import TYPE_CHECKING, AbstractSet, Mapping, Optional, Union, cast # noqa: UP035 +from collections.abc import Iterator, Sequence +from typing import TYPE_CHECKING, AbstractSet, Optional, Union, cast # noqa: UP035 import dagster_shared.seven as seven from dagster import ( @@ -15,6 +17,7 @@ from dagster._core.definitions.partitions.definition import ( MultiPartitionsDefinition, PartitionsDefinition, + TimeWindowPartitionsDefinition, ) from dagster._core.definitions.partitions.subset import PartitionsSubset, TimeWindowPartitionsSubset from dagster._core.definitions.partitions.utils.time_window import ( @@ -28,11 +31,6 @@ from dagster._core.remote_representation.external import RemoteRepository from dagster._core.storage.event_log.sql_event_log import get_max_event_records_limit -from dagster_graphql.implementation.partition_status_utils import ( - build_multi_partition_ranges_generic, - extract_partition_keys_by_status, -) - if TYPE_CHECKING: from dagster_graphql.schema.asset_graph import ( GrapheneAssetNode, @@ -53,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 @@ -82,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( @@ -97,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, @@ -119,26 +156,23 @@ 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], @@ -471,9 +505,7 @@ def build_partition_statuses( ) if isinstance(materialized_partitions_subset, TimeWindowPartitionsSubset): - # Use flattened ranges to handle overlapping statuses - # (e.g., a partition can be both MATERIALIZED and FAILED) - flattened_ranges = fetch_flattened_time_window_ranges( + ranges = fetch_flattened_time_window_ranges( { PartitionRangeStatus.MATERIALIZED: materialized_partitions_subset, PartitionRangeStatus.FAILED: cast( @@ -482,24 +514,23 @@ def build_partition_statuses( PartitionRangeStatus.MATERIALIZING: cast( "TimeWindowPartitionsSubset", in_progress_partitions_subset ), - } + }, ) - - # Convert flattened ranges to GraphQL types - graphene_ranges = [ - GrapheneTimePartitionRangeStatus( - startTime=r.time_window.start.timestamp(), - endTime=r.time_window.end.timestamp(), - startKey=materialized_partitions_subset.partitions_def.get_partition_key_range_for_time_window( - r.time_window - ).start, - endKey=materialized_partitions_subset.partitions_def.get_partition_key_range_for_time_window( - r.time_window - ).end, - status=r.status, + graphene_ranges = [] + for r in ranges: + partition_key_range = cast( + "TimeWindowPartitionsDefinition", + materialized_partitions_subset.partitions_def, + ).get_partition_key_range_for_time_window(r.time_window) + graphene_ranges.append( + GrapheneTimePartitionRangeStatus( + startTime=r.time_window.start.timestamp(), + endTime=r.time_window.end.timestamp(), + startKey=partition_key_range.start, + endKey=partition_key_range.end, + status=r.status, + ) ) - for r in flattened_ranges - ] return GrapheneTimePartitionStatuses(ranges=graphene_ranges) elif isinstance(partitions_def, MultiPartitionsDefinition): return get_2d_run_length_encoded_partitions( @@ -510,35 +541,23 @@ def build_partition_statuses( partitions_def, ) elif partitions_def: - # Use shared utility to extract partition keys - keys_by_status = extract_partition_keys_by_status( - { - PartitionRangeStatus.MATERIALIZED: materialized_partitions_subset, - PartitionRangeStatus.FAILED: failed_partitions_subset, - PartitionRangeStatus.MATERIALIZING: in_progress_partitions_subset, - }, - partitions_def, - dynamic_partitions_store, - ) - - materialized_keys = keys_by_status.get(PartitionRangeStatus.MATERIALIZED, []) - failed_keys = keys_by_status.get(PartitionRangeStatus.FAILED, []) - in_progress_keys = keys_by_status.get(PartitionRangeStatus.MATERIALIZING, []) - - # Calculate unmaterialized partitions - with partition_loading_context(dynamic_partitions_store=dynamic_partitions_store): - unmaterialized_keys = materialized_partitions_subset.get_partition_keys_not_in_subset( - partitions_def=partitions_def + with partition_loading_context( + dynamic_partitions_store=dynamic_partitions_store, + ): + materialized_keys = materialized_partitions_subset.get_partition_keys() + failed_keys = failed_partitions_subset.get_partition_keys() + in_progress_keys = in_progress_partitions_subset.get_partition_keys() + + return GrapheneDefaultPartitionStatuses( + materializedPartitions=sorted( + set(materialized_keys) - set(failed_keys) - set(in_progress_keys) + ), + failedPartitions=failed_keys, + unmaterializedPartitions=materialized_partitions_subset.get_partition_keys_not_in_subset( + partitions_def=partitions_def + ), + materializingPartitions=in_progress_keys, ) - - return GrapheneDefaultPartitionStatuses( - materializedPartitions=sorted( - set(materialized_keys) - set(failed_keys) - set(in_progress_keys) - ), - failedPartitions=failed_keys, - unmaterializedPartitions=unmaterialized_keys, - materializingPartitions=in_progress_keys, - ) else: check.failed("Should not reach this point") @@ -560,48 +579,124 @@ def get_2d_run_length_encoded_partitions( "Partitions definition should be multipartitioned", ) - # Define recursive builder for secondary dimension - def build_secondary_dim(secondary_subsets: Mapping[PartitionRangeStatus, PartitionsSubset]): - # Create empty subset for use as default when status has no partitions - empty_secondary_subset = partitions_def.secondary_dimension.partitions_def.empty_subset() + with partition_loading_context( + dynamic_partitions_store=dynamic_partitions_store, + ): + primary_dim = partitions_def.primary_dimension + secondary_dim = partitions_def.secondary_dimension - # Convert status mapping to positional parameters for build_partition_statuses - # Use empty subset as default to ensure non-None values - return build_partition_statuses( - dynamic_partitions_store, - secondary_subsets.get(PartitionRangeStatus.MATERIALIZED, empty_secondary_subset), - secondary_subsets.get(PartitionRangeStatus.FAILED, empty_secondary_subset), - secondary_subsets.get(PartitionRangeStatus.MATERIALIZING, empty_secondary_subset), - partitions_def.secondary_dimension.partitions_def, - ) + dim2_materialized_partition_keys_by_dim1 = defaultdict(set) - # Use shared utility to build generic multi-partition ranges - generic_ranges = build_multi_partition_ranges_generic( - { - PartitionRangeStatus.MATERIALIZED: materialized_partitions_subset, - PartitionRangeStatus.FAILED: failed_partitions_subset, - PartitionRangeStatus.MATERIALIZING: in_progress_partitions_subset, - }, - partitions_def, - dynamic_partitions_store, - build_secondary_dim, - ) + for partition_key in materialized_partitions_subset.get_partition_keys(): + multipartition_key = partitions_def.get_partition_key_from_str(partition_key) + dim2_materialized_partition_keys_by_dim1[ + multipartition_key.keys_by_dimension[primary_dim.name] + ].add(multipartition_key.keys_by_dimension[secondary_dim.name]) - # Convert generic ranges to GraphQL types - graphene_ranges = [ - GrapheneMultiPartitionRangeStatuses( - primaryDimStartKey=r.primary_dim_start_key, - primaryDimEndKey=r.primary_dim_end_key, - primaryDimStartTime=r.primary_dim_start_time, - primaryDimEndTime=r.primary_dim_end_time, - secondaryDim=r.secondary_dim, - ) - for r in generic_ranges - ] + dim2_materialized_partition_subset_by_dim1 = { + key: secondary_dim.partitions_def.empty_subset().with_partition_keys(keys) + for key, keys in dim2_materialized_partition_keys_by_dim1.items() + } - return GrapheneMultiPartitionStatuses( - ranges=graphene_ranges, primaryDimensionName=partitions_def.primary_dimension.name - ) + dim2_failed_partition_keys_by_dim1 = defaultdict(set) + + for partition_key in failed_partitions_subset.get_partition_keys(): + multipartition_key = partitions_def.get_partition_key_from_str(partition_key) + dim2_failed_partition_keys_by_dim1[ + multipartition_key.keys_by_dimension[primary_dim.name] + ].add(multipartition_key.keys_by_dimension[secondary_dim.name]) + + dim2_failed_partition_subset_by_dim1 = { + key: secondary_dim.partitions_def.empty_subset().with_partition_keys(keys) + for key, keys in dim2_failed_partition_keys_by_dim1.items() + } + + dim2_in_progress_partition_keys_by_dim1 = defaultdict(set) + + for partition_key in in_progress_partitions_subset.get_partition_keys(): + multipartition_key = partitions_def.get_partition_key_from_str(partition_key) + dim2_in_progress_partition_keys_by_dim1[ + multipartition_key.keys_by_dimension[primary_dim.name] + ].add(multipartition_key.keys_by_dimension[secondary_dim.name]) + + dim2_in_progress_partition_subset_by_dim1 = { + key: secondary_dim.partitions_def.empty_subset().with_partition_keys(keys) + for key, keys in dim2_in_progress_partition_keys_by_dim1.items() + } + + materialized_2d_ranges = [] + + dim1_keys = primary_dim.partitions_def.get_partition_keys() + unevaluated_idx = 0 + range_start_idx = 0 # pointer to first dim1 partition with same dim2 materialization status + + if len(dim1_keys) == 0 or len(secondary_dim.partitions_def.get_partition_keys()) == 0: + return GrapheneMultiPartitionStatuses(ranges=[], primaryDimensionName=primary_dim.name) + + empty_dim2_subset = secondary_dim.partitions_def.empty_subset() + + while unevaluated_idx <= len(dim1_keys): + if ( + unevaluated_idx == len(dim1_keys) + or dim2_materialized_partition_keys_by_dim1[dim1_keys[unevaluated_idx]] + != dim2_materialized_partition_keys_by_dim1[dim1_keys[range_start_idx]] + or dim2_failed_partition_keys_by_dim1[dim1_keys[unevaluated_idx]] + != dim2_failed_partition_keys_by_dim1[dim1_keys[range_start_idx]] + or dim2_in_progress_partition_keys_by_dim1[dim1_keys[unevaluated_idx]] + != dim2_in_progress_partition_keys_by_dim1[dim1_keys[range_start_idx]] + ): + # Add new multipartition range if we've reached the end of the dim1 keys or if the + # second dimension subsets are different than for the previous dim1 key + if ( + len(dim2_materialized_partition_keys_by_dim1[dim1_keys[range_start_idx]]) > 0 + or len(dim2_failed_partition_keys_by_dim1[dim1_keys[range_start_idx]]) > 0 + or len(dim2_in_progress_partition_keys_by_dim1[dim1_keys[range_start_idx]]) > 0 + ): + # Do not add to materialized_2d_ranges if the dim2 partition subset is empty + start_key = dim1_keys[range_start_idx] + end_key = dim1_keys[unevaluated_idx - 1] + + primary_partitions_def = primary_dim.partitions_def + if isinstance(primary_partitions_def, TimeWindowPartitionsDefinition): + time_windows = cast( + "TimeWindowPartitionsDefinition", primary_partitions_def + ).time_windows_for_partition_keys( + frozenset([start_key, end_key]), + validate=False, # we already know these keys are in the partition set + ) + start_time = time_windows[0].start.timestamp() + end_time = time_windows[-1].end.timestamp() + else: + start_time = None + end_time = None + + materialized_2d_ranges.append( + GrapheneMultiPartitionRangeStatuses( + primaryDimStartKey=start_key, + primaryDimEndKey=end_key, + primaryDimStartTime=start_time, + primaryDimEndTime=end_time, + secondaryDim=build_partition_statuses( + dynamic_partitions_store, + dim2_materialized_partition_subset_by_dim1.get( + start_key, empty_dim2_subset + ), + dim2_failed_partition_subset_by_dim1.get( + start_key, empty_dim2_subset + ), + dim2_in_progress_partition_subset_by_dim1.get( + start_key, empty_dim2_subset + ), + secondary_dim.partitions_def, + ), + ) + ) + range_start_idx = unevaluated_idx + unevaluated_idx += 1 + + return GrapheneMultiPartitionStatuses( + ranges=materialized_2d_ranges, primaryDimensionName=primary_dim.name + ) def get_freshness_info( 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 1c8e0cf57088c..a0d4853287f7b 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 @@ -163,7 +163,10 @@ def get_inner_queryer_for_back_compat(self) -> "CachingInstanceQueryer": return self._queryer def _get_partitions_def(self, key: T_EntityKey) -> Optional["PartitionsDefinition"]: - return self.asset_graph.get(key).partitions_def + if isinstance(key, AssetKey): + return self.asset_graph.get(key).partitions_def + else: + return None @cached_method @use_partition_loading_context @@ -372,16 +375,6 @@ def get_asset_subset_from_asset_partitions( ) return EntitySubset(self, key=key, value=_ValidatedEntitySubsetValue(value)) - @use_partition_loading_context - def get_subset_from_partition_keys( - self, - key: T_EntityKey, - partitions_def: "PartitionsDefinition", - partition_keys: AbstractSet[str], - ) -> EntitySubset[T_EntityKey]: - value = partitions_def.subset_with_partition_keys(partition_keys) - return EntitySubset(self, key=key, value=_ValidatedEntitySubsetValue(value)) - @use_partition_loading_context def compute_parent_subset_and_required_but_nonexistent_subset( self, parent_key, subset: EntitySubset[T_EntityKey] @@ -556,22 +549,11 @@ def compute_latest_time_window_subset( check.failed(f"Unsupported partitions_def: {partitions_def}") async def compute_subset_with_status( - self, - key: AssetCheckKey, - status: Optional["AssetCheckExecutionResolvedStatus"], - from_subset: EntitySubset, - ) -> EntitySubset[AssetCheckKey]: - """Returns the subset of an asset check that matches a given status.""" + self, key: AssetCheckKey, status: Optional["AssetCheckExecutionResolvedStatus"] + ): from dagster._core.storage.event_log.base import AssetCheckSummaryRecord - # Handle partitioned asset checks - if self._get_partitions_def(key): - with partition_loading_context(new_ctx=self._partition_loading_context): - return await self._get_partitioned_check_subset_with_status( - key, status, from_subset - ) - - # Handle non-partitioned asset checks with existing logic + """Returns the subset of an asset check that matches a given status.""" summary = await AssetCheckSummaryRecord.gen(self, key) latest_record = summary.last_check_execution_record if summary else None resolved_status = ( @@ -605,64 +587,31 @@ async def compute_subset_with_freshness_state( return self.get_empty_subset(key=key) async def _compute_run_in_progress_check_subset( - self, key: AssetCheckKey, from_subset: EntitySubset + self, key: AssetCheckKey ) -> EntitySubset[AssetCheckKey]: from dagster._core.storage.asset_check_execution_record import ( AssetCheckExecutionResolvedStatus, ) return await self.compute_subset_with_status( - key, AssetCheckExecutionResolvedStatus.IN_PROGRESS, from_subset + key, AssetCheckExecutionResolvedStatus.IN_PROGRESS ) async def _compute_execution_failed_check_subset( - self, key: AssetCheckKey, from_subset: EntitySubset + self, key: AssetCheckKey ) -> EntitySubset[AssetCheckKey]: from dagster._core.storage.asset_check_execution_record import ( AssetCheckExecutionResolvedStatus, ) return await self.compute_subset_with_status( - key, AssetCheckExecutionResolvedStatus.EXECUTION_FAILED, from_subset + key, AssetCheckExecutionResolvedStatus.EXECUTION_FAILED ) async def _compute_missing_check_subset( - self, key: AssetCheckKey, from_subset: EntitySubset - ) -> EntitySubset[AssetCheckKey]: - return await self.compute_subset_with_status(key, None, from_subset) - - @use_partition_loading_context - async def _get_partitioned_check_subset_with_status( - self, - key: AssetCheckKey, - status: Optional["AssetCheckExecutionResolvedStatus"], - from_subset: EntitySubset, + self, key: AssetCheckKey ) -> EntitySubset[AssetCheckKey]: - from dagster._core.storage.asset_check_state import AssetCheckState - - check_node = self.asset_graph.get(key) - if not check_node or not check_node.partitions_def: - check.failed(f"Asset check {key} not found or not partitioned.") - - cache_value = ( - await AssetCheckState.gen(self, (key, check_node.partitions_def)) - or AssetCheckState.empty() - ) - - if status is None: - known_statuses = self.get_empty_subset(key=key) - for serializable_subset in cache_value.subsets.values(): - subset = self.get_subset_from_serializable_subset(serializable_subset) - if subset: - known_statuses = known_statuses.compute_union(subset) - return from_subset.compute_difference(known_statuses) or self.get_empty_subset(key=key) - else: - serializable_subset = cache_value.subsets.get(status) - if serializable_subset is None: - return self.get_empty_subset(key=key) - return self.get_subset_from_serializable_subset( - serializable_subset - ) or self.get_empty_subset(key=key) + return await self.compute_subset_with_status(key, None) async def _compute_run_in_progress_asset_subset(self, key: AssetKey) -> EntitySubset[AssetKey]: from dagster._core.storage.partition_status_cache import AssetStatusCacheValue @@ -776,32 +725,45 @@ 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, from_subset: EntitySubset - ) -> EntitySubset: + async def compute_run_in_progress_subset(self, *, key: EntityKey) -> EntitySubset: return await _dispatch( key=key, - check_method=functools.partial( - self._compute_run_in_progress_check_subset, from_subset=from_subset - ), + check_method=self._compute_run_in_progress_check_subset, asset_method=self._compute_run_in_progress_asset_subset, ) @cached_method - async def compute_backfill_in_progress_subset( - self, *, key: EntityKey, from_subset: EntitySubset - ) -> EntitySubset: + async def compute_backfill_in_progress_subset(self, *, key: EntityKey) -> EntitySubset: async def get_empty_subset(key: EntityKey) -> EntitySubset: return self.get_empty_subset(key=key) @@ -813,14 +775,10 @@ async def get_empty_subset(key: EntityKey) -> EntitySubset: ) @cached_method - async def compute_execution_failed_subset( - self, *, key: EntityKey, from_subset: EntitySubset - ) -> EntitySubset: + async def compute_execution_failed_subset(self, *, key: EntityKey) -> EntitySubset: return await _dispatch( key=key, - check_method=functools.partial( - self._compute_execution_failed_check_subset, from_subset=from_subset - ), + check_method=self._compute_execution_failed_check_subset, asset_method=self._compute_execution_failed_asset_subset, ) @@ -830,9 +788,7 @@ async def compute_missing_subset( ) -> EntitySubset: return await _dispatch( key=key, - check_method=functools.partial( - self._compute_missing_check_subset, from_subset=from_subset - ), + check_method=self._compute_missing_check_subset, asset_method=functools.partial( self._compute_missing_asset_subset, from_subset=from_subset ), 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 9dfa54f2f3a58..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 @@ -217,7 +217,6 @@ def __init__( description: Optional[str], automation_condition: Optional["AutomationCondition[AssetCheckKey]"], metadata: ArbitraryMetadataMapping, - partitions_def: Optional[PartitionsDefinition], ): self.key = key self.blocking = blocking @@ -225,7 +224,6 @@ def __init__( self._additional_deps = additional_deps self._description = description self._metadata = metadata - self._partitions_def = partitions_def @property def parent_entity_keys(self) -> AbstractSet[AssetKey]: @@ -237,7 +235,8 @@ def child_entity_keys(self) -> AbstractSet[EntityKey]: @property def partitions_def(self) -> Optional[PartitionsDefinition]: - return self._partitions_def + # all checks are unpartitioned + return None @property def partition_mappings(self) -> Mapping[EntityKey, PartitionMapping]: @@ -267,10 +266,6 @@ class BaseAssetGraph(ABC, Generic[T_AssetNode]): def asset_nodes(self) -> Iterable[T_AssetNode]: return self._asset_nodes_by_key.values() - @property - def asset_check_nodes(self) -> Iterable[AssetCheckNode]: - return self._asset_check_nodes_by_key.values() - @property def nodes(self) -> Iterable[BaseEntityNode]: return [ @@ -310,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 @@ -673,28 +676,6 @@ def validate_partitions(self): f"Invalid partition mapping from {node.key.to_user_string()} to {parent.key.to_user_string()}" ) from e - # Validate that asset checks have compatible partitions_def with their target asset - for node in self.asset_check_nodes: - if node.partitions_def is None: - continue - - target_asset_key = node.key.asset_key - if not self.has(target_asset_key): - raise DagsterInvalidDefinitionError( - f"Partitioned asset check '{node.key.to_user_string()}' targets " - f"asset '{target_asset_key.to_user_string()}' " - "but the asset does not exist in the graph." - ) - # If the check is partitioned, it must have the same partitions_def as the asset - if node.partitions_def != self.get(target_asset_key).partitions_def: - raise DagsterInvalidDefinitionError( - f"Asset check '{node.key.to_user_string()}' targets asset '{target_asset_key.to_user_string()}' " - "but has a different partitions definition. " - f"Asset check partitions_def: {node.partitions_def}, " - f"Asset partitions_def: {self.get(target_asset_key).partitions_def}. " - "Partitioned asset checks must have the same partitions definition as their target asset." - ) - def upstream_key_iterator(self, asset_key: AssetKey) -> Iterator[AssetKey]: """Iterates through all asset keys which are upstream of the given key.""" visited: set[AssetKey] = set() 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 ef5a6b367d251..2ebfbac106a29 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" @@ -1843,16 +1877,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 411dde48062aa..809d63a43cdae 100644 --- a/python_modules/dagster/dagster/_core/storage/event_log/base.py +++ b/python_modules/dagster/dagster/_core/storage/event_log/base.py @@ -19,7 +19,6 @@ EventLogRecord, EventRecordsFilter, EventRecordsResult, - PartitionKeyFilter, RunStatusChangeRecordsFilter, ) from dagster._core.events import DagsterEventType @@ -36,7 +35,6 @@ from dagster._core.storage.asset_check_execution_record import ( AssetCheckExecutionRecord, AssetCheckExecutionRecordStatus, - AssetCheckPartitionInfo, ) from dagster._core.storage.dagster_run import DagsterRunStatsSnapshot from dagster._core.storage.partition_status_cache import get_and_update_asset_status_cache_value @@ -50,7 +48,6 @@ if TYPE_CHECKING: from dagster._core.events.log import EventLogEntry - from dagster._core.storage.asset_check_state import AssetCheckState from dagster._core.storage.partition_status_cache import AssetStatusCacheValue @@ -536,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 @@ -636,52 +650,17 @@ def get_asset_check_execution_history( limit: int, cursor: Optional[int] = None, status: Optional[Set[AssetCheckExecutionRecordStatus]] = None, - partition_filter: Optional[PartitionKeyFilter] = None, ) -> Sequence[AssetCheckExecutionRecord]: """Get executions for one asset check, sorted by recency.""" pass @abstractmethod def get_latest_asset_check_execution_by_key( - self, - check_keys: Sequence[AssetCheckKey], - partition_filter: Optional[PartitionKeyFilter] = None, + self, check_keys: Sequence[AssetCheckKey] ) -> Mapping[AssetCheckKey, AssetCheckExecutionRecord]: """Get the latest executions for a list of asset checks.""" pass - @abstractmethod - def get_asset_check_partition_info( - self, - keys: Sequence[AssetCheckKey], - after_storage_id: Optional[int] = None, - partition_keys: Optional[Sequence[str]] = None, - ) -> Sequence[AssetCheckPartitionInfo]: - """Get asset check partition records with execution status and planned run info.""" - pass - - def get_checkpointed_asset_check_state( - self, keys: Sequence[AssetCheckKey] - ) -> Mapping[AssetCheckKey, "AssetCheckState"]: - """Get the current stored asset check state for a list of asset checks and their - associated partitions definitions. This method is not guaranteed to return a - state object that is up to date with the latest events. - """ - from dagster._core.storage.asset_check_state import AssetCheckState - - return {key: AssetCheckState.empty() for key in keys} - - def get_asset_check_state( - self, keys: Sequence[tuple[AssetCheckKey, Optional[PartitionsDefinition]]] - ) -> Mapping[AssetCheckKey, "AssetCheckState"]: - from dagster._core.storage.asset_check_state import bulk_update_asset_check_state - - return bulk_update_asset_check_state( - self._instance, - keys, - initial_states=self.get_checkpointed_asset_check_state([key for key, _ in keys]), - ) - @abstractmethod def fetch_materializations( self, @@ -778,23 +757,3 @@ def get_pool_config(self) -> PoolConfig: # Base implementation of fetching pool config. To be overriden for remote storage # implementations where the local instance might not match the remote instance. return self._instance.get_concurrency_config().pool_config - - def _get_latest_unpartitioned_materialization_storage_ids( - self, keys: Sequence[AssetKey] - ) -> Mapping[AssetKey, int]: - # Returns a mapping of asset key to the latest recorded materialization storage id for the asset, - # ignoring partitioned assets. Used purely for the `get_asset_check_partition_info` method across - # different storage implementations. - asset_records = self.get_asset_records(keys) - latest_unpartitioned_materialization_storage_ids = {} - for asset_record in asset_records: - if ( - asset_record.asset_entry.last_materialization_record is not None - and asset_record.asset_entry.last_materialization_record.event_log_entry.get_dagster_event().partition - is None - ): - latest_unpartitioned_materialization_storage_ids[ - asset_record.asset_entry.asset_key - ] = asset_record.asset_entry.last_materialization_storage_id - - return latest_unpartitioned_materialization_storage_ids 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 2da51be8d6a5b..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 @@ -38,7 +38,6 @@ ) from dagster._core.event_api import ( EventRecordsResult, - PartitionKeyFilter, RunShardedEventsCursor, RunStatusChangeRecordsFilter, ) @@ -59,7 +58,6 @@ COMPLETED_ASSET_CHECK_EXECUTION_RECORD_STATUSES, AssetCheckExecutionRecord, AssetCheckExecutionRecordStatus, - AssetCheckPartitionInfo, ) from dagster._core.storage.dagster_run import DagsterRunStatsSnapshot from dagster._core.storage.event_log.base import ( @@ -2133,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: @@ -2174,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) @@ -2994,25 +3026,15 @@ def _store_asset_check_evaluation_planned( planned = cast( "AssetCheckEvaluationPlanned", check.not_none(event.dagster_event).event_specific_data ) - partition_keys = ( - planned.partitions_subset.get_partition_keys() if planned.partitions_subset else [None] - ) with self.index_connection() as conn: conn.execute( AssetCheckExecutionsTable.insert().values( - [ - dict( - asset_key=planned.asset_key.to_string(), - check_name=planned.check_name, - partition=partition_key, - run_id=event.run_id, - execution_status=AssetCheckExecutionRecordStatus.PLANNED.value, - evaluation_event=serialize_value(event), - evaluation_event_timestamp=self._event_insert_timestamp(event), - evaluation_event_storage_id=event_id, - ) - for partition_key in partition_keys - ] + asset_key=planned.asset_key.to_string(), + check_name=planned.check_name, + run_id=event.run_id, + execution_status=AssetCheckExecutionRecordStatus.PLANNED.value, + evaluation_event=serialize_value(event), + evaluation_event_timestamp=self._event_insert_timestamp(event), ) ) @@ -3045,7 +3067,6 @@ def _store_runless_asset_check_evaluation( if evaluation.target_materialization_data else None ), - partition=evaluation.partition, ) ) @@ -3062,9 +3083,6 @@ def _update_asset_check_evaluation(self, event: EventLogEntry, event_id: Optiona AssetCheckExecutionsTable.c.asset_key == evaluation.asset_key.to_string(), AssetCheckExecutionsTable.c.check_name == evaluation.check_name, AssetCheckExecutionsTable.c.run_id == event.run_id, - self._get_asset_check_partition_filter_clause( - PartitionKeyFilter(key=evaluation.partition) - ), ) ) .values( @@ -3081,7 +3099,6 @@ def _update_asset_check_evaluation(self, event: EventLogEntry, event_id: Optiona if evaluation.target_materialization_data else None ), - partition=evaluation.partition, ) ).rowcount @@ -3106,7 +3123,6 @@ def _update_asset_check_evaluation(self, event: EventLogEntry, event_id: Optiona if evaluation.target_materialization_data else None ), - partition=evaluation.partition, ) ).rowcount @@ -3118,23 +3134,12 @@ def _update_asset_check_evaluation(self, event: EventLogEntry, event_id: Optiona "as a result of duplicate AssetCheckPlanned events." ) - def _get_asset_check_partition_filter_clause( - self, partition_filter: Optional[PartitionKeyFilter] - ): - if partition_filter is None: - return True - elif partition_filter.key is None: - return AssetCheckExecutionsTable.c.partition.is_(None) - else: - return AssetCheckExecutionsTable.c.partition == partition_filter.key - def get_asset_check_execution_history( self, check_key: AssetCheckKey, limit: int, cursor: Optional[int] = None, status: Optional[AbstractSet[AssetCheckExecutionRecordStatus]] = None, - partition_filter: Optional[PartitionKeyFilter] = None, ) -> Sequence[AssetCheckExecutionRecord]: check.inst_param(check_key, "key", AssetCheckKey) check.int_param(limit, "limit") @@ -3148,14 +3153,12 @@ def get_asset_check_execution_history( AssetCheckExecutionsTable.c.execution_status, AssetCheckExecutionsTable.c.evaluation_event, AssetCheckExecutionsTable.c.create_timestamp, - AssetCheckExecutionsTable.c.partition, ] ) .where( db.and_( AssetCheckExecutionsTable.c.asset_key == check_key.asset_key.to_string(), AssetCheckExecutionsTable.c.check_name == check_key.name, - self._get_asset_check_partition_filter_clause(partition_filter), ) ) .order_by(AssetCheckExecutionsTable.c.id.desc()) @@ -3175,13 +3178,8 @@ def get_asset_check_execution_history( return [AssetCheckExecutionRecord.from_db_row(row, key=check_key) for row in rows] def get_latest_asset_check_execution_by_key( - self, - check_keys: Sequence[AssetCheckKey], - partition_filter: Optional[PartitionKeyFilter] = None, + self, check_keys: Sequence[AssetCheckKey] ) -> Mapping[AssetCheckKey, AssetCheckExecutionRecord]: - """Returns the latest AssetCheckExecutionRecord for each check key. By default, returns the latest - record regardless of partitioning. - """ if not check_keys: return {} @@ -3197,7 +3195,6 @@ def get_latest_asset_check_execution_by_key( [key.asset_key.to_string() for key in check_keys] ), AssetCheckExecutionsTable.c.check_name.in_([key.name for key in check_keys]), - self._get_asset_check_partition_filter_clause(partition_filter), ) ) .group_by( @@ -3215,7 +3212,6 @@ def get_latest_asset_check_execution_by_key( AssetCheckExecutionsTable.c.execution_status, AssetCheckExecutionsTable.c.evaluation_event, AssetCheckExecutionsTable.c.create_timestamp, - AssetCheckExecutionsTable.c.partition, ] ).select_from( AssetCheckExecutionsTable.join( @@ -3238,135 +3234,6 @@ def get_latest_asset_check_execution_by_key( results[check_key] = AssetCheckExecutionRecord.from_db_row(row, key=check_key) return results - def _get_asset_check_partition_info_for_key( - self, - check_key: AssetCheckKey, - after_storage_id: Optional[int], - partition_keys: Optional[Sequence[str]], - latest_unpartitioned_materialization_storage_ids: Mapping[AssetKey, int], - ) -> Sequence[AssetCheckPartitionInfo]: - # Build the base filter conditions - filter_conditions = [ - AssetCheckExecutionsTable.c.asset_key == check_key.asset_key.to_string(), - AssetCheckExecutionsTable.c.check_name == check_key.name, - # Historical records may have NULL in the evaluation_event_storage_id column for - # PLANNED events - AssetCheckExecutionsTable.c.evaluation_event_storage_id.isnot(None), - ] - if partition_keys is not None: - filter_conditions.append(AssetCheckExecutionsTable.c.partition.in_(partition_keys)) - - # Subquery to find the max id for each partition - latest_check_ids_subquery = db_subquery( - db_select( - [ - db.func.max(AssetCheckExecutionsTable.c.id).label("id"), - AssetCheckExecutionsTable.c.partition.label("partition"), - ] - ) - .where(db.and_(*filter_conditions)) - .group_by(AssetCheckExecutionsTable.c.partition), - "latest_check_ids_subquery", - ) - - # Subquery to find the latest materialization storage id for each partition of the - # target asset. Note: we don't filter by after_storage_id here because we always want - # to return the latest materialization storage id, even if it's older than after_storage_id. - latest_materialization_ids_subquery = self._latest_event_ids_by_partition_subquery( - check_key.asset_key, - [DagsterEventType.ASSET_MATERIALIZATION], - asset_partitions=partition_keys, - ) - - # Main query to get all columns for the latest records, joined with latest - # materialization storage ids - query = db_select( - [ - AssetCheckExecutionsTable.c.id, - AssetCheckExecutionsTable.c.partition, - AssetCheckExecutionsTable.c.execution_status, - AssetCheckExecutionsTable.c.evaluation_event_storage_id, - AssetCheckExecutionsTable.c.materialization_event_storage_id, - AssetCheckExecutionsTable.c.run_id, - latest_materialization_ids_subquery.c.id.label("latest_materialization_storage_id"), - ] - ).select_from( - AssetCheckExecutionsTable.join( - latest_check_ids_subquery, - AssetCheckExecutionsTable.c.id == latest_check_ids_subquery.c.id, - ).join( - latest_materialization_ids_subquery, - AssetCheckExecutionsTable.c.partition - == latest_materialization_ids_subquery.c.partition, - isouter=True, - ) - ) - - # these filters are applied to the main query rather than the individual subqueries to ensure - # we don't miss records that only have a new materialization or a new check execution but not both - if after_storage_id is not None: - query = query.where( - db.or_( - AssetCheckExecutionsTable.c.evaluation_event_storage_id > after_storage_id, - latest_materialization_ids_subquery.c.id > after_storage_id, - ) - ) - - with self.index_connection() as conn: - rows = db_fetch_mappings(conn, query) - - return [ - AssetCheckPartitionInfo( - check_key=check_key, - partition_key=row["partition"], - latest_execution_status=AssetCheckExecutionRecordStatus(row["execution_status"]), - latest_target_materialization_storage_id=row["materialization_event_storage_id"], - latest_planned_run_id=row["run_id"], - latest_check_event_storage_id=row["evaluation_event_storage_id"], - latest_materialization_storage_id=max( - filter( - None, - [ - row["latest_materialization_storage_id"], - latest_unpartitioned_materialization_storage_ids.get( - check_key.asset_key - ), - ], - ), - default=None, - ), - ) - for row in rows - ] - - def get_asset_check_partition_info( - self, - keys: Sequence[AssetCheckKey], - after_storage_id: Optional[int] = None, - partition_keys: Optional[Sequence[str]] = None, - ) -> Sequence[AssetCheckPartitionInfo]: - check.list_param(keys, "keys", of_type=AssetCheckKey) - check.opt_int_param(after_storage_id, "after_storage_id") - - infos = [] - latest_unpartitioned_materialization_storage_ids = ( - self._get_latest_unpartitioned_materialization_storage_ids( - list(set(key.asset_key for key in keys)) - ) - ) - # the inner query is not feasible to join in a single query because the latest materialization ids subquery, - # so for now we fetch the info for each key separately - for key in keys: - infos.extend( - self._get_asset_check_partition_info_for_key( - key, - after_storage_id, - partition_keys, - latest_unpartitioned_materialization_storage_ids, - ) - ) - return infos - @property def supports_asset_checks(self): # pyright: ignore[reportIncompatibleMethodOverride] return self.has_table(AssetCheckExecutionsTable.name) diff --git a/python_modules/dagster/dagster/_daemon/monitoring/run_monitoring.py b/python_modules/dagster/dagster/_daemon/monitoring/run_monitoring.py index 4d067e2fff7b0..41ee75b949402 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,69 @@ 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 not run_records: + 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 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( @@ -235,11 +260,7 @@ def check_run_timeout( MAX_RUNTIME_SECONDS_TAG, run_record.dagster_run.tags.get("dagster/max_runtime_seconds") ) if max_time_str: - try: - max_time = float(max_time_str) - except ValueError: - logger.warning(f"Invalid max runtime value: {max_time_str}") - max_time = None + max_time = float(max_time_str) else: max_time = default_timeout_seconds diff --git a/python_modules/dagster/dagster/_daemon/sensor.py b/python_modules/dagster/dagster/_daemon/sensor.py index 6293bea43d149..911760c0a8392 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,7 @@ 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-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..7bd4befa38153 100644 --- a/python_modules/libraries/dagster-dg-cli/tox.ini +++ b/python_modules/libraries/dagster-dg-cli/tox.ini @@ -9,6 +9,7 @@ passenv = 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 e3328eec2f1b0..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" ), @@ -50,7 +50,7 @@ def get_version() -> str: "setuptools", # Needed to parse setup.cfg "packaging", "python-dotenv", - "typer>=0.15.1,<1.0", + "typer<0.17.0", f"dagster-shared{pin}", f"dagster-cloud-cli{pin}", ], 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-omni/tox.ini b/python_modules/libraries/dagster-omni/tox.ini index 21fcc24c7ece2..8e1690e100e75 100644 --- a/python_modules/libraries/dagster-omni/tox.ini +++ b/python_modules/libraries/dagster-omni/tox.ini @@ -9,6 +9,7 @@ passenv = 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 +22,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/dagster-spark/dagster_spark/configs_spark.py b/python_modules/libraries/dagster-spark/dagster_spark/configs_spark.py index 96e31bf71aaca..8e97d72682c18 100644 --- a/python_modules/libraries/dagster-spark/dagster_spark/configs_spark.py +++ b/python_modules/libraries/dagster-spark/dagster_spark/configs_spark.py @@ -9,6 +9,8 @@ from dagster import Bool, Field, Float, IntSource, Permissive, StringSource +from dagster_spark.types import SparkMemory, SparkTime + def spark_config(): return Field( @@ -37,17 +39,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 +191,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 +199,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 +207,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 +348,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 +496,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 +536,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 +640,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 +669,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 +734,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 +781,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 +1425,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 +1662,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 +1735,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..03946fce3f558 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,35 @@ ) +_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 diff --git a/python_modules/libraries/dagstermill/dagstermill/examples/repository.py b/python_modules/libraries/dagstermill/dagstermill/examples/repository.py index f37fed8c6dd95..0e03087f1a8c3 100644 --- a/python_modules/libraries/dagstermill/dagstermill/examples/repository.py +++ b/python_modules/libraries/dagstermill/dagstermill/examples/repository.py @@ -83,7 +83,8 @@ 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 + # 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 +306,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..8f2831525b2c7 100644 --- a/scripts/templates_create_dagster_package/api-docs.rst.tmpl +++ b/scripts/templates_create_dagster_package/api-docs.rst.tmpl @@ -2,14 +2,17 @@ ----------------------------------------------------- .. 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..1c9a3f7feb1cf 100644 --- a/scripts/templates_create_dagster_package/setup.py.tmpl +++ b/scripts/templates_create_dagster_package/setup.py.tmpl @@ -11,6 +11,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 +22,8 @@ setup( author="Dagster Labs", author_email="hello@dagsterlabs.com", license="Apache-2.0", - description="", # TODO - fill out description + # TODO - fill out description + description="", # NOTE: Fill out description. url=( "https://github.com/dagster-io/dagster/tree/master/python_modules/libraries/" "{{ hyphen_name }}" @@ -40,6 +42,7 @@ setup( 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..5b92e8e9a642d 100644 --- a/scripts/templates_create_dagster_package/tox.ini.tmpl +++ b/scripts/templates_create_dagster_package/tox.ini.tmpl @@ -10,11 +10,13 @@ passenv = 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