/** A vector of double values. */
@PublicEvolving
public interface Vector extends Serializable {

    /** Gets the size of the vector. */
    int size();

    /** Gets the value of the ith element. */
    double get(int i);

    /** Sets the value of the ith element. */
    void set(int i, double value);

    /** Converts the instance to a double array. */
    double[] toArray();

    /** Converts the instance to a dense vector. */
    DenseVector toDense();

    /** Converts the instance to a sparse vector. */
    SparseVector toSparse();

    /** Makes a deep copy of the vector. */
    Vector clone();
}

/** A dense vector of double values. */
@PublicEvolving
public class DenseVector implements Vector {
    public final double[] values;

    public DenseVector(double[] values) {...}

    public DenseVector(int size) {...}
}

/** A sparse vector of double values. */
@PublicEvolving
public class SparseVector implements Vector {
    public final int n;
    public int[] indices;
    public double[] values;

    public SparseVector(int n, int[] indices, double[] values) {...}
}


/** Utility methods for instantiating Vector. */
@PublicEvolving
public class Vectors {

    /** Creates a dense vector from its values. */
    public static Vector dense(double... values) {...}

    /** Creates a sparse vector from its values. */
    public static Vector sparse(int size, int[] indices, double[] values) {...}
}


/** A matrix of double values. */
@PublicEvolving
public interface Matrix extends Serializable {

    /** Gets number of rows. */
    int numRows();

    /** Gets number of columns. */
    int numCols();

    /** Gets value of the (i,j) element. */
    double get(int i, int j);

    /** Converts the instance to a dense matrix. */
    DenseMatrix toDense();
}

/**
 * Column-major dense matrix. The entry values are stored in a single array of doubles with columns
 * listed in sequence.
 */
@PublicEvolving
public class DenseMatrix implements Matrix {

    /**
     * Array for internal storage of elements.
     *
     * <p>The matrix data is stored in column major format internally.
     */
    public final double[] values;

    /**
     * Constructs an m-by-n matrix of zeros.
     *
     * @param numRows Number of rows.
     * @param numCols Number of columns.
     */
    public DenseMatrix(int numRows, int numCols) {...}

    /**
     * Constructs a matrix from a 1-D array. The data in the array should be organized in column
     * major.
     *
     * @param numRows Number of rows.
     * @param numCols Number of cols.
     * @param values One-dimensional array of doubles.
     */
    public DenseMatrix(int numRows, int numCols, double[] values) {...}
}

/** This enum class lists primitive types such as boolean, int, long, etc. */
@PublicEvolving
public enum BasicType {
    BOOLEAN,
    BYTE,
    SHORT,
    INT,
    LONG,
    FLOAT,
    DOUBLE,
    STRING,
    BYTE_STRING;
}


/** This class describes the data type of a value. */
@PublicEvolving
public abstract class DataType {}


/** A DataType representing a single element of the given BasicType. */
@PublicEvolving
public final class ScalarType extends DataType {

    public ScalarType(BasicType elementType) {...}

    public BasicType getElementType() {...}
}


/** A DataType representing a Vector. */
@PublicEvolving
public final class VectorType extends DataType {

    public VectorType(BasicType elementType) {...}

    public BasicType getElementType() {...}
}

 
/** A DataType representing a Matrix. */
@PublicEvolving
public final class MatrixType extends DataType {

    public MatrixType(BasicType elementType) {...}

    public BasicType getElementType() {...}
}

/** Represents an ordered list of values. */
@PublicEvolving
public class Row {
    public Row(List<Object> values) {...}

    /** Returns the value at the given index. */
    public Object get(int index) {...}

    /** Returns the value at the given index as the given type. */
    @SuppressWarnings("unchecked")
    public <T> T getAs(int index) {...}

    /** Adds the value to the end of this row and returns this row. */
    public Row add(Object value) {...}

    /** Returns the number of values in this row. */
    public int size() {...}
}

/**
 * A DataFrame consists of some number of rows, each of which has the same list of column names and
 * data types.
 *
 * <p>All values in a column must have the same data type: integer, float, string etc.
 */
@PublicEvolving
public class DataFrame {
    public DataFrame(List<String> columnNames, List<DataType> dataTypes, List<Row> rows) {...}

    /** Returns a list of the names of all the columns in this DataFrame. */
    public List<String> getColumnNames() {...}

    /**
     * Returns the index of the column with the given name.
     *
     * @throws IllegalArgumentException if the column is not present in this table
     */
    public int getIndex(String name) {...}

    /**
     * Returns the data type of the column with the given name.
     *
     * @throws IllegalArgumentException if the column is not present in this table
     */
    public DataType getDataType(String name) {...}

    /**
     * Adds to this DataFrame a column with the given name, data type, and values.
     *
     * @throws IllegalArgumentException if the number of values is different from the number of rows.
     */
    public DataFrame addColumn(String columnName, DataType dataType, List<Object> values) {...}

    /** Returns all rows of this table. */
    public List<Row> collect() {...}
}

