@Override
    public DataType project(DataType dataType) {
      final List<RowType.RowField> updatedFields = new ArrayList<>();
      final List<DataType> updatedChildren = new ArrayList<>();
      Set<String> nameDomain = new HashSet<>();
      int duplicateCount = 0;
      for (int[] indexPath : this.projection) {
        DataType fieldType = dataType.getChildren().get(indexPath[0]);
        LogicalType fieldLogicalType = fieldType.getLogicalType();
        StringBuilder builder =
            new StringBuilder(
                ((RowType) dataType.getLogicalType()).getFieldNames().get(indexPath[0]));
        for (int index = 1; index < indexPath.length; index++) {
          Preconditions.checkArgument(
              fieldLogicalType.getTypeRoot() == ROW, "Row data type expected.");
          RowType rowtype = ((RowType) fieldLogicalType);
          builder.append("_").append(rowtype.getFieldNames().get(indexPath[index]));
          fieldLogicalType = rowtype.getFields().get(indexPath[index]).getType();
          fieldType = fieldType.getChildren().get(indexPath[index]);
        }
        String path = builder.toString();
        while (nameDomain.contains(path)) {
          path = builder.append("_$").append(duplicateCount++).toString();
        }
        updatedFields.add(new RowType.RowField(path, fieldLogicalType));
        updatedChildren.add(fieldType);
        nameDomain.add(path);
      }
      return new FieldsDataType(
          new RowType(dataType.getLogicalType().isNullable(), updatedFields),
          dataType.getConversionClass(),
          updatedChildren);
    }
Create a {@link Projection} of a field range.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/util/Projection.java