Drop column from child table test

Author: anuunchin

docs/website/docs/reference/command-line-interface.md

@@ -495,8 +495,8 @@ Selectively drop columns from specified tables.
 
 **Usage**
 ```sh
-dlt pipeline [pipeline_name] drop-columns [-h] [--from [FROM_RESOURCES ...]]
-    [--columns [COLUMNS ...]]
+dlt pipeline [pipeline_name] drop-columns [-h] [--from-resources [FROM_RESOURCES
+    ...]] [--from-tables [FROM_TABLES ...]] [--columns [COLUMNS ...]]
 ```
 
 **Description**
@@ -520,8 +520,9 @@ Inherits arguments from [`dlt pipeline`](#dlt-pipeline).
 
 **Options**
 * `-h, --help` - Show this help message and exit
-* `--from [FROM_RESOURCES ...]` - (with --columns) resource names to drop columns from.
-* `--columns [COLUMNS ...]` - (with --from) column names to drop from the specified resources.
+* `--from-resources [FROM_RESOURCES ...]` - Resource names to drop columns from.
+* `--from-tables [FROM_TABLES ...]` - Table names to drop columns from.
+* `--columns [COLUMNS ...]` - Column names to drop from the specified resources.
 
 </details>
 

tests/load/pipeline/test_drop_column.py

@@ -29,7 +29,9 @@ def test_drop_column_command_resources(destination_config: DestinationTestConfig
     """Test the drop command with resource and verify correct data is deleted from destination"""
     source: Any = droppable_source()
 
-    pipeline = destination_config.setup_pipeline("droppbale", dev_mode=True)
+    pipeline = destination_config.setup_pipeline(
+        "test_drop_column_command_resources", dev_mode=True
+    )
     info = pipeline.run(source, **destination_config.run_kwargs)
     assert_load_info(info)
 
@@ -81,3 +83,37 @@ def test_drop_column_command_resources(destination_config: DestinationTestConfig
         remaining_cols = pipeline.default_schema.get_table_columns(must_drop_info["from_table"])
         for dropped_col in must_drop_info["drop_columns"]:
             assert dropped_col not in remaining_cols
+
+
+@pytest.mark.parametrize(
+    "destination_config",
+    destinations_configs(
+        default_sql_configs=True,
+        local_filesystem_configs=True,
+        all_buckets_filesystem_configs=True,
+        table_format_local_configs=True,
+    ),
+    ids=lambda x: x.name,
+)
+def test_drop_column_from_child_table(destination_config: DestinationTestConfiguration) -> None:
+    """Test the drop command with column and child table and verify correct data is deleted from destination"""
+    source: Any = droppable_source()
+
+    pipeline = destination_config.setup_pipeline("test_drop_column_from_child_table", dev_mode=True)
+    info = pipeline.run(source, **destination_config.run_kwargs)
+    assert_load_info(info)
+
+    original_parent_tbl_cols = pipeline.default_schema.get_table_columns("droppable_b")
+    assert "m" in pipeline.default_schema.get_table_columns("droppable_b__items")
+
+    drop_cmd = helpers.DropCommand(
+        pipeline, from_resources=["droppable_b"], from_tables=["droppable_b__items"], columns=["m"]
+    )
+    must_drop_infos = drop_cmd.from_tables_drop_cols
+
+    assert [{"from_table": "droppable_b__items", "drop_columns": ["m"]}] == must_drop_infos
+
+    drop_cmd()
+
+    assert "m" not in pipeline.default_schema.get_table_columns("droppable_b__items")
+    assert original_parent_tbl_cols == pipeline.default_schema.get_table_columns("droppable_b")