Package org.apache.druid.frame.processor

Class SuperSorter

java.lang.Object
org.apache.druid.frame.processor.SuperSorter
public class SuperSorter extends Object
Sorts and partitions a dataset using parallel, possibly-external merge sort. Input is provided as a set of ReadableFrameChannel and output is provided as OutputChannels. Work is performed on a provided FrameProcessorExecutor. The most central point for SuperSorter logic is the runWorkersIfPossible() method, which determines what needs to be done next based on the current state of the SuperSorter. First, input channels are read into inputBuffer using FrameChannelBatcher, launched via runNextBatcher(). If the sorter finishes reading its input before reaching getMaxInputBufferFramesForDirectMerging() number of frames in the inputBuffer, the sorter operates in "direct mode". In this mode, totalMergingLevels is 1. Mergers launched by runNextDirectMerger() directly merge input frames into output channels, fully in-memory. No temporary files are used. Each direct merger reads *all* frames, but only rows corresponding to a single output partition. Otherwise, the sorter operates in "external mode". In this mode, totalMergingLevels is 2 or more. Logic for external mode is: 1) Read inputs into inputBuffer using FrameChannelBatcher (same as direct mode). 2) Merge and write frames from inputBuffer into FrameFile scratch files using FrameChannelMerger launched via runNextLevelZeroMerger(). 3a) Merge level 0 scratch files into level 1 scratch files using FrameChannelMerger launched from runNextMiddleMerger(), processing up to maxChannelsPerMerger files per merger. Continue this process through increasing level numbers, with the size of scratch files increasing by a factor of maxChannelsPerMerger each level. 3b) For the penultimate level, the FrameChannelMerger launched by runNextMiddleMerger() writes partitioned FrameFile scratch files. The penultimate level cannot be written until outputPartitionsFuture resolves, so if it has not resolved yet by this point, the SuperSorter pauses. The SuperSorter resumes and writes the penultimate level's files when the future resolves. 4) Write the final level using FrameChannelMerger launched from runNextUltimateMerger(). Outputs for this level are written to channels provided by outputChannelFactory, rather than scratch files. At all points, higher level processing is preferred over lower-level processing. Writing to final output files is preferred over intermediate, and writing to intermediate files is preferred over reading inputs. These preferences ensure that the amount of data buffered up in memory does not grow too large. Potential future work (things we could optimize if necessary): - Identify sorted runs of frames in the inputBuffer, group them into smaller sets of channels. - Combine (for example: aggregate) while merging.

  • Field Details

  • Constructor Details

    • SuperSorter

      public SuperSorter(List<ReadableFrameChannel> inputChannels, FrameReader frameReader, List<KeyColumn> sortKey, com.google.common.util.concurrent.ListenableFuture<ClusterByPartitions> outputPartitionsFuture, FrameProcessorExecutor exec, FrameProcessorDecorator processorDecorator, OutputChannelFactory outputChannelFactory, OutputChannelFactory intermediateOutputChannelFactory, FrameType outputFrameType, int maxActiveProcessors, int maxChannelsPerMerger, long rowLimit, @Nullable String cancellationId, SuperSorterProgressTracker superSorterProgressTracker, boolean removeNullBytes)
      Initializes a SuperSorter.
      Parameters:
      inputChannels - input channels. All frames in these channels must be sorted according to the ClusterBy.getColumns(), or else sorting will not produce correct output.
      frameReader - frame reader for the input channels
      sortKey - desired sorting order
      outputPartitionsFuture - a future that resolves to the desired output partitions. Sorting will block prior to writing out final outputs until this future resolves. However, the sorter will be able to read all inputs even if this future is unresolved. If output need not be partitioned, use ClusterByPartitions.oneUniversalPartition(). In this case a single sorted channel is generated.
      exec - executor to perform work in
      outputChannelFactory - factory for partitioned, sorted output channels
      intermediateOutputChannelFactory - factory for intermediate data produced by sorting levels
      outputFrameType - frame type for all output channels (intermediate and final)
      maxActiveProcessors - maximum number of merging processors to execute at once in the provided FrameProcessorExecutor
      maxChannelsPerMerger - maximum number of channels to merge at once, for regular mergers (does not apply to direct mergers; see getMaxInputBufferFramesForDirectMerging())
      rowLimit - limit to apply during sorting. The limit is applied across all partitions, not to each partition individually. Use UNLIMITED if there is no limit.
      cancellationId - cancellation id to use when running processors in the provided FrameProcessorExecutor.
      superSorterProgressTracker - progress tracker

  • Method Details

    • run

      public com.google.common.util.concurrent.ListenableFuture<OutputChannels> run()
      Starts sorting. Can only be called once. Work is performed in the FrameProcessorExecutor that was passed to the constructor. Returns a future containing partitioned sorted output channels.

    • stateString

      public String stateString()
      Returns a string encapsulating the current state of this object.