void translate(final LogicalJoin join, final TranslatorContext translatorContext) {
    JoinInputNode.InputType inputTypeOnLeft = JoinInputNode.getInputType(join.getLeft(),
        translatorContext.getExecutionContext().getSamzaSqlApplicationConfig().getInputSystemStreamConfigBySource());
    JoinInputNode.InputType inputTypeOnRight = JoinInputNode.getInputType(join.getRight(),
        translatorContext.getExecutionContext().getSamzaSqlApplicationConfig().getInputSystemStreamConfigBySource());

    // Do the validation of join query
    validateJoinQuery(join, inputTypeOnLeft, inputTypeOnRight);

    // At this point, one of the sides is a table. Let's figure out if it is on left or right side.
    boolean isTablePosOnRight = inputTypeOnRight != JoinInputNode.InputType.STREAM;

    // stream and table keyIds are used to extract the join condition field (key) names and values out of the stream
    // and table records.
    List<Integer> streamKeyIds = new LinkedList<>();
    List<Integer> tableKeyIds = new LinkedList<>();

    // Fetch the stream and table indices corresponding to the fields given in the join condition.

    final int leftSideSize = join.getLeft().getRowType().getFieldCount();
    final int tableStartIdx = isTablePosOnRight ? leftSideSize : 0;
    final int streamStartIdx = isTablePosOnRight ? 0 : leftSideSize;
    final int tableEndIdx = isTablePosOnRight ? join.getRowType().getFieldCount() : leftSideSize;
    join.getCondition().accept(new RexShuttle() {
      @Override
      public RexNode visitInputRef(RexInputRef inputRef) {
        validateJoinKeyType(inputRef); // Validate the type of the input ref.
        int index = inputRef.getIndex();
        if (index >= tableStartIdx && index < tableEndIdx) {
          tableKeyIds.add(index - tableStartIdx);
        } else {
          streamKeyIds.add(index - streamStartIdx);
        }
        return inputRef;
      }
    });
    Collections.sort(tableKeyIds);
    Collections.sort(streamKeyIds);

    // Get the two input nodes (stream and table nodes) for the join.
    JoinInputNode streamNode = new JoinInputNode(isTablePosOnRight ? join.getLeft() : join.getRight(), streamKeyIds,
        isTablePosOnRight ? inputTypeOnLeft : inputTypeOnRight, !isTablePosOnRight);
    JoinInputNode tableNode = new JoinInputNode(isTablePosOnRight ? join.getRight() : join.getLeft(), tableKeyIds,
        isTablePosOnRight ? inputTypeOnRight : inputTypeOnLeft, isTablePosOnRight);

    MessageStream<SamzaSqlRelMessage> inputStream = translatorContext.getMessageStream(streamNode.getRelNode().getId());
    Table table = getTable(tableNode, translatorContext);

    MessageStream<SamzaSqlRelMessage> outputStream =
        joinStreamWithTable(inputStream, table, streamNode, tableNode, join, translatorContext);

    translatorContext.registerMessageStream(join.getId(), outputStream);

    outputStream.map(outputMetricsMF);
  }
Translator to translate the LogicalJoin node in the relational graph to the corresponding StreamGraph
implementation.
Join is supported with the following caveats:
  1. Only stream-table joins are supported. No stream-stream joins.
  2. Only Equi-joins are supported. No theta-joins.
  3. Inner joins, Left and Right outer joins are supported. No cross joins, full outer joins or natural joins.
  4. Join condition with a constant is not supported.
  5. Compound join condition with only AND operator is supported. AND operator with a constant is not supported. No
     support for OR operator or any other operator in the join condition.
For local table, we always repartition both the stream to be joined and the stream denoted as table by the key(s)
specified in the join condition.
samza-sql/src/main/java/org/apache/samza/sql/translator/JoinTranslator.java