private DataStream<RowData> wrapKrb(DataStream<RowData> ds) {
      IcebergClassUtil.clean(env);
      Transformation origin = ds.getTransformation();
      int scanParallelism =
          flinkConf
              .getOptional(MixedFormatValidator.SCAN_PARALLELISM)
              .orElse(origin.getParallelism());

      if (origin instanceof OneInputTransformation) {
        OneInputTransformation<RowData, RowData> tf =
            (OneInputTransformation<RowData, RowData>) ds.getTransformation();
        OneInputStreamOperatorFactory op = (OneInputStreamOperatorFactory) tf.getOperatorFactory();
        ProxyFactory<FlinkInputFormat> inputFormatProxyFactory =
            IcebergClassUtil.getInputFormatProxyFactory(op, mixedTable.io(), mixedTable.schema());

        if (tf.getInputs().isEmpty()) {
          return env.addSource(
                  new UnkeyedInputFormatSourceFunction(inputFormatProxyFactory, tf.getOutputType()))
              .setParallelism(scanParallelism);
        }

        LegacySourceTransformation tfSource = (LegacySourceTransformation) tf.getInputs().get(0);
        StreamSource source = tfSource.getOperator();
        SourceFunction function = IcebergClassUtil.getSourceFunction(source);

        SourceFunction functionProxy =
            (SourceFunction) ProxyUtil.getProxy(function, mixedTable.io());
        DataStreamSource sourceStream =
            env.addSource(functionProxy, tfSource.getName(), tfSource.getOutputType());
        context.generateUid(MIXED_FORMAT_FILE_TRANSFORMATION).ifPresent(sourceStream::uid);
        if (sourceStream instanceof ParallelSourceFunction) {
          sourceStream.setParallelism(scanParallelism);
        }
        return sourceStream.transform(
            tf.getName(),
            tf.getOutputType(),
            new UnkeyedInputFormatOperatorFactory(inputFormatProxyFactory));
      }

      LegacySourceTransformation tfSource = (LegacySourceTransformation) origin;
      StreamSource source = tfSource.getOperator();
      InputFormatSourceFunction function =
          (InputFormatSourceFunction) IcebergClassUtil.getSourceFunction(source);

      InputFormat inputFormatProxy =
          (InputFormat) ProxyUtil.getProxy(function.getFormat(), mixedTable.io());
      DataStreamSource sourceStream =
          env.createInput(inputFormatProxy, tfSource.getOutputType())
              .setParallelism(scanParallelism);
      context.generateUid(MIXED_FORMAT_FILE_TRANSFORMATION).ifPresent(sourceStream::uid);
      return sourceStream;
    }
extract op from dataStream, and wrap krb support
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/table/FlinkSource.java