[Analytics Backend / DataFusion] Don't let substrait AssertionError kill the cluster

Substrait's plan validators (VariadicParameterConsistencyValidator,
RelOptUtil.eq via Litmus.THROW, etc.) throw AssertionError directly via
explicit `throw new AssertionError(...)` rather than via the `assert`
keyword, so the JVM -da flag doesn't gate them. When a malformed plan
triggers one inside a search-thread call to SubstraitRelVisitor.apply,
the AssertionError propagates uncaught up the analytics-engine fragment
handler stack, OpenSearchUncaughtExceptionHandler classifies it as fatal,
and the entire cluster JVM exits.

Wrap the visitor.apply call in a narrow try/catch that re-raises the
AssertionError as IllegalStateException with the original message and
cause preserved. The analytics-engine error path already buckets
IllegalStateException at the fragment boundary into a normal HTTP 500
response — the cluster stays up and the failure shows in the per-query
report instead.

This came up while diagnosing CalciteMVAppendFunctionIT failures: malformed
ARRAY<ANY> plans were taking down the cluster mid-test instead of producing
per-test failures, masking the underlying substrait conversion error.

Signed-off-by: Kai Huang <ahkcs@amazon.com>

Author: ahkcs

sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionFragmentConvertor.java

@@ -230,7 +230,20 @@ private byte[] convertToSubstrait(RelNode fragment) {
         RelNode preprocessed = UntypedNullPreprocessor.rewrite(fragment);
         RelRoot root = RelRoot.of(preprocessed, SqlKind.SELECT);
         SubstraitRelVisitor visitor = createVisitor(preprocessed);
-        Rel substraitRel = visitor.apply(root.rel);
+        Rel substraitRel;
+        try {
+            substraitRel = visitor.apply(root.rel);
+        } catch (AssertionError e) {
+            // Substrait validators (e.g. VariadicParameterConsistencyValidator,
+            // RelOptUtil.eq via Litmus.THROW) throw AssertionError directly via Java
+            // code rather than via the `assert` keyword, so JVM -da doesn't gate them.
+            // If one fires inside a search thread, OpenSearchUncaughtExceptionHandler
+            // exits the cluster JVM. Convert to IllegalStateException so the analytics-
+            // engine error path treats it as a normal per-query failure (HTTP 500 with
+            // a bucketable message) instead of taking down the cluster.
+            throw new IllegalStateException(
+                "Substrait conversion rejected the plan: " + e.getMessage(), e);
+        }
 
         List<String> fieldNames = root.fields.stream().map(field -> field.getValue()).toList();