title: Apache Drill: Write general-purpose array_agg UDF
body: <p>I would like to create an <code>array_agg</code> UDF for Apache Drill to be able to aggregate all values of a group to a list of values.
  This should work with any major types (required, optional) and minor types (varchar, dict, map, int, etc.)</p>
  <p>However, I get the impression that Apache Drill's UDF API does not really make use of inheritance and generics. Each type has its own writer and handler, and they cannot be abstracted to handle any type. E.g., the <code>ValueHolder</code> interface seems to be purely cosmetic and cannot be used to have type-agnostic hooking of UDFs to any type.</p>
  <h2>My current implementation</h2>
  <p>I tried to solve this by using Java's reflection so I could use the <code>ListHolder</code>'s write function independent of the holder of the original value.</p>
  <p>However, I then ran into the limitations of the <code>@FunctionTemplate</code> annotation.
  I cannot create a general UDF annotation for any value (I tried it with the interface <code>ValueHolder</code>: <code>@param ValueHolder input</code>.</p>
  <p>So to me it seems like the only way to support different types to have separate classes for each type. But I can't even abstract much and work on any <code>@Param input</code>, because <code>input</code> is only visible in the class where its defined (i.e. type specific).</p>
  <p>I based my implementation on <a href="https://issues.apache.org/jira/browse/DRILL-6963" rel="nofollow noreferrer">https://issues.apache.org/jira/browse/DRILL-6963</a>
  and created the following two classes for required and optional varchars (how can this be unified in the first place?)</p>
  <pre class="lang-java prettyprint-override"><code>@FunctionTemplate(
      name = &quot;array_agg&quot;,
      scope = FunctionScope.POINT_AGGREGATE,
      nulls = NullHandling.INTERNAL
  )
  public static class VarChar_Agg implements DrillAggFunc {
      @Param org.apache.drill.exec.expr.holders.VarCharHolder input;
      @Workspace ObjectHolder agg;
      @Output org.apache.drill.exec.vector.complex.writer.BaseWriter.ComplexWriter out;
  
      @Override
      public void setup() {
          agg = new ObjectHolder();
      }
  
      @Override
      public void reset() {
          agg = new ObjectHolder();
      }
  
      @Override public void add() {
          if (agg.obj == null) {
              // Initialise list object for output
              agg.obj = out.rootAsList();
          }
  
          org.apache.drill.exec.vector.complex.writer.BaseWriter.ListWriter listWriter =
                  (org.apache.drill.exec.vector.complex.writer.BaseWriter.ListWriter)agg.obj;
  
          listWriter.varChar().write(input);
      }
  
      @Override
      public void output() {
          ((org.apache.drill.exec.vector.complex.writer.BaseWriter.ListWriter)agg.obj).endList();
      }
  }
  
  @FunctionTemplate(
      name = &quot;array_agg&quot;,
      scope = FunctionScope.POINT_AGGREGATE,
      nulls = NullHandling.INTERNAL
  )
  public static class NullableVarChar_Agg implements DrillAggFunc {
      @Param NullableVarCharHolder input;
      @Workspace ObjectHolder agg;
      @Output org.apache.drill.exec.vector.complex.writer.BaseWriter.ComplexWriter out;
  
      @Override
      public void setup() {
          agg = new ObjectHolder();
      }
  
      @Override
      public void reset() {
          agg = new ObjectHolder();
      }
  
      @Override public void add() {
          if (agg.obj == null) {
              // Initialise list object for output
              agg.obj = out.rootAsList();
          }
  
          if (input.isSet != 1) {
              return;
          }
  
          org.apache.drill.exec.vector.complex.writer.BaseWriter.ListWriter listWriter =
                  (org.apache.drill.exec.vector.complex.writer.BaseWriter.ListWriter)agg.obj;
  
          org.apache.drill.exec.expr.holders.VarCharHolder outHolder = new org.apache.drill.exec.expr.holders.VarCharHolder();
          outHolder.start = input.start;
          outHolder.end = input.end;
          outHolder.buffer = input.buffer;
  
          listWriter.varChar().write(outHolder);
      }
  
      @Override
      public void output() {
          ((org.apache.drill.exec.vector.complex.writer.BaseWriter.ListWriter)agg.obj).endList();
      }
  }
  </code></pre>
  <p>Interestingly, I can't import <code>org.apache.drill.exec.vector.complex.writer.BaseWriter</code> to make the whole thing easier because then Apache Drill would not find it.
  So I have to put the entire package path for everything in <code>org.apache.drill.exec.vector.complex.writer</code> in the code.
  Furthermore, I'm using the depcreated ObjectHolder. Any better solution?</p>
  <p>Anyway: These work so far, e.g. with this query:</p>
  <pre><code>SELECT
      MIN(tbl.`timestamp`) AS start_view,
      MAX(tbl.`timestamp`) AS end_view,
      array_agg(tbl.eventLabel) AS label_agg
  FROM `dfs.root`.`/path/to/avro/folder` AS tbl
  WHERE tbl.data.slug IS NOT NULL
  GROUP BY tbl.data.slug
  </code></pre>
  <p>however, when I use <code>ORDER BY</code>, I get this:</p>
  <pre><code>org.apache.drill.common.exceptions.UserRemoteException: SYSTEM ERROR: UnsupportedOperationException: NULL
  
  Fragment 0:0
  </code></pre>
  <p>Additionally, I tried more complex types, namely maps/dicts.
  Interestingly, when I call <code>SELECT sqlTypeOf(tbl.data) FROM tbl</code>, I get MAP.
  But when I write UDFs, the query planner complains about having no UDF <code>array_agg</code> for type <code>dict</code>.</p>
  <p>Anyway, I wrote a version for dicts:</p>
  <pre class="lang-java prettyprint-override"><code>@FunctionTemplate(
      name = &quot;array_agg&quot;,
      scope = FunctionScope.POINT_AGGREGATE,
      nulls = NullHandling.INTERNAL
  )
  public static class Map_Agg implements DrillAggFunc {
      @Param MapHolder input;
      @Workspace ObjectHolder agg;
      @Output org.apache.drill.exec.vector.complex.writer.BaseWriter.ComplexWriter out;
  
      @Override
      public void setup() {
          agg = new ObjectHolder();
      }
  
      @Override
      public void reset() {
          agg = new ObjectHolder();
      }
  
      @Override public void add() {
          if (agg.obj == null) {
              // Initialise list object for output
              agg.obj = out.rootAsList();
          }
  
          org.apache.drill.exec.vector.complex.writer.BaseWriter.ListWriter listWriter =
                  (org.apache.drill.exec.vector.complex.writer.BaseWriter.ListWriter) agg.obj;
          //listWriter.copyReader(input.reader);
          input.reader.copyAsValue(listWriter);
      }
  
      @Override
      public void output() {
          ((org.apache.drill.exec.vector.complex.writer.BaseWriter.ListWriter)agg.obj).endList();
      }
  }
  
  @FunctionTemplate(
          name = &quot;array_agg&quot;,
          scope = FunctionScope.POINT_AGGREGATE,
          nulls = NullHandling.INTERNAL
  )
  public static class Dict_agg implements DrillAggFunc {
      @Param DictHolder input;
      @Workspace ObjectHolder agg;
      @Output org.apache.drill.exec.vector.complex.writer.BaseWriter.ComplexWriter out;
  
      @Override
      public void setup() {
          agg = new ObjectHolder();
      }
  
      @Override
      public void reset() {
          agg = new ObjectHolder();
      }
  
      @Override public void add() {
          if (agg.obj == null) {
              // Initialise list object for output
              agg.obj = out.rootAsList();
          }
  
          org.apache.drill.exec.vector.complex.writer.BaseWriter.ListWriter listWriter =
                  (org.apache.drill.exec.vector.complex.writer.BaseWriter.ListWriter) agg.obj;
          //listWriter.copyReader(input.reader);
          input.reader.copyAsValue(listWriter);
      }
  
      @Override
      public void output() {
          ((org.apache.drill.exec.vector.complex.writer.BaseWriter.ListWriter)agg.obj).endList();
      }
  }
  </code></pre>
  <p>But here, I get an empty list in the field <code>data_agg</code> for my query:</p>
  <pre><code>SELECT
      MIN(tbl.`timestamp`) AS start_view,
      MAX(tbl.`timestamp`) AS end_view,
      array_agg(tbl.data) AS data_agg
  FROM `dfs.root`.`/path/to/avro/folder` AS tbl
  GROUP BY tbl.data.viewSlag
  </code></pre>
  <h2>Summary of questions</h2>
  <ul>
  <li><strong>Most importantly: How do I create an <code>array_agg</code> UDF for Apache Drill?</strong></li>
  <li>How to make UDFs type-agnostic/general purpose? Do I really have to implement an entire class for each Nullable, Required and Repeated version of all types? That's a lot to do and quite tedious. Isn't there a way to handle values in an UDF agnostic to the underlying types?
  I wish Apache Drill would just use what Java offers here with function generic types, specialised function overloading and inheritence of their own type system. Am I missing something on how to do that?</li>
  <li>How can I fix the NULL problem when I use <code>ORDER BY</code> on my varchar version of the aggregate?</li>
  <li>How can I fix the problem where my aggregate of maps/dicts is an empty list?</li>
  <li>Is there an alternative to using the deprecated <code>ObjectHolder</code>?</li>
  </ul>