[FLINK-34620][formats][protobuf] Add serialize-side handling for recursive message schemas

docs/content/docs/connectors/table/formats/protobuf.md

@@ -153,6 +153,7 @@ Format Options
           If the value is set to false, the format will generate null values if the data element does not exist in the binary protobuf message.
           With Flink's current protobuf version (4.32.1), field presence is properly supported for proto3, allowing null handling for non-primitive types.
           Please be aware that setting this to true will cause the deserialization performance to be much slower depending on schema complexity and message size.
+          Note: a recursive message type that declares <code>required</code> fields cannot be serialized when an absent recursive field is read with this set to true. The absent sub-message is materialized as empty bytes, which cannot be parsed back because its required fields are missing, so serialization fails with a runtime error. Use false for such messages, or ensure the recursive field is always present.
       </td>
     </tr>
     <tr>

flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/deserialize/PbCodegenBytesDeserializer.java

@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.formats.protobuf.deserialize;
+
+import org.apache.flink.formats.protobuf.PbCodegenException;
+import org.apache.flink.formats.protobuf.util.PbCodegenAppender;
+
+import com.google.protobuf.Descriptors.FieldDescriptor;
+
+/**
+ * Deserializer that converts a protobuf message to its raw binary bytes. Used when a recursive
+ * message type is detected and represented as BYTES in the Flink schema, preserving the data for
+ * optional later unpacking via a UDF.
+ */
+public class PbCodegenBytesDeserializer implements PbCodegenDeserializer {
+    private final FieldDescriptor fd;
+
+    public PbCodegenBytesDeserializer(FieldDescriptor fd) {
+        this.fd = fd;
+    }
+
+    @Override
+    public String codegen(String resultVar, String pbObjectCode, int indent)
+            throws PbCodegenException {
+        PbCodegenAppender appender = new PbCodegenAppender(indent);
+        appender.appendLine(resultVar + " = " + pbObjectCode + ".toByteArray()");
+        return appender.code();
+    }
+}

flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/deserialize/PbCodegenDeserializeFactory.java

@@ -23,16 +23,25 @@
 import org.apache.flink.formats.protobuf.util.PbFormatUtils;
 import org.apache.flink.table.types.logical.ArrayType;
 import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
 import org.apache.flink.table.types.logical.MapType;
 import org.apache.flink.table.types.logical.RowType;
 
 import com.google.protobuf.Descriptors;
