public DataStream<RowData> build() {
      Preconditions.checkNotNull(env, "StreamExecutionEnvironment should not be null");
      loadTableIfNeeded();

      if (mixedTable.isUnkeyedTable()) {
        String scanStartupMode = properties.get(MixedFormatValidator.SCAN_STARTUP_MODE.key());
        return buildUnkeyedTableSource(scanStartupMode);
      }

      boolean dimTable =
          CompatibleFlinkPropertyUtil.propertyAsBoolean(
              properties,
              MixedFormatValidator.DIM_TABLE_ENABLE.key(),
              MixedFormatValidator.DIM_TABLE_ENABLE.defaultValue());
      RowType rowType;

      if (projectedSchema == null) {
        contextBuilder.project(mixedTable.schema());
        rowType = FlinkSchemaUtil.convert(mixedTable.schema());
      } else {
        contextBuilder.project(
            FlinkSchemaUtil.convert(
                mixedTable.schema(),
                org.apache.amoro.flink.FlinkSchemaUtil.filterWatermark(projectedSchema)));
        // If dim table is enabled, we reserve a RowTime field in Emitter.
        if (dimTable) {
          rowType = org.apache.amoro.flink.FlinkSchemaUtil.toRowType(projectedSchema);
        } else {
          rowType =
              org.apache.amoro.flink.FlinkSchemaUtil.toRowType(
                  org.apache.amoro.flink.FlinkSchemaUtil.filterWatermark(projectedSchema));
        }
      }
      MixedFormatScanContext scanContext =
          contextBuilder.fromProperties(properties).batchMode(batchMode).build();

      RowDataReaderFunction rowDataReaderFunction =
          new RowDataReaderFunction(
              flinkConf,
              mixedTable.schema(),
              scanContext.project(),
              mixedTable.asKeyedTable().primaryKeySpec(),
              scanContext.nameMapping(),
              scanContext.caseSensitive(),
              mixedTable.io());

      int scanParallelism =
          flinkConf.getOptional(MixedFormatValidator.SCAN_PARALLELISM).orElse(env.getParallelism());
      DataStreamSource<RowData> sourceStream =
          env.fromSource(
                  new MixedFormatSource<>(
                      tableLoader,
                      scanContext,
                      rowDataReaderFunction,
                      InternalTypeInfo.of(rowType),
                      mixedTable.name(),
                      dimTable),
                  watermarkStrategy,
                  MixedFormatSource.class.getName())
              .setParallelism(scanParallelism);
      context.generateUid(MIXED_FORMAT_FILE_TRANSFORMATION).ifPresent(sourceStream::uid);
      return sourceStream;
    }
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