[azure-ai-ml] Add hdfs input mode mapping for job input serialization

Author: GnanasundaramSampath

sdk/ml/azure-ai-ml/azure/ai/ml/constants/_common.py

@@ -836,6 +836,8 @@ class InputOutputModes:
     """Evaluation download asset type."""
     DIRECT = "direct"
     """Direct asset type."""
+    HDFS = "hdfs"
+    """HDFS asset type."""
 
 
 class ConnectionTypes:

sdk/ml/azure-ai-ml/azure/ai/ml/entities/_job/_input_output_helpers.py

@@ -69,6 +69,8 @@
     InputDeliveryMode.DIRECT: InputOutputModes.DIRECT,
     InputDeliveryMode.EVAL_MOUNT: InputOutputModes.EVAL_MOUNT,
     InputDeliveryMode.EVAL_DOWNLOAD: InputOutputModes.EVAL_DOWNLOAD,
+    "Hdfs": InputOutputModes.HDFS,
+    "hdfs": InputOutputModes.HDFS,
 }
 
 INPUT_MOUNT_MAPPING_TO_REST = {
@@ -79,6 +81,7 @@
     InputOutputModes.EVAL_MOUNT: InputDeliveryMode.EVAL_MOUNT,
     InputOutputModes.EVAL_DOWNLOAD: InputDeliveryMode.EVAL_DOWNLOAD,
     InputOutputModes.DIRECT: InputDeliveryMode.DIRECT,
+    InputOutputModes.HDFS: "Hdfs",
 }
 
 
@@ -249,7 +252,7 @@ def to_rest_dataset_literal_inputs(
                     input_data = LiteralJobInput(value=input_value.path)
                     # set mode attribute manually for binding job input
                     if input_value.mode:
-                        input_data.mode = INPUT_MOUNT_MAPPING_TO_REST[input_value.mode]
+                        input_data.mode = INPUT_MOUNT_MAPPING_TO_REST[input_value.mode.lower()]
                     if getattr(input_value, "path_on_compute", None) is not None:
                         input_data.pathOnCompute = input_value.path_on_compute
                     input_data.job_input_type = JobInputType.LITERAL
@@ -279,7 +282,8 @@ def to_rest_dataset_literal_inputs(
                     input_data = LiteralJobInput(value=str(input_value["value"]))
                     # set mode attribute manually for binding job input
                     if "mode" in input_value:
-                        input_data.mode = input_value["mode"]
+                        input_mode = str(input_value["mode"])
+                        input_data.mode = INPUT_MOUNT_MAPPING_TO_REST.get(input_mode.lower(), input_mode)
                 else:
                     input_data = LiteralJobInput(value=str(input_value))
                 input_data.job_input_type = JobInputType.LITERAL

sdk/ml/azure-ai-ml/tests/pipeline_job/unittests/test_pipeline_job_schema.py

@@ -32,6 +32,8 @@
 from azure.ai.ml.entities._inputs_outputs import Input, Output
 from azure.ai.ml.entities._job._input_output_helpers import (
     INPUT_MOUNT_MAPPING_FROM_REST,
+    from_rest_inputs_to_dataset_literal,
+    to_rest_dataset_literal,
     validate_pipeline_input_key_characters,
 )
 from azure.ai.ml.entities._job.automl.search_space_utils import _convert_sweep_dist_dict_to_str_dict
@@ -53,6 +55,24 @@
 @pytest.mark.unittest
 @pytest.mark.pipeline_test
 class TestPipelineJobSchema:
+    def test_hdfs_input_mode_round_trip(self):
+        rest_inputs = to_rest_dataset_literal(
+            {
+                "hdfs_input": Input(
+                    type=AssetTypes.URI_FOLDER,
+                    path="azureml://datastores/workspaceblobstore/paths/data/",
+                    mode=InputOutputModes.HDFS,
+                )
+            },
+            job_type=None,
+        )
+
+        assert rest_inputs["hdfs_input"].mode == "Hdfs"
+
+        sdk_inputs = from_rest_inputs_to_dataset_literal(rest_inputs)
+        assert isinstance(sdk_inputs["hdfs_input"], Input)
+        assert sdk_inputs["hdfs_input"].mode == InputOutputModes.HDFS
+
     def test_validate_pipeline_job_keys(self):
         def validator(key, assert_valid=True):
             if assert_valid: