does not allow unknown tables to be passed in flush statements

Signed-off-by: GuptaManan100 <[email protected]>

go/vt/vtgate/planbuilder/builder.go

@@ -256,24 +256,14 @@ func buildFlushTables(stmt *sqlparser.Flush, vschema ContextVSchema) (engine.Pri
 		table, _, _, _, destinationTab, err = vschema.FindTableOrVindex(tab)
 
 		if err != nil {
-			_, isNotFound := err.(vindexes.NotFoundError)
-			if !isNotFound {
-				return nil, err
-			}
+			return nil, err
 		}
 		if table == nil {
-			destinationTab, keyspaceTab, _, err = vschema.TargetDestination(tab.Qualifier.String())
-			if err != nil {
-				return nil, err
-			}
-			stmt.TableNames[i] = sqlparser.TableName{
-				Name: tab.Name,
-			}
-		} else {
-			keyspaceTab = table.Keyspace
-			stmt.TableNames[i] = sqlparser.TableName{
-				Name: table.Name,
-			}
+			return nil, vindexes.NotFoundError{TableName: tab.Name.String()}
+		}
+		keyspaceTab = table.Keyspace
+		stmt.TableNames[i] = sqlparser.TableName{
+			Name: table.Name,
 		}
 		if destinationTab == nil {
 			destinationTab = key.DestinationAllShards{}

go/vt/vtgate/planbuilder/testdata/flush_cases_no_default_keyspace.txt

@@ -69,3 +69,25 @@
     ]
   }
 }
+
+# Incorrect tables in flush
+"flush tables user.a with read lock"
+"table a not found"
+
+# Unknown tables in unsharded keyspaces are allowed
+"flush tables main.a with read lock"
+{
+  "QueryType": "FLUSH",
+  "Original": "flush tables main.a with read lock",
+  "Instructions": {
+    "OperatorType": "Send",
+    "Keyspace": {
+      "Name": "main",
+      "Sharded": false
+    },
+    "TargetDestination": "AllShards()",
+    "IsDML": false,
+    "Query": "flush tables a with read lock",
+    "SingleShardOnly": false
+  }
+}