<h3 dir="auto">Motivation</h3>
<p dir="auto">Native indexing tasks currently support only text file formats because <code class="notranslate">FirehoseFactory</code> is tightly coupled with <code class="notranslate">InputRowParser</code> (<a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="311783745" data-permission-text="Title is private" data-url="https://github.com/apache/druid/issues/5584" data-hovercard-type="issue" data-hovercard-url="/apache/druid/issues/5584/hovercard" href="https://github.com/apache/druid/issues/5584">#5584</a>). Other issues around <code class="notranslate">FirehoseFactory</code> and <code class="notranslate">InputRowParser</code> are:</p>
<ul dir="auto">
<li>Sampler and indexing task use the same <code class="notranslate">Firehose</code> interface even though their use cases are pretty different.</li>
<li><code class="notranslate">InputRowParser</code> doesn't have to be exposed in the user spec.</li>
<li><code class="notranslate">Parser</code> is only for text file formats, but all <code class="notranslate">ParseSpec</code>s have <code class="notranslate">makeParser()</code>.</li>
<li>Parallel indexing tasks will need to be able to read a portion of a file for finer-grained parallelism in the future. To do this, the interface abstracting file formats should have a trait of <code class="notranslate">splittable</code>.</li>
<li>Kafka/Kinesis tasks can use only implementations of <code class="notranslate">ByteBufferInputRowParser</code>. This forces us to have duplicate implementations for batch and realtime tasks (<code class="notranslate">AvroHadoopInputRowParser</code> vs <code class="notranslate">AvroStreamInputRowParser</code>).</li>
</ul>
<p dir="auto">Since It's not easy to modify or improve existing interfaces without huge change, we need new interfaces that can be used instead of <code class="notranslate">FirehoseFactory</code>, <code class="notranslate">Firehose</code>, <code class="notranslate">InputRowParser</code>, and <code class="notranslate">ParseSpec</code>. The new interfaces should support the following storage types and file formats.</p>
<ul dir="auto">
<li>Storage types
<ul dir="auto">
<li>HDFS</li>
<li>cloud (s3, gcp, etc)</li>
<li>local</li>
<li>http</li>
<li>sql</li>
<li>byte (inline, kafka/kinesis tasks)</li>
<li>Druid (reingtestion)</li>
</ul>
</li>
<li>File formats
<ul dir="auto">
<li>csv</li>
<li>tsv</li>
<li>json</li>
<li>regex</li>
<li>influx</li>
<li>javascript</li>
<li>avro</li>
<li>orc</li>
<li>parquet</li>
<li>protobuf</li>
<li>thrift</li>
</ul>
</li>
</ul>
<h3 dir="auto">Proposed changes</h3>
<p dir="auto">The proposed new interfaces are:</p>
<h4 dir="auto"><code class="notranslate">InputSource</code></h4>
<p dir="auto"><code class="notranslate">InputSource</code> abstracts the storage where input data comes from for batch ingestion. This will replace <code class="notranslate">FiniteFirehoseFactory</code>.</p>
<div class="highlight highlight-source-java notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="public interface InputSource
{
  /**
   * Returns true if this inputSource can be processed in parallel using ParallelIndexSupervisorTask.
   */
  boolean isSplittable();

  /**
   * Returns true if this inputSource supports different {@link InputFormat}s.
   */
  boolean needsFormat();

  InputSourceReader reader(
      InputRowSchema inputRowSchema,
      @Nullable InputFormat inputFormat,
      @Nullable File temporaryDirectory
  );
}"><pre class="notranslate"><span class="pl-k">public</span> <span class="pl-k">interface</span> <span class="pl-smi">InputSource</span>
{
  <span class="pl-c">/**</span>
<span class="pl-c">   * Returns true if this inputSource can be processed in parallel using ParallelIndexSupervisorTask.</span>
<span class="pl-c">   */</span>
  <span class="pl-smi">boolean</span> <span class="pl-en">isSplittable</span>();

  <span class="pl-c">/**</span>
<span class="pl-c">   * Returns true if this inputSource supports different {@link InputFormat}s.</span>
<span class="pl-c">   */</span>
  <span class="pl-smi">boolean</span> <span class="pl-en">needsFormat</span>();

  <span class="pl-smi">InputSourceReader</span> <span class="pl-en">reader</span>(
      <span class="pl-smi">InputRowSchema</span> <span class="pl-s1">inputRowSchema</span>,
      <span class="pl-c1">@</span><span class="pl-c1">Nullable</span> <span class="pl-smi">InputFormat</span> <span class="pl-s1">inputFormat</span>,
      <span class="pl-c1">@</span><span class="pl-c1">Nullable</span> <span class="pl-smi">File</span> <span class="pl-s1">temporaryDirectory</span>
  );
}</pre></div>
<p dir="auto"><code class="notranslate">InputRowSchema</code> is the schema for <code class="notranslate">InputRow</code> to be created.</p>
<div class="highlight highlight-source-java notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="public class InputRowSchema
{
  private final TimestampSpec timestampSpec;
  private final DimensionsSpec dimensionsSpec;
  private final List&lt;String&gt; metricsNames;
}"><pre class="notranslate"><span class="pl-k">public</span> <span class="pl-k">class</span> <span class="pl-smi">InputRowSchema</span>
{
  <span class="pl-k">private</span> <span class="pl-k">final</span> <span class="pl-smi">TimestampSpec</span> <span class="pl-s1">timestampSpec</span>;
  <span class="pl-k">private</span> <span class="pl-k">final</span> <span class="pl-smi">DimensionsSpec</span> <span class="pl-s1">dimensionsSpec</span>;
  <span class="pl-k">private</span> <span class="pl-k">final</span> <span class="pl-smi">List</span>&lt;<span class="pl-smi">String</span>&gt; <span class="pl-s1">metricsNames</span>;
}</pre></div>
<p dir="auto"><code class="notranslate">SplittableSource</code> is the splittable <code class="notranslate">InputSource</code> that can be processed in parallel.</p>
<div class="highlight highlight-source-java notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="public interface SplittableInputSource&lt;T&gt; extends InputSource
{
  @JsonIgnore
  @Override
  default boolean isSplittable()
  {
    return true;
  }

  Stream&lt;InputSplit&lt;T&gt;&gt; createSplits(InputFormat inputFormat, @Nullable SplitHintSpec splitHintSpec) throws IOException;

  int getNumSplits(InputFormat inputFormat, @Nullable SplitHintSpec splitHintSpec) throws IOException;

  SplittableInputSource&lt;T&gt; withSplit(InputSplit&lt;T&gt; split);
}
"><pre class="notranslate"><span class="pl-k">public</span> <span class="pl-k">interface</span> <span class="pl-smi">SplittableInputSource</span>&lt;<span class="pl-smi">T</span>&gt; <span class="pl-k">extends</span> <span class="pl-smi">InputSource</span>
{
  <span class="pl-c1">@</span><span class="pl-c1">JsonIgnore</span>
  <span class="pl-c1">@</span><span class="pl-c1">Override</span>
  <span class="pl-k">default</span> <span class="pl-smi">boolean</span> <span class="pl-en">isSplittable</span>()
  {
    <span class="pl-k">return</span> <span class="pl-c1">true</span>;
  }

  <span class="pl-smi">Stream</span>&lt;<span class="pl-smi">InputSplit</span>&lt;<span class="pl-smi">T</span>&gt;&gt; <span class="pl-en">createSplits</span>(<span class="pl-smi">InputFormat</span> <span class="pl-s1">inputFormat</span>, <span class="pl-c1">@</span><span class="pl-c1">Nullable</span> <span class="pl-smi">SplitHintSpec</span> <span class="pl-s1">splitHintSpec</span>) <span class="pl-k">throws</span> <span class="pl-smi">IOException</span>;

  <span class="pl-smi">int</span> <span class="pl-en">getNumSplits</span>(<span class="pl-smi">InputFormat</span> <span class="pl-s1">inputFormat</span>, <span class="pl-c1">@</span><span class="pl-c1">Nullable</span> <span class="pl-smi">SplitHintSpec</span> <span class="pl-s1">splitHintSpec</span>) <span class="pl-k">throws</span> <span class="pl-smi">IOException</span>;

  <span class="pl-smi">SplittableInputSource</span>&lt;<span class="pl-smi">T</span>&gt; <span class="pl-en">withSplit</span>(<span class="pl-smi">InputSplit</span>&lt;<span class="pl-smi">T</span>&gt; <span class="pl-s1">split</span>);
}</pre></div>
<p dir="auto">Check <a href="https://github.com/jihoonson/druid/blob/file-format/core/src/main/java/org/apache/druid/data/input/HttpInputSource.java"><code class="notranslate">HttpInputSource</code></a> as an example.</p>
<h4 dir="auto"><code class="notranslate">InputSourceReader</code> and <code class="notranslate">InputSourceSampler</code></h4>
<p dir="auto">You can create <code class="notranslate">InputSourceReader</code> and <code class="notranslate">InputSourceSampler</code> from <code class="notranslate">InputSource</code>. <code class="notranslate">InputSourceReader</code> is for reading inputs and creating segments while <code class="notranslate">InputSourceSampler</code> is for sampling inputs.</p>
<div class="highlight highlight-source-java notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="public interface InputSourceReader
{
  CloseableIterator&lt;InputRow&gt; read() throws IOException;

  CloseableIterator&lt;InputRowPlusRaw&gt; sample() throws IOException;
}"><pre class="notranslate"><span class="pl-k">public</span> <span class="pl-k">interface</span> <span class="pl-smi">InputSourceReader</span>
{
  <span class="pl-smi">CloseableIterator</span>&lt;<span class="pl-smi">InputRow</span>&gt; <span class="pl-en">read</span>() <span class="pl-k">throws</span> <span class="pl-smi">IOException</span>;

  <span class="pl-smi">CloseableIterator</span>&lt;<span class="pl-smi">InputRowPlusRaw</span>&gt; <span class="pl-en">sample</span>() <span class="pl-k">throws</span> <span class="pl-smi">IOException</span>;
}</pre></div>
<p dir="auto">These reader and sampler are the interfaces what users will use directly. <a href="https://github.com/jihoonson/druid/blob/file-format/core/src/main/java/org/apache/druid/data/input/impl/SplitIteratingReader.java"><code class="notranslate">SplitIteratingReader</code></a> is an example of <code class="notranslate">InputSourceReader</code>.</p>
<h4 dir="auto"><code class="notranslate">ObjectSource</code> and <code class="notranslate">InputFormat</code></h4>
<p dir="auto"><code class="notranslate">InputSourceReader</code> and <code class="notranslate">InputSourceSampler</code> will internally use <code class="notranslate">ObjectSource</code> and <code class="notranslate">InputFormat</code>. <code class="notranslate">ObjectSource</code> knows how to read bytes from the given object.</p>
<div class="highlight highlight-source-java notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="public interface ObjectSource&lt;T&gt;
{
  int FETCH_BUFFER_SIZE = 4 * 1024;
  int MAX_FETCH_RETRY = 3;

  interface CleanableFile
  {
    File file();

    void cleanup();
  }

  CleanableFile fetch(File temporaryDirectory, byte[] fetchBuffer) throws IOException;

  T getObject();

  InputStream open() throws IOException;

  Predicate&lt;Throwable&gt; getRetryCondition();
}"><pre class="notranslate"><span class="pl-k">public</span> <span class="pl-k">interface</span> <span class="pl-smi">ObjectSource</span>&lt;<span class="pl-smi">T</span>&gt;
{
  <span class="pl-smi">int</span> <span class="pl-c1">FETCH_BUFFER_SIZE</span> = <span class="pl-c1">4</span> * <span class="pl-c1">1024</span>;
  <span class="pl-smi">int</span> <span class="pl-c1">MAX_FETCH_RETRY</span> = <span class="pl-c1">3</span>;

  <span class="pl-k">interface</span> <span class="pl-smi">CleanableFile</span>
  {
    <span class="pl-smi">File</span> <span class="pl-en">file</span>();

    <span class="pl-smi">void</span> <span class="pl-en">cleanup</span>();
  }

  <span class="pl-smi">CleanableFile</span> <span class="pl-en">fetch</span>(<span class="pl-smi">File</span> <span class="pl-s1">temporaryDirectory</span>, <span class="pl-smi">byte</span>[] <span class="pl-s1">fetchBuffer</span>) <span class="pl-k">throws</span> <span class="pl-smi">IOException</span>;

  <span class="pl-smi">T</span> <span class="pl-en">getObject</span>();

  <span class="pl-smi">InputStream</span> <span class="pl-en">open</span>() <span class="pl-k">throws</span> <span class="pl-smi">IOException</span>;

  <span class="pl-smi">Predicate</span>&lt;<span class="pl-smi">Throwable</span>&gt; <span class="pl-en">getRetryCondition</span>();
}</pre></div>
<p dir="auto">You can directly open an <code class="notranslate">InputStrema</code> on the <code class="notranslate">ObjectSource</code> or <code class="notranslate">fetch()</code> the remote object into a local disk and open a <code class="notranslate">FileInputStream</code> on it. This may be useful to avoid expensive random access on remote storage (e.g., Orc file in s3) or holding connections for too long time (as in <code class="notranslate">SqlFirehoseFactory</code>). Check <a href="https://github.com/jihoonson/druid/blob/file-format/core/src/main/java/org/apache/druid/data/input/impl/HttpSource.java"><code class="notranslate">HttpSource</code></a> as an example <code class="notranslate">ObjectSource</code>.</p>
<p dir="auto"><code class="notranslate">InputFormat</code> knows how to parse bytes.</p>
<div class="highlight highlight-source-java notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="public interface InputFormat
{
  boolean isSplittable();

  ObjectReader createReader(InputRowSchema inputRowSchema);
}"><pre class="notranslate"><span class="pl-k">public</span> <span class="pl-k">interface</span> <span class="pl-smi">InputFormat</span>
{
  <span class="pl-smi">boolean</span> <span class="pl-en">isSplittable</span>();

  <span class="pl-smi">ObjectReader</span> <span class="pl-en">createReader</span>(<span class="pl-smi">InputRowSchema</span> <span class="pl-s1">inputRowSchema</span>);
}</pre></div>
<p dir="auto"><code class="notranslate">ObjectReader</code> actually reads and parses data and returns an interator of <code class="notranslate">InputRow</code>.</p>
<div class="highlight highlight-source-java notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="public interface ObjectReader
{
  CloseableIterator&lt;InputRow&gt; read(ObjectSource&lt;?&gt; source, File temporaryDirectory) throws IOException;

  CloseableIterator&lt;InputRowPlusRaw&gt; sample(ObjectSource&lt;?&gt; source, File temporaryDirectory) throws IOException;
}"><pre class="notranslate"><span class="pl-k">public</span> <span class="pl-k">interface</span> <span class="pl-smi">ObjectReader</span>
{
  <span class="pl-smi">CloseableIterator</span>&lt;<span class="pl-smi">InputRow</span>&gt; <span class="pl-en">read</span>(<span class="pl-smi">ObjectSource</span>&lt;?&gt; <span class="pl-s1">source</span>, <span class="pl-smi">File</span> <span class="pl-s1">temporaryDirectory</span>) <span class="pl-k">throws</span> <span class="pl-smi">IOException</span>;

  <span class="pl-smi">CloseableIterator</span>&lt;<span class="pl-smi">InputRowPlusRaw</span>&gt; <span class="pl-en">sample</span>(<span class="pl-smi">ObjectSource</span>&lt;?&gt; <span class="pl-s1">source</span>, <span class="pl-smi">File</span> <span class="pl-s1">temporaryDirectory</span>) <span class="pl-k">throws</span> <span class="pl-smi">IOException</span>;
}</pre></div>
<p dir="auto">For example, <a href="https://github.com/jihoonson/druid/blob/file-format/extensions-core/orc-extensions/src/main/java/org/apache/druid/data/input/orc/OrcInputFormat.java"><code class="notranslate">OrcInputFormat</code></a> creates <a href="https://github.com/jihoonson/druid/blob/file-format/extensions-core/orc-extensions/src/main/java/org/apache/druid/data/input/orc/OrcReader.java"><code class="notranslate">OrcReader</code></a>. Note that this implementation is really not optimized but will show how it could be implemented.</p>
<h4 dir="auto">Deprecated <code class="notranslate">ParseSpec</code></h4>
<p dir="auto">The existing <code class="notranslate">ParseSpec</code> will be split into <code class="notranslate">TimestampSpec</code>, <code class="notranslate">DimensionsSpec</code>, and <code class="notranslate">Inputformat</code>. <code class="notranslate">TimestampSpec</code> and <code class="notranslate">DimensionsSpec</code> will be at the top level of <code class="notranslate">DataSchema</code>. <code class="notranslate">InputFormat</code> will be in <code class="notranslate">ioConfig</code>.</p>
<p dir="auto">An example spec is:</p>
<div class="highlight highlight-source-json notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="{
  &quot;type&quot;: &quot;index_parallel&quot;,
  &quot;spec&quot;: {
    &quot;dataSchema&quot;: {
      &quot;dataSource&quot;: &quot;wikipedia&quot;,
      &quot;timestampSpec&quot;: {
        &quot;column&quot;: &quot;timestamp&quot;,
        &quot;format&quot;: &quot;iso&quot;
      },
      &quot;dimensionsSpec&quot;: {
        &quot;dimensions&quot;: [
          &quot;channel&quot;,
          &quot;cityName&quot;,
          &quot;comment&quot;,
          &quot;countryIsoCode&quot;,
          &quot;countryName&quot;,
          &quot;diffUrl&quot;,
          &quot;flags&quot;,
          &quot;isAnonymous&quot;,
          &quot;isMinor&quot;,
          &quot;isNew&quot;,
          &quot;isRobot&quot;,
          &quot;isUnpatrolled&quot;,
          &quot;namespace&quot;,
          &quot;page&quot;,
          &quot;regionIsoCode&quot;,
          &quot;regionName&quot;,
          &quot;user&quot;
        ]
      },
      &quot;metricsSpec&quot;: [
        {
          &quot;type&quot;: &quot;count&quot;,
          &quot;name&quot;: &quot;count&quot;
        }
      ],
      &quot;granularitySpec&quot;: {
        &quot;type&quot;: &quot;uniform&quot;,
        &quot;segmentGranularity&quot;: &quot;DAY&quot;,
        &quot;rollup&quot;: true,
        &quot;intervals&quot;: [&quot;2018/2019&quot;]
      }
    },
    &quot;ioConfig&quot;: {
      &quot;type&quot;: &quot;index_parallel&quot;,
      &quot;inputSource&quot;: {
        &quot;type&quot;: &quot;http&quot;,
        &quot;uris&quot;: [
          &quot;https://path/to/wikipedia.json.gz&quot;
        ]
      },
      &quot;inputFormat&quot;: {
        &quot;type&quot;: &quot;json&quot;
      },
      &quot;appendToExisting&quot;: false
    },
    &quot;tuningConfig&quot;: {
      &quot;type&quot;: &quot;index_parallel&quot;,
      &quot;maxNumConcurrentSubTasks&quot;: 10
    }
  }
}"><pre class="notranslate">{
  <span class="pl-ent">"type"</span>: <span class="pl-s"><span class="pl-pds">"</span>index_parallel<span class="pl-pds">"</span></span>,
  <span class="pl-ent">"spec"</span>: {
    <span class="pl-ent">"dataSchema"</span>: {
      <span class="pl-ent">"dataSource"</span>: <span class="pl-s"><span class="pl-pds">"</span>wikipedia<span class="pl-pds">"</span></span>,
      <span class="pl-ent">"timestampSpec"</span>: {
        <span class="pl-ent">"column"</span>: <span class="pl-s"><span class="pl-pds">"</span>timestamp<span class="pl-pds">"</span></span>,
        <span class="pl-ent">"format"</span>: <span class="pl-s"><span class="pl-pds">"</span>iso<span class="pl-pds">"</span></span>
      },
      <span class="pl-ent">"dimensionsSpec"</span>: {
        <span class="pl-ent">"dimensions"</span>: [
          <span class="pl-s"><span class="pl-pds">"</span>channel<span class="pl-pds">"</span></span>,
          <span class="pl-s"><span class="pl-pds">"</span>cityName<span class="pl-pds">"</span></span>,
          <span class="pl-s"><span class="pl-pds">"</span>comment<span class="pl-pds">"</span></span>,
          <span class="pl-s"><span class="pl-pds">"</span>countryIsoCode<span class="pl-pds">"</span></span>,
          <span class="pl-s"><span class="pl-pds">"</span>countryName<span class="pl-pds">"</span></span>,
          <span class="pl-s"><span class="pl-pds">"</span>diffUrl<span class="pl-pds">"</span></span>,
          <span class="pl-s"><span class="pl-pds">"</span>flags<span class="pl-pds">"</span></span>,
          <span class="pl-s"><span class="pl-pds">"</span>isAnonymous<span class="pl-pds">"</span></span>,
          <span class="pl-s"><span class="pl-pds">"</span>isMinor<span class="pl-pds">"</span></span>,
          <span class="pl-s"><span class="pl-pds">"</span>isNew<span class="pl-pds">"</span></span>,
          <span class="pl-s"><span class="pl-pds">"</span>isRobot<span class="pl-pds">"</span></span>,
          <span class="pl-s"><span class="pl-pds">"</span>isUnpatrolled<span class="pl-pds">"</span></span>,
          <span class="pl-s"><span class="pl-pds">"</span>namespace<span class="pl-pds">"</span></span>,
          <span class="pl-s"><span class="pl-pds">"</span>page<span class="pl-pds">"</span></span>,
          <span class="pl-s"><span class="pl-pds">"</span>regionIsoCode<span class="pl-pds">"</span></span>,
          <span class="pl-s"><span class="pl-pds">"</span>regionName<span class="pl-pds">"</span></span>,
          <span class="pl-s"><span class="pl-pds">"</span>user<span class="pl-pds">"</span></span>
        ]
      },
      <span class="pl-ent">"metricsSpec"</span>: [
        {
          <span class="pl-ent">"type"</span>: <span class="pl-s"><span class="pl-pds">"</span>count<span class="pl-pds">"</span></span>,
          <span class="pl-ent">"name"</span>: <span class="pl-s"><span class="pl-pds">"</span>count<span class="pl-pds">"</span></span>
        }
      ],
      <span class="pl-ent">"granularitySpec"</span>: {
        <span class="pl-ent">"type"</span>: <span class="pl-s"><span class="pl-pds">"</span>uniform<span class="pl-pds">"</span></span>,
        <span class="pl-ent">"segmentGranularity"</span>: <span class="pl-s"><span class="pl-pds">"</span>DAY<span class="pl-pds">"</span></span>,
        <span class="pl-ent">"rollup"</span>: <span class="pl-c1">true</span>,
        <span class="pl-ent">"intervals"</span>: [<span class="pl-s"><span class="pl-pds">"</span>2018/2019<span class="pl-pds">"</span></span>]
      }
    },
    <span class="pl-ent">"ioConfig"</span>: {
      <span class="pl-ent">"type"</span>: <span class="pl-s"><span class="pl-pds">"</span>index_parallel<span class="pl-pds">"</span></span>,
      <span class="pl-ent">"inputSource"</span>: {
        <span class="pl-ent">"type"</span>: <span class="pl-s"><span class="pl-pds">"</span>http<span class="pl-pds">"</span></span>,
        <span class="pl-ent">"uris"</span>: [
          <span class="pl-s"><span class="pl-pds">"</span>https://path/to/wikipedia.json.gz<span class="pl-pds">"</span></span>
        ]
      },
      <span class="pl-ent">"inputFormat"</span>: {
        <span class="pl-ent">"type"</span>: <span class="pl-s"><span class="pl-pds">"</span>json<span class="pl-pds">"</span></span>
      },
      <span class="pl-ent">"appendToExisting"</span>: <span class="pl-c1">false</span>
    },
    <span class="pl-ent">"tuningConfig"</span>: {
      <span class="pl-ent">"type"</span>: <span class="pl-s"><span class="pl-pds">"</span>index_parallel<span class="pl-pds">"</span></span>,
      <span class="pl-ent">"maxNumConcurrentSubTasks"</span>: <span class="pl-c1">10</span>
    }
  }
}</pre></div>
<h4 dir="auto">Prefetch and cache</h4>
<p dir="auto">The <code class="notranslate">FirehoseFactory</code>s extending <code class="notranslate">PrefetchableTextFilesFirehoseFactory</code> currently support prefetching and caching which I don't find very useful. I ran a couple of tests in a cluster and my laptop. The total time taken to download 2 files (200 MB each) from s3 into local storage was 4 sec and 20 sec, whereas the total ingestion time was 20 min and 30 min, respectively. I think the more important issue is probably the indexing or the segment merge speed. The prefetch and cache will be not supported with new interfaces until it becomes a real bottleneck.</p>
<h4 dir="auto"><code class="notranslate">FirehoseFactory</code></h4>
<p dir="auto"><code class="notranslate">FirehoseFactory</code> will remain for <code class="notranslate">RealtimeIndexTask</code> and <code class="notranslate">AppenderatorDriverRealtimeIndexTask</code>.</p>
<h3 dir="auto">Rationale</h3>
<p dir="auto">One possible alternative would be modifying existing interfaces. I think adding new ones will be better because the new ones are pretty different from existing ones.</p>
<h3 dir="auto">Operational impact</h3>
<p dir="auto">A couple of interfaces will be deprecated but kept for a couple of future releases. This means, the old spec will be still respected.</p>
<ul dir="auto">
<li><code class="notranslate">ParseSpec</code> will be deprecated and split into <code class="notranslate">TimestampSpec</code>, <code class="notranslate">DimensionsSpec</code>, and <code class="notranslate">InputFormat</code>.</li>
<li><code class="notranslate">DataSchema</code> will have <code class="notranslate">TimestampSpec</code> and <code class="notranslate">DimensionsSpec</code> from the deprecated <code class="notranslate">ParseSpec</code>.</li>
<li><code class="notranslate">IOConfig</code> will have <code class="notranslate">InputFormat</code> and <code class="notranslate">InputSource</code>.</li>
<li><code class="notranslate">FirehoseFactory</code> will be deprecated for all batch tasks in favor of <code class="notranslate">InputSource</code>. This will not be applied to the compaction task since it doesn't have firehoseFactory in its spec.</li>
<li><code class="notranslate">InputRowParser</code> will be deprecated.</li>
</ul>
<h3 dir="auto">Test plan (optional)</h3>
<p dir="auto">Unit tests will be added to test backward compatibility and the implementations of new interfaces.</p>
<h3 dir="auto">Future work (optional)</h3>
<p dir="auto">A new method can be added to <code class="notranslate">ObjectSource</code> for more optimized data scan.</p>
<div class="highlight highlight-source-java notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="int read(ByteBuffer buffer, int offset, int length) throws IOException;"><pre class="notranslate"><span class="pl-smi">int</span> <span class="pl-s1">read</span>(<span class="pl-smi">ByteBuffer</span> <span class="pl-s1">buffer</span>, <span class="pl-smi">int</span> <span class="pl-s1">offset</span>, <span class="pl-smi">int</span> <span class="pl-s1">length</span>) <span class="pl-k">throws</span> <span class="pl-s1">IOException</span>;</pre></div>
<p dir="auto">This proposal is about extending Druid to support multi-stage distributed queries, something that I think will be really exciting and amp up what Druid is capable of.</p>
<h2 dir="auto">Motivation</h2>
<p dir="auto">Today Druid's distributed query stack is single-stage: the Broker receives a query, slices it up into pieces, and sends one piece to each Historical. The Historicals produce partial results for their piece of the query, send them back to the Broker, and the Broker merges them into the final result set.</p>
<p dir="auto">This venerable design works well when the bulk of processing can be done at the leaf servers that store the actual data. It scales well: Druid can handle thousands of QPS of queries, very CPU and memory efficiently, so long as there are enough Historicals running and so long as the amount of data that makes it to the Broker is relatively small.</p>
<p dir="auto">Druid has a lot of functionality that is designed to make this single-stage distributed query stack as useful as possible. Whenever possible, we push down projections, filters, aggregations, limits, and joins. We also have a suite of builtin approximations, like approximate topN and various sketches, that people can use to minimize the amount of state that must be tracked beyond the leaf servers.</p>
<p dir="auto">But there are some cases where this query stack doesn't work well, and which provide the motivation for multi-stage queries:</p>
<ol dir="auto">
<li>Queries with very large result sets. Imagine a GROUP BY query with no LIMIT that would return billions of rows, because it groups on something with very high cardinality. Today these result sets must flow through the Broker, which creates a bottleneck.</li>
<li>Additional SQL support. We'd like to keep expanding Druid's ability to support SQL until we can do it all. Some operations that'd we like to support, like joining together two distributed tables, cannot be done in the single-stage design.</li>
<li>Complex query structures. Today, a query with lots of JOINs and subqueries will execute certain queries in a fully distributed manner -- basically, anything that can be represented as a base distributed table joined with broadcasted tables -- and then finish the remainder of query processing on the Broker. In some cases this works OK, because most of the work can be done as part of the distributed queries. But in other cases it results in a lot of data being collected on the Broker, which leads to an error like "Subquery generated results beyond maximum".</li>
<li>Ingestion. Some databases use their query stack to handle ingestion. It makes sense if you think about an ingestion as a query that writes its results to a table. It would be nice if we could do this too, so we wouldn't need to maintain an ingestion execution stack that is separate from the query execution stack.</li>
</ol>
<h2 dir="auto">Design</h2>
<p dir="auto">There are a lot of interesting pieces here, so I just want to touch on each one in this main proposal. Each one should then be fleshed out separately.</p>
<h3 dir="auto">Query representation and SQL planning</h3>
<p dir="auto"><strong>Background</strong></p>
<p dir="auto">Druid models native queries as "fat" query types that represent a full distributed query. There are four query types used by the SQL engine: scan, timeseries, topN, and groupBy. Each query type represents an entire single-stage distributed computation: there is a first piece that can run distributed, and a second piece that must run on the Broker. They all internally handle filtering, projection, sorting, limiting, etc. They all have a "dataSource" field that describes where they should get their data from, which can be "table", representing an actual Druid datasource; "query", representing a subquery, and which can be any other query type; "join", representing two other datasources joined together; or a handful of other less-common datasource types.</p>
<p dir="auto">The SQL planner's main job is to morph a tree of relational operators either a single native query or a tree of native queries. In the latter case, it uses "query" or "join" datasources to link the native queries together. It has limited flexibility in how it does this, because native queries have a rigid computational structure: they always do broadcast join first, then projection, then filter, then aggregation, then sort.</p>
<p dir="auto"><strong>Proposal</strong></p>
<p dir="auto">To support ever-more complex query structures, we will need a way of representing data flow that is more flexible than native queries. The tried and true DAG approach will work well here. So I propose that the multi-stage engine should model queries as a DAG of "stages". Each stage:</p>
<ul dir="auto">
<li>Runs distributed across as many servers as makes sense.</li>
<li>May consume partitioned data from some set of input stages. There will always be some stages with no input stages; these stages would be reading from Druid segments, external data, or something like that.</li>
<li>Produces data either aligned with its input partitions, or reshuffled in some way.</li>
</ul>
<p dir="auto">The stages would be finer-grained than native queries. For example, a stage might do scan-filter-project, or aggregate, or sort, or limit. This is a common approach in relational databases.</p>
<p dir="auto">As a first step, I suggest we keep native queries in the picture by translating SQL -&gt; native query -&gt; DAG-based query. This can be done with minimal changes to the SQL planner, and would enable classic native queries to exist at the same time as DAG-based queries without much code duplication. But at some point, we'll want to go directly from SQL -&gt; DAG-based query, since that will give the SQL planner more flexibility to reorder operations.</p>
<p dir="auto">Here's an example of how a query would look. This SQL query:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="SELECT
  session,
  COUNT(*) AS cnt
