Merge remote-tracking branch 'origin/develop' into feature/elasticsearch-handler-update

Author: richardokonicha

.github/workflows/build_deploy_staging.yml

@@ -10,7 +10,6 @@ on:
     types:
       - closed
     branches:
-      - 'main'
       - 'develop'
       - 'release/*'
 

mindsdb/api/executor/command_executor.py

@@ -1286,7 +1286,9 @@ def answer_drop_database(self, statement: DropDatabase | DropDatasource) -> Exec
 
     def answer_alter_database(self, statement: AlterDatabase) -> ExecuteAnswer:
         db_name = match_one_part_name(statement.name)
-        self.session.database_controller.update(db_name, data=statement.params, strict_case=statement.name.is_quoted[0])
+        self.session.database_controller.update(
+            db_name, data=statement.params, strict_case=statement.name.is_quoted[0], check_connection=True
+        )
         return ExecuteAnswer()
 
     def answer_drop_tables(self, statement, database_name):

mindsdb/api/executor/planner/plan_join.py

@@ -1,4 +1,3 @@
-from typing import List
 import copy
 from dataclasses import dataclass, field
 
@@ -36,8 +35,8 @@
 class TableInfo:
     integration: str
     table: Identifier
-    aliases: List[str] = field(default_factory=List)
-    conditions: List = None
+    aliases: list[tuple[str, ...]] = field(default_factory=list)
+    conditions: list = None
     sub_select: ast.ASTNode = None
     predictor_info: dict = None
     join_condition = None
@@ -279,49 +278,27 @@ def _check_node_condition(node, **kwargs):
         self.query_context["binary_ops"] = binary_ops
 
     def check_use_limit(self, query_in, join_sequence):
-        """
-        Determine if LIMIT can be pushed down to the first table fetch.
-
-        LIMIT pushdown means: fetch only N rows from the first table, then join.
-        This optimization is ONLY correct when the first table determines the final row count.
-
-        LIMIT pushdown is CORRECT for:
-        - Single table queries (no join)
-        - LEFT JOIN (left table determines row count - each left row appears exactly once)
-        - Joins with ML predictors
-
-        LIMIT pushdown is SLOW for:
-        - INNER JOIN between tables
-        - RIGHT JOIN (right table determines row count, not left)
-
-        When LIMIT pushdown is disabled, we fetch all data and apply LIMIT after the join.
-        This is slower but guarantees correct results.
-        """
+        # if only models (predictors), not for regular table joins
         use_limit = False
         if query_in.having is None and query_in.group_by is None and query_in.limit is not None:
             use_limit = True
-
-            # Check what we're joining
-            has_predictor = False
-            cannot_pushdown_limit = False
+            has_join = False
+            regular_table_count = 0
 
             for item in join_sequence:
                 if isinstance(item, TableInfo):
-                    if item.predictor_info is not None:
-                        has_predictor = True
+                    # Check if it's a regular table (not a predictor, not a subselect)
+                    if item.predictor_info is None and item.sub_select is None:
+                        regular_table_count += 1
                 elif isinstance(item, Join):
-                    join_type = (
-                        item.join_type.upper() if hasattr(item.join_type, "upper") else str(item.join_type).upper()
-                    )
-
                     # LEFT JOIN preserves left table row count - LIMIT pushdown is safe
-                    if join_type in ("LEFT JOIN", "LEFT OUTER JOIN"):
-                        continue
+                    join_type = str(item.join_type).upper() if item.join_type else ""
+                    if join_type not in ("LEFT JOIN", "LEFT OUTER JOIN"):
+                        has_join = True
 
-                    # INNER/RIGHT JOIN: can't push LIMIT down
-                    cannot_pushdown_limit = True
-
-            if cannot_pushdown_limit and not has_predictor:
+            # Disable limit pushdown only if joining MULTIPLE regular database tables
+            # Allow it for: single table, or table + predictor (predictor generates on-demand)
+            if has_join and regular_table_count > 1:
                 use_limit = False
 
         self.query_context["use_limit"] = use_limit
@@ -351,6 +328,7 @@ def replace_subselects(node, **args):
 
         # get all join tables, form join sequence
         join_sequence = self.get_join_sequence(query.from_table)
+        self.join_sequence = join_sequence
 
         # find tables for identifiers used in query
         def _check_identifiers(node, is_table, **kwargs):
@@ -382,7 +360,7 @@ def _check_identifiers(node, is_table, **kwargs):
         for item in join_sequence:
             if isinstance(item, TableInfo):
                 if item.sub_select is not None:
-                    self.process_subselect(item)
+                    self.process_subselect(item, query_in)
                 elif item.predictor_info is not None:
                     self.process_predictor(item, query_in)
                 else:
@@ -409,16 +387,26 @@ def _check_identifiers(node, is_table, **kwargs):
         self.close_partition()
         return self.step_stack.pop()
 
-    def process_subselect(self, item):
+    def process_subselect(self, item, query_in):
         # is sub select
         item.sub_select.alias = None
         item.sub_select.parentheses = False
         step = self.planner.plan_select(item.sub_select)
 
         where = filters_to_bin_op(item.conditions)
 
+        # Column pruning for subselects:
+        # - If subselect has pure SELECT *, we can prune to only needed columns
+        # - If subselect has explicit columns (SELECT a, b, c), pass through all (don't prune)
+        # This preserves column aliases and prevents breaking explicit projections
+        targets = [Star()]
+        if self._can_prune_columns(item):
+            needed_columns = self.get_columns_for_table(item, query_in, self.join_sequence)
+            if needed_columns:
+                targets = needed_columns
+
         # apply table alias
-        query2 = Select(targets=[Star()], where=where)
+        query2 = Select(targets=targets, where=where)
         if item.table.alias is None:
             raise PlanningException(f"Subselect in join have to be aliased: {item.sub_select.to_string()}")
         table_name = item.table.alias.parts[-1]
@@ -431,15 +419,181 @@ def process_subselect(self, item):
         step2 = self.add_plan_step(step2)
         self.step_stack.append(step2)
 
+    def _collect_from_order_by(self, query_in, alias_map, add_column_callback):
+        """Helper to collect columns from ORDER BY clause, resolving aliases and ordinals."""
+        for order_col in query_in.order_by:
+            field = order_col.field
+
+            # Handle ORDER BY ordinal (e.g., ORDER BY 1)
+            if isinstance(field, Constant) and isinstance(field.value, int):
+                ordinal = field.value
+                if 1 <= ordinal <= len(query_in.targets):
+                    target_expr = query_in.targets[ordinal - 1]
+                    query_traversal(target_expr, add_column_callback)
+                continue
+
+            # Handle ORDER BY alias (e.g., ORDER BY alias_name)
+            if isinstance(field, Identifier) and len(field.parts) == 1:
+                alias_name = field.parts[0].lower()
+                if alias_name in alias_map:
+                    query_traversal(alias_map[alias_name], add_column_callback)
+                    continue
+
+            # Regular column reference
+            query_traversal(field, add_column_callback)
+
+    def _join_has_predictor(self, join_sequence) -> bool:
+        """Check if the join sequence contains any predictor."""
+        for item in join_sequence:
+            if isinstance(item, TableInfo) and item.predictor_info is not None:
+                return True
+        return False
+
+    def _can_prune_columns(self, table_info) -> bool:
+        """
+        Determine if column pruning can be applied to this table.
+
+        Returns:
+            True if column pruning can be applied
+            False if we should skip pruning (use SELECT *)
+        """
+        # Predictors/models: cannot prune (need all input features)
+        if table_info.predictor_info is not None:
+            return False
+
+        # If this table is part of a join with a predictor: cannot prune
+        # Predictors may need all columns from joined tables as input features
+        if hasattr(self, "join_sequence") and self._join_has_predictor(self.join_sequence):
+            return False
+
+        # For subselects: can only prune if they have pure SELECT * (no other columns)
+        sub = table_info.sub_select
+        if sub is not None and isinstance(sub, Select):
+            targets = getattr(sub, "targets", None) or []
+            # Can prune only if subselect has PURE SELECT * (one target that is Star)
+            # Cannot prune if:
+            #   - Mixed: SELECT *, col1 (has Star but also other columns)
+            if len(targets) == 1 and isinstance(targets[0], Star):
+                return True  # Pure SELECT * - can prune
+            return False
+
+        # For project tables (KB tables, views, etc.): cannot prune
+        # Project tables need SELECT * for proper column mapping
+        if table_info.integration and table_info.integration in self.planner.projects:
+            return False
+
+        # Regular integration tables: can prune
+        return True
+
+    def get_columns_for_table(self, table_info, query_in, join_sequence):
+        """
+        Collect all columns needed from a specific table for column pruning optimization.
+
+        Note: Caller should check _can_prune_columns() before calling this method.
+
+        Returns a list of column Identifiers or None if we should fetch all columns.
+        """
+        columns = {}
+        has_qualified_star_for_table = False
+
+        alias_map = {}
+        if query_in.targets:
+            for target in query_in.targets:
+                if isinstance(target, Identifier) and target.alias:
+                    alias_map[target.alias.parts[-1].lower()] = target
+
+        def add_column(node, **kwargs):
+            if isinstance(node, Identifier):
+                col_table = self.get_table_for_column(node)
+                if not col_table or col_table.index != table_info.index:
+                    return
+
+                # Check for qualified star: t1.* or alias.*
+                if node.parts and (node.parts[-1] == "*" or isinstance(node.parts[-1], Star)):
+                    nonlocal has_qualified_star_for_table
+                    has_qualified_star_for_table = True
+                    return
+
+                col_name = node.parts[-1]
+                # Make sure col_name is a string, not a Star object
+                if not isinstance(col_name, str):
+                    return
+
+                is_quoted = node.is_quoted[-1] if node.is_quoted and len(node.is_quoted) == len(node.parts) else False
+                # Store - if already exists, keep it quoted if either reference was quoted
+                if col_name in columns:
+                    columns[col_name] = columns[col_name] or is_quoted
+                else:
+                    columns[col_name] = is_quoted
+            elif isinstance(node, Function):
+                # Traverse window function clauses (PARTITION BY, ORDER BY)
+                if hasattr(node, "partition_by") and node.partition_by:
+                    query_traversal(node.partition_by, add_column)
+                if hasattr(node, "order_by") and node.order_by:
+                    for order_item in node.order_by:
+                        query_traversal(order_item.field if hasattr(order_item, "field") else order_item, add_column)
+
+        # Check for bare Star() in targets
+        if query_in.targets:
+            for target in query_in.targets:
+                if isinstance(target, Star):
+                    return None
+
+        # Collect columns from SELECT targets
+        if query_in.targets:
+            query_traversal(query_in.targets, add_column)
+
+        # Collect columns from WHERE clause
+        if query_in.where:
+            query_traversal(query_in.where, add_column)
+
+        # Collect columns from ORDER BY (resolve aliases and ordinals)
+        if query_in.order_by:
+            self._collect_from_order_by(query_in, alias_map, add_column)
+
+        # Collect columns from GROUP BY
+        if query_in.group_by:
+            query_traversal(query_in.group_by, add_column)
+
+        # Collect columns from HAVING
+        if query_in.having:
+            query_traversal(query_in.having, add_column)
+
+        # Collect columns from JOIN conditions
+        for seq_item in join_sequence:
+            if isinstance(seq_item, TableInfo) and seq_item.join_condition:
+                query_traversal(seq_item.join_condition, add_column)
+
+        # If qualified star found for this table, fetch all columns
+        if has_qualified_star_for_table:
+            return None
+
+        # If we found no columns, fetch all
+        if not columns:
+            return None
+
+        # Convert column names to Identifier objects, we need to preserve quoting
+        result = []
+        for col, is_quoted in sorted(columns.items()):
+            ident = Identifier(parts=[col])
+            ident.is_quoted = [is_quoted]
+            result.append(ident)
+        return result
+
     def process_table(self, item, query_in):
         table = copy.deepcopy(item.table)
         table.parts.insert(0, item.integration)
         table.is_quoted.insert(0, False)
