From fff0d91b4977e13c109b58c44754224d2a14be54 Mon Sep 17 00:00:00 2001 From: Gabriel Bressan Date: Wed, 8 Jul 2026 18:43:37 -0400 Subject: [PATCH] fix(multi_format_api): explode nested JSON record arrays into rows An API response shaped as an object with a nested record array (e.g. {"tickers": [...13k...], "status": "OK", "count": 12962}) collapsed into a single row with the array dumped into one ~10MB string cell. That blob propagated to downstream code-execution and blew the context budget (4,656,292 tokens > 1,000,000). The crash is caused by shape, not raw size: the same data as a proper 12,962-row table costs ~8.5K tokens. parse_json only recognized a record array under a hardcoded whitelist (data/results/items/records/rows/entries); domain keys like `tickers` missed it and fell through to a single-row json_normalize. Changes: - Generic record-array detection: explicit record_path -> top-level list -> whitelist -> bounded auto-detect (list-of-objects, <=3 levels), longest-wins with a warning on ambiguity. - Reattach top-level scalar siblings (status, count, ...) as constant columns; meta_ prefix on name collisions. - New config args record_path (STR) and auto_explode (STR 'true'/'false', default 'true'), settable at connection level and overridable per query. - Preserve legacy behavior: empty whitelist arrays -> empty DataFrame; auto_explode='false' keeps the single-row shape. - Net-new unit tests (13 cases) + docs. Co-Authored-By: Claude Opus 4.8 --- .../multi_format_api_handler/README.md | 57 +++++++ .../USAGE_EXAMPLES.md | 16 ++ .../connection_args.py | 20 +++ .../format_parsers.py | 153 ++++++++++++++--- .../multi_format_table.py | 23 ++- .../tests/__init__.py | 0 .../tests/test_format_parsers.py | 160 ++++++++++++++++++ 7 files changed, 401 insertions(+), 28 deletions(-) create mode 100644 mindsdb/integrations/handlers/multi_format_api_handler/tests/__init__.py create mode 100644 mindsdb/integrations/handlers/multi_format_api_handler/tests/test_format_parsers.py diff --git a/mindsdb/integrations/handlers/multi_format_api_handler/README.md b/mindsdb/integrations/handlers/multi_format_api_handler/README.md index de3e960895d..23163f41197 100644 --- a/mindsdb/integrations/handlers/multi_format_api_handler/README.md +++ b/mindsdb/integrations/handlers/multi_format_api_handler/README.md @@ -189,6 +189,11 @@ The handler detects the format in the following order: ### JSON Handling +The handler detects the primary **record array** in a JSON response and +explodes it into one row per record — even when the array sits under a +domain-specific key rather than a common envelope key like `data`/`results`. +Nested objects are flattened with dot notation. + **Input:** ```json { @@ -205,6 +210,54 @@ The handler detects the format in the following order: | 1 | Alice | 30 | | 2 | Bob | 25 | +#### Envelope payloads (array + scalar siblings) + +When the record array is wrapped alongside scalar metadata, those top-level +scalars are reattached to every row as constant columns, so nothing is lost. + +**Input:** +```json +{ + "tickers": [ + {"symbol": "A", "price": 1.0}, + {"symbol": "B", "price": 2.0} + ], + "status": "OK", + "count": 2 +} +``` + +**Output DataFrame:** +| symbol | price | status | count | +|--------|-------|--------|-------| +| A | 1.0 | OK | 2 | +| B | 2.0 | OK | 2 | + +If a scalar sibling's name collides with a record column, it is reattached +under a `meta_` prefix (e.g. `meta_status`). + +#### Detection, `record_path`, and `auto_explode` + +The record array is resolved in this order: an explicit `record_path`, a +top-level list, a common envelope key (`data`, `results`, `items`, `records`, +`rows`, `entries`), then auto-detection of any list-of-objects (searched up to +3 levels deep). If a payload contains **multiple** candidate arrays, the +longest is used and a warning is logged — set `record_path` to disambiguate. + +- **`record_path`** — dot-path to the record array, e.g. `record_path = 'tickers'` + or `record_path = 'data.results'` for a nested array. Overrides auto-detection. +- **`auto_explode`** — `'true'` (default) or `'false'`. Set to `'false'` to keep + the legacy single-row shape for object payloads. + +Both can be set at the connection level (PARAMETERS) or per query in the WHERE +clause: + +```sql +SELECT * FROM my_api.data +WHERE url = 'https://api.example.com/snapshot.json' + AND record_path = 'tickers'; +``` + ### XML Handling **Input:** @@ -298,6 +351,8 @@ Error: Failed to fetch data from URL: 404 Client Error | headers | dict | Default headers for all requests | {} | | timeout | int | Default request timeout in seconds | 30 | | max_content_size | int | Maximum response size in MB (prevents large downloads) | 100 | +| record_path | string | Dot-path to the JSON record array (overrides auto-detection) | None | +| auto_explode | string | Explode JSON record arrays into rows: `'true'`/`'false'` | 'true' | ### Query Parameters @@ -307,6 +362,8 @@ Error: Failed to fetch data from URL: 404 Client Error | headers | string | JSON string of request-specific headers | {} | | timeout | int | Request timeout in seconds | 30 | | max_content_size | int | Maximum response size in MB (overrides connection-level setting) | 100 | +| record_path | string | Dot-path to the JSON record array (overrides auto-detection) | None | +| auto_explode | string | Explode JSON record arrays into rows: `'true'`/`'false'` | 'true' | ## Limitations diff --git a/mindsdb/integrations/handlers/multi_format_api_handler/USAGE_EXAMPLES.md b/mindsdb/integrations/handlers/multi_format_api_handler/USAGE_EXAMPLES.md index d91d2e43cd9..a1255e712bc 100644 --- a/mindsdb/integrations/handlers/multi_format_api_handler/USAGE_EXAMPLES.md +++ b/mindsdb/integrations/handlers/multi_format_api_handler/USAGE_EXAMPLES.md @@ -115,6 +115,22 @@ WHERE url = 'https://jsonplaceholder.typicode.com/posts' LIMIT 20; ``` +#### Envelope payload with a domain-specific record array +When a response wraps the records under a domain key (e.g. `{"tickers": [...], +"status": "OK", "count": 12962}`), the handler explodes `tickers` into rows and +reattaches `status`/`count` as constant columns. Auto-detection handles this, +but you can pin it explicitly with `record_path`: + +```sql +SELECT symbol, price, status, count +FROM my_api.data +WHERE url = 'https://api.example.com/snapshot.json' + AND record_path = 'tickers'; +``` + +For a nested array use a dot-path (`record_path = 'data.results'`). To keep the +legacy single-row shape instead of exploding, set `auto_explode = 'false'`. + ### CSV Examples #### Iris Dataset diff --git a/mindsdb/integrations/handlers/multi_format_api_handler/connection_args.py b/mindsdb/integrations/handlers/multi_format_api_handler/connection_args.py index 35494d944a4..ebf7193cd4e 100644 --- a/mindsdb/integrations/handlers/multi_format_api_handler/connection_args.py +++ b/mindsdb/integrations/handlers/multi_format_api_handler/connection_args.py @@ -39,6 +39,26 @@ 'required': False, 'label': 'Max Content Size (MB)', }, + record_path={ + 'type': ARG_TYPE.STR, + 'description': ( + 'Dot-path to the array of records inside a JSON response (e.g. "tickers" ' + 'or "data.results"). Use to override auto-detection when a payload has ' + 'ambiguous or nested record arrays. Leave empty to auto-detect.' + ), + 'required': False, + 'label': 'JSON Record Path', + }, + auto_explode={ + 'type': ARG_TYPE.STR, + 'description': ( + "Whether to explode a JSON record array into rows ('true'/'false'). " + "Default 'true'. Set to 'false' to keep the legacy single-row shape " + 'for object payloads.' + ), + 'required': False, + 'label': 'Auto-Explode JSON Arrays', + }, ) 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 eda22bce092..9b46ab61da9 100644 --- a/mindsdb/integrations/handlers/multi_format_api_handler/format_parsers.py +++ b/mindsdb/integrations/handlers/multi_format_api_handler/format_parsers.py @@ -62,45 +62,142 @@ def detect_format(response, url: str) -> Optional[str]: return None -def parse_json(content: str) -> pd.DataFrame: +# Common envelope keys that historically wrapped the record array. Kept for +# back-compat so payloads like {"data": [...]} keep exploding as before. +WHITELIST = ['data', 'results', 'items', 'records', 'rows', 'entries'] + + +def _is_object_array(value: Any) -> bool: + """True if value is a non-empty list whose (sampled) elements are dicts.""" + return ( + isinstance(value, list) + and len(value) > 0 + and all(isinstance(item, dict) for item in value[:20]) + ) + + +def _dig(data: Any, path: str) -> Any: + """Follow a dot-separated path through nested dicts. Returns None on miss.""" + node = data + for part in path.split('.'): + if isinstance(node, dict) and part in node: + node = node[part] + else: + return None + return node + + +def _find_object_arrays(node, prefix: str = '', depth: int = 0, max_depth: int = 3): + """Recursively collect (dot_path, list) for every list-of-objects reachable + within `max_depth` levels of nested dicts.""" + found = [] + if isinstance(node, dict) and depth <= max_depth: + for k, v in node.items(): + path = f'{prefix}.{k}' if prefix else k + if _is_object_array(v): + found.append((path, v)) + elif isinstance(v, dict): + found.extend(_find_object_arrays(v, path, depth + 1, max_depth)) + return found + + +def _resolve_records(data, record_path: Optional[str] = None): + """Determine the primary record array to explode into rows. + + Resolution priority: + 1. explicit `record_path` dot-path; + 2. top-level list; + 3. whitelist envelope keys (list-of-objects only); + 4. auto-detect via bounded recursion — one candidate wins, multiple -> + longest wins with a warning, none -> treat the dict as a single record. + + Returns (records_list_or_None, chosen_top_or_nested_path_or_None). + """ + if record_path: + node = _dig(data, record_path) + if isinstance(node, list): + return node, record_path + logger.warning("record_path '%s' did not resolve to a list; auto-detecting", record_path) + + if isinstance(data, list): + return data, None + + if isinstance(data, dict): + for key in WHITELIST: + val = data.get(key) + # Whitelist keys match a list of objects OR an empty list (the + # latter preserves the original "{'data': []} -> empty DF" behavior). + if isinstance(val, list) and (len(val) == 0 or _is_object_array(val)): + logger.info(f"Extracting list from '{key}' field") + return val, key + candidates = _find_object_arrays(data) + if len(candidates) == 1: + return candidates[0][1], candidates[0][0] + if len(candidates) > 1: + best = max(candidates, key=lambda c: len(c[1])) + logger.warning( + "Multiple record arrays %s; using longest '%s'. Set record_path to override.", + [p for p, _ in candidates], best[0], + ) + return best[1], best[0] + # No record array anywhere: the dict itself is a single record. + return [data], None + + return None, None + + +def parse_json(content: str, record_path: Optional[str] = None, auto_explode: bool = True) -> pd.DataFrame: """ Parse JSON content and convert to DataFrame. - Handles nested structures using json_normalize. + + Detects the primary record array generically and explodes it into rows, + reattaching top-level scalar siblings (e.g. status, count) as constant + columns. Set `auto_explode=False` to keep the legacy single-row shape for + object payloads, or `record_path` to point at an ambiguous/nested array. Args: content: JSON string + record_path: optional dot-path to the record array (overrides detection) + auto_explode: when False, object payloads normalize to a single row Returns: pandas DataFrame """ try: data = json.loads(content) - - if isinstance(data, list): - if len(data) == 0: - return pd.DataFrame() - df = pd.json_normalize(data) - elif isinstance(data, dict): - # Check if dict contains a list that should be the main data - # Common patterns: {"data": [...], "results": [...], "items": [...]} - df = None - for key in ['data', 'results', 'items', 'records', 'rows', 'entries']: - if key in data and isinstance(data[key], list): - logger.info(f"Extracting list from '{key}' field") - df = pd.json_normalize(data[key]) - break - if df is None: - df = pd.json_normalize(data) - else: - # Primitive type, wrap in DataFrame - return pd.DataFrame({'value': [data]}) - - return _ensure_scalar_columns(df) - except json.JSONDecodeError as e: logger.error(f"JSON parsing error: {e}") raise ValueError(f"Invalid JSON content: {e}") + if not isinstance(data, (list, dict)): + # Primitive type, wrap in DataFrame + return pd.DataFrame({'value': [data]}) + + # Legacy single-row behavior for object payloads when explosion is disabled. + if not auto_explode and isinstance(data, dict): + return _ensure_scalar_columns(pd.json_normalize(data)) + + records, chosen_path = _resolve_records(data, record_path) + if records is None: + return pd.DataFrame({'value': [data]}) + if len(records) == 0: + return pd.DataFrame() + + df = pd.json_normalize(records, sep='.') + + # Reattach top-level scalar siblings (status, count, request_id, ...) as + # constant columns so envelope metadata is not lost. Skip the array key; + # prefix `meta_` on name collisions with normalized record columns. + if isinstance(data, dict) and chosen_path: + top_key = chosen_path.split('.')[0] + for k, v in data.items(): + if k == top_key: + continue + if v is None or isinstance(v, (str, int, float, bool)): + df[k if k not in df.columns else f'meta_{k}'] = v + + return _ensure_scalar_columns(df) + def parse_xml(content: str) -> pd.DataFrame: """ @@ -373,13 +470,15 @@ def parse_csv(content: str) -> pd.DataFrame: raise ValueError(f"Invalid CSV content: {e}") -def parse_response(response, url: str) -> pd.DataFrame: +def parse_response(response, url: str, record_path: Optional[str] = None, auto_explode: bool = True) -> pd.DataFrame: """ Auto-detect format and parse response to DataFrame. Args: response: requests.Response object url: URL string + record_path: optional dot-path to the record array (JSON only) + auto_explode: when False, JSON object payloads normalize to a single row Returns: pandas DataFrame @@ -388,7 +487,7 @@ def parse_response(response, url: str) -> pd.DataFrame: if format_type == 'json': logger.info("Detected JSON format") - return parse_json(response.text) + return parse_json(response.text, record_path, auto_explode) elif format_type == 'xml': logger.info("Detected XML format") return parse_xml(response.text) @@ -399,7 +498,7 @@ def parse_response(response, url: str) -> pd.DataFrame: # Try JSON as default fallback logger.warning(f"Could not detect format, trying JSON as fallback") try: - return parse_json(response.text) + return parse_json(response.text, record_path, auto_explode) except ValueError: # Try XML as second fallback try: 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 25fd8560339..c0cb8f58d87 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 @@ -40,6 +40,19 @@ def _coerce_dict_arg(value, arg_name: str) -> dict: return {} +def _coerce_bool_arg(value, default: bool = True) -> bool: + """Coerce a STR/bool arg (e.g. 'auto_explode') to a boolean. Falls back to + `default` when unset; recognises common falsey strings.""" + if value is None: + return default + if isinstance(value, bool): + return value + text = str(value).strip().lower() + if text == '': + return default + return text not in ('false', '0', 'no', 'off') + + class MultiFormatAPITable(APIResource): """ Generic API table that fetches and parses data from URLs. @@ -72,6 +85,8 @@ def list( body = {} timeout = self.handler.connection_args.get('timeout', 30) max_content_size_mb = self.handler.connection_args.get('max_content_size', 100) + record_path = self.handler.connection_args.get('record_path') or None + auto_explode = _coerce_bool_arg(self.handler.connection_args.get('auto_explode'), default=True) if conditions: for condition in conditions: @@ -107,6 +122,12 @@ def list( except Exception: logger.warning(f"Invalid max_content_size value: {condition.value}") condition.applied = True + elif column == 'record_path': + record_path = condition.value or None + condition.applied = True + elif column == 'auto_explode': + auto_explode = _coerce_bool_arg(condition.value, default=auto_explode) + condition.applied = True # Use connection-level URL if query-level not provided if not url: @@ -215,7 +236,7 @@ def __init__(self, content, headers): # Parse response based on detected format try: - df = parse_response(mock_response, url) + df = parse_response(mock_response, url, record_path, auto_explode) except ValueError as e: logger.error(f"Parsing failed: {e}") raise diff --git a/mindsdb/integrations/handlers/multi_format_api_handler/tests/__init__.py b/mindsdb/integrations/handlers/multi_format_api_handler/tests/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/mindsdb/integrations/handlers/multi_format_api_handler/tests/test_format_parsers.py b/mindsdb/integrations/handlers/multi_format_api_handler/tests/test_format_parsers.py new file mode 100644 index 00000000000..4c2e2183ecd --- /dev/null +++ b/mindsdb/integrations/handlers/multi_format_api_handler/tests/test_format_parsers.py @@ -0,0 +1,160 @@ +"""Unit tests for multi_format_api_handler JSON record-array flattening. + +These exercise `parse_json` directly (no network, no MindsDB executor) and lock +in the generic record-array detection, scalar-sibling reattachment, and the +`record_path` / `auto_explode` overrides. +""" + +import json +import logging + +import pandas as pd + +from mindsdb.integrations.handlers.multi_format_api_handler.format_parsers import parse_json + + +def _dumps(obj): + return json.dumps(obj) + + +def test_envelope_array_under_domain_key_explodes(): + """The bug: {"tickers": [...], "status": "OK", "count": N} collapsed into + one row. It must now explode into one row per record.""" + payload = { + "tickers": [ + {"symbol": "A", "price": 1.0}, + {"symbol": "B", "price": 2.0}, + {"symbol": "C", "price": 3.0}, + ], + "status": "OK", + "count": 3, + "request_id": "abc123", + } + df = parse_json(_dumps(payload)) + + assert len(df) == 3 + assert set(["symbol", "price"]).issubset(df.columns) + # Scalar siblings reattached as constant columns. + assert (df["status"] == "OK").all() + assert (df["count"] == 3).all() + assert (df["request_id"] == "abc123").all() + # The array key itself is not reattached as a column. + assert "tickers" not in df.columns + + +def test_whitelist_key_still_explodes(): + """Back-compat: an envelope under a whitelist key ('data') keeps exploding.""" + payload = {"data": [{"id": 1}, {"id": 2}], "status": "ok"} + df = parse_json(_dumps(payload)) + + assert len(df) == 2 + assert list(df["id"]) == [1, 2] + assert (df["status"] == "ok").all() + + +def test_top_level_list(): + payload = [{"id": 1}, {"id": 2}, {"id": 3}] + df = parse_json(_dumps(payload)) + + assert len(df) == 3 + assert list(df["id"]) == [1, 2, 3] + + +def test_single_object(): + payload = {"name": "widget", "qty": 5} + df = parse_json(_dumps(payload)) + + assert len(df) == 1 + assert df.iloc[0]["name"] == "widget" + assert df.iloc[0]["qty"] == 5 + + +def test_list_of_scalars_not_exploded(): + """A dict whose only list is a list of scalars has no record array; it is a + single record and the scalar list is serialized, not exploded.""" + payload = {"tags": ["a", "b", "c"], "status": "ok"} + df = parse_json(_dumps(payload)) + + assert len(df) == 1 + assert df.iloc[0]["status"] == "ok" + # List-of-scalars serialized to a string cell (not exploded to 3 rows). + assert df.iloc[0]["tags"] == "a, b, c" + + +def test_multiple_arrays_picks_longest_and_warns(caplog): + payload = { + "alpha": [{"a": 1}, {"a": 2}], + "beta": [{"b": 1}, {"b": 2}, {"b": 3}], + } + with caplog.at_level(logging.WARNING): + df = parse_json(_dumps(payload)) + + # Longest array ('beta') wins. + assert len(df) == 3 + assert "b" in df.columns + assert any("Multiple record arrays" in rec.message for rec in caplog.records) + + +def test_explicit_record_path_nested(): + payload = {"result": {"records": [{"x": 1}, {"x": 2}]}} + df = parse_json(_dumps(payload), record_path="result.records") + + assert len(df) == 2 + assert list(df["x"]) == [1, 2] + + +def test_record_path_miss_warns_and_autodetects(caplog): + payload = {"data": [{"a": 1}]} + with caplog.at_level(logging.WARNING): + df = parse_json(_dumps(payload), record_path="does.not.exist") + + assert len(df) == 1 + assert df.iloc[0]["a"] == 1 + assert any("did not resolve to a list" in rec.message for rec in caplog.records) + + +def test_auto_explode_false_keeps_single_row(): + payload = { + "tickers": [{"symbol": "A"}, {"symbol": "B"}], + "status": "OK", + } + df = parse_json(_dumps(payload), auto_explode=False) + + assert len(df) == 1 + assert df.iloc[0]["status"] == "OK" + + +def test_nested_dict_produces_dotted_columns(): + payload = {"items": [{"a": {"b": 1}}, {"a": {"b": 2}}]} + df = parse_json(_dumps(payload)) + + assert len(df) == 2 + assert "a.b" in df.columns + assert list(df["a.b"]) == [1, 2] + + +def test_sibling_collision_gets_meta_prefix(): + """A scalar sibling whose name collides with a normalized record column is + reattached under a 'meta_' prefix.""" + payload = {"data": [{"status": "active"}, {"status": "inactive"}], "status": "OK"} + df = parse_json(_dumps(payload)) + + assert len(df) == 2 + # Record-level column preserved. + assert list(df["status"]) == ["active", "inactive"] + # Envelope scalar reattached under meta_ prefix. + assert (df["meta_status"] == "OK").all() + + +def test_empty_array_returns_empty_dataframe(): + payload = {"data": []} + df = parse_json(_dumps(payload)) + + assert len(df) == 0 + + +def test_primitive_wrapped_in_value_column(): + df = parse_json(_dumps(42)) + + assert len(df) == 1 + assert df.iloc[0]["value"] == 42