@Override
  public TaskWriter<RowData> create() {
    Preconditions.checkNotNull(
        mask, "Mask should be set first. Invoke setMask() before this method");

    FlinkTaskWriterBuilder builder =
        FlinkTaskWriterBuilder.buildFor(table)
            .withTaskId(taskId)
            .withMask(mask)
            .withTransactionId(transactionId)
            .withFlinkSchema(flinkSchema)
            .withPartitionId(attemptId);
    if (overwrite) {
      return builder.buildWriter(WriteOperationKind.OVERWRITE);
    } else {
      return builder.buildWriter(WriteOperationKind.APPEND);
    }
  }
This is an mixed-format table writer factory.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/write/MixedFormatRowDataTaskWriterFactory.java