Repository: mindsdb/mindsdb
Pull request: LAST keyword support & spanner update
## Description

Added LAST keyword. It is created for handling new data from sources and organize pipelines to process new pieces of data

Updated spanner handler: https://docs.google.com/document/d/13gxXva1KLaY_NOg3JIFvRhs2Q8Gwq4hjf6WfWrezQfE/edit#heading=h.yk4h1jjpgwmd

Simple example:
```sql
select * from db.table 
where id > last
```
First run of this query
- executes query `select id from db.table order by id desc limit 1` and will find current max value  for `id`.
- returns empty data

Second run of the query:
- finds current stored value for id
- put it in in query instead of last
- executes query and gets new max value for id from it
User will see only delta of data that appeared in table since previous run of the query.

Variables are stored in the context, current list of contexts:
- job context. different for every job
- view context. different for every view . if view is executed inside of the job it still has view context
- default. 

Because new context variable should be updated this type of query won't work because it doesn't have id on output 
```sql
select name from db.table 
where id > last
```

Other examples:

1. Job to
- get new data from home_rentals
- apply prediction
- insert result to home_rentals_res
```sql
create job j1 (
  insert into my_pg.home_rentals_res
  (
    SELECT
     t.rental_price as real_price, m.rental_price as predicted_price,  t.id
    FROM my_pg.home_rentals as t
    JOIN home_rentals_model as m
    where t.id > last
  )
)
every minute;
```

2. View which returns only new data on every call
```sql
create view v1(
select * from my_pg.branches where id > last
)
```

**Demo**: https://www.loom.com/share/399dec869e0b4438946c5c716926fa7e

**Demo with spanner**: https://www.loom.com/share/51068191cde44999a3e8a707a71b278e

**Design doc**: https://docs.google.com/document/d/1hKAGLcibH7apt-QlIO_Rgv2Wi_YiPsEQIEKgfdjlSSQ/edit#heading=h.lvdv8b942awy


Linked PR: https://github.com/mindsdb/mindsdb_sql/pull/313
Fixes: https://github.com/mindsdb/mindsdb_private/issues/234
Fixes: #7104

## Type of change
- [x] ⚡ New feature (non-breaking change which adds functionality)


## Verification Process

To ensure the changes are working as expected:

 - [ ]   Test Location: Specify the URL or path for testing.
 - [ ]   Verification Steps: Outline the steps or queries needed to validate the change. Include any data, configurations, or actions required to reproduce or see the new functionality.

## Additional Media:

- [ ] I have attached a brief loom video or screenshots showcasing the new functionality or change.

## Checklist:

- [x] My code follows the style guidelines(PEP 8) of MindsDB.
- [x] I have appropriately commented on my code, especially in complex areas.
- [ ] Necessary documentation updates are either made or tracked in issues.
- [x] Relevant unit and integration tests are updated or added.





Problem statement:
[Job] stateful job (add cursor state)
### Short description and motivation for the proposed feature

Remember states across different invocations of the job. 
This feature is useful in implementing a use case like incremental data fetching.



### Video or screenshots

_No response_

### Describe some possible solutions

Users need to specify the cursor fields when creating a job. So the job will record the cursor value after the invocation.

```sql
CREATE JOB <some_job_name>
(
  SELECT *
  FROM twitter_source.tweets
  WHERE
    updated_at > STATE.updated_at
    AND 
    created_at > STATE.created_at
)
```

After the job is created, from the inner query, we infer that `updated_at` and `created_at` are two cursor states to be persisted. 

Under the hood, we add a `state` object to the `job` object. 
After each invocation of the job, we refresh the latest state from the result dataframe. E.g., `state.created_at = df['created_at'].max()`

Before the invocation of the job, `LATEST.created_at` and `LATEST.updated_at` will be substituted by the fields in the current state object. Then the query will be sent to the actual data source to be run.

We may also want to support manually resetting the `state` on jobs as well.
```sql
ALTER JOB <some_job_name>
SET
    STATE.created_at = '2020-01-01'
    STATE.updated_at = '2020-01-01'

```



### Anything else?

_No response_

Hints from issue discussion:

Commit message: rollback sql dep

Patch to review:
diff --git a/mindsdb/api/mysql/mysql_proxy/classes/sql_query.py b/mindsdb/api/mysql/mysql_proxy/classes/sql_query.py
index c8523320f63..7ea0c00cf96 100644
--- a/mindsdb/api/mysql/mysql_proxy/classes/sql_query.py
+++ b/mindsdb/api/mysql/mysql_proxy/classes/sql_query.py
@@ -64,7 +64,6 @@
 from mindsdb_sql.render.sqlalchemy_render import SqlalchemyRender
 from mindsdb_sql.planner import query_planner
 from mindsdb_sql.planner.utils import query_traversal
-from mindsdb_sql.parser.ast.base import ASTNode
 
 from mindsdb.api.mysql.mysql_proxy.utilities.sql import query_df
 from mindsdb.interfaces.model.functions import (
@@ -78,6 +77,7 @@
     ErLogicError,
     ErSqlWrongArguments
 )
