feat: propagate Java stack traces from JVM upcalls into DataFusionError

core/src/main/java/org/apache/datafusion/ExceptionVerbosity.java

@@ -0,0 +1,61 @@
+/*
+ * 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.datafusion;
+
+/**
+ * Controls how detailed the representation of a Java throwable is when a JVM upcall (UDF, {@link
+ * TableProvider}) throws and the failure has to cross back into native code.
+ *
+ * <p>Set on {@link SessionContextBuilder#exceptionVerbosity}. The setting is locked at
+ * session-construction time and applies uniformly to every UDF and table provider registered
+ * against the session.
+ */
+public enum ExceptionVerbosity {
+  /** Class name + {@code getMessage()} + standard Java stack trace. The default. */
+  FULL,
+  /**
+   * Class name + {@code getMessage()} only. Matches the behaviour of `datafusion-java` before this
+   * setting existed; useful when full traces are too verbose for production logs.
+   */
+  MESSAGE,
+  /**
+   * Class name only. For callers who treat exception bodies as untrusted user input that shouldn't
+   * reach logs.
+   */
+  NONE;
+
+  /**
+   * Byte tag passed through {@code SessionContext.registerScalarUdf} / {@code registerTableNative}
+   * to the native side. {@code 1=FULL}, {@code 2=MESSAGE}, {@code 3=NONE}; {@code 0} is reserved as
+   * a "bogus" sentinel and rejected by the Rust decoder.
+   */
+  byte toByte() {
+    switch (this) {
+      case FULL:
+        return 1;
+      case MESSAGE:
+        return 2;
+      case NONE:
+        return 3;
+      default:
+        throw new IllegalStateException("unreachable: unknown ExceptionVerbosity " + this);
+    }
+  }
+}

core/src/main/java/org/apache/datafusion/SessionContext.java

@@ -49,19 +49,27 @@ public final class SessionContext implements AutoCloseable {
   }
 
   private long nativeHandle;
+  // Snapshot of the builder's exceptionVerbosity, passed through to UDF /
+  // TableProvider registrations. Locked at construction time; not mutable.
+  // Default FULL keeps stack traces flowing for callers that don't touch the
+  // setter. Stored as a byte to keep the JNI signature ABI-stable across any
+  // future ExceptionVerbosity additions.
+  private final byte exceptionVerbosity;
 
   public SessionContext() {
     this.nativeHandle = createSessionContext();
     if (this.nativeHandle == 0) {
       throw new RuntimeException("Failed to create native SessionContext");
     }
+    this.exceptionVerbosity = ExceptionVerbosity.FULL.toByte();
   }
 
