static VertexWithInputConfig join(
            DAG dag,
            String mapName,
            String tableName,
            JetJoinInfo joinInfo,
            KvRowProjector.Supplier rightRowProjectorSupplier
    ) {
        int leftEquiJoinPrimitiveKeyIndex = leftEquiJoinPrimitiveKeyIndex(joinInfo, rightRowProjectorSupplier.paths());
        if (leftEquiJoinPrimitiveKeyIndex > -1) {
            // This branch handles the case when there's an equi-join condition for the __key field.
            // For example: SELECT * FROM left [LEFT] JOIN right ON left.field1=right.__key
            // In this case we'll use map.get() for the right map to get the matching entry by key and evaluate the
            // remaining conditions on the returned row.
            return new VertexWithInputConfig(
                    dag.newUniqueVertex(
                            "Join(Lookup-" + tableName + ")",
                            new JoinByPrimitiveKeyProcessorSupplier(
                                    joinInfo.isInner(),
                                    leftEquiJoinPrimitiveKeyIndex,
                                    joinInfo.condition(),
                                    mapName,
                                    rightRowProjectorSupplier
                            )
                    ),
                    edge -> edge.distributed().partitioned(extractPrimitiveKeyFn(leftEquiJoinPrimitiveKeyIndex))
            );
        } else if (joinInfo.isEquiJoin()) {
            // This branch handles the case when there's an equi-join, but not for __key (that was handled above)
            // For example: SELECT * FROM left JOIN right ON left.field1=right.field1
            // In this case we'll construct a com.hazelcast.query.Predicate that will find matching rows using
            // the `map.entrySet(predicate)` method.
            assert joinInfo.isLeftOuter() || joinInfo.isInner();
            return new VertexWithInputConfig(
                    dag.newUniqueVertex(
                            "Join(Predicate-" + tableName + ")",
                            JoinByEquiJoinProcessorSupplier.supplier(joinInfo, mapName, rightRowProjectorSupplier)
                    ),
                    edge -> {
                        // In case of an inner join we'll use `entrySet(predicate, partitionIdSet)` - we'll fan-out each
                        // left item to all members and each member will query a subset of partitions (the local ones).
                        // Otherwise, a default edge is used (local unicast)
                        if (joinInfo.isInner()) {
                            edge.distributed().fanout();
                        }
                    });
        } else {
            // This is the fallback case when there's not an equi-join: it can be a cross-join or join on
            // another condition. For example:
            // SELECT * FROM houses h JOIN renters r WHERE h.rent BETWEEN r.min_rent AND r.max_rent
            return new VertexWithInputConfig(
                    dag.newUniqueVertex(
                            "Join(Scan-" + tableName + ")",
                            new JoinScanProcessorSupplier(joinInfo, mapName, rightRowProjectorSupplier)
                    )
            );
        }
        // TODO: detect and handle always-false condition ?
    }
@Test
    @Parameters(method = "joinTypes")
    public void test_joinByPrimitiveKey(JoinRelType joinType) {
        // given
        given(rightRowProjectorSupplier.paths()).willReturn(new QueryPath[]{KEY_PATH});
        given(dag.newUniqueVertex(contains("Lookup"), isA(JoinByPrimitiveKeyProcessorSupplier.class))).willReturn(vertex);

        // when
        VertexWithInputConfig vertexWithConfig = Joiner.join(
                dag,
                "imap-name",
                "table-name",
                joinInfo(joinType, new int[]{0}, new int[]{0}),
                rightRowProjectorSupplier
        );

        // then
        assertThat(vertexWithConfig.vertex()).isEqualTo(vertex);
        assertThat(vertexWithConfig.configureEdgeFn()).isNotNull();
    }