[FLINK-35014] SqlNode to operation conversion for models

flink-table/flink-sql-parser/src/main/codegen/data/Parser.tdd

@@ -49,6 +49,9 @@
     "org.apache.flink.sql.parser.ddl.SqlAlterMaterializedTableResume"
     "org.apache.flink.sql.parser.ddl.SqlAlterMaterializedTableSuspend"
     "org.apache.flink.sql.parser.ddl.SqlAlterModel"
+    "org.apache.flink.sql.parser.ddl.SqlAlterModelRename"
+    "org.apache.flink.sql.parser.ddl.SqlAlterModelReset"
+    "org.apache.flink.sql.parser.ddl.SqlAlterModelSet"
     "org.apache.flink.sql.parser.ddl.SqlAlterTable"
     "org.apache.flink.sql.parser.ddl.SqlAlterTable.AlterTableContext"
     "org.apache.flink.sql.parser.ddl.SqlAlterTableAdd"

flink-table/flink-sql-parser/src/main/codegen/includes/parserImpls.ftl

@@ -3274,7 +3274,7 @@ SqlTruncateTable SqlTruncateTable() :
 }
 
 /**
-* SHOW MODELS [FROM [catalog.] database] sql call.
+* SHOW MODELS [FROM [catalog.] database] [[NOT] LIKE pattern]; sql call.
 */
 SqlShowModels SqlShowModels() :
 {
@@ -3313,6 +3313,7 @@ SqlShowModels SqlShowModels() :
 /**
 * ALTER MODEL [IF EXISTS] modelName SET (property_key = property_val, ...)
 * ALTER MODEL [IF EXISTS] modelName RENAME TO newModelName
+* ALTER MODEL [IF EXISTS] modelName RESET (property_key, ...)
 */
 SqlAlterModel SqlAlterModel() :
 {
@@ -3321,6 +3322,7 @@ SqlAlterModel SqlAlterModel() :
     SqlIdentifier modelIdentifier;
     SqlIdentifier newModelIdentifier = null;
     SqlNodeList propertyList = SqlNodeList.EMPTY;
+    SqlNodeList propertyKeyList = SqlNodeList.EMPTY;
 }
 {
     <ALTER> <MODEL> { startPos = getPos(); }
@@ -3331,7 +3333,7 @@ SqlAlterModel SqlAlterModel() :
         <RENAME> <TO>
         newModelIdentifier = CompoundIdentifier()
         {
-            return new SqlAlterModel(
+            return new SqlAlterModelRename(
                         startPos.plus(getPos()),
                         modelIdentifier,
                         newModelIdentifier,
@@ -3341,11 +3343,21 @@ SqlAlterModel SqlAlterModel() :
         <SET>
         propertyList = Properties()
         {
-            return new SqlAlterModel(
+            return new SqlAlterModelSet(
                         startPos.plus(getPos()),
                         modelIdentifier,
-                        propertyList,
-                        ifExists);
+                        ifExists,
+                        propertyList);
+        }
+    |
+        <RESET>
+        propertyKeyList = PropertyKeys()
+        {
+            return new SqlAlterModelReset(
+                        startPos.plus(getPos()),
+                        modelIdentifier,
+                        ifExists,
+                        propertyKeyList);
         }
     )
 }

flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlAlterModel.java

@@ -18,58 +18,31 @@
 
 package org.apache.flink.sql.parser.ddl;
 
-import org.apache.flink.sql.parser.SqlUnparseUtils;
-
 import org.apache.calcite.sql.SqlCall;
 import org.apache.calcite.sql.SqlIdentifier;
 import org.apache.calcite.sql.SqlKind;
-import org.apache.calcite.sql.SqlNode;
-import org.apache.calcite.sql.SqlNodeList;
 import org.apache.calcite.sql.SqlOperator;
 import org.apache.calcite.sql.SqlSpecialOperator;
 import org.apache.calcite.sql.SqlWriter;
 import org.apache.calcite.sql.parser.SqlParserPos;
-import org.apache.calcite.util.ImmutableNullableList;
-
-import java.util.List;
 
 import static java.util.Objects.requireNonNull;
 
 /**
- * ALTER MODEL [IF EXISTS] [[catalogName.] dataBasesName].modelName SET ( name=value [,
- * name=value]*).
+ * Abstract class to describe statements like ALTER MODEL [IF EXISTS] [[catalogName.]
+ * dataBasesName.]modelName ...
  */
-public class SqlAlterModel extends SqlCall {
+public abstract class SqlAlterModel extends SqlCall {
 
     public static final SqlSpecialOperator OPERATOR =
             new SqlSpecialOperator("ALTER MODEL", SqlKind.OTHER_DDL);
 
     protected final SqlIdentifier modelName;
-    protected final SqlIdentifier newModelName;
     protected final boolean ifModelExists;
-    private final SqlNodeList propertyList;
-
-    public SqlAlterModel(
-            SqlParserPos pos,
-            SqlIdentifier modelName,
-            SqlNodeList propertyList,
-            boolean ifModelExists) {
-        super(pos);
-        this.modelName = requireNonNull(modelName, "modelName should not be null");
-        this.newModelName = null;
-        this.propertyList = requireNonNull(propertyList, "propertyList should not be null");
-        this.ifModelExists = ifModelExists;
-    }
 
-    public SqlAlterModel(
-            SqlParserPos pos,
-            SqlIdentifier modelName,
-            SqlIdentifier newModelName,
-            boolean ifModelExists) {
+    public SqlAlterModel(SqlParserPos pos, SqlIdentifier modelName, boolean ifModelExists) {
         super(pos);
         this.modelName = requireNonNull(modelName, "modelName should not be null");
-        this.newModelName = requireNonNull(newModelName, "newModelName should not be null");
-        this.propertyList = null;
         this.ifModelExists = ifModelExists;
     }
 
@@ -95,52 +68,12 @@ public boolean ifModelExists() {
         return ifModelExists;
     }
 
-    public SqlIdentifier getNewModelName() {
-        return newModelName;
-    }
-
-    public String[] fullNewModelName() {
-        if (newModelName != null) {
-            return newModelName.names.toArray(new String[0]);
-        }
-        return new String[0];
-    }
-
-    public SqlNodeList getPropertyList() {
-        return propertyList;
-    }
-
-    @Override
-    public List<SqlNode> getOperandList() {
-        // Rename Model.
-        if (newModelName != null) {
-            return ImmutableNullableList.of(modelName, newModelName);
-        }
-        return ImmutableNullableList.of(modelName, propertyList);
-    }
-
     @Override
     public void unparse(SqlWriter writer, int leftPrec, int rightPrec) {
         writer.keyword("ALTER MODEL");
         if (ifModelExists) {
             writer.keyword("IF EXISTS");
         }
         modelName.unparse(writer, leftPrec, rightPrec);
-        if (newModelName != null) {
-            // Rename Model.
-            writer.keyword("RENAME TO");
-            newModelName.unparse(writer, leftPrec, rightPrec);
-        } else {
-            writer.keyword("SET");
-            SqlWriter.Frame withFrame = writer.startList("(", ")");
-            if (propertyList != null) {
-                for (SqlNode modelOption : propertyList) {
-                    SqlUnparseUtils.printIndent(writer);
-                    modelOption.unparse(writer, leftPrec, rightPrec);
-                }
-            }
-            writer.newlineAndIndent();
-            writer.endList(withFrame);
-        }
     }
 }

flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlAlterModelRename.java

@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.sql.parser.ddl;
+
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlWriter;
+import org.apache.calcite.sql.parser.SqlParserPos;
+
+import java.util.List;
+
+import static java.util.Objects.requireNonNull;
+
+/** ALTER MODEL [IF EXISTS] [[catalogName.] dataBasesName.]modelName RENAME TO newModelName. */
+public class SqlAlterModelRename extends SqlAlterModel {
+
+    private final SqlIdentifier newModelName;
+
+    public SqlAlterModelRename(
+            SqlParserPos pos,
+            SqlIdentifier modelName,
+            SqlIdentifier newModelName,
+            boolean ifModelExists) {
+        super(pos, modelName, ifModelExists);
+        this.newModelName = requireNonNull(newModelName, "newModelName should not be null");
+    }
+
+    public SqlIdentifier getNewModelName() {
+        return newModelName;
+    }
+
+    public String[] fullNewModelName() {
+        return newModelName.names.toArray(new String[0]);
+    }
+
+    @Override
+    public List<SqlNode> getOperandList() {
+        return List.of(modelName, newModelName);
+    }
+
+    @Override
+    public void unparse(SqlWriter writer, int leftPrec, int rightPrec) {
+        super.unparse(writer, leftPrec, rightPrec);
+        writer.keyword("RENAME TO");
+        newModelName.unparse(writer, leftPrec, rightPrec);
+    }
+}

flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlAlterModelReset.java

@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.sql.parser.ddl;
+
+import org.apache.flink.sql.parser.SqlUnparseUtils;
+
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlLiteral;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlNodeList;
+import org.apache.calcite.sql.SqlWriter;
+import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.calcite.util.NlsString;
+
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * ALTER MODEL [IF EXISTS] [[catalogName.] dataBasesName.]modelName RESET ( 'key1' [, 'key2']...).
+ */
+public class SqlAlterModelReset extends SqlAlterModel {
+    private final SqlNodeList optionKeyList;
+
+    public SqlAlterModelReset(
+            SqlParserPos pos,
+            SqlIdentifier modelName,
+            boolean ifModelExists,
+            SqlNodeList optionKeyList) {
+        super(pos, modelName, ifModelExists);
+        this.optionKeyList = requireNonNull(optionKeyList, "optionKeyList should not be null");
+    }
+
+    @Override
+    public List<SqlNode> getOperandList() {
+        return List.of(modelName, optionKeyList);
+    }
+
+    public Set<String> getResetKeys() {
+        return optionKeyList.getList().stream()
+                .map(key -> ((NlsString) SqlLiteral.value(key)).getValue())
+                .collect(Collectors.toSet());
+    }
+
+    @Override
+    public void unparse(SqlWriter writer, int leftPrec, int rightPrec) {
+        super.unparse(writer, leftPrec, rightPrec);
+        writer.keyword("RESET");
+        SqlWriter.Frame withFrame = writer.startList("(", ")");
+        for (SqlNode optionKey : optionKeyList) {
+            SqlUnparseUtils.printIndent(writer);
+            optionKey.unparse(writer, leftPrec, rightPrec);
+        }
+        writer.newlineAndIndent();
+        writer.endList(withFrame);
+    }
+}

flink-table/flink-sql-parser/src/main/java/org/apache/flink/sql/parser/ddl/SqlAlterModelSet.java

@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.sql.parser.ddl;
+
+import org.apache.flink.sql.parser.SqlUnparseUtils;
+
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlNodeList;
+import org.apache.calcite.sql.SqlWriter;
+import org.apache.calcite.sql.parser.SqlParserPos;
+
+import java.util.List;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * ALTER MODEL [IF EXISTS] [[catalogName.] dataBasesName.]modelName SET ( name=value [,
+ * name=value]*).
+ */
+public class SqlAlterModelSet extends SqlAlterModel {
+
+    private final SqlNodeList modelOptionList;
+
+    public SqlAlterModelSet(
+            SqlParserPos pos,
+            SqlIdentifier modelName,
+            boolean ifModelExists,
+            SqlNodeList modelOptionList) {
+        super(pos, modelName, ifModelExists);
+        this.modelOptionList =
+                requireNonNull(modelOptionList, "modelOptionList should not be null");
+    }
+
+    public SqlNodeList getOptionList() {
+        return modelOptionList;
+    }
+
+    @Override
+    public List<SqlNode> getOperandList() {
+        return List.of(modelName, modelOptionList);
+    }
+
+    @Override
+    public void unparse(SqlWriter writer, int leftPrec, int rightPrec) {
+        super.unparse(writer, leftPrec, rightPrec);
+        writer.keyword("SET");
+        SqlWriter.Frame withFrame = writer.startList("(", ")");
+        for (SqlNode modelOption : modelOptionList) {
+            SqlUnparseUtils.printIndent(writer);
+            modelOption.unparse(writer, leftPrec, rightPrec);
+        }
+        writer.newlineAndIndent();
+        writer.endList(withFrame);
+    }
+}