public DataStream<RowData> buildUnkeyedTableSource(String scanStartupMode) {
      scanStartupMode = scanStartupMode == null ? null : scanStartupMode.toLowerCase();
      Preconditions.checkArgument(
          Objects.isNull(scanStartupMode)
              || Objects.equals(scanStartupMode, MixedFormatValidator.SCAN_STARTUP_MODE_EARLIEST)
              || Objects.equals(scanStartupMode, MixedFormatValidator.SCAN_STARTUP_MODE_LATEST),
          String.format(
              "only support %s, %s when %s is %s",
              MixedFormatValidator.SCAN_STARTUP_MODE_EARLIEST,
              MixedFormatValidator.SCAN_STARTUP_MODE_LATEST,
              MixedFormatValidator.MIXED_FORMAT_READ_MODE,
              MixedFormatValidator.MIXED_FORMAT_READ_FILE));
      org.apache.iceberg.flink.source.FlinkSource.Builder builder =
          org.apache.iceberg.flink.source.FlinkSource.forRowData()
              .env(env)
              .project(org.apache.amoro.flink.FlinkSchemaUtil.filterWatermark(projectedSchema))
              .tableLoader(tableLoader)
              .filters(filters)
              .properties(properties)
              .flinkConf(flinkConf)
              .limit(limit);
      if (MixedFormatValidator.SCAN_STARTUP_MODE_LATEST.equalsIgnoreCase(scanStartupMode)) {
        Optional<Snapshot> startSnapshotOptional =
            Optional.ofNullable(tableLoader.loadTable().currentSnapshot());
        if (startSnapshotOptional.isPresent()) {
          Snapshot snapshot = startSnapshotOptional.get();
          LOG.info(
              "Get starting snapshot id {} based on scan startup mode {}",
              snapshot.snapshotId(),
              scanStartupMode);
          builder.startSnapshotId(snapshot.snapshotId());
        }
      }
      DataStream<RowData> origin = builder.build();
      return wrapKrb(origin).assignTimestampsAndWatermarks(watermarkStrategy);
    }
An util class create mixed-format source data stream.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/table/FlinkSource.java