[SPARK-53593][SDP] Add response field for DefineDataset and DefineFlow RPC

python/pyspark/sql/connect/proto/pipelines_pb2.pyi

@@ -110,40 +110,6 @@ class PipelineCommand(google.protobuf.message.Message):
                 self, field_name: typing_extensions.Literal["key", b"key", "value", b"value"]
             ) -> None: ...
 
-        class Response(google.protobuf.message.Message):
-            DESCRIPTOR: google.protobuf.descriptor.Descriptor
-
-            DATAFLOW_GRAPH_ID_FIELD_NUMBER: builtins.int
-            dataflow_graph_id: builtins.str
-            """The ID of the created graph."""
-            def __init__(
-                self,
-                *,
-                dataflow_graph_id: builtins.str | None = ...,
-            ) -> None: ...
-            def HasField(
-                self,
-                field_name: typing_extensions.Literal[
-                    "_dataflow_graph_id",
-                    b"_dataflow_graph_id",
-                    "dataflow_graph_id",
-                    b"dataflow_graph_id",
-                ],
-            ) -> builtins.bool: ...
-            def ClearField(
-                self,
-                field_name: typing_extensions.Literal[
-                    "_dataflow_graph_id",
-                    b"_dataflow_graph_id",
-                    "dataflow_graph_id",
-                    b"dataflow_graph_id",
-                ],
-            ) -> None: ...
-            def WhichOneof(
-                self,
-                oneof_group: typing_extensions.Literal["_dataflow_graph_id", b"_dataflow_graph_id"],
-            ) -> typing_extensions.Literal["dataflow_graph_id"] | None: ...
-
         DEFAULT_CATALOG_FIELD_NUMBER: builtins.int
         DEFAULT_DATABASE_FIELD_NUMBER: builtins.int
         SQL_CONF_FIELD_NUMBER: builtins.int
@@ -753,6 +719,69 @@ class PipelineCommandResult(google.protobuf.message.Message):
 
     DESCRIPTOR: google.protobuf.descriptor.Descriptor
 
+    class CatalogIdentifier(google.protobuf.message.Message):
+        DESCRIPTOR: google.protobuf.descriptor.Descriptor
+
+        CATALOG_FIELD_NUMBER: builtins.int
+        DATABASE_FIELD_NUMBER: builtins.int
+        NAME_FIELD_NUMBER: builtins.int
+        catalog: builtins.str
+        database: builtins.str
+        name: builtins.str
+        def __init__(
+            self,
+            *,
+            catalog: builtins.str | None = ...,
+            database: builtins.str | None = ...,
+            name: builtins.str | None = ...,
+        ) -> None: ...
+        def HasField(
+            self,
+            field_name: typing_extensions.Literal[
+                "_catalog",
+                b"_catalog",
+                "_database",
+                b"_database",
+                "_name",
+                b"_name",
+                "catalog",
+                b"catalog",
+                "database",
+                b"database",
+                "name",
+                b"name",
+            ],
+        ) -> builtins.bool: ...
+        def ClearField(
+            self,
+            field_name: typing_extensions.Literal[
+                "_catalog",
+                b"_catalog",
+                "_database",
+                b"_database",
+                "_name",
+                b"_name",
+                "catalog",
+                b"catalog",
+                "database",
+                b"database",
+                "name",
+                b"name",
+            ],
+        ) -> None: ...
+        @typing.overload
+        def WhichOneof(
+            self, oneof_group: typing_extensions.Literal["_catalog", b"_catalog"]
+        ) -> typing_extensions.Literal["catalog"] | None: ...
+        @typing.overload
+        def WhichOneof(
+            self, oneof_group: typing_extensions.Literal["_database", b"_database"]
+        ) -> typing_extensions.Literal["database"] | None: ...
+        @typing.overload
+        def WhichOneof(
+            self, oneof_group: typing_extensions.Literal["_name", b"_name"]
+        ) -> typing_extensions.Literal["name"] | None: ...
+
     class CreateDataflowGraphResult(google.protobuf.message.Message):
         DESCRIPTOR: google.protobuf.descriptor.Descriptor
 
