update

Author: ArnavBalyan

parquet-avro/src/main/java/org/apache/parquet/avro/AvroSchemaConverter.java

@@ -83,26 +83,12 @@ public class AvroSchemaConverter {
 
   public static final String ADD_LIST_ELEMENT_RECORDS = "parquet.avro.add-list-element-records";
   private static final boolean ADD_LIST_ELEMENT_RECORDS_DEFAULT = true;
-  public static final String AVRO_MAX_RECURSION = "parquet.avro.max-recursion";
-  private static final int AVRO_MAX_RECURSION_DEFAULT = 10;
 
   private final boolean assumeRepeatedIsListElement;
   private final boolean writeOldListStructure;
   private final boolean writeParquetUUID;
   private final boolean readInt96AsFixed;
   private final Set<String> pathsToInt96;
-  private final int maxRecursion;
-
-  /**
-   * Sets the maximum recursion depth for recursive schemas.
-   *
-   * @param config       The hadoop configuration to be updated.
-   * @param maxRecursion The maximum recursion depth schemas are allowed to go before terminating
-   *                     with an UnsupportedOperationException instead of their actual schema.
-   */
-  public static void setMaxRecursion(Configuration config, int maxRecursion) {
-    config.setInt(AVRO_MAX_RECURSION, maxRecursion);
-  }
 
   public AvroSchemaConverter() {
     this(ADD_LIST_ELEMENT_RECORDS_DEFAULT, READ_INT96_AS_FIXED_DEFAULT);
@@ -121,7 +107,6 @@ public AvroSchemaConverter() {
     this.writeParquetUUID = WRITE_PARQUET_UUID_DEFAULT;
     this.readInt96AsFixed = readInt96AsFixed;
     this.pathsToInt96 = Collections.emptySet();
-    this.maxRecursion = AVRO_MAX_RECURSION_DEFAULT;
   }
 
   public AvroSchemaConverter(Configuration conf) {
@@ -134,7 +119,6 @@ public AvroSchemaConverter(ParquetConfiguration conf) {
     this.writeParquetUUID = conf.getBoolean(WRITE_PARQUET_UUID, WRITE_PARQUET_UUID_DEFAULT);
     this.readInt96AsFixed = conf.getBoolean(READ_INT96_AS_FIXED, READ_INT96_AS_FIXED_DEFAULT);
     this.pathsToInt96 = new HashSet<>(Arrays.asList(conf.getStrings(WRITE_FIXED_AS_INT96, new String[0])));
-    this.maxRecursion = conf.getInt(AVRO_MAX_RECURSION, AVRO_MAX_RECURSION_DEFAULT);
   }
 
   /**
@@ -213,13 +197,13 @@ private Type convertField(
     LogicalType logicalType = schema.getLogicalType();
 
     if (type.equals(Schema.Type.RECORD) || type.equals(Schema.Type.ENUM) || type.equals(Schema.Type.FIXED)) {
-      Integer depth = seenSchemas.get(schema);
-      if (depth != null && depth >= maxRecursion) {
-        throw new UnsupportedOperationException("Recursive Avro schemas are not supported by parquet-avro: "
-            + schema.getFullName() + " (exceeded maximum recursion depth of " + maxRecursion + ")");
+      // If this schema has already been seen in the current branch, we have a recursion loop
+      if (seenSchemas.containsKey(schema)) {
+        throw new UnsupportedOperationException(
+            "Recursive Avro schemas are not supported by parquet-avro: " + schema.getFullName());
       }
       seenSchemas = new IdentityHashMap<>(seenSchemas);
-      seenSchemas.put(schema, depth == null ? 1 : depth + 1);
+      seenSchemas.put(schema, 1);
     }
 
     Types.PrimitiveBuilder<PrimitiveType> builder;

parquet-avro/src/test/java/org/apache/parquet/avro/TestAvroSchemaConverter.java

@@ -980,7 +980,7 @@ public void testRecursiveSchemaThrowsException() {
     Schema recursiveSchema = new Schema.Parser().parse(recursiveSchemaJson);
 
     assertThrows(
-        "Recursive Avro schema should throw UnsupportedOperationException",
+        "Recursive Avro schema should throw UnsupportedOperationException for cycles",
         UnsupportedOperationException.class,
         () -> new AvroSchemaConverter().convert(recursiveSchema));
   }
@@ -1011,7 +1011,7 @@ public void testRecursiveSchemaFromGitHubIssue() {
     Schema issueSchema = new Schema.Parser().parse(issueSchemaJson);
 
     assertThrows(
-        "Schema should throw UnsupportedOperationException",
+        "Schema hould throw UnsupportedOperationException for cycles",
         UnsupportedOperationException.class,
         () -> new AvroSchemaConverter().convert(issueSchema));
   }
@@ -1025,41 +1025,11 @@ public void testRecursiveSchemaErrorMessage() {
 
     Schema recursiveSchema = new Schema.Parser().parse(recursiveSchemaJson);
 
-    try {
-      new AvroSchemaConverter().convert(recursiveSchema);
-      Assert.fail("Expected UnsupportedOperationException");
-    } catch (UnsupportedOperationException e) {
-      String message = e.getMessage();
-      Assert.assertTrue(
-          "Error message should mention recursion",
-          message.contains("Recursive Avro schemas are not supported"));
-      Assert.assertTrue("Error message should mention schema name", message.contains("TestRecord"));
-      Assert.assertTrue(
-          "Error message should mention max recursion depth", message.contains("maximum recursion depth"));
-    }
-  }
-
-  @Test
-  public void testConfigurableMaxRecursion() {
-    String recursiveSchemaJson = "{"
-        + "\"type\": \"record\", \"name\": \"Node\", \"fields\": ["
-        + "  {\"name\": \"child\", \"type\": [\"null\", \"Node\"], \"default\": null}"
-        + "]}";
-
-    Schema recursiveSchema = new Schema.Parser().parse(recursiveSchemaJson);
-    Configuration conf = new Configuration();
-
-    AvroSchemaConverter.setMaxRecursion(conf, 1);
+    // With our cycle detection fix, this should throw UnsupportedOperationException
     assertThrows(
-        "Should fail with max recursion 1",
+        "Recursive schema should throw UnsupportedOperationException with clear error message",
         UnsupportedOperationException.class,
-        () -> new AvroSchemaConverter(conf).convert(recursiveSchema));
-
-    AvroSchemaConverter.setMaxRecursion(conf, 5);
-    assertThrows(
-        "Should fail with max recursion 5",
-        UnsupportedOperationException.class,
-        () -> new AvroSchemaConverter(conf).convert(recursiveSchema));
+        () -> new AvroSchemaConverter().convert(recursiveSchema));
   }
 
   @Test