public List<Record> results() {
      Map<Object, Record> targetMap =
          target.stream().collect(Collectors.toMap(keyExtractor, Function.identity()));
      Map<Object, Record> sourceMap =
          source.stream().collect(Collectors.toMap(keyExtractor, Function.identity()));

      for (Object key : sourceMap.keySet()) {
        Record source = sourceMap.get(key);
        Record target = targetMap.get(key);
        if (target != null) {
          matchActions.stream()
              .filter(a -> a.getLeft().apply(target, source))
              .map(Pair::getRight)
              .findFirst()
              .ifPresent(
                  trans -> {
                    Record r = trans.apply(target, source);
                    if (r == null) {
                      targetMap.remove(key);
                    } else {
                      targetMap.put(key, r);
                    }
                  });

        } else {
          notMatchedActions.stream()
              .filter(a -> a.getLeft().test(source))
              .map(Pair::getRight)
              .findFirst()
              .ifPresent(
                  trans -> {
                    targetMap.put(key, trans.apply(source));
                  });
        }
      }

      return Lists.newArrayList(targetMap.values());
    }
if condition(source) test for true, then action(source) will be added to target records

@param condition condition(source): bool
@param action action(source) for insert, result is not-null
@return this
amoro-format-mixed/amoro-mixed-spark/amoro-mixed-spark-3-common/src/test/java/org/apache/amoro/spark/test/utils/ExpectResultUtil.java