+from mindsdb.interfaces.query_context.context_controller import contextController
 from mindsdb.utilities.cache import get_cache, json_checksum
 import mindsdb.utilities.profiler as profiler
 from mindsdb.utilities.fs import create_process_mark, delete_process_mark
@@ -570,15 +570,21 @@ def fill_params(node, **kwargs):
 
             query_traversal(query, fill_params)
 
+            query, context_callback = contextController.handle_db_context_vars(query, dn, self.session)
+
             data, columns_info = dn.query(
                 query=query,
                 session=self.session
             )
 
+            if context_callback:
+                context_callback(data, columns_info)
+
+        # is it used?
         # if this is query: execute it
-        if isinstance(data, ASTNode):
-            subquery = SQLQuery(data, session=self.session)
-            return subquery.fetched_data
+        # if isinstance(data, ASTNode):
+        #     subquery = SQLQuery(data, session=self.session)
+        #     return subquery.fetched_data
 
         result = ResultSet()
         for column in columns_info:
diff --git a/mindsdb/api/mysql/mysql_proxy/datahub/datanodes/information_schema_datanode.py b/mindsdb/api/mysql/mysql_proxy/datahub/datanodes/information_schema_datanode.py
index 5a0a03e4d9b..7cb103c52aa 100644
--- a/mindsdb/api/mysql/mysql_proxy/datahub/datanodes/information_schema_datanode.py
+++ b/mindsdb/api/mysql/mysql_proxy/datahub/datanodes/information_schema_datanode.py
@@ -38,7 +38,7 @@ class InformationSchemaDataNode(DataNode):
         'DATABASES': ['NAME', 'TYPE', 'ENGINE', 'CONNECTION_DATA'],
         'ML_ENGINES': ['NAME', 'HANDLER', 'CONNECTION_DATA'],
         'HANDLERS': ['NAME', 'TYPE', 'TITLE', 'DESCRIPTION', 'VERSION', 'CONNECTION_ARGS', 'IMPORT_SUCCESS', 'IMPORT_ERROR'],
-        'JOBS': ['NAME', 'PROJECT', 'START_AT', 'END_AT', 'NEXT_RUN_AT', 'SCHEDULE_STR', 'QUERY'],
+        'JOBS': ['NAME', 'PROJECT', 'START_AT', 'END_AT', 'NEXT_RUN_AT', 'SCHEDULE_STR', 'QUERY', 'VARIABLES'],
         'MDB_TRIGGERS': ['NAME', 'PROJECT', 'DATABASE', 'TABLE', 'QUERY', 'LAST_ERROR'],
         'JOBS_HISTORY': ['NAME', 'PROJECT', 'RUN_START', 'RUN_END', 'ERROR', 'QUERY'],
         'CHATBOTS': ['NAME', 'PROJECT', 'DATABASE', 'MODEL_NAME', 'PARAMS', 'IS_RUNNING', 'LAST_ERROR'],
diff --git a/mindsdb/api/mysql/mysql_proxy/datahub/datanodes/project_datanode.py b/mindsdb/api/mysql/mysql_proxy/datahub/datanodes/project_datanode.py
index a76d20645a7..f1bf0ef778c 100644
--- a/mindsdb/api/mysql/mysql_proxy/datahub/datanodes/project_datanode.py
+++ b/mindsdb/api/mysql/mysql_proxy/datahub/datanodes/project_datanode.py
@@ -12,6 +12,7 @@
 from mindsdb.api.mysql.mysql_proxy.datahub.classes.tables_row import TablesRow
 from mindsdb.api.mysql.mysql_proxy.classes.sql_query import SQLQuery
 from mindsdb.api.mysql.mysql_proxy.utilities.sql import query_df
+from mindsdb.interfaces.query_context.context_controller import contextController
 
 
 class ProjectDataNode(DataNode):
@@ -91,16 +92,22 @@ def query(self, query=None, native_query=None, session=None):
         view_query_ast = self.project.query_view(query)
 
         renderer = SqlalchemyRender('mysql')
-        query_str = renderer.get_string(view_query_ast, with_failback=True)
+        view_meta = renderer.get_string(view_query_ast, with_failback=True)
 
-        sqlquery = SQLQuery(
-            query_str,
-            session=session
-        )
+        contextController.set_context('view', view_meta['id'])
+
+        try:
+            sqlquery = SQLQuery(
+                view_meta['query_ast'],
+                session=session
+            )
+            result = sqlquery.fetch(view='dataframe')
+
+        finally:
+            contextController.release_context('view', view_meta['id'])
 
-        result = sqlquery.fetch(view='dataframe')
         if result['success'] is False:
-            raise Exception(f'Cant execute view query: {query_str}')
+            raise Exception(f"Cant execute view query: {view_meta['query_ast']}")
         df = result['result']
 
         df = query_df(df, query)
