Spark: Support recursive delegate unwrapping to find ExtendedParser in parser chains

Author: majian1998

spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/ExtendedParser.java

@@ -52,10 +52,10 @@ public NullOrder nullOrder() {
   }
 
   static List<RawOrderField> parseSortOrder(SparkSession spark, String orderString) {
-    if (spark.sessionState().sqlParser() instanceof ExtendedParser) {
-      ExtendedParser parser = (ExtendedParser) spark.sessionState().sqlParser();
+    ExtendedParser extParser = findParser(spark.sessionState().sqlParser(), ExtendedParser.class);
+    if (extParser != null) {
       try {
-        return parser.parseSortOrder(orderString);
+        return extParser.parseSortOrder(orderString);
       } catch (AnalysisException e) {
         throw new IllegalArgumentException(
             String.format("Unable to parse sortOrder: %s", orderString), e);
@@ -66,5 +66,42 @@ static List<RawOrderField> parseSortOrder(SparkSession spark, String orderString
     }
   }
 
+  static <T> T findParser(ParserInterface parser, Class<T> clazz) {
+    ParserInterface current = parser;
+    while (current != null) {
+      if (clazz.isInstance(current)) {
+        return clazz.cast(current);
+      }
+      Object next = null;
+      try {
+        for (String methodName : new String[]{"delegate", "getDelegate"}) {
+          try {
+            java.lang.reflect.Method delegateMethod = current.getClass().getMethod(methodName);
+            next = delegateMethod.invoke(current);
+            break;
+          } catch (NoSuchMethodException ignore) {
+            // go on
+          }
+        }
+        if (next == null) {
+          try {
+            java.lang.reflect.Field delegateField = current.getClass().getDeclaredField("delegate");
+            delegateField.setAccessible(true);
+            next = delegateField.get(current);
+          } catch (NoSuchFieldException ignore) {
+            // No delegate
+          }
+        }
+        if (next == null || next == current) {
+          break;
+        }
+        current = (ParserInterface) next;
+      } catch (Exception e) {
+        break;
+      }
+    }
+    return null;
+  }
+
   List<RawOrderField> parseSortOrder(String orderString) throws AnalysisException;
 }