Add an opt-in env var to re-enable emitting partition subsets from planned events in OSS (#29288)

## Summary & Motivation
Ths fix in #28113 trades off
correctness for performance - failed status on assets is reliably
correct now, but single-run backfills may take a lot longer to create
runs since they are creating

Until we can make the requisite schema changes to give us the best of
both worlds (single event per asset per run, and also correct partition
failure calculations) make the default to be slower but correct, with
the ability oto opt in if you prefer faster but not always correct.

## How I Tested These Changes
Ran TestSqliteEventLogStorage with this new env var set to True

Author: gibsondan

python_modules/dagster/dagster/_core/storage/event_log/base.py

@@ -1,3 +1,4 @@
+import os
 from abc import ABC, abstractmethod
 from collections.abc import Iterable, Mapping, Sequence
 from typing import TYPE_CHECKING, AbstractSet, NamedTuple, Optional, Union  # noqa: UP035
@@ -40,6 +41,7 @@
 from dagster._record import record
 from dagster._utils import PrintFn
 from dagster._utils.concurrency import ConcurrencyClaimStatus, ConcurrencyKeyInfo
+from dagster._utils.tags import get_boolean_tag_value
 from dagster._utils.warnings import deprecation_warning
 
 if TYPE_CHECKING:
@@ -563,7 +565,10 @@ def supports_global_concurrency_limits(self) -> bool:
 
     @property
     def supports_partition_subset_in_asset_materialization_planned_events(self) -> bool:
-        return False
+        # Setting this environment variable will cause a single planned event to be emitted for
+        # each asset for a run with a single run backfill, instead of one per partition
+        # (but will also cause partitions to not be marked as failed if the run fails
+        return get_boolean_tag_value(os.getenv("DAGSTER_EMIT_PARTITION_SUBSET_IN_PLANNED_EVENTS"))
 
     @property
     def asset_records_have_last_observation(self) -> bool: