[FLINK-37789][3/N] introduce ml builtin sql functions and sql validator changes

[lihaosky]

What is the purpose of the change

Add Builtin SqlTableFunction for ml_predict and make sql validator pass

Brief change log

• Add SqlMlTableFunction and SqlMlPredictTableFunction
• Make sql validator pass

Verifying this change

Skeleton. Will add test in following up PRs during integration

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): (no)
• The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
• The serializers: (no)
• The runtime per-record code paths (performance sensitive): (no)
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
• The S3 file system connector: (no)

Documentation

• Does this pull request introduce a new feature? (yes)
• If yes, how is the feature documented? (JavaDocs)

[flinkbot]

CI report:

• c311f98 UNKNOWN
• 7fb90cb UNKNOWN
• 8f6fb96 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[davidradl]

@lihaosky could we add unit tests please for the new methods please

[fsk119]

Thanks for your contribution. I left some comments

[fsk119]

Thanks for your update.

[fsk119]

I am confuesd about the change and I can not produce a case have the same error as SqlWindowFunction. I am not sure FLINK-35619 is a good solution here, which tries to validate a node when validating namespace. However, in most cases, validator should register the namespace for the node and then validate.

    private SqlNode validateScopedExpression(SqlNode topNode, SqlValidatorScope scope) {
        SqlNode outermostNode = performUnconditionalRewrites(topNode, false);
        cursorSet.add(outermostNode);
        top = outermostNode;
        TRACER.trace("After unconditional rewrite: {}", outermostNode);
        if (outermostNode.isA(SqlKind.TOP_LEVEL)) {
           // 1. REGISTER namespace
            registerQuery(scope, null, outermostNode, outermostNode, null, false);
        }
       // 2. VALIDATE node
        outermostNode.validate(this, scope);
        if (!outermostNode.isA(SqlKind.TOP_LEVEL)) {
            // force type derivation so that we can provide it to the
            // caller later without needing the scope
            deriveType(scope, outermostNode);
        }
        TRACER.trace("After validation: {}", outermostNode);
        return outermostNode;
    }

How about removing the change here?

[lihaosky]

Yeah. This one doesn't seem to have effect. The first select scope should be used when validating descriptor's operands. I made changes there

[fsk119]

How about we only rewrite the call if the model is not null here. For example, SqlValidatorImpl#performUnconditionalRewrites override the operator if the function is found.

if (call.getOperator() instanceof SqlUnresolvedFunction) {
            assert call instanceof SqlBasicCall;
            final SqlUnresolvedFunction function = (SqlUnresolvedFunction) call.getOperator();
            // This function hasn't been resolved yet.  Perform
            // a half-hearted resolution now in case it's a
            // builtin function requiring special casing.  If it's
            // not, we'll handle it later during overload resolution.
            final List<SqlOperator> overloads = new ArrayList<>();
            opTab.lookupOperatorOverloads(
                    function.getNameAsId(),
                    function.getFunctionType(),
                    SqlSyntax.FUNCTION,
                    overloads,
                    catalogReader.nameMatcher());
            if (overloads.size() == 1) {
                ((SqlBasicCall) call).setOperator(overloads.get(0));
            }
        }

I think we can have the same logic here

        if (node instanceof SqlExplicitModelCall) {
            // Convert it so that model can be accessed in planner. SqlExplicitModelCall
            // from parser can't access model.
            FlinkCalciteCatalogReader catalogReader =
                    (FlinkCalciteCatalogReader) getCatalogReader();
            CatalogSchemaModel model = catalogReader.getModel(((SqlExplicitModelCall) node).getIdentifier());
            if (model != null) {
                return new SqlModelCall(model);
            }
        }

[lihaosky]

Getting model here should be fine. But SqlExplicitModelCall with model identifier expects the model to exist so we should enforce that here. If there are cases where model doesn't need to exist in other cases, we can revisit?