-        query2 = Select(from_table=table, targets=[Star()])
-        # parts = tuple(map(str.lower, table_name.parts))
+
+        if self._can_prune_columns(item):
+            needed_columns = self.get_columns_for_table(item, query_in, self.join_sequence)
+            targets = needed_columns if needed_columns else [Star()]
+        else:
+            targets = [Star()]
+
+        query2 = Select(from_table=table, targets=targets)
         conditions = item.conditions
         if "or" in self.query_context["binary_ops"]:
-            # not use conditions
             conditions = []
 
         # For cross-database joins, skip the IN clause optimization

mindsdb/api/http/namespaces/databases.py

@@ -1,4 +1,3 @@
-import time
 import shutil
 import tempfile
 from http import HTTPStatus
@@ -212,23 +211,12 @@ def put(self, database_name):
             new_integration = session.database_controller.get_integration(new_integration_id)
             return new_integration, HTTPStatus.CREATED
 
-        if check_connection:
-            existing_integration = session.integration_controller.get(database_name)
-            temp_name = f"{database_name}_{time.time()}".replace(".", "")
-            try:
-                handler = session.integration_controller.create_tmp_handler(
-                    temp_name, existing_integration["engine"], parameters
-                )
-                status = handler.check_connection()
-            except ImportError as import_error:
-                status = HandlerStatusResponse(success=False, error_message=str(import_error))
-
-            if status.success is not True:
-                return http_error(
-                    HTTPStatus.BAD_REQUEST, "Connection error", status.error_message or "Connection error"
-                )
+        try:
+            session.integration_controller.modify(database_name, parameters, check_connection=check_connection)
+        except Exception as e:
+            status = HandlerStatusResponse(success=False, error_message=str(e))
+            return http_error(HTTPStatus.BAD_REQUEST, "Connection error", status.error_message or "Connection error")
 
-        session.integration_controller.modify(database_name, parameters)
         return session.integration_controller.get(database_name)
 
     @ns_conf.doc("delete_database")