bug-fix: support idempotency for framework and spark processors

Author: Brock Wade

src/sagemaker/processing.py

@@ -1827,14 +1827,19 @@ def _patch_inputs_with_payload(self, inputs, s3_payload) -> List[ProcessingInput
         #   a7399455f5386d83ddc5cb15c0db00c04bd518ec/src/sagemaker/processing.py#L425-L426
         if inputs is None:
             inputs = []
-        inputs.append(
+
+        # make a shallow copy of user inputs
+        patched_inputs = []
+        for user_input in inputs:
+            patched_inputs.append(user_input)
+        patched_inputs.append(
             ProcessingInput(
                 input_name="code",
                 source=s3_payload,
                 destination="/opt/ml/processing/input/code/",
             )
         )
-        return inputs
+        return patched_inputs
 
     def _set_entrypoint(self, command, user_script_name):
         """Framework processor override for setting processing job entrypoint.

src/sagemaker/spark/processing.py

@@ -940,9 +940,18 @@ def _extend_processing_args(self, inputs, outputs, **kwargs):
             outputs: Processing outputs.
             kwargs: Additional keyword arguments passed to `super()`.
         """
+
+        if inputs is None:
+            inputs = []
+
+        # make a shallow copy of user inputs
+        extended_inputs = []
+        for user_input in inputs:
+            extended_inputs.append(user_input)
+
         self.command = [_SparkProcessorBase._default_command]
         extended_inputs = self._handle_script_dependencies(
-            inputs, kwargs.get("submit_py_files"), FileType.PYTHON
+            extended_inputs, kwargs.get("submit_py_files"), FileType.PYTHON
         )
         extended_inputs = self._handle_script_dependencies(
             extended_inputs, kwargs.get("submit_jars"), FileType.JAR
@@ -1199,8 +1208,16 @@ def _extend_processing_args(self, inputs, outputs, **kwargs):
         else:
             raise ValueError("submit_class is required")
 
+        if inputs is None:
+            inputs = []
+
+        # make a shallow copy of user inputs
+        extended_inputs = []
+        for user_input in inputs:
+            extended_inputs.append(user_input)
+
         extended_inputs = self._handle_script_dependencies(
-            inputs, kwargs.get("submit_jars"), FileType.JAR
+            extended_inputs, kwargs.get("submit_jars"), FileType.JAR
         )
         extended_inputs = self._handle_script_dependencies(
             extended_inputs, kwargs.get("submit_files"), FileType.FILE

src/sagemaker/workflow/utilities.py

@@ -168,7 +168,7 @@ def get_processing_code_hash(code: str, source_dir: str, dependencies: List[str]
         str: A hash string representing the unique code artifact(s) for the step
     """
 
-    # FrameworkProcessor
+    # If FrameworkProcessor contains source_dir
     if source_dir:
         source_dir_url = urlparse(source_dir)
         if source_dir_url.scheme == "" or source_dir_url.scheme == "file":
@@ -400,5 +400,5 @@ def execute_job_functions(step_args: _StepArguments):
     """
 
     chained_args = step_args.func(*step_args.func_args, **step_args.func_kwargs)
-    if chained_args:
+    if isinstance(chained_args, _StepArguments):
         execute_job_functions(chained_args)