<h3 dir="auto">Affected Version</h3>
<p dir="auto">0.22.2</p>
<h3 dir="auto">Description</h3>
<p dir="auto">Starting up a broker with HTTP typed serverview has been resulting with multiple HttpServerInventoryView objects being created. This would seem to create a bunch of duplicate data on heap + a bunch of duplicate HTTP requests to data servers</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="2022-08-30T18:02:20,058 INFO [main] org.apache.druid.client.HttpServerInventoryView - Starting HttpServerInventoryView.
2022-08-30T18:02:20,175 INFO [CuratorDruidNodeDiscoveryProvider-ListenerExecutor] org.apache.druid.server.coordination.ChangeRequestHttpSyncer - Starting ChangeRequestHttpSyncer[https://XXXXX:8283/_1661882540175].
2022-08-30T18:02:25,510 INFO [HttpServerInventoryView-4] org.apache.druid.server.coordination.ChangeRequestHttpSyncer - [https://XXXXX:8283/_1661882540175] synced successfully for the first time.
2022-08-30T18:02:20,072 INFO [main] org.apache.druid.client.HttpServerInventoryView - Started HttpServerInventoryView.
2022-08-30T18:02:58,244 INFO [main] org.apache.druid.client.HttpServerInventoryView - Starting HttpServerInventoryView.
2022-08-30T18:02:58,248 INFO [main] org.apache.druid.server.coordination.ChangeRequestHttpSyncer - Starting ChangeRequestHttpSyncer[https://XXXXX:8283/_1661882578248].
2022-08-30T18:02:58,897 INFO [HttpServerInventoryView-7] org.apache.druid.server.coordination.ChangeRequestHttpSyncer - [https://XXXXX:8283/_1661882578248] synced successfully for the first time.
2022-08-30T18:02:58,253 INFO [main] org.apache.druid.client.HttpServerInventoryView - Started HttpServerInventoryView."><pre class="notranslate"><code class="notranslate">2022-08-30T18:02:20,058 INFO [main] org.apache.druid.client.HttpServerInventoryView - Starting HttpServerInventoryView.
2022-08-30T18:02:20,175 INFO [CuratorDruidNodeDiscoveryProvider-ListenerExecutor] org.apache.druid.server.coordination.ChangeRequestHttpSyncer - Starting ChangeRequestHttpSyncer[https://XXXXX:8283/_1661882540175].
2022-08-30T18:02:25,510 INFO [HttpServerInventoryView-4] org.apache.druid.server.coordination.ChangeRequestHttpSyncer - [https://XXXXX:8283/_1661882540175] synced successfully for the first time.
2022-08-30T18:02:20,072 INFO [main] org.apache.druid.client.HttpServerInventoryView - Started HttpServerInventoryView.
2022-08-30T18:02:58,244 INFO [main] org.apache.druid.client.HttpServerInventoryView - Starting HttpServerInventoryView.
2022-08-30T18:02:58,248 INFO [main] org.apache.druid.server.coordination.ChangeRequestHttpSyncer - Starting ChangeRequestHttpSyncer[https://XXXXX:8283/_1661882578248].
2022-08-30T18:02:58,897 INFO [HttpServerInventoryView-7] org.apache.druid.server.coordination.ChangeRequestHttpSyncer - [https://XXXXX:8283/_1661882578248] synced successfully for the first time.
2022-08-30T18:02:58,253 INFO [main] org.apache.druid.client.HttpServerInventoryView - Started HttpServerInventoryView.
</code></pre></div>
<p dir="auto">Upon further investigation it appears this only happens when Druid SQL is enabled on the broker.</p>
<h3 dir="auto">Motivation</h3>
<p dir="auto">General motivation for native batch indexing is described in <a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="309164097" data-permission-text="Title is private" data-url="https://github.com/apache/druid/issues/5543" data-hovercard-type="issue" data-hovercard-url="/apache/druid/issues/5543/hovercard" href="https://github.com/apache/druid/issues/5543">#5543</a>.</p>
<p dir="auto">We now have the parallel index task, but it doesn't support perfect rollup yet because of lack of the shuffle system.</p>
<h3 dir="auto">Proposed changes</h3>
<p dir="auto">I would propose to add a new mode for parallel index task which supports perfect rollup with two-phase shuffle.</p>
<h4 dir="auto">Two phase partitioning with shuffle</h4>
<p dir="auto"><a target="_blank" rel="noopener noreferrer nofollow" href="https://user-images.githubusercontent.com/2322288/59528209-2b746900-8ecd-11e9-8024-5b40f7521f49.png"><img src="https://user-images.githubusercontent.com/2322288/59528209-2b746900-8ecd-11e9-8024-5b40f7521f49.png" alt="Phase 1" style="max-width: 100%;"></a></p>
<p dir="auto">Phase 1: each task partitions data by segmentGranularity and then by hash or range key of some dimensions.</p>
<p dir="auto"><a target="_blank" rel="noopener noreferrer nofollow" href="https://user-images.githubusercontent.com/2322288/59528211-2d3e2c80-8ecd-11e9-80f0-a504449eef81.png"><img src="https://user-images.githubusercontent.com/2322288/59528211-2d3e2c80-8ecd-11e9-80f0-a504449eef81.png" alt="Phase 2" style="max-width: 100%;"></a></p>
<p dir="auto">Phase 2: each task reads a set of partitions created by the tasks of Phase 1 and creates a segment per partition.</p>
<h4 dir="auto"><code class="notranslate">PartitionsSpec</code> support for <code class="notranslate">IndexTask</code> and <code class="notranslate">ParallelIndexTask</code></h4>
<p dir="auto"><code class="notranslate">PartitionsSpec</code> is the way to define the secondary partitioning and is currently being used by <code class="notranslate">HadoopIndexTask</code>. This interface should be adjusted to be more general as below.</p>
<div class="highlight highlight-source-java notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="public interface PartitionsSpec
{
  @Nullable
  Integer getNumShards();
  
  @Nullable
  Integer getMaxRowsPerSegment(); // or getTargetRowsPerSegment()
  
  @Nullable
  List&lt;String&gt; getPartitionDimensions();
}"><pre class="notranslate"><span class="pl-k">public</span> <span class="pl-k">interface</span> <span class="pl-smi">PartitionsSpec</span>
{
  <span class="pl-c1">@</span><span class="pl-c1">Nullable</span>
  <span class="pl-smi">Integer</span> <span class="pl-en">getNumShards</span>();
  
  <span class="pl-c1">@</span><span class="pl-c1">Nullable</span>
  <span class="pl-smi">Integer</span> <span class="pl-en">getMaxRowsPerSegment</span>(); <span class="pl-c">// or getTargetRowsPerSegment()</span>
  
  <span class="pl-c1">@</span><span class="pl-c1">Nullable</span>
  <span class="pl-smi">List</span>&lt;<span class="pl-smi">String</span>&gt; <span class="pl-en">getPartitionDimensions</span>();
}</pre></div>
<p dir="auto">Hadoop tasks can use an extended interface which is more specialized for Hadoop.</p>
<div class="highlight highlight-source-java notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="public interface HadoopPartitionsSpec extends PartitionsSpec
{
  Jobby getPartitionJob(HadoopDruidIndexerConfig config);
  boolean isAssumeGrouped();
  boolean isDeterminingPartitions();
}"><pre class="notranslate"><span class="pl-k">public</span> <span class="pl-k">interface</span> <span class="pl-smi">HadoopPartitionsSpec</span> <span class="pl-k">extends</span> <span class="pl-smi">PartitionsSpec</span>
{
  <span class="pl-smi">Jobby</span> <span class="pl-en">getPartitionJob</span>(<span class="pl-smi">HadoopDruidIndexerConfig</span> <span class="pl-s1">config</span>);
  <span class="pl-smi">boolean</span> <span class="pl-en">isAssumeGrouped</span>();
  <span class="pl-smi">boolean</span> <span class="pl-en">isDeterminingPartitions</span>();
}</pre></div>
<p dir="auto"><code class="notranslate">IndexTask</code> currently provides duplicate configurations for partitioning in its tuningConfig such as <code class="notranslate">maxRowsPerSegment</code>, <code class="notranslate">maxTotalRows</code>, <code class="notranslate">numShards</code>, and <code class="notranslate">partitionDimensions</code>. These configurations will be deprecated and the indexTask will support <code class="notranslate">PartitionsSpec</code> instead.</p>
<p dir="auto">To support <code class="notranslate">maxRowsPerSegment</code> and <code class="notranslate">maxTotalRows</code>, a new partitionsSpec could be introduced.</p>
<div class="highlight highlight-source-java notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="/**
 * PartitionsSpec for best-effort rollup
 */
public class DynamicPartitionsSpec implements PartitionsSpec
{
  private final int maxRowsPerSegment;
  private final int maxTotalRows;
}"><pre class="notranslate"><span class="pl-c">/**</span>
<span class="pl-c"> * PartitionsSpec for best-effort rollup</span>
<span class="pl-c"> */</span>
<span class="pl-k">public</span> <span class="pl-k">class</span> <span class="pl-smi">DynamicPartitionsSpec</span> <span class="pl-k">implements</span> <span class="pl-smi">PartitionsSpec</span>
{
  <span class="pl-k">private</span> <span class="pl-k">final</span> <span class="pl-smi">int</span> <span class="pl-s1">maxRowsPerSegment</span>;
  <span class="pl-k">private</span> <span class="pl-k">final</span> <span class="pl-smi">int</span> <span class="pl-s1">maxTotalRows</span>;
}</pre></div>
<p dir="auto">This partitionsSpec will be supported as a new configuration in the tuningConfig of <code class="notranslate">IndexTask</code> and <code class="notranslate">ParallelIndexTask</code>.</p>
<h4 dir="auto">New parallel index task runner to support secondary partitioning</h4>
<p dir="auto"><code class="notranslate">ParallelIndexSupervisorTask</code> is the supervisor task which orchestrates the parallel ingestion. It's responsible for spawning and monitoring sub tasks, and publishing created segments at the end of ingestion.</p>
<p dir="auto">It uses <code class="notranslate">ParallelIndexTaskRunner</code> to run single-phase parallel ingestion without shuffle. To support two-phase ingestion, we can add a new implementation of <code class="notranslate">ParallelIndexTaskRunner</code>, <code class="notranslate">TwoPhaseParallelIndexTaskRunner</code>. <code class="notranslate">ParallelIndexSupervisorTask</code> will choose the new runner if partitionsSpec in tuningConfig is <code class="notranslate">HashedPartitionsSpec</code> or <code class="notranslate">RangePartitionsSpec</code>.</p>
<p dir="auto">This new taskRunner does the followings:</p>
<ul dir="auto">
<li>Add <code class="notranslate">TwoPhasesParallelIndexTaskRunner</code> as a new runner for the supervisor task
<ul dir="auto">
<li>Spawns tasks for determining partitions (if <code class="notranslate">numShards</code> is missing in tuningConfig)</li>
<li>Spawns tasks for building partial segments (phase 1)</li>
<li>When all tasks of the phase 1 finish, spawns new tasks for building the complete segments (phase 2)</li>
<li>Each Phase 2 task is assigned one or multiple partitions
<ul dir="auto">
<li>The assigned partition is represented as an HTTP URL</li>
</ul>
</li>
</ul>
</li>
<li>Publish the segments reported by phase 2 tasks.</li>
<li>Triggers intermediary data cleanup when the supervisor task is finished regardless of its last status.</li>
</ul>
<p dir="auto">The supervisor task provides an additional configuration in its tuningConfig, i.e., <code class="notranslate">numSecondPhaseTasks</code> or  <code class="notranslate">inputRowsPerSecondPhaseTask</code>, to support control of parallelism of the phase 2. This will be improved to automatically determine the optimal parallelism in the future.</p>
<h4 dir="auto">New sub task types</h4>
<h5 dir="auto">Partition determine task</h5>
<ul dir="auto">
<li>Similar to what indexTask or HadoopIndexTask do.</li>
<li>Scan the whole input data and collect <code class="notranslate">HyperLogLog</code> per interval to compute approximate cardinality.</li>
<li>numShards could be computed as below:</li>
</ul>
<div class="highlight highlight-source-java notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="        numShards = (int) Math.ceil(
            (double) numRows / Preconditions.checkNotNull(maxRowsPerSegment, &quot;maxRowsPerSegment&quot;)
        );"><pre class="notranslate">        <span class="pl-s1">numShards</span> = (<span class="pl-smi">int</span>) <span class="pl-smi">Math</span>.<span class="pl-en">ceil</span>(
            (<span class="pl-smi">double</span>) <span class="pl-s1">numRows</span> / <span class="pl-smi">Preconditions</span>.<span class="pl-en">checkNotNull</span>(<span class="pl-s1">maxRowsPerSegment</span>, <span class="pl-s">"maxRowsPerSegment"</span>)
        );</pre></div>
<h5 dir="auto">Phase 1 task</h5>
<ul dir="auto">
<li>Read data via the given firehose</li>
<li>Partition data by segmentGranularity by hash or range (and aggregates if rollup)</li>
<li>Should be able to access by (supervisorTaskId, timeChunk, partitionId)</li>
<li>Write partitioned <em>segments</em> in local disk. Multiple disks can be configured, and each task would write partitions in a round-robin manner to utilize disk bandwidth efficiently</li>
</ul>
<h5 dir="auto">Phase 2 task</h5>
<ul dir="auto">
<li>Download all partial segments from middleManagers where phase 1 tasks ran.</li>
<li>Merge all fetched segments into a single segment per partitionId.</li>
<li>Push the merged segments and report them to the supervisor task.</li>
</ul>
<h4 dir="auto">MiddleManager as Intermediary data server</h4>
<p dir="auto">MiddleManager (and new Indexer) should be responsible for serving intermediary data during shuffle.</p>
<p dir="auto">Each phase 1 task partitions input data and generates partitioned segments. These partitioned segments are stored in local disk of middleManager (or indexer proposed in <a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="456473887" data-permission-text="Title is private" data-url="https://github.com/apache/druid/issues/7900" data-hovercard-type="issue" data-hovercard-url="/apache/druid/issues/7900/hovercard" href="https://github.com/apache/druid/issues/7900">#7900</a>). The partitioned segment location would be <code class="notranslate">/configured/prefix/supervisorTaskId/</code> directory. The same configurations with <code class="notranslate">StorageLocationConfig</code> would be provided for intermediary segment location.</p>
<p dir="auto">MiddleManagers and indexers would clean up intermediary segments using the below mechanism.</p>
<ul dir="auto">
<li>MM will keep expiration time in memory. This expiration time is initialized with current time + configured timeout.</li>
<li>MM periodically checks there are any new partitions created for new supervisorTasks and initializes the expiration time if it finds any.</li>
<li>When a subtask accesses a partition, the expiration time for the supervisorTask is initialized or updated if it's already there.</li>
<li>MM periodically checks those expiration times for supervisorTasks. If it finds any expired supervisorTask, then it will ask the overlord if the task is still running. If not, MM will remove all partitions for the supervisorTask.</li>
<li>The overlord will also send a cleanup request to MM when the supervisorTask is finished. This will clean up the expiration time.</li>
</ul>
<h5 dir="auto">New API lists of MiddleManager</h5>
<ul dir="auto">
<li>GET <code class="notranslate">/druid/worker/v1/shuffle/tasks/{supervisorTaskId}/partition?start={startTimeOfSegment}&amp;end={endTimeOfSegment}&amp;partitionId={partitionId}</code></li>
</ul>
<p dir="auto">Return all partial segments generated by sub tasks of the given supervisor task, falling in the given interval, and having the given partitionId.</p>
<ul dir="auto">
<li>DELETE <code class="notranslate">/druid/worker/v1/shuffle/tasks/{supervisorTaskId}</code></li>
</ul>
<p dir="auto">Removes all partial segments generated by sub tasks of the given supervisor task.</p>
<h4 dir="auto">New metrics &amp; task reports</h4>
<ul dir="auto">
<li><code class="notranslate">ingest/task/time</code>: how long each task took</li>
<li><code class="notranslate">ingest/task/bytes/processed</code>: how large data each task processed</li>
<li><code class="notranslate">ingest/shuffle/bytes</code>: how large data each middleManager served</li>
<li><code class="notranslate">ingest/shuffle/requests</code>: how many requests each middleManager served</li>
</ul>
<h4 dir="auto">Task failure handling</h4>
<p dir="auto">Task failure handling is same with the current behavior.</p>
<ul dir="auto">
<li>If the supervisorTask process is killed normally, stopGracefully method is called which kills all running subtasks. If it's killed abnormally, then parallel index task doesn't handle this case for now.</li>
<li>SupervisorTask monitors subtask statuses and counts how many subtasks have failed to process the same input. If it notices more failures than configured maxRetry, it regards that input can't be processed and exists with an error. Otherwise, it respawns a new task which processes the same input.</li>
</ul>
<h3 dir="auto">Rationale</h3>
<p dir="auto">There could be two alternate designs for the shuffle system, especially for intermediary data server.</p>
<p dir="auto">MiddleManager (or indexer) as intermediary data server is the simplest design. In an alternative design, phase 1 tasks could serve intermediary data for shuffle. In this alternate, phase 1 tasks should be guaranteed to run until the phase 2 is finished, which means task 1 resources should be held until the phase 2 is finished. This is rejected for better resource utilization.</p>
<p dir="auto">Another alternate is a single set of tasks would process both phase 1 and phase 2. This design is rejected because it's not very flexible to use cluster resource efficiently.</p>
<h3 dir="auto">Operational impact</h3>
<p dir="auto"><code class="notranslate">maxRowsPerSegment</code>, <code class="notranslate">numShards</code>, <code class="notranslate">partitionDimensions</code>, and <code class="notranslate">maxTotalRows</code> in tuningConfig will be deprecated for indexTask. <code class="notranslate">partitionsSpec</code> will be provided instead. The deprecated values will be removed in the next major release after the upcoming one.</p>
<h3 dir="auto">Test plan</h3>
<p dir="auto">Unit tests and integration tests will be implemented. I will also test this with our internal cluster once it's ready.</p>
<h3 dir="auto">Future work</h3>
<ul dir="auto">
<li>The optimal parallelism for the phase 2 should be able to be determined automatically by collecting statistics during the phase 1.</li>
<li>To avoid "too many open files" problem, middleManager should be able to smoosh the intermediary segments into several large files.</li>
<li>If rollup is set, it could be better to combine intermediate data in middleManager before sending them. It would be similar to Hadoop's combiner.
<ul dir="auto">
<li>This could be implemented to support seamless incremental segment merge in middleManager.</li>
</ul>
</li>
<li>In Phase 1, tasks might skip index generation for faster shuffle. In this case, Phase 2 tasks should be able to generate the complete indexes.</li>
</ul>