Repository: apache/flink
File: flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/utils/DeltaJoinUtil.java (Java)

Code before:
            return false;
        }

        TableSourceTable tableSourceTable =
                ((StreamPhysicalTableSourceScan) tableScan).tableSourceTable();

        // source with ability specs are not supported yet
        if (tableSourceTable.abilitySpecs().length != 0) {
            return false;
        }

        DynamicTableSource source = tableSourceTable.tableSource();
        // the source must also be a lookup source
        if (!(source instanceof LookupTableSource)) {
            return false;
        }

        int[][] idxsOfAllIndexes = getColumnIndicesOfAllTableIndexes(tableSourceTable);
        if (idxsOfAllIndexes.length == 0) {
            return false;
        }
        // the source must have at least one index, and the join key contains one index
        Set<Integer> lookupKeysSet = Arrays.stream(lookupKeys).boxed().collect(Collectors.toSet());

        boolean lookupKeyContainsOneIndex = false;
        for (int[] idxsOfIndex : idxsOfAllIndexes) {
            Preconditions.checkState(idxsOfIndex.length > 0);

            // ignore the field order of the index
            boolean containsIndex = Arrays.stream(idxsOfIndex).allMatch(lookupKeysSet::contains);
            if (containsIndex) {
                lookupKeyContainsOneIndex = true;
                break;
            }
        }
        if (!lookupKeyContainsOneIndex) {
            return false;
        }

        // the lookup source must support async lookup
        return LookupJoinUtil.isAsyncLookup(
                tableSourceTable,
                lookupKeysSet,
                null, // hint
                false, // upsertMaterialize
                false // preferCustomShuffle
                );
    }

    private static TableScan getTableScan(RelNode node) {
        node = unwrapNode(node, true);

Review comment (refactor):
Can we simplify this for-loop into simple call, e.g., 
```
boolean lookupKeyContainsOneIndex = Arrays.stream(idxsOfAllIndexes)
        .peek(idxs -> Preconditions.checkState(idxs.length > 0))
        .anyMatch(idxs -> Arrays.stream(idxs).allMatch(lookupKeysSet::contains));
```
?

Code after:
            return false;
        }

        TableSourceTable tableSourceTable =
                ((StreamPhysicalTableSourceScan) tableScan).tableSourceTable();

        // source with ability specs are not supported yet
        if (tableSourceTable.abilitySpecs().length != 0) {
            return false;
        }

        DynamicTableSource source = tableSourceTable.tableSource();
        // the source must also be a lookup source
        if (!(source instanceof LookupTableSource)) {
            return false;
        }

        int[][] idxsOfAllIndexes = getColumnIndicesOfAllTableIndexes(tableSourceTable);
        if (idxsOfAllIndexes.length == 0) {
            return false;
        }
        // the source must have at least one index, and the join key contains one index
        Set<Integer> lookupKeysSet = Arrays.stream(lookupKeys).boxed().collect(Collectors.toSet());

        boolean lookupKeyContainsOneIndex =
                Arrays.stream(idxsOfAllIndexes)
                        .peek(idxsOfIndex -> Preconditions.checkState(idxsOfIndex.length > 0))
                        .anyMatch(
                                idxsOfIndex ->
                                        Arrays.stream(idxsOfIndex)
                                                .allMatch(lookupKeysSet::contains));
        if (!lookupKeyContainsOneIndex) {
            return false;
        }

        // the lookup source must support async lookup
        return LookupJoinUtil.isAsyncLookup(
                tableSourceTable,
                lookupKeysSet,
                null, // hint
                false, // upsertMaterialize
                false // preferCustomShuffle
                );
    }

    private static TableScan getTableScan(RelNode node) {
        node = unwrapNode(node, true);
        // support to get table across more nodes if we support more nodes in
        // `ALL_SUPPORTED_DELTA_JOIN_UPSTREAM_NODES`
        if (node instanceof StreamPhysicalExchange) {
            return getTableScan(((StreamPhysicalExchange) node).getInput());