/** Interface for a model in a catalog. */
@PublicEvolving
public interface CatalogModel {

    /**
     * Get the unresolved input schema of the model.
     *
     * @return unresolved input schema of the model.
     */
    Schema getInputSchema();

    /**
     * Get the unresolved output schema of the model.
     *
     * @return unresolved output schema of the model.
     */
    Schema getOutputSchema();

    /**
     * Get comment of the model.
     *
     * @return comment of the model.
     */
    String getComment();

    /**
     * Get a deep copy of the CatalogModel instance.
     *
     * @return a copy of the CatalogModel instance
     */
    CatalogModel copy();

    /**
     * Copy the input model options into the CatalogModel instance.
     *
     * @return a copy of the CatalogModel instance with new model options.
     */
    CatalogModel copy(Map<String, String> modelOptions);
}

@PublicEvolving
public interface ResolvedCatalogModel extends CatalogModel {
    /**
     * Returns the original, unresolved metadata object from the {@link Catalog}.
     *
     * <p>This method might be useful if catalog-specific object instances should be directly
     * forwarded from the catalog to a factory.
     */
    CatalogModel getOrigin();

    /** Returns a fully resolved and validated {@link ResolvedSchema} inputSchema. */
    ResolvedSchema getResolvedInputSchema();

    /** Returns a fully resolved and validated {@link ResolvedSchema} outputSchema. */
    ResolvedSchema getResolvedOutputSchema();

    /**
     * Serializes this instance into a map of string-based properties.
     *
     * <p>Compared to the pure table options in {@link #getModelOptions()}, the map includes input
     * schema, output schema, kind, task, comment and options.
     */
    Map<String, String> toProperties();

    /**
     * Creates an instance of {@link CatalogModel} from a map of string properties that were
     * previously created with {@link ResolvedCatalogModel#toProperties()}.
     *
     * @param properties serialized version of a {@link ResolvedCatalogModel} that includes input
     *     schema, output schema, kind, task, comment and options.
     */
    static CatalogModel fromProperties(Map<String, String> properties) {
        return CatalogPropertiesUtil.deserializeCatalogModel(properties);
    }
}

public interface Catalog {
/**
     * Get names of all tables models under this database. An empty list is returned if none exists.
     *
     * @return a list of the names of all models in this database
     * @throws DatabaseNotExistException if the database does not exist
     * @throws CatalogException in case of any runtime exception
     */
    default List<String> listModels(String databaseName)
            throws DatabaseNotExistException, CatalogException {
        throw new UnsupportedOperationException(
                String.format("listModel(String) is not implemented for %s.", this.getClass()));
    }

    /**
     * Returns a {@link CatalogModel} identified by the given {@link ObjectPath}.
     *
     * @param modelPath Path of the model
     * @return The requested model
     * @throws ModelNotExistException if the target does not exist
     * @throws CatalogException in case of any runtime exception
     */
    default CatalogModel getModel(ObjectPath modelPath)
            throws ModelNotExistException, CatalogException {
        throw new UnsupportedOperationException(
                String.format("getModel(ObjectPath) is not implemented for %s.", this.getClass()));
    }

    /**
     * Check if a model exists in this catalog.
     *
     * @param modelPath Path of the model
     * @return true if the given model exists in the catalog false otherwise
     * @throws CatalogException in case of any runtime exception
     */
    default boolean modelExists(ObjectPath modelPath) throws CatalogException {
        throw new UnsupportedOperationException(
                String.format(
                        "modelExists(ObjectPath) is not implemented for %s.", this.getClass()));
    }

    /**
     * Drop a model.
     *
     * @param modelPath Path of the model to be dropped
     * @param ignoreIfNotExists Flag to specify behavior when the model does not exist: if set to
     *     false, throw an exception, if set to true, do nothing.
     * @throws ModelNotExistException if the model does not exist
     * @throws CatalogException in case of any runtime exception
     */
    default void dropModel(ObjectPath modelPath, boolean ignoreIfNotExists)
            throws ModelNotExistException, CatalogException {
        throw new UnsupportedOperationException(
                String.format(
                        "dropModel(ObjectPath, boolean) is not implemented for %s.",
                        this.getClass()));
    }

    /**
     * Rename an existing model.
     *
     * @param modelPath Path of the model to be renamed
     * @param newModelName the new name of the model
     * @param ignoreIfNotExists Flag to specify behavior when the model does not exist: if set to
     *     false, throw an exception, if set to true, do nothing.
     * @throws ModelNotExistException if the model does not exist
     * @throws CatalogException in case of any runtime exception
     */
    default void renameModel(ObjectPath modelPath, String newModelName, boolean ignoreIfNotExists)
            throws ModelNotExistException, ModelAlreadyExistException, CatalogException {
        throw new UnsupportedOperationException(
                String.format(
                        "renameModel(ObjectPath, String, boolean) is not implemented for %s.",
                        this.getClass()));
    }

