Added cli command to delete file.

Author: subkanthi

pyiceberg/cli/console.py

@@ -31,7 +31,7 @@
 from pyiceberg.cli.output import ConsoleOutput, JsonOutput, Output
 from pyiceberg.exceptions import NoSuchNamespaceError, NoSuchPropertyException, NoSuchTableError
 from pyiceberg.table import TableProperties
-from pyiceberg.table.refs import SnapshotRef, SnapshotRefType
+from pyiceberg.table.refs import MAIN_BRANCH, SnapshotRef, SnapshotRefType
 from pyiceberg.utils.properties import property_as_int
 
 
@@ -179,6 +179,59 @@ def files(ctx: Context, identifier: str, history: bool) -> None:
     output.files(catalog_table, history)
 
 
+@run.command("delete-files")
+@click.argument("identifier")
+@click.argument("file_paths", nargs=-1)
+@click.option("--branch", default=None, help="Branch to delete files from (default: main).")
+@click.option(
+    "--property",
+    "-p",
+    "properties",
+    multiple=True,
+    help="Snapshot property key=value (repeatable).",
+)
+@click.pass_context
+@catch_exception()
+def delete_files(
+    ctx: Context,
+    identifier: str,
+    file_paths: tuple[str, ...],
+    branch: str | None,
+    properties: tuple[str, ...],
+) -> None:
+
+
+    """
+        Remove one or more data files from the table by path
+    """
+
+    """Remove one or more data files from the table by path."""
+    if not file_paths:
+        raise click.UsageError("At least one file path is required.")
+
+    catalog, output = _catalog_and_output(ctx)
+
+    snapshot_properties: dict[str, str] = {}
+    for prop in properties:
+        if "=" not in prop:
+            raise click.UsageError(f"Property must be in key=value form, got: {prop!r}")
+        key, _, value = prop.partition("=")
+        snapshot_properties[key] = value
+
+    table = catalog.load_table(identifier)
+
+    file_paths_list = []
+    for item in file_paths:
+        file_paths_list.append(item)
+
+    table.delete_files(
+            file_paths=list(file_paths),
+            branch=branch or MAIN_BRANCH,
+            snapshot_properties=snapshot_properties
+    )
+    output.text(f"Deleted {len(file_paths)} file(s) from {identifier}")
+
+
 @run.command()
 @click.argument("identifier")
 @click.pass_context
@@ -470,3 +523,7 @@ def _retention_properties(ref: SnapshotRef, table_properties: dict[str, str]) ->
     retention_properties["max_ref_age_ms"] = str(ref.max_ref_age_ms) if ref.max_ref_age_ms else "forever"
 
     return retention_properties
+
+
+if __name__ == "__main__":
+    run()

tests/cli/test_console.py

@@ -1029,3 +1029,36 @@ def test_log_level_cli_overrides_env(mocker: MockFixture) -> None:
     mock_basicConfig.assert_called_once()
     call_kwargs = mock_basicConfig.call_args[1]
     assert call_kwargs["level"] == logging.ERROR
+
+def test_delete_files_requires_at_least_one_path(catalog: InMemoryCatalog) -> None:
+    runner = CliRunner()
+    result = runner.invoke(run, ["delete-files", "default.my_table"])
+    assert result.exit_code == 2
+    out = (result.output or "") + (getattr(result, "stderr", "") or "")
+    assert "file path" in out.lower() or "At least one" in out
+
+
+def test_delete_files_invalid_property_format(catalog: InMemoryCatalog, mocker: MockFixture) -> None:
+    catalog.create_namespace(TEST_TABLE_NAMESPACE)
+    catalog.create_table(
+        identifier=TEST_TABLE_IDENTIFIER,
+        schema=TEST_TABLE_SCHEMA,
+        partition_spec=TEST_TABLE_PARTITION_SPEC,
+    )
+    runner = CliRunner()
+    result = runner.invoke(
+        run,
+        ["delete-files", "default.my_table", "s3://bucket/file.parquet", "--property", "invalid_no_equals"],
+    )
+    assert result.exit_code == 2
+    out = (result.output or "") + (getattr(result, "stderr", "") or "")
+    assert "key=value" in out or "invalid_no_equals" in out
+
+
+def test_delete_files_table_does_not_exist(catalog: InMemoryCatalog) -> None:
+    catalog.create_namespace(TEST_TABLE_NAMESPACE)
+    runner = CliRunner()
+    result = runner.invoke(run, ["delete-files", "default.doesnotexist", "s3://bucket/file.parquet"])
+    assert result.exit_code == 1
+    out = (result.output or "") + (getattr(result, "stderr", "") or "")
+    assert "default.doesnotexist" in out and ("Table does not exist" in out or "does not exist" in out)