FROM tbl
WHERE browser = 'Firefox'
GROUP BY session
ORDER BY cnt DESC
LIMIT 10"><pre class="notranslate"><code class="notranslate">SELECT
  session,
  COUNT(*) AS cnt
FROM tbl
WHERE browser = 'Firefox'
GROUP BY session
ORDER BY cnt DESC
LIMIT 10
</code></pre></div>
<p dir="auto">Would have five stages:</p>
<ol dir="auto">
<li>Scan: Read <code class="notranslate">tbl</code>, apply the filter <code class="notranslate">browser = 'Firefox'</code>, and project out <code class="notranslate">session</code>. No shuffle.</li>
<li>Aggregate I: Locally group by <code class="notranslate">session</code> and compute <code class="notranslate">count(*)</code> for each. Shuffle by <code class="notranslate">session</code>.</li>
<li>Aggregate II: Continue grouping by <code class="notranslate">session</code> within each partition, and sum the partial counts to get full counts. No shuffle. This produces a fully grouped resultset, still partitioned by <code class="notranslate">session</code>.</li>
<li>Sort: Locally order by <code class="notranslate">cnt DESC</code>. Shuffle everything into a single partition.</li>
<li>Limit: Take the first 10 rows from the single partition generated by the prior stage.</li>
</ol>
<p dir="auto">When stages connect without shuffling, we can pipeline execution locally on the same servers, so there is no need for buffering or cross-server traffic. When stages connect <em>with</em> shuffling, we'll need to exchange data across servers. Depending on the needs of the query, the producing stage could stream to the consuming stage, or the producing stage could buffer up all results before the consuming stage starts.</p>
<p dir="auto">For the query above, if we know (or are willing to bet) that there are not too many distinct <code class="notranslate">session</code> values then we can run it just as efficiently as the current single-stage approach. First, we'll make sure that Scan and Aggregate I are scheduled on the same set of workers, so the output of Scan can be pipelined into Aggregate I locally in memory. Then, we'll configure the shuffle in Aggregate I to shuffle everything down to a single partition. Next, we'll make sure that Aggregate II, Sort, and Limit all run on the same server. That server would be responsible for gathering all the partitioned Aggregate I outputs and preparing the final result. Finally, we'll set things up so all stages run concurrently, and so Aggregate II streams from Aggregate I. Put together, this is exactly what the Historicals and Brokers do today.</p>
<h3 dir="auto">Ingestion</h3>
<p dir="auto">In <a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="1055279806" data-permission-text="Title is private" data-url="https://github.com/apache/druid/issues/11929" data-hovercard-type="issue" data-hovercard-url="/apache/druid/issues/11929/hovercard" href="https://github.com/apache/druid/issues/11929">#11929</a> there is a proposal for adding a SQL INSERT statement. It could be implemented on top of our existing batch ingestion tasks: the SQL layer could convert relational operators to indexing tasks just as well as it can convert them to native queries. But I think it would be nicer to implement it on top of a multi-stage query stack.</p>
<p dir="auto">We'd just need to do two things:</p>
<ol dir="auto">
<li>Provide a way for the query stack to read external data. There's a natural way to do this through an "external" DataSource that maps onto RowBasedSegments. It would be similar to how we plug lookups into the query engines, except a little more complicated because we'll need to split the external datasource before sending it down to various servers.</li>
<li>Provide a way for the query stack to generate segments. Once we have this multi-stage structure this is also natural: we can add a final stage to any query that shuffles data to match the target segment size and then generates and publishes segments.</li>
</ol>
<p dir="auto">So I propose that we work towards this instead of having the SQL planner generate batch indexing tasks. It has a nice side benefit: part (1) alone means that we'd be able to query external data in a regular query too. I don't think that'll be a core use case for Druid, but it has some usefulness, like previewing what an INSERT <em>might</em> do, or doing an ad-hoc join of Druid datasources with some external data.</p>
<p dir="auto">We'll also need to figure out what to do about streaming ingest at some point. I'm not sure what to do there but I think there are a few options that make sense. Even if the multi-stage query stack doesn't have builtin support for streaming queries, we can layer streamingness on top in the same way that we do today with indexing tasks: there is a supervisor that manages a series of tasks, each of which reads a chunk of data from Kafka and then publishes it.</p>
<h3 dir="auto">Server setup</h3>
<p dir="auto">There is an obvious question: where will this query engine run? Today we have Brokers, Historicals, and indexing stuff (MiddleManager or Indexers). I don't want to add a net new service or process type in the long run, because we have a lot already. But I think in the short run we should add a new process type.</p>
<p dir="auto">I think at first the Broker should remain in charge of SQL planning, and may route certain queries to these new multi-stage query processes if they are running in that particular cluster. It could do this based on user request (like a context parameter) or based on aspects of the query (like presence of INSERT, or external data, or a "complex enough" query).</p>
<p dir="auto">However, like I said, in the long run I don't think it's a good plan to have an extra process type on top of all the ones we already have. So I think in the long run we should shoot for this process type actually being able to serve as a replacement for the Broker, Historicals, and MMs/Indexers. By that I mean that in the fullness of time, it should be able to:</p>
<ul dir="auto">
<li>Receive and plan SQL queries, like a Broker.</li>
<li>Cache segments locally, like a Historical.</li>
<li>Run simple single-stage queries in an extremely efficient way, like the Historical + Broker combo.</li>
<li>Run ingestions (INSERT statements) in a distributed and scalable manner, like MMs or Indexers.</li>
</ul>
<p dir="auto">At that point, a small Druid cluster would be pretty simple: just a Coordinator and this new process type (or a few of them for scale-out). And a large cluster could still specialize. Imagine setting up a tier of these new processes that cache segments but don't perform ingestion, and another tier that <em>don't</em> cache segments but <em>do</em> perform ingestion. That's similar to setting up Historicals and MMs today, but more configuration-driven. It would allow simple clusters to be simple and complex clusters to be complex.</p>
<h2 dir="auto">Code changes</h2>
<p dir="auto">A sketch of the code changes I think we'll need to make the above work out:</p>
<ol dir="auto">
<li>Low-level mechanism for shuffling data between servers. This should include an efficient way to transfer large amounts of data over the network. Currently we use Smile-over-HTTP for this, which works well for relatively small amounts of data, but I think we can do better. It should also include both a pipelined implementation (producing stage streams to the consuming stage), and a buffered implementation (producing stage generates all data before consuming stage starts) for operations that cannot be pipelined.</li>
<li>Cluster-level execution coordinator for a multi-stage query.</li>
<li>Server-local execution coordinator for a single stage of a multi-stage query.</li>
<li>An actual process type for the above three things to run in.</li>
<li>Converter that generates multi-stage (DAG-based) queries from today's native query types. (This enables hooking into the existing SQL layer, since it generates native queries.)</li>
<li>To make ingestion-through-query work: adapters from external data into the query stack, and from the query stack to the segment-generation code.</li>
</ol>
<p dir="auto">At Imply we've started to prototype a multi-stage query engine with the above design. We've implemented some of the pieces: not all of them, but enough to perform some basic queries. At this point I thought it would be a good time to open up a discussion with the wider community, since as we continue working on this stuff, we'd like to integrate it into Apache Druid.</p>
<h2 dir="auto">Rationale</h2>
<p dir="auto">A.K.A. the "why not" section.</p>
<p dir="auto"><strong>Why not integrate with an existing open-source query engine?</strong></p>
<p dir="auto">I wouldn't want to stop anyone from integrating with another open-source query engine. Some people might prefer to deploy that way.</p>
<p dir="auto">But I think we will want this kind of functionality as something native in Druid, for two reasons. First: I think the user experience will be nicer if we don't require a dependency on another system. Second: I expect there will be opportunities for optimization that we can use if everything is built in to Druid, and that would be tough to implement with an external query stack.</p>
<p dir="auto">Besides, the work we'll need to do in order to build a multi-stage query engine will also benefit integrations with other engines.</p>
<p dir="auto"><strong>Why not gradually add multi-stage capabilities to the existing query stack?</strong></p>
<p dir="auto">I think it's a question of complexity.</p>
<p dir="auto">The core query stack has three main hook points: individual segment scan, result merge on Historicals, and result merge on Broker. The implementations of the native query types, especially groupBy, have got quite complex over the years as new features have been added and needed to be mapped onto these hook points. For example: subtotals, ordering, limiting, having, are all typically done as part of "result merge on the Broker". The hook points are doing more work than originally envisioned, which makes the code more difficult to follow and extend.</p>
<p dir="auto">So I think it's time to try a different approach with the query stack, rather than adding new capabilities. As we create a DAG-based query stack, we can split queries up into stages and move more of the "fancy stuff" to the framework, which will simplify the query-specific logic.</p>
<p dir="auto">We can still share code, though. It won't be 100% new. We can share all of the StorageAdapter stuff (cursors, etc), the ColumnSelectorFactory stuff, and the single-segment processing code for groupBy and topN. This lower-level code is super optimized and works well. It's more the mid/high level stuff that IMO would benefit from a different approach.</p>