    /**
     * Creates a new model.
     *
     * @param modelPath path of the model to be created
     * @param model the CatalogModel definition
     * @param ignoreIfExists flag to specify behavior when a model already exists at the given path:
     *     if set to false, it throws a ModelAlreadyExistException, if set to true, do nothing.
     * @throws ModelAlreadyExistException if model already exists and ignoreIfExists is false
     * @throws DatabaseNotExistException if the database in tablePath doesn't exist
     * @throws CatalogException in case of any runtime exception
     */
    default void createModel(ObjectPath modelPath, CatalogModel model, boolean ignoreIfExists)
            throws ModelAlreadyExistException, DatabaseNotExistException, CatalogException {
        throw new UnsupportedOperationException(
                String.format(
                        "createModel(ObjectPath, CatalogModel, boolean) is not implemented for %s.",
                        this.getClass()));
    }

    /**
     * Modifies an existing model. Note that the new and old {@link CatalogModel} must be of the
     * same kind. For example, this doesn't allow altering a remote model to import model or native
     * model, and vice versa.
     *
     * @param modelPath path of the model to be modified
     * @param newModel the new CatalogModel definition
     * @param ignoreIfNotExists flag to specify behavior when the model does not exist: if set to
     *     false, throw an exception, if set to true, do nothing.
     * @throws ModelNotExistException if the model does not exist
     * @throws CatalogException in case of any runtime exception
     */
    default void alterModel(ObjectPath modelPath, CatalogModel newModel, boolean ignoreIfNotExists)
            throws ModelNotExistException, CatalogException {
        throw new UnsupportedOperationException(
                String.format(
                        "alterModel(ObjectPath, CatalogModel, boolean) is not implemented for %s.",
                        this.getClass()));
    }
}

# Create Model DDL Syntax

CREATE MODEL [IF NOT EXISTS] [catalog_name.][db_name.]model_name
[INPUT (input_column_list)]
[OUTPUT (outpt_column_list)]
[COMMENT model_comment]
WITH(model_option_list)
[AS query_statement]

#Example I: import a model from customer storage.

CREATE MODEL `my_import_model`
INPUT (f1 INT, f2 STRING)
OUTPUT (label FLOAT)
WITH(
  'task' = 'regression',
  'type' = 'import',
  'format' = 'ONNX',
  'ONNX.path' = 'http://storage.googleapis.com/storage/t.onnx',
)

#Example II: reference a remote model from customer endpoint.

CREATE MODEL `my_remote_model`
INPUT (f1 INT, f2 STRING)
OUTPUT (label STRING, probs ARRAY<FLOAT>)
WITH(
  'task' = 'classification',
  'type' = 'remote',
  'provider' = 'openai',
  'openai.endpoint' = 'https://api.openai.com/v1/llm/v1/chat',
  'openai.api_key' = 'abcdefg'
)

#Example III: train a kmeans clustering model in Flink.

CREATE MODEL `my_native_model`
WITH(
  'task' = 'clustering',
  'type' = 'training',
  'algorithm' = 'kmeans',
  'kmeans.num_clusters' = '3'
) AS SELECT F1, F2 FROM `my_data`

# Alter Model DDL Syntax

# rename model syntax
ALTER MODEL [IF EXISTS][catalog_name.][database_name.]model_name
RENAME TO [catalog_name.][database_name.]new_model_name

# Alter model options syntax
ALTER MODEL [IF EXISTS] [catalog_name.][database_name.]model_name
SET (key1=val1[, key2=val2]...)

# Example I: rename model
ALTER MODEL `my_model` RENAME TO `my_new_model`
ALTER MODEL IF EXISTS `my_model` RENAME TO `my_new_model`

# Example II: alter model options
ALTER MODEL `my_model` SET (
   tag = 'prod', 
   description = "new_description"
)

# Drop Model DDL Syntax

DROP MODEL [IF EXISTS] [catalog_name.][db_name.]model_name

#Example I: drop model

DROP MODEL `my_model`
DROP MODEL IF EXISTS `my_model`

# Show Models Syntax

SHOW MODELS [ ( FROM | IN ) [catalog_name.]database_name ]
[ [NOT] LIKE <sql_like_pattern> ]

# Describe Model Syntax

{ DESCRIBE | DESC } MODEL [catalog_name.][database_name.]model_name

# Model Prediction

SELECT f1, f2, label FROM ML_PREDICT(`my_data`, `classifier_model`, DESCRIPTOR(f1, f2))

# Prediction with named arguments

SELECT f1, f2, label FROM 
ML_PREDICT(
  input => `my_data`, 
  model => `classifier_model`, 
  args => DESCRIPTOR(f1, f2)
)

# Model Evaluation

SELECT * FROM ML_EVALUATE(`eval_data`, `classifier_model`, DESCRIPTOR(f1, f2))

# Evaluation with named arguments

SELECT * FROM ML_EVALUATE(
  input => `eval_data`, 
  model => `classifier_model`, 
  args => DESCRIPTOR(f1, f2)
)