+import com.google.protobuf.Descriptors.FieldDescriptor.JavaType;
 
 /** Codegen factory class which return {@link PbCodegenDeserializer} of different data type. */
 public class PbCodegenDeserializeFactory {
     public static PbCodegenDeserializer getPbCodegenDes(
             Descriptors.FieldDescriptor fd, LogicalType type, PbFormatContext formatContext)
             throws PbCodegenException {
+        // Check for recursive message type represented as raw bytes:
+        // the proto field is a MESSAGE but the logical type has been resolved to VARBINARY
+        // by PbToRowTypeUtil's cycle detection.
+        if (fd.getJavaType() == JavaType.MESSAGE
+                && type.getTypeRoot() == LogicalTypeRoot.VARBINARY) {
+            return new PbCodegenBytesDeserializer(fd);
+        }
         // We do not use FieldDescriptor to check because there's no way to get
         // element field descriptor of array type.
         if (type instanceof RowType) {

flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/deserialize/PbCodegenRowDeserializer.java

@@ -72,7 +72,11 @@ public String codegen(String resultVar, String pbObjectCode, int indent)
                     PbCodegenDeserializeFactory.getPbCodegenDes(elementFd, subType, formatContext);
             splitAppender.appendLine("Object " + flinkRowEleVar + " = null");
             boolean readDefaultValues = formatContext.getPbFormatConfig().isReadDefaultValues();
-            if (PbFormatUtils.isSimpleType(subType)) {
+            // A recursive message field is stored as VARBINARY, so it looks like a "simple type",
+            // but it is still a MESSAGE. Don't force the proto3 primitive default path on it, or an
+            // absent sub-message would round-trip back as a present (empty) one. See FLINK-34620.
+            if (PbFormatUtils.isSimpleType(subType)
+                    && elementFd.getJavaType() != FieldDescriptor.JavaType.MESSAGE) {
                 readDefaultValues = formatContext.getReadDefaultValuesForPrimitiveTypes();
             }
 

flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/serialize/PbCodegenBytesSerializer.java

@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.formats.protobuf.serialize;
+
+import org.apache.flink.formats.protobuf.PbCodegenException;
+import org.apache.flink.formats.protobuf.util.PbCodegenAppender;
+import org.apache.flink.formats.protobuf.util.PbCodegenVarId;
+import org.apache.flink.formats.protobuf.util.PbFormatUtils;
+
+import com.google.protobuf.Descriptors.FieldDescriptor;
+
+/**
+ * Serializer for a recursive message field, which {@link
+ * org.apache.flink.formats.protobuf.util.PbToRowTypeUtil} represents as BYTES through its cycle
+ * detection. It is the inverse of {@link
+ * org.apache.flink.formats.protobuf.deserialize.PbCodegenBytesDeserializer}: the deserializer emits
+ * the sub-message as {@code message.toByteArray()}, and here we parse those bytes back into the
+ * message so the enclosing {@link PbCodegenRowSerializer} can set it on the builder.
+ */
+public class PbCodegenBytesSerializer implements PbCodegenSerializer {
+    private final FieldDescriptor fd;
+
+    public PbCodegenBytesSerializer(FieldDescriptor fd) {
+        this.fd = fd;
+    }
+
+    @Override
+    public String codegen(String resultVar, String flinkObjectCode, int indent)
+            throws PbCodegenException {
+        // flinkObjectCode is a non-null flink BYTES value holding a serialized protobuf message.
+        PbCodegenAppender appender = new PbCodegenAppender(indent);
+        int uid = PbCodegenVarId.getInstance().getAndIncrement();
+        String bytesVar = "recursiveBytes" + uid;
+        String messageTypeStr = PbFormatUtils.getFullJavaName(fd.getMessageType());
+
+        appender.appendLine("byte[] " + bytesVar + " = " + flinkObjectCode);
+        codegenRequiredFieldGuard(appender, bytesVar, messageTypeStr);
+        appender.begin("try{");
+        appender.appendLine(resultVar + " = " + messageTypeStr + ".parseFrom(" + bytesVar + ")");
+        appender.end("}");
+        appender.begin("catch(com.google.protobuf.InvalidProtocolBufferException e){");
+        appender.appendLine(
+                "throw new RuntimeException("
+                        + "\"failed to parse recursive protobuf bytes for field '"
+                        + fd.getName()
+                        + "'\", e)");
+        appender.end("}");
+        return appender.code();
+    }
+
+    // A recursive message type with required fields cannot round-trip an absent field read with
+    // read-default-values=true: it is materialized on read as byte[0], which parseFrom cannot
+    // rebuild. getDefaultInstance().isInitialized() is false only for such types, so this guard is
+    // a no-op for proto3 and for proto2 without required fields.
+    private void codegenRequiredFieldGuard(
+            PbCodegenAppender appender, String bytesVar, String messageTypeStr) {
+        appender.begin(
+                "if("
+                        + bytesVar
+                        + ".length == 0 && !"
+                        + messageTypeStr
+                        + ".getDefaultInstance().isInitialized()){");
+        appender.appendLine(
+                "throw new RuntimeException("
+                        + "\"cannot serialize field '"
+                        + fd.getName()
+                        + "': recursive proto2 message has required fields and was read with "
+                        + "read-default-values=true; set read-default-values=false, or ensure the "
+                        + "field is present\")");
+        appender.end("}");
+    }
+}

flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/serialize/PbCodegenSerializeFactory.java

@@ -23,16 +23,25 @@
 import org.apache.flink.formats.protobuf.util.PbFormatUtils;
 import org.apache.flink.table.types.logical.ArrayType;
 import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
 import org.apache.flink.table.types.logical.MapType;
 import org.apache.flink.table.types.logical.RowType;
 
 import com.google.protobuf.Descriptors;
+import com.google.protobuf.Descriptors.FieldDescriptor.JavaType;
 
 /** Codegen factory class which return {@link PbCodegenSerializer} of different data type. */
 public class PbCodegenSerializeFactory {
     public static PbCodegenSerializer getPbCodegenSer(
             Descriptors.FieldDescriptor fd, LogicalType type, PbFormatContext formatContext)
             throws PbCodegenException {
+        // Recursive message type represented as raw bytes: the proto field is a MESSAGE but the
+        // logical type was resolved to VARBINARY by PbToRowTypeUtil's cycle detection. Mirrors the
+        // matching branch in PbCodegenDeserializeFactory.
+        if (fd.getJavaType() == JavaType.MESSAGE
+                && type.getTypeRoot() == LogicalTypeRoot.VARBINARY) {
+            return new PbCodegenBytesSerializer(fd);
+        }
         if (type instanceof RowType) {
             return new PbCodegenRowSerializer(fd.getMessageType(), (RowType) type, formatContext);
         } else if (PbFormatUtils.isSimpleType(type)) {

flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/util/PbSchemaValidationUtils.java

@@ -100,7 +100,11 @@ private static void validateTypeMatch(FieldDescriptor fd, LogicalType logicalTyp
                 // simple type
                 validateSimpleType(fd, logicalType.getTypeRoot());
             } else {
-                // message type
+                // message type - may be RowType (normal) or VarBinaryType (recursive, as bytes)
+                if (logicalType.getTypeRoot() == LogicalTypeRoot.VARBINARY) {
+                    // Recursive message type represented as raw bytes - valid mapping
+                    return;
+                }
                 if (!(logicalType instanceof RowType)) {
                     throw new ValidationException(
                             "Unexpected LogicalType: " + logicalType + ". It should be RowType");
@@ -131,6 +135,10 @@ private static void validateTypeMatch(FieldDescriptor fd, LogicalType logicalTyp
                 if (fd.getJavaType() == JavaType.MESSAGE) {
                     // array message type
                     LogicalType elementType = arrayType.getElementType();
+                    if (elementType.getTypeRoot() == LogicalTypeRoot.VARBINARY) {
+                        // Recursive message type as raw bytes - valid
+                        return;
+                    }
                     if (!(elementType instanceof RowType)) {
                         throw new ValidationException(
                                 "Unexpected logicalType: "

flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/util/PbToRowTypeUtil.java

@@ -36,27 +36,53 @@
 import com.google.protobuf.Descriptors.FieldDescriptor;
 import com.google.protobuf.Descriptors.FieldDescriptor.JavaType;
 
+import java.util.HashSet;
+import java.util.Set;
+
 /** Generate Row type information according to pb descriptors. */
 public class PbToRowTypeUtil {
     public static RowType generateRowType(Descriptors.Descriptor root) {
         return generateRowType(root, false);
     }
 
     public static RowType generateRowType(Descriptors.Descriptor root, boolean enumAsInt) {
+        // Track message types currently being resolved in the ancestor chain to detect
+        // recursive references (e.g., A -> B -> A). Without this, recursive proto
+        // definitions cause infinite recursion and StackOverflowError.
+        Set<String> ancestors = new HashSet<>();
+        return generateRowTypeInternal(root, enumAsInt, ancestors);
+    }
+
+    /**
+     * @param ancestors message type full names currently being resolved in the call stack. Used to
+     *     detect cycles: if a field's message type is already in this set, it's a recursive
+     *     reference and gets emitted as BYTES instead of recursing infinitely.
+     */
+    private static RowType generateRowTypeInternal(
+            Descriptors.Descriptor root, boolean enumAsInt, Set<String> ancestors) {
         int size = root.getFields().size();
         LogicalType[] types = new LogicalType[size];
         String[] rowFieldNames = new String[size];
 
-        for (int i = 0; i < size; i++) {
-            FieldDescriptor field = root.getFields().get(i);
-            rowFieldNames[i] = field.getName();
-            types[i] = generateFieldTypeInformation(field, enumAsInt);
+        // Mark this type as "being resolved" before processing its fields
+        String fullName = root.getFullName();
+        ancestors.add(fullName);
+
+        try {
+            for (int i = 0; i < size; i++) {
+                FieldDescriptor field = root.getFields().get(i);
+                rowFieldNames[i] = field.getName();
+                types[i] = generateFieldTypeInformation(field, enumAsInt, ancestors);
+            }
+        } finally {
+            // Unmark when we're done - sibling branches shouldn't see this type as an ancestor
+            ancestors.remove(fullName);
         }
         return RowType.of(types, rowFieldNames);
     }
 
     private static LogicalType generateFieldTypeInformation(
-            FieldDescriptor field, boolean enumAsInt) {
+            FieldDescriptor field, boolean enumAsInt, Set<String> ancestors) {
         JavaType fieldType = field.getJavaType();
         LogicalType type;
         if (fieldType.equals(JavaType.MESSAGE)) {
@@ -66,16 +92,36 @@ private static LogicalType generateFieldTypeInformation(
                                 generateFieldTypeInformation(
                                         field.getMessageType()
                                                 .findFieldByName(PbConstant.PB_MAP_KEY_NAME),
-                                        enumAsInt),
+                                        enumAsInt,
+                                        ancestors),
                                 generateFieldTypeInformation(
                                         field.getMessageType()
                                                 .findFieldByName(PbConstant.PB_MAP_VALUE_NAME),
-                                        enumAsInt));
+                                        enumAsInt,
+                                        ancestors));
                 return mapType;
-            } else if (field.isRepeated()) {
-                return new ArrayType(generateRowType(field.getMessageType()));
+            }
+
+            // Cycle detection: if this field's message type is already being resolved
+            // in the ancestor chain, we have a recursive proto definition
+            // (e.g., Node -> Child -> Node). Columnar schemas cannot represent
+            // infinite recursion, so we emit the field as raw BYTES. The protobuf
+            // binary is preserved and can be deserialized on demand if consumers
+            // need the recursive data.
+            String msgFullName = field.getMessageType().getFullName();
+            if (ancestors.contains(msgFullName)) {
+                LogicalType bytesType = new VarBinaryType(Integer.MAX_VALUE);
+                if (field.isRepeated()) {
+                    return new ArrayType(bytesType);
+                }
+                return bytesType;
+            }
+
+            if (field.isRepeated()) {
+                return new ArrayType(
+                        generateRowTypeInternal(field.getMessageType(), enumAsInt, ancestors));
             } else {
-                return generateRowType(field.getMessageType());
+                return generateRowTypeInternal(field.getMessageType(), enumAsInt, ancestors);
             }
         } else {
             if (fieldType.equals(JavaType.STRING)) {