-  SessionContext(byte[] optionsBytes) {
+  SessionContext(byte[] optionsBytes, ExceptionVerbosity exceptionVerbosity) {
     this.nativeHandle = createSessionContextWithOptions(optionsBytes);
     if (this.nativeHandle == 0) {
       throw new RuntimeException("Failed to create native SessionContext");
     }
+    this.exceptionVerbosity = exceptionVerbosity.toByte();
   }
 
   /** Start configuring a {@link SessionContext}. */
@@ -550,7 +558,8 @@ public void registerUdf(ScalarUdf udf) {
     fields.addAll(udf.argFields());
     Schema signatureSchema = new Schema(fields);
     byte[] signatureBytes = serializeSchemaIpc(signatureSchema);
-    registerScalarUdf(nativeHandle, name, signatureBytes, volatility.code(), impl);
+    registerScalarUdf(
+        nativeHandle, name, signatureBytes, volatility.code(), impl, exceptionVerbosity);
   }
 
   /**
@@ -585,7 +594,7 @@ public void registerTable(String name, TableProvider provider) {
       throw new IllegalStateException("TableProvider.schema returned null");
     }
     byte[] schemaIpc = serializeSchemaIpc(schema);
-    registerTableNative(nativeHandle, name, schemaIpc, provider);
+    registerTableNative(nativeHandle, name, schemaIpc, provider, exceptionVerbosity);
   }
 
   private static byte[] serializeSchemaIpc(Schema schema) {
@@ -660,8 +669,17 @@ private static native long readJsonWithOptions(
   private static native void closeSessionContext(long handle);
 
   private static native void registerScalarUdf(
-      long handle, String name, byte[] signatureSchemaBytes, byte volatility, ScalarFunction impl);
+      long handle,
+      String name,
+      byte[] signatureSchemaBytes,
+      byte volatility,
+      ScalarFunction impl,
+      byte exceptionVerbosity);
 
   private static native void registerTableNative(
-      long handle, String name, byte[] schemaIpcBytes, TableProvider provider);
+      long handle,
+      String name,
+      byte[] schemaIpcBytes,
+      TableProvider provider,
+      byte exceptionVerbosity);
 }

core/src/main/java/org/apache/datafusion/SessionContextBuilder.java

@@ -43,6 +43,7 @@ public final class SessionContextBuilder {
   private String tempDirectory;
   private boolean spillDisabled;
   private Long maxTempDirectorySize;
+  private ExceptionVerbosity exceptionVerbosity;
   private CacheManagerOptions cacheManager;
   private final LinkedHashMap<String, String> options = new LinkedHashMap<>();
   private final List<ObjectStoreOptions> objectStores = new ArrayList<>();
@@ -208,6 +209,24 @@ public SessionContextBuilder setOptions(Map<String, String> entries) {
     return this;
   }
 
+  /**
+   * Configure how detailed the representation of a Java throwable is when a JVM upcall (UDF, {@link
+   * TableProvider}) throws and the failure has to cross back into native code.
+   *
+   * <p>The setting is locked at session-construction time and applies uniformly to every UDF and
+   * table provider registered against the session. See {@link ExceptionVerbosity} for the three
+   * values; default (unset) is {@link ExceptionVerbosity#FULL}.
+   *
+   * @throws IllegalArgumentException if {@code verbosity} is {@code null}.
+   */
+  public SessionContextBuilder exceptionVerbosity(ExceptionVerbosity verbosity) {
+    if (verbosity == null) {
+      throw new IllegalArgumentException("exceptionVerbosity must be non-null");
+    }
+    this.exceptionVerbosity = verbosity;
+    return this;
+  }
+
   /**
    * Configure DataFusion's built-in {@code CacheManager} for the new context. Build the {@link
    * CacheManagerOptions} via {@link CacheManagerOptions#builder()}; each cache slot is independent,
@@ -264,7 +283,7 @@ public SessionContext build() {
       throw new IllegalStateException(
           "disableSpill() is mutually exclusive with tempDirectory(...)");
     }
-    return new SessionContext(toBytes());
+    return new SessionContext(toBytes(), exceptionVerbosityOrDefault());
   }
 
   byte[] toBytes() {
@@ -306,6 +325,9 @@ byte[] toBytes() {
     if (cacheManager != null) {
       b.setCacheManager(cacheManager.toProto());
     }
+    // exceptionVerbosity flows via SessionContext's snapshot field, not the
+    // SessionOptions proto -- it's only consumed at registerUdf /
+    // registerTable JNI calls, where it rides on its own arg.
     for (Map.Entry<String, String> e : options.entrySet()) {
       b.addOptions(ConfigOption.newBuilder().setKey(e.getKey()).setValue(e.getValue()).build());
     }
@@ -314,4 +336,9 @@ byte[] toBytes() {
     }
     return b.build().toByteArray();
   }
+
+  /** Visible for {@link SessionContext} so it can fall back to the FULL byte when unset. */
+  ExceptionVerbosity exceptionVerbosityOrDefault() {
+    return exceptionVerbosity != null ? exceptionVerbosity : ExceptionVerbosity.FULL;
+  }
 }