[SPARK-54435][SDP] spark-pipelines init should avoid overwriting existing directory

### What changes were proposed in this pull request?

If the name provided to `spark-pipelines init` matches an existing directory, raises an error instead of overwriting the existing directory's contents.

### Why are the changes needed?

Help users avoid accidentally overwriting their code.

### Does this PR introduce _any_ user-facing change?

Yes, to an unreleased version.

### How was this patch tested?

Added unit test for the cases where init is invoked with an existing directory.

### Was this patch authored or co-authored using generative AI tooling?

Closes apache#53140 from sryza/initoverwrite.

Authored-by: Sandy Ryza <sandy.ryza@databricks.com>
Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>

Author: sryza

python/pyspark/pipelines/init_cli.py

@@ -44,6 +44,11 @@ def example_python_materialized_view() -> DataFrame:
 def init(name: str) -> None:
     """Generates a simple pipeline project."""
     project_dir = Path.cwd() / name
+    if project_dir.exists():
+        raise FileExistsError(
+            f"Directory '{name}' already exists. "
+            "Please choose a different name or remove the existing directory."
+        )
     project_dir.mkdir(parents=True, exist_ok=False)
 
     # Create the storage directory

python/pyspark/pipelines/tests/test_init_cli.py

@@ -72,6 +72,21 @@ def test_init(self):
                     Path("transformations") / "example_sql_materialized_view.sql",
                 )
 
+    def test_init_existing_directory(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            project_name = "test_project"
+            with change_dir(Path(temp_dir)):
+                init(project_name)
+
+                with self.assertRaises(FileExistsError) as context:
+                    init(project_name)
+
+                expected_message = (
+                    f"Directory '{project_name}' already exists. "
+                    "Please choose a different name or remove the existing directory."
+                )
+                self.assertEqual(str(context.exception), expected_message)
+
 
 if __name__ == "__main__":
     try: