Datanode returns list of lists instead of dict.

mindsdb/api/executor/command_executor.py

@@ -2024,7 +2024,9 @@ def answer_update_model_version(self, statement):
             where=statement.where,
         )
 
-        models, _ = project_datanode.query(query=query, session=self.session)
+        data, columns_info = project_datanode.query(query=query, session=self.session)
+        col_names = [col['name'] for col in columns_info]
+        models = [dict(zip(col_names, item)) for item in data]
 
         # get columns for update
         kwargs = {}
@@ -2053,7 +2055,9 @@ def answer_delete_model_version(self, statement):
             where=statement.where,
         )
 
-        models, _ = project_datanode.query(query=query, session=self.session)
+        data, columns_info = project_datanode.query(query=query, session=self.session)
+        col_names = [col['name'] for col in columns_info]
+        models = [dict(zip(col_names, item)) for item in data]
 
         self.session.model_controller.delete_model_version(models)
         return ExecuteAnswer(ANSWER_TYPE.OK)

mindsdb/api/executor/datahub/datanodes/information_schema_datanode.py

@@ -1043,4 +1043,4 @@ def query(self, query: ASTNode, session=None):
 
         columns_info = [{"name": k, "type": v} for k, v in data.dtypes.items()]
 
-        return data.to_dict(orient="records"), columns_info
+        return data.to_dict(orient="split")['data'], columns_info

mindsdb/api/executor/datahub/datanodes/integration_datanode.py

@@ -190,5 +190,5 @@ def query(self, query=None, native_query=None, session=None):
             }
             for k, v in df.dtypes.items()
         ]
-        data = df.to_dict(orient='records')
+        data = df.to_dict(orient='split')['data']
         return data, columns_info

mindsdb/api/executor/datahub/datanodes/project_datanode.py

@@ -1,7 +1,5 @@
 from copy import deepcopy
 
-import pandas as pd
-
 from mindsdb_sql import parse_sql
 from mindsdb_sql.parser.ast import (
     BinaryOperation,
@@ -75,7 +73,7 @@ def query(self, query=None, native_query=None, session=None):
             if kb_table:
                 # this is the knowledge db
                 kb_table.update_query(query)
-                return pd.DataFrame(), []
+                return [], []
 
             raise NotImplementedError(f"Can't update object: {query_table}")
 
@@ -85,7 +83,7 @@ def query(self, query=None, native_query=None, session=None):
             if kb_table:
                 # this is the knowledge db
                 kb_table.delete_query(query)
-                return pd.DataFrame(), []
+                return [], []
 
             raise NotImplementedError(f"Can't delete object: {query_table}")
 
@@ -147,7 +145,7 @@ def query(self, query=None, native_query=None, session=None):
                     for k, v in df.dtypes.items()
                 ]
 
-                return df.to_dict(orient='records'), columns_info
+                return df.to_dict(orient='split')['data'], columns_info
 
             kb_table = session.kb_controller.get_table(query_table, self.project.id)
             if kb_table:
@@ -161,7 +159,7 @@ def query(self, query=None, native_query=None, session=None):
                     for k, v in df.dtypes.items()
                 ]
 
-                return df.to_dict(orient='records'), columns_info
+                return df.to_dict(orient='split')['data'], columns_info
 
             raise EntityNotExistsError(f"Can't select from {query_table} in project")
         else:

mindsdb/api/executor/sql_query/steps/fetch_dataframe.py

@@ -113,6 +113,7 @@ def call(self, step):
                 table_alias=table_alias[2],
                 database=table_alias[0]
             ))
-        result.add_records(data)
+        for record in data:
+            result.add_record_raw(record)
 
         return result

mindsdb/interfaces/query_context/context_controller.py

@@ -96,7 +96,7 @@ def _result_callback(self, l_query: LastQuery,
         if len(data) == 0:
             return
 
-        df = pd.DataFrame(data)
+        df = pd.DataFrame(data, columns=[col['name'] for col in columns_info])
         values = {}
         # get max values
         for info in l_query.get_last_columns():
@@ -161,7 +161,7 @@ def __get_init_last_values(self, l_query: LastQuery, dn, session) -> dict:
             if len(data) == 0:
                 value = None
             else:
-                value = list(data[0].values())[0]
+                value = data[0][0]
             if value is not None:
                 last_values[info['table_name']] = {info['column_name']: value}
 

tests/unit/executor_test_base.py

@@ -327,7 +327,10 @@ def native_query_f(query):
             for table, df in tables.items():
                 con.register(table, df)
             try:
-                result_df = con.execute(query).fetchdf()
+                con.execute(query)
+                columns = [c[0] for c in con.description]
+                result_df = pd.DataFrame(con.fetchall(), columns=columns)
+
                 result_df = result_df.replace({np.nan: None})
             except Exception:
                 # it can be not supported command like update or insert

tests/unit/test_project_structure.py

@@ -811,6 +811,27 @@ def test_last_in_job(self, data_handler, scheduler):
         assert 'a > 2' in sql
         assert "b = 'b'" in sql
 
+    @patch('mindsdb.integrations.handlers.postgres_handler.Handler')
+    def test_duplicated_cols(self, data_handler):
+        df1 = pd.DataFrame([
+            {'id': 1, 'a': 1},
+            {'id': 2, 'a': 2},
+            {'id': 3, 'a': 3},
+        ])
+        df2 = pd.DataFrame([
+            {'id': 1, 'a': 10},
+            {'id': 2, 'a': 20},
+        ])
+        self.set_handler(data_handler, name='pg', tables={'tbl1': df1, 'tbl2': df2})
+
+        ret = self.run_sql('''
+            select * from pg.tbl1 as a
+            join pg.tbl2 as b on a.id=b.id
+        ''')
+
+        first_row = ret.to_dict('split')['data'][0]
+        assert first_row == [1, 1, 1, 10]
+
 
 class TestJobs(BaseExecutorDummyML):