rebase and address comments

Author: fsk119

flink-table/flink-table-planner/src/main/java/org/apache/calcite/sql/validate/SqlValidatorImpl.java

@@ -16,6 +16,7 @@
  */
 package org.apache.calcite.sql.validate;
 
+import org.apache.flink.table.api.ValidationException;
 import org.apache.flink.table.planner.calcite.FlinkSqlCallBinding;
 import org.apache.flink.table.planner.functions.sql.ml.SqlVectorSearchTableFunction;
 
@@ -178,10 +179,12 @@
  *
  * <p>Lines 2571 ~ 2588, CALCITE-7217, should be removed after upgrading Calcite to 1.41.0.
  *
- * <p>Lines 3840 ~ 3844, 6511 ~ 6517 Flink improves Optimize the retrieval of sub-operands in
+ * <p>Line 2618 ~2631, set the correct scope for VECTOR_SEARCH.
+ *
+ * <p>Lines 3920 ~ 3925, 6599 ~ 6606 Flink improves Optimize the retrieval of sub-operands in
  * SqlCall when using NamedParameters at {@link SqlValidatorImpl#checkRollUp}.
  *
- * <p>Lines 5315 ~ 5321, FLINK-24352 Add null check for temporal table check on SqlSnapshot.
+ * <p>Lines 5340 ~ 5347, FLINK-24352 Add null check for temporal table check on SqlSnapshot.
  */
 public class SqlValidatorImpl implements SqlValidatorWithHints {
     // ~ Static fields/initializers ---------------------------------------------
@@ -2571,6 +2574,10 @@ private SqlNode registerFrom(
             case LATERAL:
                 // ----- FLINK MODIFICATION BEGIN -----
                 SqlBasicCall sbc = (SqlBasicCall) node;
+                // Put the usingScope which is a JoinScope,
+                // in order to make visible the left items
+                // of the JOIN tree.
+                scopes.put(node, usingScope);
                 registerFrom(
                         parentScope,
                         usingScope,
@@ -2581,10 +2588,6 @@ private SqlNode registerFrom(
                         extendList,
                         forceNullable,
                         true);
-                // Put the usingScope which is a JoinScope,
-                // in order to make visible the left items
-                // of the JOIN tree.
-                scopes.put(node, usingScope);
                 return sbc;
             // ----- FLINK MODIFICATION END -----
 
@@ -2625,8 +2628,14 @@ private SqlNode registerFrom(
                                     .isA(
                                             new HashSet<>(
                                                     Collections.singletonList(SqlKind.SELECT)))) {
+                        boolean queryColumnIsNotLiteral =
+                                binding.operand(2).getKind() != SqlKind.LITERAL;
+                        if (!queryColumnIsNotLiteral && !lateral) {
+                            throw new ValidationException(
+                                    "The query column is not literal, please use LATERAL TABLE to run VECTOR_SEARCH.");
+                        }
                         SqlValidatorScope scope = getSelectScope((SqlSelect) binding.operand(0));
-                        scopes.put(node, scope);
+                        scopes.put(enclosingNode, scope);
                         return newNode;
                     }
                     // ----- FLINK MODIFICATION END -----

flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/FlinkSqlOperatorTable.java

@@ -22,7 +22,6 @@
 import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
 import org.apache.flink.table.planner.functions.sql.internal.SqlAuxiliaryGroupAggFunction;
 import org.apache.flink.table.planner.functions.sql.ml.SqlMLEvaluateTableFunction;
-import org.apache.flink.table.planner.functions.sql.ml.SqlMLPredictTableFunction;
 import org.apache.flink.table.planner.functions.sql.ml.SqlVectorSearchTableFunction;
 import org.apache.flink.table.planner.plan.type.FlinkReturnTypes;
 import org.apache.flink.table.planner.plan.type.NumericExceptFirstOperandChecker;

flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/VectorSearchTableFunctionTest.java

@@ -89,6 +89,18 @@ void testLiteralValue() {
                                         + "+- FlinkLogicalTableSourceScan(table=[[default_catalog, default_database, VectorTable]], fields=[e, f, g])"));
     }
 
+    @Test
+    void testLiteralValueWithoutLateralKeyword() {
+        String sql =
+                "SELECT * FROM TABLE(VECTOR_SEARCH(TABLE VectorTable, DESCRIPTOR(`g`), ARRAY[1.5, 2.0], 10))";
+        assertThatThrownBy(() -> util.verifyRelPlan(sql))
+                .satisfies(
+                        FlinkAssertions.anyCauseMatches(
+                                TableException.class,
+                                "FlinkLogicalTableFunctionScan(invocation=[VECTOR_SEARCH(TABLE(#0), DESCRIPTOR(_UTF-16LE'g'), ARRAY(1.5:DECIMAL(2, 1), 2.0:DECIMAL(2, 1)), 10)], rowType=[RecordType(INTEGER e, BIGINT f, FLOAT ARRAY g, DOUBLE score)])\n"
+                                        + "+- FlinkLogicalTableSourceScan(table=[[default_catalog, default_database, VectorTable]], fields=[e, f, g])"));
+    }
+
     @Test
     void testNamedArgument() {
         String sql =