private TableSchema getProjectedSchema() {
    if (projectedFields == null) {
      return tableSchema;
    } else {
      String[] fullNames = tableSchema.getFieldNames();
      DataType[] fullTypes = tableSchema.getFieldDataTypes();

      String[] projectedColumns =
          Arrays.stream(projectedFields).mapToObj(i -> fullNames[i]).toArray(String[]::new);
      TableSchema.Builder builder =
          TableSchema.builder()
              .fields(
                  projectedColumns,
                  Arrays.stream(projectedFields)
                      .mapToObj(i -> fullTypes[i])
                      .toArray(DataType[]::new));
      boolean dimTable =
          CompatibleFlinkPropertyUtil.propertyAsBoolean(
              table.properties(),
              MixedFormatValidator.DIM_TABLE_ENABLE.key(),
              MixedFormatValidator.DIM_TABLE_ENABLE.defaultValue());
      if (dimTable) {
        builder.watermark(tableSchema.getWatermarkSpecs().get(0));
      }

      TableSchema ts = builder.build();
      LOG.info("TableSchema after projection:{}", ts);
      return ts;
    }
  }
Flink table api that generates mixed-format base/change file source operators.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/table/MixedFormatFileSource.java