Store cause for invalid field exceptions

multi-stage-query/src/main/java/org/apache/druid/msq/querykit/scan/ScanQueryFrameProcessor.java

@@ -433,6 +433,7 @@ private void populateFrameWriterAndFlushIfNeededWithExceptionHandling()
       throw
           builder.source(ParseExceptionUtils.generateReadableInputSourceNameFromMappedSegment(this.segment)) // frame segment
                  .rowNumber(this.cursorOffset.getOffset() + 1)
+                 .cause(ffwe.getCause())
                  .build();
     }
     catch (Exception e) {

processing/src/main/java/org/apache/druid/frame/write/InvalidFieldException.java

@@ -122,6 +122,8 @@ public static class Builder
     private String source;
     @Nullable
     private String errorMsg;
+    @Nullable
+    private Throwable cause;
 
     public Builder()
     {
@@ -133,11 +135,16 @@ public Builder(InvalidFieldException invalidFieldException)
       rowNumber = invalidFieldException.rowNumber;
       column = invalidFieldException.column;
       errorMsg = invalidFieldException.errorMsg;
+      cause = invalidFieldException.getCause();
     }
 
     public InvalidFieldException build()
     {
-      return new InvalidFieldException(source, column, rowNumber, errorMsg);
+      InvalidFieldException invalidFieldException = new InvalidFieldException(source, column, rowNumber, errorMsg);
+      if (cause != null) {
+        invalidFieldException.initCause(cause);
+      }
+      return invalidFieldException;
     }
 
     public Builder column(String val)
@@ -163,5 +170,11 @@ public Builder errorMsg(String val)
       errorMsg = val;
       return this;
     }
+
+    public Builder cause(Throwable val)
+    {
+      cause = val;
+      return this;
+    }
   }
 }

processing/src/main/java/org/apache/druid/frame/write/RowBasedFrameWriter.java

@@ -310,6 +310,7 @@ private boolean writeDataUsingFieldWriters()
       catch (Exception e) {
         throw InvalidFieldException.builder().column(signature.getColumnName(i))
                                    .errorMsg(e.getMessage())
+                                   .cause(e)
                                    .build();
       }
 

processing/src/test/java/org/apache/druid/frame/write/RowBasedFrameWriterTest.java

@@ -98,12 +98,14 @@ public void test_addSelection_withException()
 
     final RowSignature signature = RowSignature.builder().add(colName, ColumnType.LONG).build();
 
+    RuntimeException realException = new RuntimeException(errorMsg);
+
     LongFieldWriter fieldWriter = EasyMock.mock(LongFieldWriter.class);
     EasyMock.expect(fieldWriter.writeTo(
         EasyMock.anyObject(),
         EasyMock.anyLong(),
         EasyMock.anyLong()
-    )).andThrow(new RuntimeException(errorMsg));
+    )).andThrow(realException);
 
     EasyMock.replay(fieldWriter);
 
@@ -128,5 +130,6 @@ public void test_addSelection_withException()
         rowBasedFrameWriter::addSelection
     );
     Assert.assertEquals(expectedException, actualException);
+    Assert.assertEquals(realException, actualException.getCause());
   }
 }