/**
 * A TransformerServable takes a DataFrame as input and produces a DataFrame as the result. It can
 * be used to encode online inference computation logic.
 *
 * <p>NOTE: Every TransformerServable subclass should have a no-arg constructor.
 *
 * <p>NOTE: Every TransformerServable subclass should implement a static method with signature
 * {@code static T load(String path) throws IOException;}, where {@code T} refers to the concrete
 * subclass. This static method should instantiate a new TransformerServable instance based on the
 * data read from the given path.
 *
 * @param <T> The class type of the TransformerServable implementation itself.
 */
@PublicEvolving
public interface TransformerServable<T extends TransformerServable<T>> extends WithParams<T> {
    /**
     * Applies the TransformerServable on the given input DataFrame and returns the result DataFrame.
     *
     * @param input the input data
     * @return the result data
     */
    DataFrame transform(DataFrame input);
}

/**
 * A ModelServable is a TransformerServable with the extra API to set model data.
 *
 * @param <T> The class type of the ModelServable implementation itself.
 */
@PublicEvolving
public interface ModelServable<T extends ModelServable<T>> extends TransformerServable<T> {

    /** Sets model data using the serialized model data from the given input stream. */
    default T setModelData(InputStream... modelData) throws IOException {
        throw new UnsupportedOperationException("this operation is not supported");
    }
}

/**
 * A PipelineModelServable acts as a ModelServable. It consists of an ordered list of
 * TransformerServable, each of which could be a TransformerServable or ModelServable.
 */
@PublicEvolving
public final class PipelineModelServable implements ModelServable<PipelineModelServable> {
    public PipelineModelServable(List<TransformerServable<?>> servables) {...}

    @Override
    public DataFrame transform(DataFrame input) {...}

    @Override
    public PipelineModelServable setModelData(InputStream... modelData) throws IOException {...}

    @Override
    public Map<Param<?>, Object> getParamMap() {...}

    public static PipelineModelServable load(String path) throws IOException {...}
}

/**
 * ...
 *
 * <p>NOTE: If a Transformer has a corresponding {@link TransformerServable}, it should implement 
 * a static method with signature {@code static T loadServable(String path)}, where {@code T} refers to 
 * the concrete subclass of {@link TransformerServable}. This static method should instantiate a 
 * new {@link TransformerServable} instance based on the data read from the given path.
 * 
 * ...
 */
@PublicEvolving
public interface Transformer<T extends Transformer<T>> extends AlgoOperator<T> {...}

@PublicEvolving
public final class PipelineModel implements Model<PipelineModel> {

    /**
     * Whether all stages in the pipeline have corresponding {@link TransformerServable} so that
     * the PipelineModel can be turned into a TransformerServable and used in an online inference program.
     *
     * @return true if all stages have corresponding TransformerServable, false if not.
     */
    public boolean supportServable() {...}
}

@PublicEvolving
public final class GraphModel implements Model<GraphModel> {

     /**
     * Whether all stages in the graph have corresponding {@link TransformerServable} so that
     * the GraphModel can be turned into a TransformerServable and used in an online inference program.
     *
     * @return true if all stages have corresponding TransformerServable, false if not.
     */ 
    public boolean supportServable() {...}
}

void runTrainingOnFlinkCluster(...) {
    Table trainingStream = ...;

    Estimator estimatorA = new EstimatorA(...);      
	Model modelA = estimatorA.fit(trainingStream); 
    PipelineModel pipelineModel = new PipelineModel(Arrays.asList(modelA));

    if (!pipelineModel.supportServable()) {
        throw new RuntimeException("some stages can not be turned into servable ")
    }

    Table modelData = modelA.getModelData()[0];

    // This method writes the data from the given Table to a Kafka topic.
    writeToKafka(modelData, "topicA");
    // Saves model's state/metadata to a remote path.
    modelA.save(remotePath);

    // Executes the operators generated by the Estimator::fit(...), which reads from trainingStream and writes to modelData.
    env.execute();
}

void runOnlineInferenceOnWebServer(...) {
    // Loads model's state/metadata to generate a Servable instance.
    ModelServable servableA = ModelServableA::load(remotePath);

    // Creates a thread to update the latest model data that generated
    // by the above code snippet.
    ((Runnable) (servableA) -> {
        InputStream modelData = readFromKafka("topicA");
        servableA.setModelData(modelData);
    }).run();

    while (True) {
        HttpRequest request = getNextRequest();
        Json input_json = request.getBody();
        // Constructs the input DataFrame from the request data         
		DataFrame input_df = new DataFrame(
                Arrays.asList("input"),
                Arrays.asList(new ScalarType(BasicType.LONG)),
                Arrays.asList(convertJsonToRow(input_json))
        );

        // Transforms with ModelServableA 
        DataFrame output_df = servableA.transform(input_df);

		// Constructs the output Json object from the DataFrame
        Json output_json = convertRowToJson(
                output_df.getDataType("output"), output_df.collect().get(0)
        );
        request.setResponseBody(output_json);
    }
}