private void initHTLoadingService(long estKeyCount) {
    if (estKeyCount < VectorMapJoinFastHashTable.FIRST_SIZE_UP) {
      // Avoid many small HTs that will rehash multiple times causing GCs
      this.numLoadThreads = 1;
    } else {
      int initialValue = HiveConf.getIntVar(hconf, HiveConf.ConfVars.HIVE_MAPJOIN_PARALEL_HASHTABLE_THREADS);
      Preconditions.checkArgument(initialValue > 0, "The number of HT-loading-threads should be positive.");

      int adjustedValue = Integer.highestOneBit(initialValue);
      if (initialValue != adjustedValue) {
        LOG.info("Adjust the number of HT-loading-threads to {}. (Previous value: {})",
            adjustedValue, initialValue);
      }

      this.numLoadThreads = adjustedValue;
    }
    this.totalEntries = new LongAccumulator(Long::sum, 0L);
    this.loadExecService = Executors.newFixedThreadPool(numLoadThreads,
        new ThreadFactoryBuilder()
            .setDaemon(true)
            .setPriority(Thread.NORM_PRIORITY)
            .setNameFormat("HT-Load-Thread-%d")
            .build());
    // Reuse HashTableElementBatches to reduce GC pressure
    this.batchPool = new FixedSizedObjectPool<>(8 * numLoadThreads, new Pool.PoolObjectHelper<HashTableElementBatch>() {
      @Override
      public HashTableElementBatch create() {
        return new HashTableElementBatch();
      }

      @Override
      public void resetBeforeOffer(HashTableElementBatch elementBatch) {
        elementBatch.reset();
      }
    });
    this.elementBatches = new HashTableElementBatch[numLoadThreads];
    this.loadBatchQueues = new BlockingQueue[numLoadThreads];
    for (int i = 0; i < numLoadThreads; i++) {
      loadBatchQueues[i] = new LinkedBlockingQueue();
      elementBatches[i] = batchPool.take();
    }
  }
HashTableLoader for Tez constructs the hashtable from records read from
a broadcast edge.
ql/src/java/org/apache/hadoop/hive/ql/exec/vector/mapjoin/fast/VectorMapJoinFastHashTableLoader.java