static SqlIOConfig resolveSQlIOForTable(RelNode relNode, Map<String, SqlIOConfig> systemStreamConfigBySource) {
    // Let's recursively get to the TableScan node to identify IO for the table.

    if (relNode instanceof HepRelVertex) {
      return resolveSQlIOForTable(((HepRelVertex) relNode).getCurrentRel(), systemStreamConfigBySource);
    }

    if (relNode instanceof LogicalProject) {
      return resolveSQlIOForTable(((LogicalProject) relNode).getInput(), systemStreamConfigBySource);
    }

    if (relNode instanceof LogicalFilter) {
      return resolveSQlIOForTable(((LogicalFilter) relNode).getInput(), systemStreamConfigBySource);
    }

    // We return null for table IO as the table seems to be involved in another join. The output of stream-table join
    // is considered a stream. Hence, we return null for the table.
    if (relNode instanceof LogicalJoin && relNode.getInputs().size() > 1) {
      return null;
    }

    if (!(relNode instanceof TableScan)) {
      throw new SamzaException(String.format("Unsupported query. relNode %s is not of type TableScan.",
          relNode.toString()));
    }

    String sourceName = SqlIOConfig.getSourceFromSourceParts(relNode.getTable().getQualifiedName());
    SqlIOConfig sourceConfig = systemStreamConfigBySource.get(sourceName);
    if (sourceConfig == null) {
      throw new SamzaException("Unsupported source found in join statement: " + sourceName);
    }
    return sourceConfig;
  }
Traverse the tree of expression and validate. Only allowed predicate is conjunction of exp1 = exp2
@param rexPredicate Rex Condition
@param conjunctionList result container to pull result form recursion stack.
samza-sql/src/main/java/org/apache/samza/sql/translator/JoinTranslator.java