Merge pull request #37376: Adds a new SchemaCoderPayload proto to use for portable SchemaCoders.

Author: kennknowles

model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/beam_runner_api.proto

@@ -1073,6 +1073,10 @@ message StandardCoders {
     // Components: None
     ROW = 13 [(beam_urn) = "beam:coder:row:v1"];
 
+    // Similar to ROW above, but for arbitrary types that can be converted
+    // to and from row objects, which can then be encoded with a schema.
+    SCHEMA = 18 [(beam_urn) = "beam:coder:schema:v1"];
+
     // Encodes a user key and a shard id which is an opaque byte string.
     //
     // The encoding for a sharded key consists of a shard id byte string and the

model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/schema.proto

@@ -48,7 +48,7 @@ message Field {
   // OPTIONAL. Human readable description of this field, such as the query that generated it.
   string description = 2;
   FieldType type = 3;
-  
+
   int32 id = 4;
    // OPTIONAL. The position of this field's data when encoded, e.g. with beam:coder:row:v1.
    // Either no fields in a given row are have encoding position populated,
@@ -237,3 +237,23 @@ message MapTypeEntry {
 message LogicalTypeValue {
   FieldValue value = 1;
 }
+
+// Information needed to represent a Coder of type SCHEMA.
+message SchemaCoderPayload {
+  // The schema to use for encoding corresponding Row types.
+  Schema schema = 1;
+
+  // Function mapping from underlying object to Row type.
+  FunctionSpec to_row_fn = 2;
+
+  // Function mapping from Row type to underlying object.
+  FunctionSpec from_row_fn = 3;
+
+  // Any additional information SDKs need to encode/decode elements.
+  repeated AdditionalCoderInfo additional_coder_infos = 4;
+
+  message AdditionalCoderInfo {
+    string urn = 1;
+    bytes payload = 2;
+  }
+}