private Iterator<InputSplit> getInputSplits() {

    return new Iterator<InputSplit>() {

      private InputSplit nextSplit;

      private boolean exhausted;

      @Override
      public boolean hasNext() {
        if (exhausted) {
          return false;
        }

        if (nextSplit != null) {
          return true;
        }

        final InputSplit split;
        try {
          split = provider.getNextInputSplit(getRuntimeContext().getUserCodeClassLoader());
        } catch (InputSplitProviderException e) {
          throw new RuntimeException("Could not retrieve next input split.", e);
        }

        if (split != null) {
          this.nextSplit = split;
          return true;
        } else {
          exhausted = true;
          return false;
        }
      }

      @Override
      public InputSplit next() {
        if (this.nextSplit == null && !hasNext()) {
          throw new NoSuchElementException();
        }

        final InputSplit tmp = this.nextSplit;
        this.nextSplit = null;
        return tmp;
      }

      @Override
      public void remove() {
        throw new UnsupportedOperationException();
      }
    };
  }
Returns the {@code InputFormat}. This is only needed because we need to set the input split
assigner on the {@code StreamGraph}.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/table/UnkeyedInputFormatSourceFunction.java