diff --git a/mindsdb/integrations/handlers/cloud_spanner_handler/README.md b/mindsdb/integrations/handlers/cloud_spanner_handler/README.md
index 3fed56e0a58..ba1e3998bfd 100644
--- a/mindsdb/integrations/handlers/cloud_spanner_handler/README.md
+++ b/mindsdb/integrations/handlers/cloud_spanner_handler/README.md
@@ -27,6 +27,7 @@ parameters={
     "instance_id":"my-instance",
     "database_id":"example-id",
     "project":"my-project",
+    "dialect": "postgres" // optinonal, default is 'googlesql'
     "credentials":"{...}"
 };
 ```
diff --git a/mindsdb/integrations/handlers/cloud_spanner_handler/cloud_spanner_handler.py b/mindsdb/integrations/handlers/cloud_spanner_handler/cloud_spanner_handler.py
index 7111bdba08b..651634a932b 100644
--- a/mindsdb/integrations/handlers/cloud_spanner_handler/cloud_spanner_handler.py
+++ b/mindsdb/integrations/handlers/cloud_spanner_handler/cloud_spanner_handler.py
@@ -1,11 +1,16 @@
+import json
 from collections import OrderedDict
 
+from google.oauth2 import service_account
 from google.cloud import spanner_dbapi
 from google.cloud.spanner_dbapi import Connection
+from google.cloud.sqlalchemy_spanner import SpannerDialect
 
 import pandas as pd
 from mindsdb_sql import parse_sql
 from mindsdb_sql.parser.ast.base import ASTNode
+from mindsdb_sql.parser.ast import CreateTable, Function
+from mindsdb_sql.render.sqlalchemy_render import SqlalchemyRender
 
 from mindsdb.integrations.libs.base import DatabaseHandler
 from mindsdb.integrations.libs.const import (
@@ -27,8 +32,13 @@ class CloudSpannerHandler(DatabaseHandler):
     def __init__(self, name: str, **kwargs):
         super().__init__(name)
         self.parser = parse_sql
-        self.dialect = 'googlesql'
         self.connection_data = kwargs.get('connection_data')
+        self.dialect = self.connection_data.get('dialect', 'googlesql')
+
+        if self.dialect == 'postgres':
+            self.renderer = SqlalchemyRender('postgres')
+        else:
+            self.renderer = SqlalchemyRender(SpannerDialect)
 
         self.connection = None
         self.is_connected = False
@@ -51,9 +61,12 @@ def connect(self) -> Connection:
             'database_id': self.connection_data.get('database_id'),
             'instance_id': self.connection_data.get('instance_id'),
             'project': self.connection_data.get('project'),
-            'credentials': self.connection_data.get('crendentials'),
+            'credentials': self.connection_data.get('credentials'),
         }
 
+        args['credentials'] = service_account.Credentials.from_service_account_info(
+            json.loads(args['credentials'])
+        )
         self.connection = spanner_dbapi.connect(**args)
         self.is_connected = True
 
@@ -144,10 +157,22 @@ def query(self, query: ASTNode) -> Response:
         Returns:
             Response: The query result.
         """
-        if isinstance(query, ASTNode):
-            query_str = query.to_string()
-        else:
-            query_str = str(query)
+
+        # check primary key for table:
+        if isinstance(query, CreateTable) and query.columns is not None:
+            id_col = None
+            has_primary = False
+            for col in query.columns:
+                if col.name.lower() == 'id':
+                    id_col = col
+                if col.is_primary_key:
+                    has_primary = True
+            # if no other primary keys use id
+            if not has_primary and id_col:
+                id_col.is_primary_key = True
+                id_col.default = Function('GENERATE_UUID', args=[])
+
+        query_str = self.renderer.get_string(query, with_failback=True)
 
         return self.native_query(query_str)
 
@@ -207,6 +232,11 @@ def get_columns(self, table_name: str) -> Response:
         'type': ARG_TYPE.STR,
         'description': 'The Cloud Spanner project indentifier.',
     },
+    dialect={
+        'type': ARG_TYPE.STR,
+        'description': 'Dialect of the database',
+        "required": False,
+    },
     credentials={
         'type': ARG_TYPE.STR,
         'description': 'The Google Cloud Platform service account key in the JSON format.',
diff --git a/mindsdb/integrations/handlers/cloud_spanner_handler/requirements.txt b/mindsdb/integrations/handlers/cloud_spanner_handler/requirements.txt
index 7113e6294e9..14aacaefd0d 100644
--- a/mindsdb/integrations/handlers/cloud_spanner_handler/requirements.txt
+++ b/mindsdb/integrations/handlers/cloud_spanner_handler/requirements.txt
@@ -1 +1,2 @@
 google-cloud-spanner
+sqlalchemy-spanner
diff --git a/mindsdb/integrations/libs/learn_process.py b/mindsdb/integrations/libs/learn_process.py
index 0facc466b12..50e038f3686 100644
--- a/mindsdb/integrations/libs/learn_process.py
+++ b/mindsdb/integrations/libs/learn_process.py
@@ -119,8 +119,8 @@ def learn_process(class_path, engine, integration_id,
                 elif data_integration_ref['type'] == 'view':
                     project = database_controller.get_project(project_name)
                     query_ast = parse_sql(fetch_data_query, dialect='mindsdb')
-                    view_query_ast = project.query_view(query_ast)
-                    sqlquery = SQLQuery(view_query_ast, session=sql_session)
+                    view_meta = project.query_view(query_ast)
+                    sqlquery = SQLQuery(view_meta['query_ast'], session=sql_session)
 
                 result = sqlquery.fetch(view='dataframe')
                 training_data_df = result['result']
diff --git a/mindsdb/interfaces/database/projects.py b/mindsdb/interfaces/database/projects.py
index 39d3c5cd3e9..e5fc044b2fe 100644
--- a/mindsdb/interfaces/database/projects.py
+++ b/mindsdb/interfaces/database/projects.py
@@ -114,8 +114,8 @@ def query_view(self, query: ASTNode) -> ASTNode:
             name=view_name,
             project_name=self.name
         )
-        subquery_ast = parse_sql(view_meta['query'], dialect='mindsdb')
-        return subquery_ast
+        view_meta['query_ast'] = parse_sql(view_meta['query'], dialect='mindsdb')
+        return view_meta
 
     @staticmethod
     def _get_model_data(predictor_record, integraion_record):
diff --git a/mindsdb/interfaces/database/views.py b/mindsdb/interfaces/database/views.py
index 552fcc99acb..271a621e63b 100644
--- a/mindsdb/interfaces/database/views.py
+++ b/mindsdb/interfaces/database/views.py
@@ -1,5 +1,6 @@
 from mindsdb.interfaces.storage import db
 from mindsdb.utilities.context import context as ctx
+from mindsdb.interfaces.query_context.context_controller import contextController
 
 
 class ViewController:
@@ -65,8 +66,11 @@ def delete(self, name, project_name):
         db.session.delete(rec)
         db.session.commit()
 
+        contextController.drop_query_context('view', rec.id)
+
     def _get_view_record_data(self, record):
         return {
+            'id': record.id,
             'name': record.name,
             'query': record.query
         }
diff --git a/mindsdb/interfaces/jobs/jobs_controller.py b/mindsdb/interfaces/jobs/jobs_controller.py
index cd018e856aa..18f8d06101e 100644
--- a/mindsdb/interfaces/jobs/jobs_controller.py
+++ b/mindsdb/interfaces/jobs/jobs_controller.py
@@ -10,6 +10,7 @@
 from mindsdb.interfaces.database.projects import ProjectController
 from mindsdb.utilities.context import context as ctx
 from mindsdb.utilities import log
+from mindsdb.interfaces.query_context.context_controller import contextController
 
 
 def split_sql(sql):
@@ -169,6 +170,9 @@ def delete(self, name, project_name):
         self._delete_record(record)
         db.session.commit()
 
+        # delete context
+        contextController.drop_query_context('job', record.id)
+
     def _delete_record(self, record):
         record.deleted_at = dt.datetime.now()
 
@@ -199,6 +203,7 @@ def get_list(self, project_name=None):
                 'next_run_at': record.next_run_at,
                 'schedule_str': record.schedule_str,
                 'query': record.query_str,
+                'variables': contextController.gen_context_vars('job', record.id)
             })
         return data
 
@@ -314,6 +319,7 @@ def execute_task_local(self, record_id, history_id=None):
         if record.user_class is not None:
             ctx.user_class = record.user_class
 
+        contextController.set_context('job', record.id)
         if history_id is None:
             history_record = db.JobsHistory(
                 job_id=record.id,
diff --git a/mindsdb/interfaces/query_context/__init__.py b/mindsdb/interfaces/query_context/__init__.py
new file mode 100644
index 00000000000..e69de29bb2d
diff --git a/mindsdb/interfaces/query_context/context_controller.py b/mindsdb/interfaces/query_context/context_controller.py
new file mode 100644
index 00000000000..301c3097e39
--- /dev/null
+++ b/mindsdb/interfaces/query_context/context_controller.py
@@ -0,0 +1,171 @@
+import pandas as pd
+
+from mindsdb_sql.parser.ast import Identifier, Select, OrderBy, NullConstant, Constant, BinaryOperation
+
+from mindsdb.interfaces.storage import db
+from mindsdb.utilities.context import context as ctx
+
+from .last_query import LastQuery
+
+
+class ContextController:
+
+    def handle_db_context_vars(self, query, dn, session):
+
+        l_query = LastQuery(query)
+        if l_query.query is None:
+            # no last keyword, exit
+            return query, None
+
+        query_str = l_query.to_string()
+        context_name = self.get_context()
+
+        rec = self.__get_context_record(context_name, query_str)
+
+        if rec is None:
+            values = self.__get_init_last_values(l_query, dn, session)
+            self.__add_context_record(context_name, query_str, values)
+        else:
+            values = rec.values
+
+        query_out = l_query.apply_values(values)
+
+        def callback(data, columns_info):
+            self._handle_results(l_query, context_name, query_str, data, columns_info)
+
+        return query_out, callback
+
+    def _handle_results(self, l_query, context_name, query_str, data, columns_info):
+
+        if len(data) == 0:
+            return
+
+        max_vals = pd.DataFrame(data).max().to_dict()
+        values = {}
+        # get max values
+        for info in l_query.get_last_columns():
+            target_idx = info['target_idx']
+            if target_idx is not None:
+                # get by index
+                col_name = columns_info[target_idx]['name']
+                value = max_vals.get(col_name)
+            else:
+                # get by name
+                value = max_vals.get(info['column_name'])
+
+            if value is not None:
+                values[info['table_name']] = {info['column_name']: value}
+
+        self.__update_context_record(context_name, query_str, values)
+
+    def drop_query_context(self, object_type, object_name):
+        context_name = self.gen_context_name(object_type, object_name)
+        for rec in db.session.query(db.QueryContext).filter_by(
+            context_name=context_name,
+            company_id=ctx.company_id
+        ).all():
+            db.session.delete(rec)
+        db.session.commit()
+
+    def __get_init_last_values(self, l_query, dn, session):
+        """
+        Get current last values for query
+        """
+        last_values = {}
+        for info in l_query.get_last_columns():
+            col = Identifier(info['column_name'])
+
+            query = Select(
+                targets=[
+                    col
+                ],
+                from_table=info['table'],
+                order_by=[
+                    OrderBy(col, direction='DESC')
+                ],
+                where=BinaryOperation(
+                    op='is not',
+                    args=[
+                        col,
+                        NullConstant()
+                    ]
+                ),
+                limit=Constant(1)
+            )
+
+            data, columns_info = dn.query(
+                query=query,
+                session=session
+            )
+
+            if len(data) == 0:
+                value = None
+            else:
+                value = list(data[0].values())[0]
+            last_values[info['table_name']] = {info['column_name']: value}
+        return last_values
+
+    # Context
+
+    def get_context(self):
+        try:
+            return ctx.context_stack[-1]
+        except Exception:
+            return ''
+
+    def set_context(self, object_type=None, object_name=None):
+        if ctx.context_stack is None:
+            ctx.context_stack = []
+        ctx.context_stack.append(self.gen_context_name(object_type, object_name))
+
+    def release_context(self, object_type=None, object_name=None):
+        if ctx.context_stack is None:
+            return
+        context_name = self.gen_context_name(object_type, object_name)
+        if ctx.context_stack[-1] == context_name:
+            ctx.context_stack.pop()
+
+    def gen_context_name(self, object_type, object_name):
+        if object_name is None:
+            return ''
+        return f'{object_type}-{object_name}'
+
+    def gen_context_vars(self, object_type, object_name):
+        context_name = self.gen_context_name(object_type, object_name)
+        vars = []
+        for rec in db.session.query(db.QueryContext).filter_by(
+            context_name=context_name,
+            company_id=ctx.company_id
+        ):
+            if rec.values is not None:
+                vars.append(rec.values)
+
+        return vars
+
+    # DB
+    def __get_context_record(self, context_name, query_str):
+
+        return db.session.query(db.QueryContext).filter_by(
+            query=query_str,
+            context_name=context_name,
+            company_id=ctx.company_id
+        ).first()
+
+    def __add_context_record(self, context_name, query_str, values):
+
+        rec = db.QueryContext(
+            query=query_str,
+            context_name=context_name,
+            company_id=ctx.company_id,
+            values=values)
+        db.session.add(rec)
+        db.session.commit()
+        return rec
+
+    def __update_context_record(self, context_name, query_str, values):
+        rec = self.__get_context_record(context_name, query_str)
+        rec.values = values
+        db.session.commit()
+
+
+contextController = ContextController()
diff --git a/mindsdb/interfaces/query_context/last_query.py b/mindsdb/interfaces/query_context/last_query.py
new file mode 100644
index 00000000000..bb19b72a473
--- /dev/null
+++ b/mindsdb/interfaces/query_context/last_query.py
@@ -0,0 +1,158 @@
+import copy
+from collections import defaultdict
+
+from mindsdb_sql.parser.ast import Identifier, Select, BinaryOperation, Last, Constant, Star
+from mindsdb_sql.planner.utils import query_traversal
+
+
+class LastQuery:
+    """
+    Tracks last values in query
+    """
+
+    def __init__(self, query):
+        self.query = None
+
+        # check query type
+        if not isinstance(query, Select):
+            # just skip it
+            return
+
+        self.query_in = query
+
+        query = copy.deepcopy(self.query_in)
+
+        last_tables = self.__find_last_columns(query)
+        if last_tables is None:
+            return
+
+        self.query = query
+        self.last_tables = last_tables
+
+    def __find_last_columns(self, query):
+
+        # index last variables in query
+        tables_idx = defaultdict(dict)
+        last_idx = defaultdict(list)
+
+        def _index_query(node, is_table, parent_query, **kwargs):
+
+            parent_query_id = id(parent_query)
+
+            if is_table and isinstance(node, Identifier):
+                # memorize table
+                tables_idx[parent_query_id][node.parts[-1]] = node
+                if node.alias is not None:
+                    tables_idx[parent_query_id][node.alias.parts[-1]] = node
+
+            # find last in where
+            if isinstance(node, BinaryOperation):
+                arg0 = node.args[0]
+                arg1 = node.args[1]
+                if isinstance(arg0, Identifier) and isinstance(arg1, Last):
+
+                    # TODO handle binary operation <, >
+
+                    # inject constant
+                    link = Constant(None)
+                    node.args[1] = link
+                    last_idx[parent_query_id].append([arg0, link])
+
+                    return node
+
+        # find lasts
+        query_traversal(query, _index_query)
+
+        if len(last_idx) == 0:
+            return
+
+        # index query targets
+        query_id = id(query)
+        tables = tables_idx[query_id]
+        is_star_in_target = False
+        target_idx = {}
+        for i, target in enumerate(query.targets):
+            if isinstance(target, Star):
+                is_star_in_target = True
+                continue
+            elif not isinstance(target, Identifier):
+                continue
+
+            col_name = target.parts[-1]
+            if len(target.parts) > 1:
+                table_name = target.parts[-2]
+                table = tables.get(table_name)
+            elif len(tables) == 1:
+                table = list(tables.values())[0]
+            else:
+                continue
+
+            target_idx[(table.parts[-1], col_name)] = i
+
+        # make info about query
+
+        last_columns = {}
+        for parent_query_id, lasts in last_idx.items():
+            for col, last in lasts:
+                tables = tables_idx[parent_query_id]
+                if len(col.parts) > 1:
+
+                    table = tables.get(col.parts[-2])
+                    if table is None:
+                        raise Exception('cant find table')
+                elif len(tables) == 1:
+                    table = tables[0]
+                else:
+                    # or just skip it?
+                    raise Exception('cant find table')
+
+                col_name = col.parts[-1]
+
+                table_name = table.parts[-1]
+                if table_name not in last_columns:
+                    # check column in target
+                    target_idx = target_idx.get((table_name, col_name))
+                    if target_idx is None:
+                        if is_star_in_target:
+                            # will try to get by name
+                            ...
+                        else:
+                            raise Exception('Last value should be in query target')
+
+                    last_columns[table_name] = {
+                        'table': table,
+                        'column': col_name,
+                        'links': [last],
+                        'target_idx': target_idx
+                    }
+
+                elif last_columns[table_name]['column'] == col_name:
+                    last_columns[table_name]['column'].append(last)
+                else:
+                    raise Exception('possible to use only one column')
+
+        return last_columns
+
+    def to_string(self):
+        return self.query_in.to_string()
+
+    def get_last_columns(self):
+        return [
+            {
+                'table': info['table'],
+                'table_name': table_name,
+                'column_name': info['column'],
+                'target_idx': info['target_idx'],
+            }
+            for table_name, info in self.last_tables.items()
+        ]
+
+    def apply_values(self, values):
+        # fill the query:
+        for table_name, info in self.last_tables.items():
+            value = values[table_name][info['column']]
+            for last in info['links']:
+                # TODO how to inject
+                last.value = value
+
+        return self.query
diff --git a/mindsdb/interfaces/storage/db.py b/mindsdb/interfaces/storage/db.py
index 81b83583615..66ee5c7ffc4 100644
--- a/mindsdb/interfaces/storage/db.py
+++ b/mindsdb/interfaces/storage/db.py
@@ -11,6 +11,8 @@
 from sqlalchemy import JSON
 from sqlalchemy.exc import OperationalError
 
+from mindsdb.utilities.json_encoder import CustomJSONEncoder
+
 
 class Base:
     __allow_unmapped__ = True
@@ -27,7 +29,8 @@ def init(connection_str: str = None):
         connection_str = os.environ['MINDSDB_DB_CON']
     base_args = {
         'pool_size': 30,
-        'max_overflow': 200
+        'max_overflow': 200,
+        'json_serializer': CustomJSONEncoder().encode
     }
     engine = create_engine(connection_str, echo=False, **base_args)
     session = scoped_session(sessionmaker(bind=engine, autoflush=True))
@@ -431,3 +434,16 @@ def as_dict(self) -> Dict:
             'updated_at': self.updated_at,
             'created_at': self.created_at
         }
+
+
+class QueryContext(Base):
+    __tablename__ = 'query_context'
+    id = Column(Integer, primary_key=True)
+    company_id = Column(Integer, nullable=True)
+
+    query = Column(String, nullable=False)
+    context_name = Column(String, nullable=False)
+    values = Column(JSON)
+
+    updated_at = Column(DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now)
+    created_at = Column(DateTime, default=datetime.datetime.now)
diff --git a/mindsdb/migrations/versions/2023-10-03_6cb02dfd7f61_query_context.py b/mindsdb/migrations/versions/2023-10-03_6cb02dfd7f61_query_context.py
new file mode 100644
index 00000000000..42bd8b45c50
--- /dev/null
+++ b/mindsdb/migrations/versions/2023-10-03_6cb02dfd7f61_query_context.py
@@ -0,0 +1,38 @@
+"""query_context
+
+Revision ID: 6cb02dfd7f61
+Revises: b0382f5be48d
+Create Date: 2023-10-03 12:36:00.008731
+
+"""
+from alembic import op
+import sqlalchemy as sa
+import mindsdb.interfaces.storage.db # noqa
+
+
+# revision identifiers, used by Alembic.
+revision = '6cb02dfd7f61'
+down_revision = 'b0382f5be48d'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        'query_context',
+        sa.Column('id', sa.Integer(), nullable=False),
+        sa.Column('company_id', sa.Integer(), nullable=True),
+        sa.Column('query', sa.String(), nullable=False),
+        sa.Column('context_name', sa.String(), nullable=False),
+        sa.Column('values', sa.JSON(), nullable=True),
+        sa.Column('updated_at', sa.DateTime(), nullable=True),
+        sa.Column('created_at', sa.DateTime(), nullable=True),
+        sa.PrimaryKeyConstraint('id')
+    )
+
+
+def downgrade():
+
+    op.drop_table('query_context')
+    # ### end Alembic commands ###


Code review comments:
diff_hunk: @@ -0,0 +1,158 @@
+import copy
+from collections import defaultdict
+
+from mindsdb_sql.parser.ast import Identifier, Select, BinaryOperation, Last, Constant, Star
+from mindsdb_sql.planner.utils import query_traversal
+
+
+class LastQuery:
+    """
+    Tracks last values in query
+    """
+
+    def __init__(self, query):
+        self.query = None
+
+        # check query type
+        if not isinstance(query, Select):
+            # just skip it
+            return
+
+        self.query_in = query
+
+        query = copy.deepcopy(self.query_in)
+
+        last_tables = self.__find_last_columns(query)
+        if last_tables is None:
+            return
+
+        self.query = query
path: mindsdb/interfaces/query_context/last_query.py
text: @user1:
`self.query_in` and `self.query` is the same? may be del one of them?

@author:
replaced, now are:
- query_orig is original query
- query is modified one

diff_hunk: @@ -0,0 +1,158 @@
+import copy
+from collections import defaultdict
+
+from mindsdb_sql.parser.ast import Identifier, Select, BinaryOperation, Last, Constant, Star
+from mindsdb_sql.planner.utils import query_traversal
+
+
+class LastQuery:
+    """
+    Tracks last values in query
+    """
+
+    def __init__(self, query):
+        self.query = None
+
+        # check query type
+        if not isinstance(query, Select):
+            # just skip it
+            return
+
+        self.query_in = query
+
+        query = copy.deepcopy(self.query_in)
path: mindsdb/interfaces/query_context/last_query.py
text: @user1:
It will be executed on any `fetch_dataframe_step`? If so, then need to rid off from `.deepcopy` - it is very slow. May be add new 'copy' method to `mindsdb_sql`?

@author:
changed: deepcopy is executed after analyzing of query and only if query contents LAST

diff_hunk: @@ -0,0 +1,171 @@
+import pandas as pd
+
+from mindsdb_sql.parser.ast import Identifier, Select, OrderBy, NullConstant, Constant, BinaryOperation
+
+from mindsdb.interfaces.storage import db
+from mindsdb.utilities.context import context as ctx
+
+from .last_query import LastQuery
+
+
+class ContextController:
+
+    def handle_db_context_vars(self, query, dn, session):
+
+        l_query = LastQuery(query)
+        if l_query.query is None:
+            # no last keyword, exit
+            return query, None
+
+        query_str = l_query.to_string()
+        context_name = self.get_context()
+
+        rec = self.__get_context_record(context_name, query_str)
+
+        if rec is None:
+            values = self.__get_init_last_values(l_query, dn, session)
+            self.__add_context_record(context_name, query_str, values)
+        else:
+            values = rec.values
+
+        query_out = l_query.apply_values(values)
+
+        def callback(data, columns_info):
+            self._handle_results(l_query, context_name, query_str, data, columns_info)
+
+        return query_out, callback
+
+    def _handle_results(self, l_query, context_name, query_str, data, columns_info):
+
+        if len(data) == 0:
+            return
+
+        max_vals = pd.DataFrame(data).max().to_dict()
+        values = {}
+        # get max values
+        for info in l_query.get_last_columns():
+            target_idx = info['target_idx']
+            if target_idx is not None:
+                # get by index
+                col_name = columns_info[target_idx]['name']
+                value = max_vals.get(col_name)
+            else:
+                # get by name
+                value = max_vals.get(info['column_name'])
+
+            if value is not None:
+                values[info['table_name']] = {info['column_name']: value}
+
+        self.__update_context_record(context_name, query_str, values)
+
+    def drop_query_context(self, object_type, object_name):
+        context_name = self.gen_context_name(object_type, object_name)
+        for rec in db.session.query(db.QueryContext).filter_by(
+            context_name=context_name,
+            company_id=ctx.company_id
+        ).all():
+            db.session.delete(rec)
+        db.session.commit()
+
+    def __get_init_last_values(self, l_query, dn, session):
+        """
+        Get current last values for query
+        """
+        last_values = {}
+        for info in l_query.get_last_columns():
+            col = Identifier(info['column_name'])
+
+            query = Select(
+                targets=[
+                    col
+                ],
+                from_table=info['table'],
+                order_by=[
+                    OrderBy(col, direction='DESC')
+                ],
+                where=BinaryOperation(
+                    op='is not',
+                    args=[
+                        col,
+                        NullConstant()
+                    ]
+                ),
+                limit=Constant(1)
+            )
+
+            data, columns_info = dn.query(
+                query=query,
+                session=session
+            )
+
+            if len(data) == 0:
+                value = None
+            else:
+                value = list(data[0].values())[0]
+            last_values[info['table_name']] = {info['column_name']: value}
+        return last_values
+
+    # Context
+
+    def get_context(self):
+        try:
+            return ctx.context_stack[-1]
+        except Exception:
+            return ''
+
+    def set_context(self, object_type=None, object_name=None):
+        if ctx.context_stack is None:
+            ctx.context_stack = []
+        ctx.context_stack.append(self.gen_context_name(object_type, object_name))
+
+    def release_context(self, object_type=None, object_name=None):
+        if ctx.context_stack is None:
+            return
+        context_name = self.gen_context_name(object_type, object_name)
+        if ctx.context_stack[-1] == context_name:
+            ctx.context_stack.pop()
+
+    def gen_context_name(self, object_type, object_name):
+        if object_name is None:
+            return ''
+        return f'{object_type}-{object_name}'
+
+    def gen_context_vars(self, object_type, object_name):
+        context_name = self.gen_context_name(object_type, object_name)
+        vars = []
+        for rec in db.session.query(db.QueryContext).filter_by(
+            context_name=context_name,
+            company_id=ctx.company_id
+        ):
+            if rec.values is not None:
+                vars.append(rec.values)
+
+        return vars
+
+    # DB
+    def __get_context_record(self, context_name, query_str):
+
+        return db.session.query(db.QueryContext).filter_by(
+            query=query_str,
+            context_name=context_name,
+            company_id=ctx.company_id
+        ).first()
+
+    def __add_context_record(self, context_name, query_str, values):
+
+        rec = db.QueryContext(
+            query=query_str,
+            context_name=context_name,
+            company_id=ctx.company_id,
+            values=values)
+        db.session.add(rec)
+        db.session.commit()
+        return rec
+
+    def __update_context_record(self, context_name, query_str, values):
+        rec = self.__get_context_record(context_name, query_str)
+        rec.values = values
+        db.session.commit()
+
+
+contextController = ContextController()
path: mindsdb/interfaces/query_context/context_controller.py
text: @user1:
Let keep instances names in snake case. For that particular may be better will be `query_context_controller` instead of just `context_controller`? That will be more obvious.

@author:
name replaced to query_context_controller

diff_hunk: @@ -570,15 +570,21 @@ def fill_params(node, **kwargs):
 
             query_traversal(query, fill_params)
 
+            query, context_callback = contextController.handle_db_context_vars(query, dn, self.session)
+
             data, columns_info = dn.query(
                 query=query,
                 session=self.session
             )
 
+            if context_callback:
+                context_callback(data, columns_info)
+
+        # is it used?
         # if this is query: execute it
-        if isinstance(data, ASTNode):
-            subquery = SQLQuery(data, session=self.session)
-            return subquery.fetched_data
+        # if isinstance(data, ASTNode):
path: mindsdb/api/mysql/mysql_proxy/classes/sql_query.py
text: @user1:
del ?

@author:
removed

diff_hunk: @@ -27,6 +27,7 @@ parameters={
     "instance_id":"my-instance",
     "database_id":"example-id",
     "project":"my-project",
+    "dialect": "postgres" // optinonal, default is 'googlesql'
path: mindsdb/integrations/handlers/cloud_spanner_handler/README.md
text: @user1:
This is sql, `//` is not comment symbol. Need to replace to `--` or `/* text */`

@author:
fixed

diff_hunk: @@ -91,16 +92,22 @@ def query(self, query=None, native_query=None, session=None):
         view_query_ast = self.project.query_view(query)
 
         renderer = SqlalchemyRender('mysql')
-        query_str = renderer.get_string(view_query_ast, with_failback=True)
+        view_meta = renderer.get_string(view_query_ast, with_failback=True)
path: mindsdb/api/mysql/mysql_proxy/datahub/datanodes/project_datanode.py
text: @user1:
because of `project.query_view` method changed below, it returns 
```
{
    'id': 123,
    'name': x,
    'query': y,
    'query_ast': z
}
```
therefore this line should raise error

@author:
removed render here, I think we can use AST instead