@Override
  public void load(MapJoinTableContainer[] mapJoinTables,
      MapJoinTableContainerSerDe[] mapJoinTableSerdes)
      throws HiveException {

    Map<Integer, String> parentToInput = desc.getParentToInput();
    Map<Integer, Long> parentKeyCounts = desc.getParentKeyCounts();

    MemoryMonitorInfo memoryMonitorInfo = desc.getMemoryMonitorInfo();
    boolean doMemCheck = false;
    long effectiveThreshold = 0;
    if (memoryMonitorInfo != null) {
      effectiveThreshold = memoryMonitorInfo.getEffectiveThreshold(desc.getMaxMemoryAvailable());

      // hash table loading happens in server side, LlapDecider could kick out some fragments to run outside of LLAP.
      // Flip the flag at runtime in case if we are running outside of LLAP
      if (!LlapDaemonInfo.INSTANCE.isLlap()) {
        memoryMonitorInfo.setLlap(false);
      }
      if (memoryMonitorInfo.doMemoryMonitoring()) {
        doMemCheck = true;
        LOG.info("Memory monitoring for hash table loader enabled. {}", memoryMonitorInfo);
      }
    }

    if (!doMemCheck) {
      LOG.info("Not doing hash table memory monitoring. {}", memoryMonitorInfo);
    }
    for (int pos = 0; pos < mapJoinTables.length; pos++) {
      if (pos == desc.getPosBigTable()) {
        continue;
      }

      String inputName = parentToInput.get(pos);
      LogicalInput input = tezContext.getInput(inputName);

      try {
        input.start();
        tezContext.getTezProcessorContext().waitForAnyInputReady(
            Collections.<Input> singletonList(input));
      } catch (Exception e) {
        throw new HiveException(e);
      }

      try {
        KeyValueReader kvReader = (KeyValueReader) input.getReader();

        Long keyCountObj = parentKeyCounts.get(pos);
        long estKeyCount = (keyCountObj == null) ? -1 : keyCountObj;

        long inputRecords = -1;
        try {
          //TODO : Need to use class instead of string.
          // https://issues.apache.org/jira/browse/HIVE-23981
          inputRecords = ((AbstractLogicalInput) input).getContext().getCounters().
                  findCounter("org.apache.tez.common.counters.TaskCounter",
                          "APPROXIMATE_INPUT_RECORDS").getValue();
        } catch (Exception e) {
          LOG.debug("Failed to get value for counter APPROXIMATE_INPUT_RECORDS", e);
        }
        long keyCount = Math.max(estKeyCount, inputRecords);
        initHTLoadingService(keyCount);

        VectorMapJoinFastTableContainer tableContainer =
            new VectorMapJoinFastTableContainer(desc, hconf, keyCount, numLoadThreads);

        LOG.info("Loading hash table for input: {} cacheKey: {} tableContainer: {} smallTablePos: {} " +
                "estKeyCount : {} keyCount : {}", inputName, cacheKey,
                tableContainer.getClass().getSimpleName(), pos, estKeyCount, keyCount);

        tableContainer.setSerde(null, null); // No SerDes here.
        // Submit parallel loading Threads
        submitQueueDrainThreads(tableContainer);

        long receivedEntries = 0;
        long startTime = System.currentTimeMillis();
        while (kvReader.next()) {
          BytesWritable currentKey = (BytesWritable) kvReader.getCurrentKey();
          BytesWritable currentValue = (BytesWritable) kvReader.getCurrentValue();
          long hashCode = tableContainer.getHashCode(currentKey);
          int partitionId = (int) ((numLoadThreads - 1) & hashCode); // numLoadThreads divisor must be a power of 2!
          // call getBytes as copy is called later
          HashTableElement h = new HashTableElement(hashCode, currentValue.copyBytes(), currentKey.copyBytes());
          if (elementBatches[partitionId].addElement(h)) {
            loadBatchQueues[partitionId].add(elementBatches[partitionId]);
            elementBatches[partitionId] = batchPool.take();
          }
          receivedEntries++;
          if (doMemCheck && (receivedEntries % memoryMonitorInfo.getMemoryCheckInterval() == 0)) {
            final long estMemUsage = tableContainer.getEstimatedMemorySize();
            if (estMemUsage > effectiveThreshold) {
              String msg = "Hash table loading exceeded memory limits for input: " + inputName +
                  " numEntries: " + receivedEntries + " estimatedMemoryUsage: " + estMemUsage +
                  " effectiveThreshold: " + effectiveThreshold + " memoryMonitorInfo: " + memoryMonitorInfo;
                LOG.error(msg);
                throw new MapJoinMemoryExhaustionError(msg);
              } else {
              LOG.info(
                  "Checking hash table loader memory usage for input: {} numEntries: {} "
                      + "estimatedMemoryUsage: {} effectiveThreshold: {}",
                  inputName, receivedEntries, estMemUsage, effectiveThreshold);
            }
          }
        }

        LOG.info("Finished loading the queue for input: {} waiting {} minutes for TPool shutdown", inputName, 2);
        addQueueDoneSentinel();
        loadExecService.shutdown();

        if (!loadExecService.awaitTermination(2, TimeUnit.MINUTES)) {
          throw new HiveException("Failed to complete the hash table loader. Loading timed out.");
        }
        batchPool.clear();
        LOG.info("Total received entries: {} Threads {} HT entries: {}", receivedEntries, numLoadThreads, totalEntries.get());

        long delta = System.currentTimeMillis() - startTime;
        htLoadCounter.increment(delta);

        tableContainer.seal();
        mapJoinTables[pos] = tableContainer;
        if (doMemCheck) {
          LOG.info("Finished loading hash table for input: {} cacheKey: {} numEntries: {} " +
              "estimatedMemoryUsage: {} Load Time : {} ", inputName, cacheKey, receivedEntries,
            tableContainer.getEstimatedMemorySize(), delta);
        } else {
          LOG.info("Finished loading hash table for input: {} cacheKey: {} numEntries: {} Load Time : {} ",
                  inputName, cacheKey, receivedEntries, delta);
        }
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new HiveException(e);
      } catch (HiveException e) {
        throw e;
      } catch (Exception e) {
        throw new HiveException(e);
      } finally {
        if (loadExecService != null && !loadExecService.isTerminated()) {
          loadExecService.shutdownNow();
        }
      }
    }
  }
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