@@ -787,22 +816,104 @@ class PipelineCommandResult(google.protobuf.message.Message):
             oneof_group: typing_extensions.Literal["_dataflow_graph_id", b"_dataflow_graph_id"],
         ) -> typing_extensions.Literal["dataflow_graph_id"] | None: ...
 
+    class DefineDatasetResult(google.protobuf.message.Message):
+        DESCRIPTOR: google.protobuf.descriptor.Descriptor
+
+        RESOLVED_IDENTIFIER_FIELD_NUMBER: builtins.int
+        @property
+        def resolved_identifier(self) -> global___PipelineCommandResult.CatalogIdentifier:
+            """Resolved name of the dataset"""
+        def __init__(
+            self,
+            *,
+            resolved_identifier: global___PipelineCommandResult.CatalogIdentifier | None = ...,
+        ) -> None: ...
+        def HasField(
+            self,
+            field_name: typing_extensions.Literal[
+                "_resolved_identifier",
+                b"_resolved_identifier",
+                "resolved_identifier",
+                b"resolved_identifier",
+            ],
+        ) -> builtins.bool: ...
+        def ClearField(
+            self,
+            field_name: typing_extensions.Literal[
+                "_resolved_identifier",
+                b"_resolved_identifier",
+                "resolved_identifier",
+                b"resolved_identifier",
+            ],
+        ) -> None: ...
+        def WhichOneof(
+            self,
+            oneof_group: typing_extensions.Literal["_resolved_identifier", b"_resolved_identifier"],
+        ) -> typing_extensions.Literal["resolved_identifier"] | None: ...
+
+    class DefineFlowResult(google.protobuf.message.Message):
+        DESCRIPTOR: google.protobuf.descriptor.Descriptor
+
+        RESOLVED_IDENTIFIER_FIELD_NUMBER: builtins.int
+        @property
+        def resolved_identifier(self) -> global___PipelineCommandResult.CatalogIdentifier:
+            """Resolved name of the flow"""
+        def __init__(
+            self,
+            *,
+            resolved_identifier: global___PipelineCommandResult.CatalogIdentifier | None = ...,
+        ) -> None: ...
+        def HasField(
+            self,
+            field_name: typing_extensions.Literal[
+                "_resolved_identifier",
+                b"_resolved_identifier",
+                "resolved_identifier",
+                b"resolved_identifier",
+            ],
+        ) -> builtins.bool: ...
+        def ClearField(
+            self,
+            field_name: typing_extensions.Literal[
+                "_resolved_identifier",
+                b"_resolved_identifier",
+                "resolved_identifier",
+                b"resolved_identifier",
+            ],
+        ) -> None: ...
+        def WhichOneof(
+            self,
+            oneof_group: typing_extensions.Literal["_resolved_identifier", b"_resolved_identifier"],
+        ) -> typing_extensions.Literal["resolved_identifier"] | None: ...
+
     CREATE_DATAFLOW_GRAPH_RESULT_FIELD_NUMBER: builtins.int
+    DEFINE_DATASET_RESULT_FIELD_NUMBER: builtins.int
+    DEFINE_FLOW_RESULT_FIELD_NUMBER: builtins.int
     @property
     def create_dataflow_graph_result(
         self,
     ) -> global___PipelineCommandResult.CreateDataflowGraphResult: ...
