fixes bug in drop command (#693)

dlt/pipeline/helpers.py

@@ -143,10 +143,12 @@ def _create_modified_state(self) -> Dict[str, Any]:
             return state  # type: ignore[return-value]
         source_states = _sources_state(state).items()
         for source_name, source_state in source_states:
-            if self.drop_state:
+            # drop table states
+            if self.drop_state and self.resource_pattern:
                 for key in _get_matching_resources(self.resource_pattern, source_state):
                     self.info['resource_states'].append(key)
                     reset_resource_state(key, source_state)
+            # drop additional state paths
             resolved_paths = resolve_paths(self.state_paths_to_drop, source_state)
             if self.state_paths_to_drop and not resolved_paths:
                 self.info['warnings'].append(f"State paths {self.state_paths_to_drop} did not select any paths in source {source_name}")

tests/load/pipeline/test_drop.py

@@ -122,6 +122,28 @@ def test_drop_command_resources_and_state(destination_config: DestinationTestCon
     assert_destination_state_loaded(pipeline)
 
 
+@pytest.mark.parametrize("destination_config", destinations_configs(default_sql_configs=True), ids=lambda x: x.name)
+def test_drop_command_only_state(destination_config: DestinationTestConfiguration) -> None:
+    """Test the drop command with resource and state path options and
+    verify correct data is deleted from destination and locally"""
+    source = droppable_source()
+    pipeline = destination_config.setup_pipeline('drop_test_' + uniq_id(), full_refresh=True)
+    pipeline.run(source)
+
+    attached = _attach(pipeline)
+    helpers.drop(attached, state_paths='data_from_d.*.bar')
+
+    attached = _attach(pipeline)
+
+    assert_dropped_resources(attached, [])
+
+    # Verify extra json paths are removed from state
+    sources_state = pipeline.state['sources']
+    assert sources_state['droppable']['data_from_d'] == {'foo1': {}, 'foo2': {}}
+
+    assert_destination_state_loaded(pipeline)
+
+
 @pytest.mark.parametrize("destination_config", destinations_configs(default_sql_configs=True), ids=lambda x: x.name)
 def test_drop_destination_tables_fails(destination_config: DestinationTestConfiguration) -> None:
     """Fail on drop tables. Command runs again."""