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

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

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

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

SELECT * FROM ML_EVALUATE(input => TABLE `eval_data`, model_input => MODEL `classifier_model`, label => DESCRIPTOR(label), features => DESCRIPTOR(f1, f2), task => 'classification')

CREATE MODEL mymodel 
INPUT (text STRING)
OUTPUT (result STRING)
WITH (
    'provider' = 'openai',
    'task' = 'text_generation',
    'endpoint' = 'https://api.openai.com/v1/llm/v1/chat',
    'apikey' = 'abcdefg',
    'system_prompt' = 'translate text to english'
)

/**
 * Base interface for configuring a model provider.
 */
@PublicEvolving
public interface ModelProviderFactory extends Factory {

    /**
     * Create ModelRuntime based on connector.
     */
    ModelProvider createModelProvider(Context context);

    /** Provides catalog and session information describing the model to be accessed. */
    @PublicEvolving
    interface Context {
        /**
         * Returns the identifier of the model in the {@link Catalog}.
         *
         * <p>This identifier describes the relationship between the model instance and the
         * associated {@link Catalog} (if any).
         */
        ObjectIdentifier getObjectIdentifier();

        /**
         * Returns the resolved model information received from the {@link Catalog} or persisted
         * plan.
         *
         * <p>The {@link ResolvedCatalogModel} forwards the metadata from the catalog but offers a
         * validated {@link ResolvedSchema}. The original metadata object is available via {@link
         * ResolvedCatalogModel#getOrigin()}.
         *
         * <p>In most cases, a factory is interested in the following characteristics:
         *
         * <pre>{@code
         * // get the physical input and output data type to initialize the connector
         * context.getCatalogModel().getResolvedInputSchema().toPhysicalRowDataType()
         * context.getCatalogModel().getResolvedInputSchema().toPhysicalRowDataType()
         *
         * // get configuration options
         * context.getCatalogModel().getOptions()
         * }</pre>
         *
         * <p>During a plan restore, usually the model information persisted in the plan is used to
         * reconstruct the catalog model.
         */
        ResolvedCatalogModel getCatalogModel();

        /** Gives read-only access to the configuration of the current session. */
        ReadableConfig getConfiguration();

        /**
         * Returns the class loader of the current session.
         *
         * <p>The class loader is in particular useful for discovering further (nested) factories.
         */
        ClassLoader getClassLoader();

        /** Whether the model is temporary. */
        boolean isTemporary();
    }
}

/**
 * Model Provider base interface.
 */
interface ModelProvider {

	/**
	 * Creates a copy of this instance during planning. The copy should be a deep copy of all
     * mutable members.
     */
    ModelProvider copy();

	/**
 	 * Context for creating runtime providers.
	 */
    interface Context {
	    
		/** Resolved catalog model. */
        ResolvedCatalogModel getCatalogModel();

		/** Runtime config provided to provider */
        ReadableConfig runtimeConfig();
    }
}

/**
 * Provider to provide synchronous model runtime for predict function.
 */
interface PredictRuntimeProvider extends ModelProvider {

    /**
     * Returns a ml_predict function.
     */
    PredictFunction createPredictFunction(Context context);
}

/**
 * Provider to provide asynchronous model runtime for predict function.
 */
interface AsyncPredictRuntimeProvider extends ModelProvider {

    /**
     * Returns a async ml_predict function.
     */
    AsyncPredictFunction createAsyncPredictFunction(Context context);
}

/**
 * A wrapper class of {@link TableFunction} for synchronously prediction.
 *
 * <p>The output type of this table function is fixed as {@link Row}.
 */
@PublicEvolving
public abstract class PredictFunction extends TableFunction<RowData> {

    /**
     * Synchronously predict result based on input row.
     *
     * @param inputRow - A {@link RowData} that wraps input for predict function.
	 *
     * @return A collection of predicted results.
     */
    public abstract Collection<RowData> predict(RowData inputRow);

    /** Invoke {@link #predict} and handle exceptions. */
    public final void eval(Object... args) {
        try {
            GenericRowData argsData = GenericRowData.of(args);              
            Collection<RowData> results = predict(argsData);
            if (results == null) {
                return;
            }
            results.forEach(this::collect);
        } catch (Exception e) {
            throw new FlinkRuntimeException("Prediction error", e);
        }
    }
}

/**
 * A wrapper class of {@link AsyncTableFunction} for asynchronously prediction.
 *
 * <p>The output type of this table function is fixed as {@link RowData}.
 */
@PublicEvolving
public abstract class AsyncPredictFunction extends AsyncTableFunction<RowData> {

    /**
     * Asynchronously predict result based on input row.
     *
     * @param inputRow - A {@link RowData} that wraps input for predict function.
	 * @param completableFuture - Future to collect prediction results.
     */
    public abstract CompletableFuture<Collection<RowData>> asyncPredict(RowData inputRow);

    public void eval(CompletableFuture<Collection<Row>> completableFuture, Object... args) {
        try {
             GenericRowData argsData = GenericRowData.of(args);
             asyncPredict(keyRow)
                .whenComplete(
                        (result, exception) -> {
                            if (exception != null) {
                                future.completeExceptionally(
                                        new TableException(
                                                "Failed to execute asynchronously prediction.",
                                                exception));
                                return;
                            }
                            completableFuture.complete(result);
                        });
        }
    }
}

interface Module {
	default Optional<ModelProviderFactory> getModelProviderFactory() {
		return Optional.empty();
	}
}

SELECT * FROM ML_PREDICT(TABLE input, MODEL mdl, descriptor(f1, f2), MAP['async', 'true']);

table.exec.async-ml-predict.max-concurrent-operations: 10
table.exec.async-ml-predict.timeout: 30s
table.exec.async-ml-predict.retry-strategy: FIXED_DELAY
table.exec.async-ml-predict.fixed-delay: 10s
table.exec.async-ml-predict.max-attempts: 3

public abstract class SqlMlFunctionTableFunction extends SqlFunction implements SqlTableFunction {
    @Override
    public SqlReturnTypeInference getRowTypeInference() {
        return this::inferRowType;
    }
    protected abstract RelDataType inferRowType(SqlOperatorBinding opBinding);
}

public class SqlMlPredictTableFunction extends SqlMlFunctionTableFunction {
    // Input/output validation and type inference
}

public class SqlMlEvaluateTableFunction extends SqlMlFunctionTableFunction {
    // Input/output validation and type inference
}