+    @property
+    def define_dataset_result(self) -> global___PipelineCommandResult.DefineDatasetResult: ...
+    @property
+    def define_flow_result(self) -> global___PipelineCommandResult.DefineFlowResult: ...
     def __init__(
         self,
         *,
         create_dataflow_graph_result: global___PipelineCommandResult.CreateDataflowGraphResult
         | None = ...,
+        define_dataset_result: global___PipelineCommandResult.DefineDatasetResult | None = ...,
+        define_flow_result: global___PipelineCommandResult.DefineFlowResult | None = ...,
     ) -> None: ...
     def HasField(
         self,
         field_name: typing_extensions.Literal[
             "create_dataflow_graph_result",
             b"create_dataflow_graph_result",
+            "define_dataset_result",
+            b"define_dataset_result",
+            "define_flow_result",
+            b"define_flow_result",
             "result_type",
             b"result_type",
         ],
@@ -812,13 +923,22 @@ class PipelineCommandResult(google.protobuf.message.Message):
         field_name: typing_extensions.Literal[
             "create_dataflow_graph_result",
             b"create_dataflow_graph_result",
+            "define_dataset_result",
+            b"define_dataset_result",
+            "define_flow_result",
+            b"define_flow_result",
             "result_type",
             b"result_type",
         ],
     ) -> None: ...
     def WhichOneof(
         self, oneof_group: typing_extensions.Literal["result_type", b"result_type"]
-    ) -> typing_extensions.Literal["create_dataflow_graph_result"] | None: ...
+    ) -> (
+        typing_extensions.Literal[
+            "create_dataflow_graph_result", "define_dataset_result", "define_flow_result"
+        ]
+        | None
+    ): ...
 
 global___PipelineCommandResult = PipelineCommandResult
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/identifiers.scala

@@ -36,17 +36,17 @@ sealed trait CatalystIdentifier {
    */
   private def quoteIdentifier(name: String): String = name.replace("`", "``")
 
+  def resolvedId: String = quoteIdentifier(identifier)
+  def resolvedDb: Option[String] = database.map(quoteIdentifier)
+  def resolvedCatalog: Option[String] = catalog.map(quoteIdentifier)
+
   def quotedString: String = {
-    val replacedId = quoteIdentifier(identifier)
-    val replacedDb = database.map(quoteIdentifier)
-    val replacedCatalog = catalog.map(quoteIdentifier)
-
-    if (replacedCatalog.isDefined && replacedDb.isDefined) {
-      s"`${replacedCatalog.get}`.`${replacedDb.get}`.`$replacedId`"
-    } else if (replacedDb.isDefined) {
-      s"`${replacedDb.get}`.`$replacedId`"
+    if (resolvedCatalog.isDefined && resolvedDb.isDefined) {
+      s"`${resolvedCatalog.get}`.`${resolvedDb.get}`.`$resolvedId`"
+    } else if (resolvedDb.isDefined) {
+      s"`${resolvedDb.get}`.`$resolvedId`"
     } else {
-      s"`$replacedId`"
+      s"`$resolvedId`"
     }
   }
 

sql/connect/common/src/main/protobuf/spark/connect/pipelines.proto

@@ -48,11 +48,6 @@ message PipelineCommand {
 
     // SQL configurations for all flows in this graph.
     map<string, string> sql_conf = 5;
-
-    message Response {
-      // The ID of the created graph.
-      optional string dataflow_graph_id = 1;
-    }
   }
 
   // Drops the graph and stops any running attached flows.
@@ -146,11 +141,26 @@ message PipelineCommand {
 message PipelineCommandResult {
   oneof result_type {
     CreateDataflowGraphResult create_dataflow_graph_result = 1;
+    DefineDatasetResult define_dataset_result = 2;
+    DefineFlowResult define_flow_result = 3;
+  }
+  message CatalogIdentifier {
+    optional string catalog = 1;
+    optional string database = 2;
+    optional string name = 3;
   }
   message CreateDataflowGraphResult {
     // The ID of the created graph.
     optional string dataflow_graph_id = 1;
   }
+  message DefineDatasetResult {
+    // Resolved name of the dataset
+    optional CatalogIdentifier resolved_identifier = 1;
+  }
+  message DefineFlowResult {
+    // Resolved name of the flow
+    optional CatalogIdentifier resolved_identifier = 1;
+  }
 }
 
 // The type of dataset.