feat: MSQ: Log full stack trace when "debug" is set.

multi-stage-query/src/main/java/org/apache/druid/msq/dart/controller/DartControllerContext.java

@@ -254,4 +254,10 @@ public int targetPartitionsPerWorker()
         DEFAULT_TARGET_PARTITIONS_PER_WORKER
     );
   }
+
+  @Override
+  public boolean isDebug()
+  {
+    return context.isDebug();
+  }
 }

multi-stage-query/src/main/java/org/apache/druid/msq/dart/worker/DartWorkerContext.java

@@ -274,6 +274,12 @@ public boolean includeAllCounters()
     return true;
   }
 
+  @Override
+  public boolean isDebug()
+  {
+    return queryContext.isDebug();
+  }
+
   @Override
   public DruidNode selfNode()
   {

multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerContext.java

@@ -135,4 +135,9 @@ default File taskTempDir()
    * shuffle specs that have {@link ShuffleSpec#isAdjustable()} set to true.
    */
   int targetPartitionsPerWorker();
+
+  /**
+   * Whether the controller should log full stack traces on error.
+   */
+  boolean isDebug();
 }

multi-stage-query/src/main/java/org/apache/druid/msq/exec/ControllerImpl.java

@@ -487,11 +487,11 @@ private MSQTaskReportPayload runInternal(final QueryListener queryListener, fina
 
       // Log the errors we encountered.
       if (controllerError != null) {
-        log.warn("Controller: %s", MSQTasks.errorReportToLogMessage(controllerError));
+        log.warn("Controller: %s", MSQTasks.errorReportToLogMessage(controllerError, context.isDebug()));
       }
 
       if (workerError != null) {
-        log.warn("Worker: %s", MSQTasks.errorReportToLogMessage(workerError));
+        log.warn("Worker: %s", MSQTasks.errorReportToLogMessage(workerError, context.isDebug()));
       }
     }
     if (queryKernel != null && queryKernel.isSuccess()) {

multi-stage-query/src/main/java/org/apache/druid/msq/exec/MSQTasks.java

@@ -203,9 +203,10 @@ static MSQErrorReport makeErrorReport(
   }
 
   /**
-   * Returns a string form of a {@link MSQErrorReport} suitable for logging.
+   * Returns a string form of a {@link MSQErrorReport} suitable for logging. When {@code forceFullStackTrace} is true,
+   * the full stack trace is always included (when available), regardless of fault type.
    */
-  static String errorReportToLogMessage(final MSQErrorReport errorReport)
+  static String errorReportToLogMessage(final MSQErrorReport errorReport, final boolean forceFullStackTrace)
   {
     final StringBuilder logMessage = new StringBuilder("Work failed");
 
@@ -222,7 +223,7 @@ static String errorReportToLogMessage(final MSQErrorReport errorReport)
     logMessage.append(": ").append(MSQFaultUtils.generateMessageWithErrorCode(errorReport.getFault()));
 
     if (errorReport.getExceptionStackTrace() != null) {
-      if (logFullStackTrace(errorReport.getFault())) {
+      if (forceFullStackTrace || logFullStackTrace(errorReport.getFault())) {
         logMessage.append('\n').append(errorReport.getExceptionStackTrace());
       } else {
         // Log first line only (error class, message) for known faults, to avoid polluting logs.

multi-stage-query/src/main/java/org/apache/druid/msq/exec/WorkerContext.java

@@ -116,6 +116,11 @@ public interface WorkerContext extends Closeable
    */
   boolean includeAllCounters();
 
+  /**
+   * Whether to log full stack traces for all errors.
+   */
+  boolean isDebug();
+
   @Override
   void close();
 }

multi-stage-query/src/main/java/org/apache/druid/msq/exec/WorkerImpl.java

@@ -195,7 +195,7 @@ public void run()
 
       if (maybeErrorReport.isPresent()) {
         final MSQErrorReport errorReport = maybeErrorReport.get();
-        final String logMessage = MSQTasks.errorReportToLogMessage(errorReport);
+        final String logMessage = MSQTasks.errorReportToLogMessage(errorReport, context.isDebug());
         log.warn("%s", logMessage);
 
         // Inform controller of any errors that occur, unless we were canceled. This prevents attempting to contact

multi-stage-query/src/main/java/org/apache/druid/msq/indexing/IndexerControllerContext.java

@@ -259,6 +259,12 @@ public int targetPartitionsPerWorker()
     );
   }
 
+  @Override
+  public boolean isDebug()
+  {
+    return taskQuerySpecContext.isDebug();
+  }
+
   /**
    * Helper method for {@link #queryKernelConfig(MSQSpec)}. Also used in tests.
    */

multi-stage-query/src/main/java/org/apache/druid/msq/indexing/IndexerWorkerContext.java

@@ -93,6 +93,7 @@ public class IndexerWorkerContext implements WorkerContext
   private final int maxConcurrentStages;
   private final boolean liveReportCounters;
   private final boolean includeAllCounters;
+  private final boolean debug;
   private final int threadCount;
 
   // Written under synchronized(this) using double-checked locking.
@@ -134,6 +135,7 @@ public IndexerWorkerContext(
     );
     this.liveReportCounters = MultiStageQueryContext.getLiveReportCounters(queryContext, DEFAULT_LIVE_REPORT_COUNTERS);
     this.includeAllCounters = MultiStageQueryContext.getIncludeAllCounters(queryContext);
+    this.debug = queryContext.isDebug();
 
     // Compute thread count once in constructor
     final int baseThreadCount = memoryIntrospector.numProcessingThreads();
@@ -327,6 +329,12 @@ public boolean includeAllCounters()
     return includeAllCounters;
   }
 
+  @Override
+  public boolean isDebug()
+  {
+    return debug;
+  }
+
   public ServiceLocator controllerLocator()
   {
     return controllerLocator;

multi-stage-query/src/main/java/org/apache/druid/msq/indexing/error/DruidExceptionFault.java

@@ -106,6 +106,7 @@ public DruidException toDruidException()
     return DruidException.forPersona(personaEnum)
                          .ofCategory(categoryEnum)
                          .withErrorCode(druidErrorCode)
+                         .wasDeserialized()
                          .build(getErrorMessage())
                          .withContext(context);
   }

multi-stage-query/src/test/java/org/apache/druid/msq/test/MSQTestControllerContext.java

@@ -485,6 +485,12 @@ public int targetPartitionsPerWorker()
     return 1;
   }
 
+  @Override
+  public boolean isDebug()
+  {
+    return true;
+  }
+
   @Override
   public ControllerContext newContext(QueryContext context)
   {

multi-stage-query/src/test/java/org/apache/druid/msq/test/MSQTestWorkerContext.java

@@ -206,6 +206,12 @@ public boolean includeAllCounters()
     return true;
   }
 
+  @Override
+  public boolean isDebug()
+  {
+    return true;
+  }
+
   @Override
   public void close()
   {

multi-stage-query/src/test/java/org/apache/druid/msq/test/TestDartControllerContextFactoryImpl.java

@@ -108,6 +108,12 @@ public void emitMetric(MSQMetricEventBuilder metricBuilder)
       {
         serviceEmitter.emit(metricBuilder.build("controller", queryId()));
       }
+
+      @Override
+      public boolean isDebug()
+      {
+        return true;
+      }
     };
   }
 

processing/src/main/java/org/apache/druid/error/DruidException.java

@@ -477,7 +477,7 @@ public DruidExceptionBuilder ofCategory(Category category)
      *
      * @return the builder
      */
-    DruidExceptionBuilder wasDeserialized()
+    public DruidExceptionBuilder wasDeserialized()
     {
       this.deserialized = true;
       return this;