Add dev_mode support to pipeline state management

- Introduced `dev_mode` attribute in TPipelineLocalState to track development mode status.
- Implemented logic to recreate initial pipeline state when toggling between dev and non-dev modes.
- Updated dataset name generation to ensure consistency based on the current mode.
- Added tests to verify dataset name behavior in different modes.

Author: alkaline-0

dlt/common/pipeline.py

@@ -479,6 +479,8 @@ class TPipelineLocalState(TypedDict, total=False):
     """Run dir when pipeline was instantiated for a first time, defaults to cwd on OSS run context"""
     last_run_context: Optional[TLastRunContext]
     """Context from the last successful pipeline run or sync"""
+    dev_mode: bool
+    """Indicates whether previous run used dev_mode; used to reset state on dev->non-dev toggle"""
 
 
 class TPipelineState(TVersionedState, total=False):

dlt/pipeline/pipeline.py

@@ -1224,6 +1224,14 @@ def _init_working_dir(self, pipeline_name: str, pipelines_dir: str) -> None:
         if self.dev_mode:
             self._wipe_working_folder()
 
+    def _recreate_initial_state(self) -> None:
+        # replace the injected state dict with a fresh default
+        injected_state = self._container[StateInjectableContext].state
+        injected_state.clear()
+        injected_state.update(default_pipeline_state())
+        self._set_dataset_name(None)
+        self._save_state(injected_state)
+
     def _configure(
         self, import_schema_path: str, export_schema_path: str, must_attach_to_local_pipeline: bool
     ) -> None:
@@ -1247,14 +1255,17 @@ def _configure(
             )
 
         self.must_attach_to_local_pipeline = must_attach_to_local_pipeline
-        # attach to pipeline if folder exists and contains state
-        if has_state:
+        prev_dev_mode: bool = self.get_local_state_val("dev_mode") if has_state else False
+        should_recreate_pipeline: bool = prev_dev_mode and not self.dev_mode
+        self.set_local_state_val("dev_mode", self.dev_mode)  # create schema storage
+        if has_state and not should_recreate_pipeline:
             self._attach_pipeline()
         else:
-            # this will erase the existing working folder
             self._create_pipeline()
 
-        # create schema storage
+        if should_recreate_pipeline:
+            self._recreate_initial_state()
+
         self._schema_storage = LiveSchemaStorage(self._schema_storage_config, makedirs=True)
 
     def _create_pipeline(self) -> None:
@@ -1521,10 +1532,12 @@ def _make_dataset_name(
     ) -> str:
         """Generates dataset name for the pipeline based on `new_dataset_name`
         1. if name is not provided, default name is created
-        2. for destinations that do not need dataset names, def. name is not created
-        3. we add serial number in dev mode
-        4. we apply layout from pipeline config if present
+        2. if destination is not provided, and no dataset name is provided, default name is created
+        3. for destinations that do not need dataset names, def. name is not created
+        4. we add serial number in dev mode
+        5. we apply layout from pipeline config if present
         """
+        # TODO: update this function to differentiate between the new_dataset_name parameter and the dataset_name property created by the function
         if not new_dataset_name:
             # dataset name is required but not provided - generate the default now
             destination_needs_dataset = False
@@ -1534,6 +1547,9 @@ def _make_dataset_name(
             if destination_needs_dataset:
                 new_dataset_name = self.pipeline_name + self.DEFAULT_DATASET_SUFFIX
 
+        if not destination and not new_dataset_name:
+            new_dataset_name = self.pipeline_name + self.DEFAULT_DATASET_SUFFIX
+
         if not new_dataset_name:
             return new_dataset_name
 

dlt/pipeline/state_sync.py

@@ -144,5 +144,6 @@ def default_pipeline_state() -> TPipelineState:
             "first_run": True,
             # keep the initial run dir when the pipeline was created
             "initial_cwd": os.path.abspath(dlt.current.run_context().local_dir),
+            "dev_mode": False,
         },
     }

tests/pipeline/test_pipeline.py

@@ -278,6 +278,12 @@ def test_pipeline_configuration_named_section(environment) -> None:
         == environment["PIPELINES__NAMED__IMPORT_SCHEMA_PATH"]
     )
 
+def test_dataset_name_consistency_with_dev_mode() -> None:
+    p = dlt.pipeline(dev_mode=True, destination="filesystem")
+    assert p.dataset_name.endswith(p._pipeline_instance_id)
+    # restore this pipeline
+    r_p = dlt.attach()
+    assert not r_p.dataset_name.endswith(p._pipeline_instance_id)
 
 def test_run_dev_mode_default_dataset() -> None:
     p = dlt.pipeline(dev_mode=True, destination="filesystem")