From 3421f2d015ef8f7afb2495cdf73bdfa811b60621 Mon Sep 17 00:00:00 2001 From: Gabriel Bressan Date: Sun, 15 Feb 2026 18:28:01 -0400 Subject: [PATCH 1/6] Enhance XML parsing: clean HTML tags and extract URLs from anchor tags --- .../format_parsers.py | 33 +++++++++++++++++-- .../multi_format_table.py | 19 ++++++++--- 2 files changed, 44 insertions(+), 8 deletions(-) diff --git a/mindsdb/integrations/handlers/multi_format_api_handler/format_parsers.py b/mindsdb/integrations/handlers/multi_format_api_handler/format_parsers.py index f3d002b92ab..5adc47e42c9 100644 --- a/mindsdb/integrations/handlers/multi_format_api_handler/format_parsers.py +++ b/mindsdb/integrations/handlers/multi_format_api_handler/format_parsers.py @@ -157,7 +157,23 @@ def parse_xml(content: str) -> pd.DataFrame: # Empty XML or no parseable structure return pd.DataFrame({'root_tag': [root.tag], 'content': [root.text or '']}) - return pd.DataFrame(records) + df = pd.DataFrame(records) + + # Final cleanup: ensure no HTML tags remain in any string columns + # This is a defensive measure in case HTML tags were not caught during parsing + cleaned_count = 0 + for col in df.columns: + if df[col].dtype == 'object': # String/object columns + # Only clean values that contain '<' (potential HTML) + html_mask = df[col].apply(lambda x: pd.notna(x) and isinstance(x, str) and '<' in x) + if html_mask.any(): + df.loc[html_mask, col] = df.loc[html_mask, col].apply(lambda x: _clean_cdata_content(str(x))) + cleaned_count += html_mask.sum() + + if cleaned_count > 0: + logger.info(f"Cleaned HTML tags from {cleaned_count} values during DataFrame post-processing") + + return df except ET.ParseError as e: logger.error(f"XML parsing error: {e}") @@ -204,8 +220,19 @@ def _clean_cdata_content(text: str) -> Union[str, pd.Timestamp]: # Strip leading/trailing whitespace (common in CDATA sections) text = text.strip() - # Remove HTML tags (e.g., URL -> URL) - text = re.sub(r'<[^>]+>', '', text) + # Extract URL from anchor tag href if present + # Matches: ... or + href_match = re.search(r']*href=["\']([^"\']+)["\'][^>]*>', text, re.IGNORECASE) + if href_match: + logger.debug(f"Extracting URL from anchor tag: {text[:100]}...") + text = href_match.group(1) # Extract just the URL from href attribute + + # Remove HTML tags (multiple passes for nested tags) + # Loop until no more tags are found + prev_text = None + while prev_text != text: + prev_text = text + text = re.sub(r'<[^>]+>', '', text) # Decode HTML entities (e.g., & -> &, < -> <) text = unescape(text) diff --git a/mindsdb/integrations/handlers/multi_format_api_handler/multi_format_table.py b/mindsdb/integrations/handlers/multi_format_api_handler/multi_format_table.py index 7e63ba28117..ca95752b61c 100644 --- a/mindsdb/integrations/handlers/multi_format_api_handler/multi_format_table.py +++ b/mindsdb/integrations/handlers/multi_format_api_handler/multi_format_table.py @@ -9,7 +9,14 @@ import logging from mindsdb.integrations.libs.api_handler import APIResource -from mindsdb.integrations.utilities.sql_utils import FilterCondition +from mindsdb.integrations.utilities.sql_utils import ( + extract_comparison_conditions, + filter_dataframe, + sort_dataframe, + FilterCondition, + FilterOperator, + SortColumn, +) from .format_parsers import parse_response @@ -24,10 +31,12 @@ class MultiFormatAPITable(APIResource): def list( self, - conditions: Optional[List[FilterCondition]] = None, - limit: Optional[int] = None, - **kwargs - ) -> pd.DataFrame: + conditions: List[FilterCondition] = None, + limit: int = None, + sort: List[SortColumn] = None, + targets: List[str] = None, + **kwargs, + ): """ Fetch data from URL and parse according to detected format. From a9028385a6d9772a82cfa6c30c3df4a5acd765d4 Mon Sep 17 00:00:00 2001 From: Gabriel Bressan Date: Sun, 15 Feb 2026 23:07:40 -0400 Subject: [PATCH 2/6] Enhance error handling and improve user feedback in query planning and execution steps --- mindsdb/api/executor/planner/plan_join.py | 67 +++++++++++++-- mindsdb/api/executor/planner/query_planner.py | 24 +++++- .../executor/sql_query/steps/insert_step.py | 5 +- .../api/executor/sql_query/steps/join_step.py | 82 ++++++++++++------- .../executor/sql_query/steps/project_step.py | 16 +++- .../sql_query/steps/subselect_step.py | 7 +- .../format_parsers.py | 53 +++++++++++- 7 files changed, 206 insertions(+), 48 deletions(-) diff --git a/mindsdb/api/executor/planner/plan_join.py b/mindsdb/api/executor/planner/plan_join.py index cc8f179b144..554319c1539 100644 --- a/mindsdb/api/executor/planner/plan_join.py +++ b/mindsdb/api/executor/planner/plan_join.py @@ -107,6 +107,7 @@ def __init__(self, planner): self.step_stack = None self.query_context = {} + self.table_columns = {} # TableInfo -> set of column names self.partition = None @@ -159,7 +160,12 @@ def resolve_table(self, table): integration = self.planner.default_namespace if integration is None and not hasattr(table, "sub_select"): - raise PlanningException(f"Database not found for: {table}") + available_dbs = ", ".join(list(self.planner.databases)[:10]) + raise PlanningException( + f"Database not found for table: {table}.\n" + f"Available databases: {available_dbs}\n" + "Hint: Use format 'database_name.table_name'" + ) sub_select = getattr(table, "sub_select", None) @@ -240,7 +246,11 @@ def check_node_condition(self, node): table_info = self.get_table_for_column(arg1) if table_info is None: - raise PlanningException(f"Table not found for identifier: {arg1.to_string()}") + known = ['.'.join(a) for t in self.tables for a in t.aliases] + raise PlanningException( + f"Table not found for identifier: {arg1.to_string()}.\n" + f"Known tables/aliases: {', '.join(known)}" + ) # keep only column name arg1.parts = [arg1.parts[-1]] @@ -308,13 +318,34 @@ def replace_subselects(node, **args): # get all join tables, form join sequence join_sequence = self.get_join_sequence(query.from_table) + # Collect column references per table from SELECT targets and JOIN ON + # conditions only (not WHERE — those are pushed down as filter params). + # This ensures API-type handlers fetch the right columns. + def _collect_fetch_columns(node, is_table, **kwargs): + if not is_table and isinstance(node, Identifier) and len(node.parts) > 1: + table_info = self.get_table_for_column(node) + if table_info is not None: + table_id = id(table_info) + if table_id not in self.table_columns: + self.table_columns[table_id] = set() + self.table_columns[table_id].add(node.parts[-1]) + + query_traversal(query.targets, _collect_fetch_columns) + for tbl in self.tables: + if tbl.join_condition is not None: + query_traversal(tbl.join_condition, _collect_fetch_columns) + # find tables for identifiers used in query def _check_identifiers(node, is_table, **kwargs): if not is_table and isinstance(node, Identifier): if len(node.parts) > 1: table_info = self.get_table_for_column(node) if table_info is None: - raise PlanningException(f"Table not found for identifier: {node.to_string()}") + known = [str(a) for t in self.tables for a in t.aliases] + raise PlanningException( + f"Table not found for identifier: {node.to_string()}.\n" + f"Known tables/aliases: {', '.join(known)}" + ) # # replace identifies name col_parts = list(table_info.aliases[-1]) @@ -391,15 +422,32 @@ 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)) + + # Build WHERE conditions first (from query WHERE + JOIN ON filter params) conditions = item.conditions if "or" in self.query_context["binary_ops"]: - # not use conditions conditions = [] - conditions += self.get_filters_from_join_conditions(item) + # Use specific columns from SELECT targets and JOIN ON conditions. + # Exclude columns that are filter parameters (pushed as WHERE conditions) + # so API handlers (e.g. Google Analytics) don't try to use filter params + # like start_date/end_date as SELECT dimensions. + referenced_cols = self.table_columns.get(id(item)) + if referenced_cols: + filter_col_names = set() + for cond in conditions: + if isinstance(cond, BinaryOperation): + for arg in cond.args: + if isinstance(arg, Identifier): + filter_col_names.add(arg.parts[-1]) + fetch_cols = referenced_cols - filter_col_names + targets = [Identifier(parts=[col]) for col in fetch_cols] if fetch_cols else [Star()] + else: + targets = [Star()] + + query2 = Select(from_table=table, targets=targets) + if self.query_context["use_limit"]: order_by = None if query_in.order_by is not None: @@ -493,7 +541,10 @@ def _check_conditions(node, **kwargs): arg1, arg2 = arg2, arg1 if isinstance(arg2, Constant): - conditions.append(node) + conditions.append(copy.deepcopy(node)) + # Neutralize in the join condition tree so DuckDB doesn't + # try to resolve filter params as DataFrame columns. + node.args = [Constant(0), Constant(0)] elif table2 is not None: data_conditions.append([arg1, arg2]) diff --git a/mindsdb/api/executor/planner/query_planner.py b/mindsdb/api/executor/planner/query_planner.py index 4cbea5685a6..59a7ebb95f4 100644 --- a/mindsdb/api/executor/planner/query_planner.py +++ b/mindsdb/api/executor/planner/query_planner.py @@ -601,20 +601,35 @@ def plan_select_from_predictor(self, select): for target in select.targets: if isinstance(target, Identifier): new_query_targets.append(disambiguate_predictor_column_identifier(target, predictor)) - elif type(target) in (Star, Constant, Function): + elif type(target) in (Star, Constant, Function, Parameter): new_query_targets.append(target) else: - raise PlanningException(f"Unknown select target {type(target)}") + raise PlanningException( + f"Unsupported expression in SELECT target: {type(target).__name__}.\n" + "When querying a predictor, targets must be column names, *, constants, or functions.\n" + f"Problematic target: {target}" + ) if select.group_by or select.having: + unsupported = [] + if select.group_by: + unsupported.append("GROUP BY") + if select.having: + unsupported.append("HAVING") raise PlanningException( - "Unsupported operation when querying predictor. Only WHERE is allowed and required." + f"Unsupported clause(s) for predictor query: {', '.join(unsupported)}.\n" + "Hint: Wrap the predictor in a subquery first:\n" + " SELECT ... FROM (SELECT * FROM predictor WHERE ...) GROUP BY ..." ) row_dict = {} where_clause = select.where if not where_clause: - raise PlanningException("WHERE clause required when selecting from predictor") + raise PlanningException( + "WHERE clause required when selecting from predictor.\n" + "Predictor queries need input data via WHERE, e.g.:\n" + f" SELECT * FROM {select.from_table} WHERE column = value" + ) predictor_identifier = utils.get_predictor_name_identifier(predictor) recursively_extract_column_values(where_clause, row_dict, predictor_identifier) @@ -748,6 +763,7 @@ def plan_project(self, query, dataframe, ignore_doubles=False): or isinstance(target, Function) or isinstance(target, Constant) or isinstance(target, BinaryOperation) + or isinstance(target, Parameter) ): out_identifiers.append(target) else: diff --git a/mindsdb/api/executor/sql_query/steps/insert_step.py b/mindsdb/api/executor/sql_query/steps/insert_step.py index 892a4129e41..6165f8b5d52 100644 --- a/mindsdb/api/executor/sql_query/steps/insert_step.py +++ b/mindsdb/api/executor/sql_query/steps/insert_step.py @@ -33,7 +33,10 @@ def call(self, step): dn = self.session.datahub.get(integration_name) if hasattr(dn, "create_table") is False: - raise NotSupportedYet(f"Creating table in '{integration_name}' is not supported") + raise NotSupportedYet( + f"Creating table in '{integration_name}' is not supported.\n" + "Hint: Create the table directly in your database, then query it through MindsDB." + ) if step.dataframe is not None: data = self.steps_data[step.dataframe.step_num] diff --git a/mindsdb/api/executor/sql_query/steps/join_step.py b/mindsdb/api/executor/sql_query/steps/join_step.py index 166521d5383..d9e744c6705 100644 --- a/mindsdb/api/executor/sql_query/steps/join_step.py +++ b/mindsdb/api/executor/sql_query/steps/join_step.py @@ -57,51 +57,73 @@ def call(self, step): join_type = 'left join' elif right_data.is_prediction: join_type = 'right join' + table_a, names_a = left_data.to_df_cols(prefix='A') + table_b, names_b = right_data.to_df_cols(prefix='B') + + query = f""" + SELECT * FROM table_a {join_type} table_b + ON {join_condition} + """ + resp_df, _description = query_df_with_type_infer_fallback(query, { + 'table_a': table_a, + 'table_b': table_b + }) else: - def adapt_condition(node, **kwargs): - if not isinstance(node, Identifier) or len(node.parts) != 2: - return - - table_alias, alias = node.parts - cols = left_data.find_columns(alias, table_alias) - if len(cols) == 1: - col_name = cols[0].get_hash_name(prefix='A') - return Identifier(parts=['table_a', col_name]) - - cols = right_data.find_columns(alias, table_alias) - if len(cols) == 1: - col_name = cols[0].get_hash_name(prefix='B') - return Identifier(parts=['table_b', col_name]) + # Register DataFrames with DuckDB using the original table aliases + # so DuckDB resolves column references in ON conditions natively, + # including functions like LOWER(), SPLIT_PART(), etc. + left_alias = left_data.columns[0].table_alias if left_data.columns else 'table_a' + right_alias = right_data.columns[0].table_alias if right_data.columns else 'table_b' + if left_alias == right_alias: + right_alias = f'{right_alias}_r' + + left_df = left_data.to_df() + right_df = right_data.to_df() + + # Build SELECT with hash-named aliases to avoid column name collisions + names_a = {} + select_parts = [] + for col in left_data.columns: + hash_name = col.get_hash_name('A') + names_a[hash_name] = col + select_parts.append(f'"{left_alias}"."{col.alias}" AS "{hash_name}"') + + names_b = {} + for col in right_data.columns: + hash_name = col.get_hash_name('B') + names_b[hash_name] = col + select_parts.append(f'"{right_alias}"."{col.alias}" AS "{hash_name}"') if step.query.condition is None: - # prevent memory overflow if len(left_data) * len(right_data) < 10 ** 7: step.query.condition = BinaryOperation(op='=', args=[Constant(0), Constant(0)]) else: - raise NotSupportedYet('Unable to join table without condition') + raise NotSupportedYet( + 'Unable to join tables without a condition: the resulting cross join ' + f'would produce {len(left_data) * len(right_data):,} rows ' + f'({len(left_data):,} x {len(right_data):,}), exceeding the 10,000,000 row limit.\n' + 'Hint: Add an ON clause, e.g.: SELECT * FROM t1 JOIN t2 ON t1.id = t2.id' + ) condition = copy.deepcopy(step.query.condition) - query_traversal(condition, adapt_condition) - join_condition = SqlalchemyRender('postgres').get_string(condition) join_type = step.query.join_type - table_a, names_a = left_data.to_df_cols(prefix='A') - table_b, names_b = right_data.to_df_cols(prefix='B') - - query = f""" - SELECT * FROM table_a {join_type} table_b - ON {join_condition} - """ - resp_df, _description = query_df_with_type_infer_fallback(query, { - 'table_a': table_a, - 'table_b': table_b - }) + select_clause = ', '.join(select_parts) + query = f""" + SELECT {select_clause} + FROM "{left_alias}" {join_type} "{right_alias}" + ON {join_condition} + """ + resp_df, _description = query_df_with_type_infer_fallback(query, { + left_alias: left_df, + right_alias: right_df, + }) resp_df.replace({np.nan: None}, inplace=True) names_a.update(names_b) - data = ResultSet.from_df_cols(df=resp_df, columns_dict=names_a) + data = ResultSet.from_df_cols(df=resp_df, columns_dict=names_a, strict=False) for col in data.find_columns('__mindsdb_row_id'): data.del_column(col) diff --git a/mindsdb/api/executor/sql_query/steps/project_step.py b/mindsdb/api/executor/sql_query/steps/project_step.py index cb5e9db46b7..c3840afe498 100644 --- a/mindsdb/api/executor/sql_query/steps/project_step.py +++ b/mindsdb/api/executor/sql_query/steps/project_step.py @@ -16,6 +16,7 @@ ) from .base import BaseStepCall +from .fetch_dataframe import get_fill_param_fnc class ProjectStepCall(BaseStepCall): @@ -36,10 +37,14 @@ def call(self, step): if col.table_name != col.table_alias: tbl_idx[col.table_alias].append(name) + # Resolve Parameter nodes from previous step results + fill_params = get_fill_param_fnc(self.steps_data) + resolved_columns = query_traversal(step.columns, fill_params) or step.columns + # analyze condition and change name of columns def check_fields(node, is_table=None, **kwargs): if is_table: - raise NotSupportedYet('Subqueries is not supported in target') + return # skip table nodes — subqueries already resolved to constants if isinstance(node, Identifier): # only column name col_name = node.parts[-1] @@ -62,13 +67,18 @@ def check_fields(node, is_table=None, **kwargs): key = (table_name, col_name) if key not in col_idx: - raise KeyColumnDoesNotExist(f'Table not found for column: {key}') + simple_cols = [k for k in col_idx.keys() if isinstance(k, str)] + raise KeyColumnDoesNotExist( + f'Column not found: {key}.\n' + f'Available columns: {", ".join(str(c) for c in simple_cols[:20])}' + + (f' ... and {len(simple_cols) - 20} more' if len(simple_cols) > 20 else '') + ) new_name = col_idx[key] return Identifier(parts=[new_name], alias=node.alias) query = Select( - targets=step.columns, + targets=resolved_columns, from_table=Identifier('df_table') ) diff --git a/mindsdb/api/executor/sql_query/steps/subselect_step.py b/mindsdb/api/executor/sql_query/steps/subselect_step.py index 9cb139a58c4..ffd21ed8916 100644 --- a/mindsdb/api/executor/sql_query/steps/subselect_step.py +++ b/mindsdb/api/executor/sql_query/steps/subselect_step.py @@ -184,7 +184,12 @@ def check_fields(node, is_target=None, **kwargs): search_idx = col_idx if column_quoted else lower_col_idx if key not in search_idx: - raise KeyColumnDoesNotExist(f"Table not found for column: {key}") + available = [k for k in col_idx.keys() if isinstance(k, str)] + raise KeyColumnDoesNotExist( + f"Column not found: {key}.\n" + f"Available columns: {', '.join(str(c) for c in available[:20])}" + + (f' ... and {len(available) - 20} more' if len(available) > 20 else '') + ) new_name = search_idx[key] return Identifier(parts=[new_name], alias=node.alias) diff --git a/mindsdb/integrations/handlers/multi_format_api_handler/format_parsers.py b/mindsdb/integrations/handlers/multi_format_api_handler/format_parsers.py index 5adc47e42c9..3181dc1b72e 100644 --- a/mindsdb/integrations/handlers/multi_format_api_handler/format_parsers.py +++ b/mindsdb/integrations/handlers/multi_format_api_handler/format_parsers.py @@ -157,7 +157,11 @@ def parse_xml(content: str) -> pd.DataFrame: # Empty XML or no parseable structure return pd.DataFrame({'root_tag': [root.tag], 'content': [root.text or '']}) - df = pd.DataFrame(records) + # Use json_normalize to flatten nested dicts into underscore-separated + # column names (e.g. author_name). Then ensure all values are scalars + # so DuckDB receives VARCHAR-compatible types, not VARCHAR[]. + df = pd.json_normalize(records, sep='_') + df = _ensure_scalar_columns(df) # Final cleanup: ensure no HTML tags remain in any string columns # This is a defensive measure in case HTML tags were not caught during parsing @@ -244,6 +248,53 @@ def _clean_cdata_content(text: str) -> Union[str, pd.Timestamp]: return _try_parse_date(text) +def _serialize_non_scalar(value: Any) -> Any: + """ + Convert non-scalar values (lists, dicts) to string representations + suitable for flat DataFrame columns compatible with DuckDB. + + Args: + value: Any cell value + + Returns: + Scalar value (string, number, timestamp, or None) + """ + if value is None or isinstance(value, (str, int, float, bool, pd.Timestamp)): + return value + if isinstance(value, list): + if len(value) == 0: + return '' + if all(isinstance(item, (str, int, float)) for item in value): + return ', '.join(str(item) for item in value) + return json.dumps(value, ensure_ascii=False, default=str) + if isinstance(value, dict): + return json.dumps(value, ensure_ascii=False, default=str) + return str(value) + + +def _ensure_scalar_columns(df: pd.DataFrame) -> pd.DataFrame: + """ + Ensure all DataFrame columns contain only scalar values. + Converts any remaining list or dict values to strings so that + DuckDB receives only VARCHAR-compatible types. + + Args: + df: DataFrame that may contain non-scalar cell values + + Returns: + DataFrame with all scalar values + """ + for col in df.columns: + if df[col].dtype == 'object': + has_non_scalar = df[col].apply( + lambda x: isinstance(x, (list, dict)) + ).any() + if has_non_scalar: + logger.debug(f"Converting non-scalar values in column '{col}' to strings") + df[col] = df[col].apply(_serialize_non_scalar) + return df + + def _xml_element_to_dict(element: ET.Element) -> Dict[str, Any]: """ Convert XML element to dictionary. From 96a68f8a74fcc78984169d1802476103610cbddc Mon Sep 17 00:00:00 2001 From: Gabriel Bressan Date: Tue, 17 Feb 2026 21:57:07 -0400 Subject: [PATCH 3/6] Enhance query planning: improve identifier collection for complex expressions and clear CTEs after planning --- mindsdb/api/executor/planner/plan_join.py | 11 ++- mindsdb/api/executor/planner/query_planner.py | 1 + .../google_analytics_data_tables.py | 67 ++++++++++++++----- 3 files changed, 59 insertions(+), 20 deletions(-) diff --git a/mindsdb/api/executor/planner/plan_join.py b/mindsdb/api/executor/planner/plan_join.py index 554319c1539..1bbc146fac0 100644 --- a/mindsdb/api/executor/planner/plan_join.py +++ b/mindsdb/api/executor/planner/plan_join.py @@ -437,10 +437,15 @@ def process_table(self, item, query_in): if referenced_cols: filter_col_names = set() for cond in conditions: - if isinstance(cond, BinaryOperation): - for arg in cond.args: + if isinstance(cond, BinaryOperation) and len(cond.args) >= 2: + # Only exclude columns that are scalar filter parameters (col = Constant). + # Do NOT exclude columns used in cross-table JOIN predicates + # (col IN (Parameter)), as those columns must still be fetched. + for i, arg in enumerate(cond.args[:2]): if isinstance(arg, Identifier): - filter_col_names.add(arg.parts[-1]) + other = cond.args[1 - i] + if isinstance(other, Constant): + filter_col_names.add(arg.parts[-1]) fetch_cols = referenced_cols - filter_col_names targets = [Identifier(parts=[col]) for col in fetch_cols] if fetch_cols else [Star()] else: diff --git a/mindsdb/api/executor/planner/query_planner.py b/mindsdb/api/executor/planner/query_planner.py index 59a7ebb95f4..1c397b1eff7 100644 --- a/mindsdb/api/executor/planner/query_planner.py +++ b/mindsdb/api/executor/planner/query_planner.py @@ -892,6 +892,7 @@ def plan_select(self, query, integration=None): if query.cte is not None: self.plan_cte(query) + query.cte = None # CTEs have been decomposed into steps; clear so DuckDB doesn't re-execute them from_table = query.from_table diff --git a/mindsdb/integrations/handlers/google_analytics_handler/google_analytics_data_tables.py b/mindsdb/integrations/handlers/google_analytics_handler/google_analytics_data_tables.py index 1435cabd3c1..f64167b722b 100644 --- a/mindsdb/integrations/handlers/google_analytics_handler/google_analytics_data_tables.py +++ b/mindsdb/integrations/handlers/google_analytics_handler/google_analytics_data_tables.py @@ -30,6 +30,37 @@ logger = log.getLogger(__name__) +def _collect_identifiers(node) -> List[str]: + """Recursively collect all Identifier column names from an AST node. + + Walks into CASE WHEN, Function args, BinaryOperation, etc. so that + columns referenced inside complex expressions are not missed. + """ + if node is None: + return [] + if isinstance(node, ast.Identifier): + return [str(node.parts[-1])] + if isinstance(node, ast.Case): + names = [] + for condition, result in node.rules: + names.extend(_collect_identifiers(condition)) + names.extend(_collect_identifiers(result)) + names.extend(_collect_identifiers(node.default)) + return names + if isinstance(node, ast.Function): + names = [] + for arg in (node.args or []): + names.extend(_collect_identifiers(arg)) + return names + if isinstance(node, ast.BinaryOperation): + return _collect_identifiers(node.args[0]) + _collect_identifiers(node.args[1]) + if isinstance(node, ast.UnaryOperation): + return _collect_identifiers(node.args[0]) + if isinstance(node, ast.TypeCast): + return _collect_identifiers(node.arg) + return [] + + class ReportsTable(APITable): """ Table for running standard Google Analytics reports using the Data API. @@ -94,27 +125,29 @@ def select(self, query: ast.Select) -> pd.DataFrame: else: dimension_filters[api_name] = (op, val) - # Extract dimensions and metrics from SELECT columns + # Extract dimensions and metrics from SELECT columns. + # Use _collect_identifiers to recurse into CASE WHEN, SUM(...), and other + # complex expressions so that all referenced columns are fetched from GA4. dimensions = [] metrics = [] if query.targets: - for target in query.targets: - if isinstance(target, ast.Star): - # If SELECT *, use default dimensions and metrics - dimensions = [Dimension(name='date'), Dimension(name='country')] - metrics = [Metric(name='activeUsers'), Metric(name='sessions')] - break - elif isinstance(target, ast.Identifier): - col_name = str(target.parts[-1]) - # Convert underscores back to colons for custom dimensions (reverse sanitization) - # customEvent_job_title -> customEvent:job_title - api_name = self._unsanitize_column_name(col_name) - # Determine if it's a dimension or metric based on common patterns - if self._is_metric(col_name): - metrics.append(Metric(name=api_name)) - else: - dimensions.append(Dimension(name=api_name)) + has_star = any(isinstance(t, ast.Star) for t in query.targets) + if has_star: + dimensions = [Dimension(name='date'), Dimension(name='country')] + metrics = [Metric(name='activeUsers'), Metric(name='sessions')] + else: + seen = set() + for target in query.targets: + for col_name in _collect_identifiers(target): + if col_name in seen: + continue + seen.add(col_name) + api_name = self._unsanitize_column_name(col_name) + if self._is_metric(col_name): + metrics.append(Metric(name=api_name)) + else: + dimensions.append(Dimension(name=api_name)) # If no dimensions/metrics specified, use defaults if not dimensions: From bf0084d686a2e726e681edaf6cda3aae260c37a8 Mon Sep 17 00:00:00 2001 From: Gabriel Bressan Date: Tue, 17 Feb 2026 22:04:26 -0400 Subject: [PATCH 4/6] Enhance query handling: return all raw columns for complex expressions in Google Calendar and Search Analytics tables --- .../google_calendar_tables.py | 19 ++++++++++++++++++- .../google_search_tables.py | 18 +++++++++++++++--- 2 files changed, 33 insertions(+), 4 deletions(-) diff --git a/mindsdb/integrations/handlers/google_calendar_handler/google_calendar_tables.py b/mindsdb/integrations/handlers/google_calendar_handler/google_calendar_tables.py index a5ca7237f5f..b1b2b5193df 100644 --- a/mindsdb/integrations/handlers/google_calendar_handler/google_calendar_tables.py +++ b/mindsdb/integrations/handlers/google_calendar_handler/google_calendar_tables.py @@ -97,7 +97,12 @@ def select(self, query: ast.Select) -> DataFrame: elif isinstance(target, ast.Identifier): selected_columns.append(target.parts[-1]) else: - raise ValueError(f"Unknown query target {type(target)}") + # Complex expression (CASE WHEN, SUM, etc.) — the outer DuckDB + # layer handles the computation; return all raw columns so it can. + selected_columns = self.get_columns() + break + if not selected_columns: + selected_columns = self.get_columns() if len(events) == 0: events = pd.DataFrame([], columns=selected_columns) @@ -371,6 +376,12 @@ def select(self, query: ast.Select) -> DataFrame: break elif isinstance(target, ast.Identifier): selected_columns.append(target.parts[-1]) + else: + # Complex expression — return all raw columns for DuckDB to process. + selected_columns = self.get_columns() + break + if not selected_columns: + selected_columns = self.get_columns() # Call handler method calendars = self.handler.call_application_api( @@ -469,6 +480,12 @@ def select(self, query: ast.Select) -> DataFrame: break elif isinstance(target, ast.Identifier): selected_columns.append(target.parts[-1]) + else: + # Complex expression — return all raw columns for DuckDB to process. + selected_columns = self.get_columns() + break + if not selected_columns: + selected_columns = self.get_columns() # Call handler method busy_times = self.handler.call_application_api( diff --git a/mindsdb/integrations/handlers/google_search_handler/google_search_tables.py b/mindsdb/integrations/handlers/google_search_handler/google_search_tables.py index 30a948db09f..d55007405e1 100644 --- a/mindsdb/integrations/handlers/google_search_handler/google_search_tables.py +++ b/mindsdb/integrations/handlers/google_search_handler/google_search_tables.py @@ -129,7 +129,11 @@ def select(self, query: ast.Select) -> DataFrame: elif isinstance(target, ast.Identifier): selected_columns.append(target.parts[-1]) else: - raise ValueError(f"Unknown query target {type(target)}") + # Complex expression — return all raw columns for DuckDB to process. + selected_columns = self.get_columns(dimensions) + break + if not selected_columns: + selected_columns = self.get_columns(dimensions) if len(traffic_data) == 0: traffic_data = pd.DataFrame([], columns=selected_columns) else: @@ -307,7 +311,11 @@ def select(self, query: ast.Select) -> DataFrame: elif isinstance(target, ast.Identifier): selected_columns.append(target.parts[-1]) else: - raise ValueError(f"Unknown query target {type(target)}") + # Complex expression — return all raw columns for DuckDB to process. + selected_columns = self.get_columns() + break + if not selected_columns: + selected_columns = self.get_columns() if len(sitemaps) == 0: sitemaps = pd.DataFrame([], columns=selected_columns) @@ -432,7 +440,11 @@ def select(self, query: ast.Select) -> DataFrame: elif isinstance(target, ast.Identifier): selected_columns.append(target.parts[-1]) else: - raise ValueError(f"Unknown query target {type(target)}") + # Complex expression — return all raw columns for DuckDB to process. + selected_columns = self.get_columns() + break + if not selected_columns: + selected_columns = self.get_columns() if len(inspection_data) == 0: inspection_data = pd.DataFrame([], columns=selected_columns) From 65b52bd166a3247665279d25533718c6f60235af Mon Sep 17 00:00:00 2001 From: Gabriel Bressan Date: Tue, 17 Feb 2026 22:06:55 -0400 Subject: [PATCH 5/6] Add MindsDB AI Coding Agent Guidelines documentation --- CLAUDE.md | 191 ++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 191 insertions(+) create mode 100644 CLAUDE.md diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 00000000000..e62fa781723 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,191 @@ +# MindsDB — AI Coding Agent Guidelines + +## Development Environment + +- **Hot-reload**: MindsDB runs in Docker with a bind mount (`./mindsdb:/mindsdb`) and `watchfiles`. Python file changes take effect immediately — no container restart needed. +- **Testing queries**: Use `mindsdb_sdk` connecting to `http://127.0.0.1:47334`. +- **Config**: `config.json` at the project root, mounted at `/root/mindsdb_config.json` in the container. +- **Environment variables**: see `.env` (do not commit secrets; `GOOGLE_API_KEY` and DB credentials live there). + +--- + +## Handler Architecture + +### How the query planner splits API handler queries + +When a handler is registered with `class_type = "api"`, MindsDB's query planner splits every SELECT into two steps: + +1. **`FetchDataframeStep`** → calls the handler's `select()` with the original query (including complex targets). The handler must return the raw DataFrame with the columns DuckDB will need. +2. **`SubSelectStep`** → DuckDB executes the full original SELECT expression (CASE WHEN, SUM, GROUP BY, etc.) on top of the DataFrame from step 1. + +**Implication**: handlers do not need to implement aggregations, CASE WHEN, or arithmetic. They only need to return the right raw columns. DuckDB handles everything else. + +--- + +## Handler `select()` — Two Patterns + +### Pattern A: Data-fetch-and-filter (most handlers) + +The handler fetches all data from the API and then drops columns that weren't requested. Calendar, Search Console, email, HubSpot, Shopify, Xero all use this pattern. + +**Correct implementation:** + +```python +selected_columns = [] +for target in query.targets: + if isinstance(target, ast.Star): + selected_columns = self.get_columns() + break + elif isinstance(target, ast.Identifier): + selected_columns.append(target.parts[-1]) + else: + # Complex expression (CASE WHEN, SUM, BinaryOperation, etc.). + # The outer SubSelectStep/DuckDB layer handles the computation. + # Return all raw columns so DuckDB has what it needs. + selected_columns = self.get_columns() + break +if not selected_columns: + selected_columns = self.get_columns() +``` + +**Bugs to avoid:** +- `raise ValueError(f"Unknown query target {type(target)}")` — breaks any CTE or aggregation query. +- Silently skipping non-Identifier targets without a fallback — `selected_columns` stays empty and `set(df.columns).difference(set([]))` drops every column, returning an empty DataFrame. + +### Pattern B: Column-selection-determines-API-params (e.g., Google Analytics) + +The handler uses the SELECT targets to decide *what* to request from the API (GA4 dimensions vs metrics, Search Console dimensions, etc.). A raw `isinstance(target, ast.Identifier)` check silently skips columns referenced inside complex expressions, causing the API to be called with incomplete parameters. + +**Correct implementation — add a recursive `_collect_identifiers` helper before the table class:** + +```python +from typing import List +from mindsdb_sql_parser import ast + + +def _collect_identifiers(node) -> List[str]: + """Recursively collect all Identifier column names from any AST node. + + Walks into CASE WHEN, Function args, BinaryOperation, etc. so that + columns referenced inside complex expressions are not missed. + """ + if node is None: + return [] + if isinstance(node, ast.Identifier): + return [str(node.parts[-1])] + if isinstance(node, ast.Case): + names = [] + for condition, result in node.rules: + names.extend(_collect_identifiers(condition)) + names.extend(_collect_identifiers(result)) + names.extend(_collect_identifiers(node.default)) + return names + if isinstance(node, ast.Function): + names = [] + for arg in (node.args or []): + names.extend(_collect_identifiers(arg)) + return names + if isinstance(node, ast.BinaryOperation): + return _collect_identifiers(node.args[0]) + _collect_identifiers(node.args[1]) + if isinstance(node, ast.UnaryOperation): + return _collect_identifiers(node.args[0]) + if isinstance(node, ast.TypeCast): + return _collect_identifiers(node.arg) + return [] +``` + +**Then use it in `select()`:** + +```python +seen = set() +for target in query.targets: + if isinstance(target, ast.Star): + # fall back to default dimensions/metrics + break + for col_name in _collect_identifiers(target): + if col_name in seen: + continue + seen.add(col_name) + # classify col_name as dimension or metric and add to API params +``` + +--- + +## Query Planner — Known Bugs Fixed in This Codebase + +### 1. CTE must be cleared after `plan_cte()` — `query_planner.py` + +After `self.plan_cte(query)` decomposes CTEs into steps, `query.cte` must be set to `None`. Otherwise the outer SELECT (which may reference a CTE name that resolves to a handler table) carries the full CTE definition into DuckDB, which fails with: + +> `Catalog Error: Table with name does not exist` + +```python +if query.cte is not None: + self.plan_cte(query) + query.cte = None # CTEs decomposed into steps; clear so DuckDB doesn't re-execute them +``` + +### 2. JOIN `filter_col_names` must only exclude scalar filters — `plan_join.py` + +In `process_table()`, `filter_col_names` is computed to strip API filter parameters (e.g., `start_date = 'yesterday'`) from the SELECT column list so they are not sent to the API as dimensions. However, `get_filters_from_join_conditions()` also generates cross-table IN predicates (`page_segment IN (VALUES FROM t1)`), and naively extracting every `Identifier` from every condition will incorrectly remove JOIN dimension columns from the SELECT list, causing the right-hand table in a FULL OUTER JOIN to be missing those columns. + +**Only exclude a column when the condition partner is a `Constant`:** + +```python +filter_col_names = set() +for cond in conditions: + if isinstance(cond, BinaryOperation) and len(cond.args) >= 2: + # Only exclude scalar filter parameters (col = Constant). + # Do NOT exclude JOIN predicates (col IN (Parameter)). + for i, arg in enumerate(cond.args[:2]): + if isinstance(arg, Identifier): + other = cond.args[1 - i] + if isinstance(other, Constant): + filter_col_names.add(arg.parts[-1]) +fetch_cols = referenced_cols - filter_col_names +``` + +--- + +## Handler Checklist + +When creating or modifying a handler's `select()` method: + +- [ ] Does the handler use target columns to control API parameters (Pattern B)? + - If yes: use `_collect_identifiers()` to recursively extract column names. +- [ ] Does the handler fetch all data and then filter by column (Pattern A)? + - If yes: add `else: selected_columns = self.get_columns(); break` and a `if not selected_columns: selected_columns = self.get_columns()` guard. +- [ ] Never `raise ValueError` on unrecognised target types — complex expressions are valid inputs from the planner. +- [ ] Never leave `selected_columns` empty after the targets loop — that silently drops all result columns. +- [ ] WHERE filter params (e.g., `start_date`, `end_date`) should be extracted from `query.where` and passed to the API, not treated as SELECT dimensions. +- [ ] `get_columns()` must list every column the API can return so Pattern A drop-logic works correctly. + +--- + +## Handlers in This Project + +| Handler | Pattern | Notes | +|---|---|---| +| `google_analytics_handler` | B | Uses `_collect_identifiers`; target columns map to GA4 dimensions/metrics | +| `google_calendar_handler` | A | Fetches all events/calendars/free-busy, then filters columns | +| `google_search_handler` | A | Fetches traffic/sitemaps/url-inspection data, then filters columns | +| `email_handler` | A (via `SELECTQueryParser`) | Delegated to utility — safe | +| `hubspot_handler` | A (via `SELECTQueryParser`) | Delegated to utility — safe | +| `shopify_handler` | A (via `SELECTQueryParser`) | Delegated to utility — safe | +| `xero_handler` | A | No target iteration — safe | +| `ms_one_drive_handler` | A | String checks only — safe | +| `web_handler` (`url_reader`) | A | Uses `FilterCondition`, no target iteration — safe | +| `s3_handler` | A | Only scans targets for `"content"` key; full query passed to DuckDB | + +--- + +## Relevant Source Paths + +| File | Purpose | +|---|---| +| `mindsdb/api/executor/planner/query_planner.py` | `plan_select`, `plan_cte`, `plan_api_db_select`, `get_integration_select_step` | +| `mindsdb/api/executor/planner/plan_join.py` | `PlanJoinTablesQuery`, `process_table`, `get_filters_from_join_conditions` | +| `mindsdb/api/executor/sql_query/steps/subselect_step.py` | `SubSelectStepCall` — runs DuckDB on handler result | +| `mindsdb/api/executor/utilities/sql.py` | `query_df`, `query_df_with_type_infer_fallback` | +| `mindsdb/integrations/utilities/query_traversal.py` | `query_traversal` — AST walker used across planner and handlers | +| `mindsdb/integrations/handlers//` | Individual handler implementations | From 67feba48a630b0002217d087b079734f4aec8f6b Mon Sep 17 00:00:00 2001 From: Gabriel Bressan Date: Wed, 18 Feb 2026 14:39:44 -0400 Subject: [PATCH 6/6] Fix query planner: prevent forwarding ORDER BY to handler in plan_api_db_select --- CLAUDE.md | 22 ++++++++++++++++++- mindsdb/api/executor/planner/query_planner.py | 4 +++- 2 files changed, 24 insertions(+), 2 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index e62fa781723..36375fd066e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -125,7 +125,27 @@ if query.cte is not None: query.cte = None # CTEs decomposed into steps; clear so DuckDB doesn't re-execute them ``` -### 2. JOIN `filter_col_names` must only exclude scalar filters — `plan_join.py` +### 2. `plan_api_db_select` must NOT forward `order_by` to the handler — `query_planner.py` + +`plan_api_db_select` splits a query into a handler fetch (`FetchDataframeStep`) and a DuckDB pass (`SubSelectStep`). It passes `order_by` from the SQL query to the handler, which is wrong: ORDER BY may reference **SQL aliases** (e.g. `SUM(sessions) AS total_sessions` → `ORDER BY total_sessions`) that are meaningless to the underlying API. The GA4 API returns: + +> `400 Field total_sessions exists in OrderBy but is not defined in input Dimensions/Metrics list` + +The outer SubSelectStep already retains `order_by` (it is not cleared like `where`/`limit`), so DuckDB applies it correctly after aggregation. + +```python +# query_planner.py — plan_api_db_select() +query2 = Select( + targets=query.targets, + from_table=query.from_table, + where=query.where, + # order_by intentionally omitted: ORDER BY may reference SQL aliases unknown + # to the underlying API. The SubSelectStep/DuckDB layer handles it correctly. + limit=query.limit, +) +``` + +### 3. JOIN `filter_col_names` must only exclude scalar filters — `plan_join.py` In `process_table()`, `filter_col_names` is computed to strip API filter parameters (e.g., `start_date = 'yesterday'`) from the SELECT column list so they are not sent to the API as dimensions. However, `get_filters_from_join_conditions()` also generates cross-table IN predicates (`page_segment IN (VALUES FROM t1)`), and naively extracting every `Identifier` from every condition will incorrectly remove JOIN dimension columns from the SELECT list, causing the right-hand table in a FULL OUTER JOIN to be missing those columns. diff --git a/mindsdb/api/executor/planner/query_planner.py b/mindsdb/api/executor/planner/query_planner.py index 1c397b1eff7..61363d1349b 100644 --- a/mindsdb/api/executor/planner/query_planner.py +++ b/mindsdb/api/executor/planner/query_planner.py @@ -434,7 +434,9 @@ def plan_api_db_select(self, query): targets=query.targets, from_table=query.from_table, where=query.where, - order_by=query.order_by, + # order_by intentionally omitted: ORDER BY may reference SQL aliases + # (e.g. SUM(sessions) AS total_sessions) that are unknown to the + # underlying API. The outer SubSelectStep/DuckDB layer handles it. limit=query.limit, ) prev_step = self.plan_integration_select(query2)