diff --git a/recce/adapter/histogram_bucketing.py b/recce/adapter/histogram_bucketing.py new file mode 100644 index 000000000..2cfd63af7 --- /dev/null +++ b/recce/adapter/histogram_bucketing.py @@ -0,0 +1,225 @@ +"""Adapter-specific exact bucketing strategies for numeric histograms. + +The ordered literal-comparison ladder in :mod:`recce.tasks.histogram` is the +portable source of truth. This module opts narrowly proven adapter/type pairs +into a compact expression; every unknown or unsafe input returns the literal +fallback marker instead. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from decimal import Decimal, InvalidOperation +from typing import Optional + +DUCKDB_SCALED_INTEGER = "duckdb_scaled_integer" +LITERAL_ORDERED_COMPARISON = "literal_ordered_comparison" + +_HUGEINT_MAX = 2**127 - 1 +_SAFE_SCALED_MAGNITUDE = 10**37 - 1 + +# DuckDB aliases whose complete declared domain is narrower than HUGEINT. +# HUGEINT/INT128 and their unsigned variants intentionally stay on fallback. +_DUCKDB_INTEGER_MAX_ABS = { + "TINYINT": 2**7, + "INT1": 2**7, + "SMALLINT": 2**15, + "INT2": 2**15, + "SHORT": 2**15, + "INTEGER": 2**31, + "INT": 2**31, + "INT4": 2**31, + "SIGNED": 2**31, + "BIGINT": 2**63, + "INT8": 2**63, + "LONG": 2**63, + "UTINYINT": 2**8 - 1, + "USMALLINT": 2**16 - 1, + "UINTEGER": 2**32 - 1, + "UBIGINT": 2**64 - 1, +} + +_DUCKDB_DECIMAL_TYPE = re.compile(r"^(?:DECIMAL|DEC|NUMERIC)\s*\(\s*(\d+)\s*(?:,\s*(\d+)\s*)?\)$") + + +@dataclass(frozen=True) +class HistogramBucketingPlan: + """Selected SQL strategy and its optional compact ``CASE`` expression.""" + + strategy: str + bin_expression: Optional[str] = None + + +def _literal_fallback() -> HistogramBucketingPlan: + return HistogramBucketingPlan(strategy=LITERAL_ORDERED_COMPARISON) + + +def _as_finite_decimal(value) -> Optional[Decimal]: + try: + decimal_value = Decimal(str(value)) + except (InvalidOperation, TypeError, ValueError): + return None + return decimal_value if decimal_value.is_finite() else None + + +def _fractional_digits(value: Decimal) -> int: + if value == 0: + return 0 + _, _, exponent = _exact_coefficient_and_exponent(value) + return max(-exponent, 0) + + +def _exact_coefficient_and_exponent(value: Decimal): + """Extract normalized Decimal parts without consulting its ambient context.""" + sign, digits, exponent = value.as_tuple() + trailing_zeros = 0 + for digit in reversed(digits): + if digit != 0: + break + trailing_zeros += 1 + + significant_digits = digits[: len(digits) - trailing_zeros] if trailing_zeros else digits + coefficient = int("".join(str(digit) for digit in significant_digits)) if significant_digits else 0 + return sign, coefficient, exponent + trailing_zeros + + +def _scaled_integer(value: Decimal, scale: int) -> Optional[int]: + """Return ``value * 10**scale`` without Decimal-context rounding.""" + if not value.is_finite(): + return None + if value == 0: + return 0 + + sign, coefficient, exponent = _exact_coefficient_and_exponent(value) + shift = exponent + scale + if shift < 0: + return None + scaled = coefficient * 10**shift + return -scaled if sign else scaled + + +def _decimal_sql_literal(value: Decimal) -> str: + if value == 0: + return "0" + literal = format(value, "f") + return literal.rstrip("0").rstrip(".") if "." in literal else literal + + +def _normalize_adapter_type(adapter_type) -> str: + return adapter_type.strip().lower() if isinstance(adapter_type, str) else "" + + +def _normalize_column_type(column_type) -> str: + return " ".join(column_type.strip().upper().split()) if isinstance(column_type, str) else "" + + +def _declared_type(column_type: str): + integer_bound = _DUCKDB_INTEGER_MAX_ABS.get(column_type) + if integer_bound is not None: + return "integer", integer_bound, 0, 0 + + match = _DUCKDB_DECIMAL_TYPE.fullmatch(column_type) + if match is None: + return None + + precision = int(match.group(1)) + scale = int(match.group(2) or 0) + if precision < 1 or precision > 38 or scale < 0 or scale > precision: + return None + return "decimal", 10**precision - 1, precision, scale + + +def select_histogram_bucketing( + *, + adapter_type, + column_sql: str, + column_type: str, + minimum, + maximum, + width, + num_bins: int, +) -> HistogramBucketingPlan: + """Select a compact exact expression or the universal literal fallback. + + DuckDB is enabled only when the declared type and concrete geometry prove + every scaled value and floor-division intermediate fits signed HUGEINT. + Fixed DECIMAL multiplication has an additional precision plus common + geometry-scale propagation constraint so DuckDB's DECIMAL(38, scale) + intermediate cannot overflow before its HUGEINT cast. + """ + if _normalize_adapter_type(adapter_type) != "duckdb" or not isinstance(column_sql, str) or not column_sql: + return _literal_fallback() + if not isinstance(num_bins, int) or isinstance(num_bins, bool) or num_bins <= 0: + return _literal_fallback() + + declared = _declared_type(_normalize_column_type(column_type)) + if declared is None: + return _literal_fallback() + kind, unscaled_declared_bound, precision, declared_scale = declared + + decimal_minimum = _as_finite_decimal(minimum) + decimal_maximum = _as_finite_decimal(maximum) + decimal_width = _as_finite_decimal(width) + if decimal_minimum is None or decimal_maximum is None or decimal_width is None: + return _literal_fallback() + if decimal_width <= 0 or decimal_minimum > decimal_maximum: + return _literal_fallback() + + common_scale = max( + declared_scale, + _fractional_digits(decimal_minimum), + _fractional_digits(decimal_maximum), + _fractional_digits(decimal_width), + ) + if common_scale > 37: + return _literal_fallback() + scale_expansion = common_scale - declared_scale + scale_factor = 10**common_scale + declared_scaled_bound = unscaled_declared_bound * 10**scale_expansion + + # One decimal digit of headroom keeps subtraction, negation, and the + # ``-offset + width - 1`` floor numerator inside signed HUGEINT. + if declared_scaled_bound > _SAFE_SCALED_MAGNITUDE: + return _literal_fallback() + + # DuckDB retains the source DECIMAL scale for this multiplication. Even + # when its eventual integer would fit HUGEINT, the DECIMAL intermediate is + # not proven safe unless its integer digits also fit DECIMAL(38, scale). + if kind == "decimal" and declared_scale > 0 and precision + common_scale > 38: + return _literal_fallback() + + minimum_scaled = _scaled_integer(decimal_minimum, common_scale) + maximum_scaled = _scaled_integer(decimal_maximum, common_scale) + width_scaled = _scaled_integer(decimal_width, common_scale) + if minimum_scaled is None or maximum_scaled is None or width_scaled is None or width_scaled <= 0: + return _literal_fallback() + if any(abs(value) > _SAFE_SCALED_MAGNITUDE for value in (minimum_scaled, maximum_scaled, width_scaled)): + return _literal_fallback() + if maximum_scaled - minimum_scaled != width_scaled * num_bins: + return _literal_fallback() + + max_offset_magnitude = declared_scaled_bound + abs(minimum_scaled) + max_floor_numerator = max_offset_magnitude + width_scaled - 1 + if max_offset_magnitude > _HUGEINT_MAX or max_floor_numerator > _HUGEINT_MAX: + return _literal_fallback() + + factor_sql = f"({scale_factor}::HUGEINT)" + if kind == "decimal" and declared_scale > 0: + column_scaled = f"CAST(CAST({column_sql} AS DECIMAL(38, {declared_scale})) * {factor_sql} AS HUGEINT)" + else: + column_scaled = f"CAST({column_sql} AS HUGEINT)" + if scale_factor != 1: + column_scaled = f"({column_scaled} * {factor_sql})" + + minimum_sql = f"({minimum_scaled}::HUGEINT)" + width_sql = f"({width_scaled}::HUGEINT)" + offset_sql = f"({column_scaled} - {minimum_sql})" + maximum_sql = _decimal_sql_literal(decimal_maximum) + bin_expression = f"""CASE + WHEN {column_sql} IS NULL THEN NULL + WHEN {column_sql} = {maximum_sql} THEN {num_bins - 1} + WHEN {offset_sql} >= 0 THEN {offset_sql} // {width_sql} + ELSE -((-{offset_sql} + {width_sql} - 1) // {width_sql}) + END""" + return HistogramBucketingPlan(strategy=DUCKDB_SCALED_INTEGER, bin_expression=bin_expression) diff --git a/recce/tasks/histogram.py b/recce/tasks/histogram.py index 3736def4c..8745978a4 100644 --- a/recce/tasks/histogram.py +++ b/recce/tasks/histogram.py @@ -1,3 +1,4 @@ +import logging import math import re from bisect import bisect_right @@ -9,12 +10,16 @@ from dateutil.relativedelta import relativedelta from pydantic import BaseModel +from recce.adapter.histogram_bucketing import select_histogram_bucketing from recce.core import default_context +from recce.event import log_performance from recce.models import Check from recce.tasks import Task from recce.tasks.core import CheckValidator, TaskResultDiffer from recce.tasks.query import QueryMixin +logger = logging.getLogger("uvicorn") + sql_datetime_types = [ "DATE", "DATETIME", @@ -279,6 +284,81 @@ def _generate_histogram_sql(node, column, min_value, max_value, num_bins, bin_si return sql, bin_size +@dataclass(frozen=True) +class _GeneratedHistogramSql: + sql: str + bin_size: Decimal + strategy: str + + +def _generate_histogram_sql_for_adapter( + node, + column, + min_value, + max_value, + num_bins, + bin_size, + *, + adapter_type, + column_type, +): + plan = select_histogram_bucketing( + adapter_type=adapter_type, + column_sql=column, + column_type=column_type, + minimum=min_value, + maximum=max_value, + width=bin_size, + num_bins=num_bins, + ) + if plan.bin_expression is None: + sql, exact_bin_size = _generate_histogram_sql( + node, + column, + min_value, + max_value, + num_bins, + bin_size, + ) + return _GeneratedHistogramSql(sql=sql, bin_size=exact_bin_size, strategy=plan.strategy) + + min_literal = _decimal_sql_literal(min_value) + max_literal = _decimal_sql_literal(max_value) + bin_size_literal = _decimal_sql_literal(bin_size) + sql = f""" + WITH value_ranges AS ( + SELECT + {min_literal} as min_value, + {max_literal} as max_value + ), + bin_parameters AS ( + SELECT + min_value, + max_value, + {bin_size_literal} AS bin_size + FROM value_ranges + ), + binned_values AS ( + SELECT + {column} as column_value, + {plan.bin_expression} AS bin + FROM {{{{ ref("{node}") }}}}, + bin_parameters + ), + bin_edges AS ( + SELECT + bin, + COUNT(*) AS count + FROM binned_values, bin_parameters + GROUP BY bin + ORDER BY bin + ) + + SELECT bin, count FROM bin_edges + """ + return _GeneratedHistogramSql(sql=sql, bin_size=_decimal_value(bin_size), strategy=plan.strategy) + + def generate_histogram_sql_integer(node, column, min_value, max_value, num_bins=50, bin_size=None): if bin_size is None: bin_size = Decimal(max(math.ceil((max_value - min_value) / num_bins), 1)) @@ -303,16 +383,97 @@ def _validate_histogram_column_type(column_type: str) -> None: raise ValueError(f"Column type {column_type} is not supported for histogram analysis") -def query_numeric_histogram(task, node, column, column_type, min_value, max_value, num_bins=50): +def _emit_histogram_sql_telemetry(adapter_type, strategy, effective_bin_count, sql_length): + try: + normalized_adapter_type = adapter_type.strip().lower() if isinstance(adapter_type, str) else "unknown" + log_performance( + "histogram_sql", + { + "adapter_type": normalized_adapter_type, + "strategy": strategy, + "effective_bin_count": effective_bin_count, + "sql_length": sql_length, + }, + ) + except Exception: + logger.debug("histogram SQL telemetry emit failed", exc_info=True) + + +def _normalize_physical_column_type(column_type): + return " ".join(column_type.strip().upper().split()) if isinstance(column_type, str) else "" + + +def _shared_compact_column_type(base_column_type, current_column_type): + base_normalized = _normalize_physical_column_type(base_column_type) + current_normalized = _normalize_physical_column_type(current_column_type) + if not base_normalized or base_normalized != current_normalized: + return None + return base_normalized + + +def _get_physical_histogram_column_types(dbt_adapter, node, column): + """Resolve both runtime relation types, returning unknowns on any uncertainty.""" + get_columns = getattr(dbt_adapter, "get_columns", None) + if not callable(get_columns): + return None, None + + physical_types = [] + for base in (True, False): + try: + columns = get_columns(node, base=base) + matches = [ + candidate + for candidate in columns + if str(getattr(candidate, "column", "")).casefold() == str(column).casefold() + ] + if len(matches) != 1: + return None, None + column_type = getattr(matches[0], "dtype", None) + if not _normalize_physical_column_type(column_type): + return None, None + physical_types.append(column_type) + except Exception: + logger.debug("histogram physical column type lookup failed", exc_info=True) + return None, None + return physical_types[0], physical_types[1] + + +def query_numeric_histogram( + task, + node, + column, + column_type, + min_value, + max_value, + num_bins=50, + *, + adapter_type=None, + base_column_type=None, + current_column_type=None, +): is_integer = is_integral_histogram_type(column_type) geometry = numeric_histogram_geometry(min_value, max_value, num_bins, is_integer=is_integer) min_edge = geometry.bin_edges[0] max_edge = geometry.bin_edges[-1] num_bins = geometry.num_bins - if is_integer: - histogram_sql, _ = generate_histogram_sql_integer(node, column, min_edge, max_edge, num_bins, geometry.width) - else: - histogram_sql, _ = generate_histogram_sql_numeric(node, column, min_edge, max_edge, num_bins, geometry.width) + compact_column_type = _shared_compact_column_type(base_column_type, current_column_type) + generated_sql = _generate_histogram_sql_for_adapter( + node, + column, + min_edge, + max_edge, + num_bins, + geometry.width, + adapter_type=adapter_type, + column_type=compact_column_type, + ) + histogram_sql = generated_sql.sql + _emit_histogram_sql_telemetry( + adapter_type, + generated_sql.strategy, + num_bins, + len(histogram_sql), + ) base = None try: @@ -498,6 +659,7 @@ def execute(self): result = {} dbt_adapter: DbtAdapter = default_context().adapter + adapter_type = dbt_adapter.adapter.type() node = self.params.model column = self.params.column_name num_bins = self.params.num_bins or 50 @@ -550,8 +712,25 @@ def get_min_max(fn, base, curr): self, node, column, min_value, max_value ) else: + base_column_type = None + current_column_type = None + if isinstance(adapter_type, str) and adapter_type.strip().lower() == "duckdb": + base_column_type, current_column_type = _get_physical_histogram_column_types( + dbt_adapter, + node, + column, + ) base_result, current_result, bin_edges, labels = query_numeric_histogram( - self, node, column, column_type, min_value, max_value, num_bins + self, + node, + column, + column_type, + min_value, + max_value, + num_bins, + adapter_type=adapter_type, + base_column_type=base_column_type, + current_column_type=current_column_type, ) if base_result: base_result["total"] = base_total diff --git a/tests/adapter/test_histogram_bucketing.py b/tests/adapter/test_histogram_bucketing.py new file mode 100644 index 000000000..784c71ca2 --- /dev/null +++ b/tests/adapter/test_histogram_bucketing.py @@ -0,0 +1,176 @@ +from decimal import Decimal + +import pytest + +from recce.adapter.histogram_bucketing import ( + DUCKDB_SCALED_INTEGER, + LITERAL_ORDERED_COMPARISON, + select_histogram_bucketing, +) + + +def select_plan( + column_type: str, + *, + adapter_type: str = "duckdb", + minimum: Decimal = Decimal("0"), + maximum: Decimal = Decimal("1"), + width: Decimal = Decimal("0.2"), + num_bins: int = 5, +): + return select_histogram_bucketing( + adapter_type=adapter_type, + column_sql="amount", + column_type=column_type, + minimum=minimum, + maximum=maximum, + width=width, + num_bins=num_bins, + ) + + +@pytest.mark.parametrize( + "column_type", + [ + "TINYINT", + "SMALLINT", + "INTEGER", + "BIGINT", + "UTINYINT", + "USMALLINT", + "UINTEGER", + "UBIGINT", + "DECIMAL(18, 6)", + "DEC(18, 6)", + "NUMERIC(18, 6)", + "DECIMAL(18)", + "DECIMAL(18, 0)", + ], +) +def test_duckdb_bounded_integral_and_fixed_decimal_types_select_compact_bucketing(column_type): + """Catch a proven DuckDB type being routed back through the linear CASE ladder.""" + plan = select_plan(column_type) + + assert plan.strategy == DUCKDB_SCALED_INTEGER + assert plan.bin_expression is not None + assert "//" in plan.bin_expression + assert "/" not in plan.bin_expression.replace("//", "") + + +@pytest.mark.parametrize("adapter_type", [None, "", "sqlite", "postgres", "snowflake", "some-future-engine"]) +def test_non_duckdb_adapters_always_select_the_exact_literal_fallback(adapter_type): + """Catch an unproven dialect receiving DuckDB-only integer-division SQL.""" + plan = select_plan("DECIMAL(18, 6)", adapter_type=adapter_type) + + assert plan.strategy == LITERAL_ORDERED_COMPARISON + assert plan.bin_expression is None + + +@pytest.mark.parametrize( + "column_type", + [ + "FLOAT", + "DOUBLE", + "DOUBLE PRECISION", + "REAL", + "DECIMAL", + "NUMERIC", + "NUMBER", + "NUMBER(18, 6)", + "HUGEINT", + "UHUGEINT", + "INT128", + "UINT128", + "VARCHAR", + "future_numeric", + ], +) +def test_duckdb_unproven_types_select_the_exact_literal_fallback(column_type): + """Catch floating, unbounded, or unknown values entering scaled-integer arithmetic.""" + plan = select_plan(column_type) + + assert plan.strategy == LITERAL_ORDERED_COMPARISON + assert plan.bin_expression is None + + +@pytest.mark.parametrize( + ("column_type", "minimum", "maximum", "width", "num_bins"), + [ + ("DECIMAL(38, 0)", Decimal("0"), Decimal("10"), Decimal("1"), 10), + ("DECIMAL(37, 2)", Decimal("0"), Decimal("1"), Decimal("0.1"), 10), + ("DECIMAL(30, 20)", Decimal("0"), Decimal("1"), Decimal("0.1"), 10), + ("BIGINT", Decimal("0"), Decimal("1E-20"), Decimal("1E-21"), 10), + ("TINYINT", Decimal("0"), Decimal("1E-37"), Decimal("1E-38"), 10), + ("INTEGER", Decimal("0"), Decimal("1"), Decimal("0.3"), 3), + ("INTEGER", Decimal("NaN"), Decimal("1"), Decimal("0.2"), 5), + ], +) +def test_overflow_or_inexact_geometry_selects_the_literal_fallback(column_type, minimum, maximum, width, num_bins): + """Catch unsafe casts, intermediates, or inconsistent geometry being treated as proven.""" + plan = select_plan( + column_type, + minimum=minimum, + maximum=maximum, + width=width, + num_bins=num_bins, + ) + + assert plan.strategy == LITERAL_ORDERED_COMPARISON + assert plan.bin_expression is None + + +def test_fixed_decimal_scale_expansion_remains_exact_and_compact(): + """Catch geometry precision beyond the declared scale being rounded away.""" + plan = select_plan( + " numeric ( 12 , 2 ) ", + minimum=Decimal("0"), + maximum=Decimal("0.02"), + width=Decimal("0.002"), + num_bins=10, + ) + + assert plan.strategy == DUCKDB_SCALED_INTEGER + assert plan.bin_expression is not None + assert "1000" in plan.bin_expression + assert "2::HUGEINT" in plan.bin_expression + + +def test_large_decimal_geometry_is_scaled_without_ambient_context_rounding(): + """Catch Decimal.normalize shifting a valid 37-digit bound by one bin.""" + minimum = Decimal("1234567890123456789012345678900000000") + maximum = Decimal("1234567890123456789012345679900000000") + plan = select_plan( + "DECIMAL(37, 0)", + minimum=minimum, + maximum=maximum, + width=Decimal("100000000"), + num_bins=10, + ) + + assert plan.strategy == DUCKDB_SCALED_INTEGER + assert plan.bin_expression is not None + assert f"({minimum}::HUGEINT)" in plan.bin_expression + + +def test_compact_expression_size_does_not_grow_with_bin_count(): + """Catch accidental reintroduction of one SQL branch per requested bin.""" + small = select_plan( + "INTEGER", + minimum=Decimal("0"), + maximum=Decimal("50"), + width=Decimal("1"), + num_bins=50, + ) + large = select_plan( + "INTEGER", + minimum=Decimal("0"), + maximum=Decimal("10000"), + width=Decimal("1"), + num_bins=10000, + ) + + assert small.strategy == large.strategy == DUCKDB_SCALED_INTEGER + assert small.bin_expression is not None + assert large.bin_expression is not None + assert len(large.bin_expression) - len(small.bin_expression) < 16 + assert len(large.bin_expression) < 1500 diff --git a/tests/tasks/test_histogram.py b/tests/tasks/test_histogram.py index a921b763f..e6ad4faed 100644 --- a/tests/tasks/test_histogram.py +++ b/tests/tasks/test_histogram.py @@ -11,6 +11,7 @@ import recce.tasks.histogram as histogram from recce.tasks.histogram import ( HistogramDiffCheckValidator, + HistogramDiffParams, HistogramDiffTask, _is_histogram_supported, ) @@ -24,6 +25,12 @@ ] +@pytest.fixture(autouse=True) +def isolate_histogram_telemetry(monkeypatch): + """Keep unit/integration tests from scheduling the process-wide event collector.""" + monkeypatch.setattr(histogram, "log_performance", MagicMock()) + + @pytest.mark.parametrize( ("raw", "expected"), [ @@ -91,6 +98,83 @@ def test_numeric_histogram_geometry_covers_literal_domains( assert geometry.width >= Decimal("1") +@pytest.mark.parametrize( + ("params_update", "stored_value", "execution_request", "effective_bins"), + [ + ({}, 50, 50, 50), + ({"num_bins": None}, None, 50, 50), + ({"num_bins": 0}, 0, 50, 50), + ({"num_bins": False}, 0, 50, 50), + ({"num_bins": -3}, -3, -3, 1), + ({"num_bins": "-3"}, -3, -3, 1), + ({"num_bins": " 7 "}, 7, 7, 7), + ({"num_bins": 1.0}, 1, 1, 1), + ({"num_bins": "1.0"}, 1, 1, 1), + ({"num_bins": True}, 1, 1, 1), + ({"num_bins": 10000}, 10000, 10000, 10000), + ], +) +def test_histogram_num_bins_accepted_compatibility_matrix( + params_update, stored_value, execution_request, effective_bins +): + """Freeze accepted coercions/defaults while compact SQL removes the growth cost.""" + params = HistogramDiffParams( + model="numbers", + column_name="amount", + column_type="INTEGER", + **params_update, + ) + + assert params.num_bins == stored_value + if params.num_bins is not None: + assert type(params.num_bins) is int + requested = params.num_bins or 50 + assert requested == execution_request + geometry = histogram.numeric_histogram_geometry( + Decimal("0"), + Decimal(str(effective_bins)), + requested, + is_integer=True, + ) + assert geometry.num_bins == effective_bins + + +@pytest.mark.parametrize("num_bins", [2.5, "2.5"]) +def test_histogram_num_bins_still_rejects_fractional_inputs(num_bins): + """Catch optimization work broadening the request contract by accident.""" + with pytest.raises(ValueError): + HistogramDiffParams( + model="numbers", + column_name="amount", + column_type="INTEGER", + num_bins=num_bins, + ) + + +def test_histogram_num_bins_has_no_new_model_or_strategy_cap(): + """Catch either request parsing or compact dispatch imposing a hidden ceiling.""" + params = HistogramDiffParams( + model="numbers", + column_name="amount", + column_type="INTEGER", + num_bins=1_000_000, + ) + plan = histogram.select_histogram_bucketing( + adapter_type="duckdb", + column_sql="amount", + column_type="INTEGER", + minimum=Decimal("0"), + maximum=Decimal("1000000"), + width=Decimal("1"), + num_bins=params.num_bins, + ) + + assert params.num_bins == 1_000_000 + assert plan.strategy == "duckdb_scaled_integer" + assert plan.bin_expression is not None + assert len(plan.bin_expression) < 1500 + + def test_histogram_sql_uses_plain_decimal_literals_and_terminal_rule(): """Catch SQL that leaks exponent literals or lets a terminal maximum escape.""" sql, _ = histogram.generate_histogram_sql_numeric( @@ -107,6 +191,62 @@ def test_histogram_sql_uses_plain_decimal_literals_and_terminal_rule(): assert "WHEN amount = (SELECT max_value FROM bin_parameters) THEN 1" in sql +@pytest.mark.parametrize( + ("adapter_type", "column_type"), + [ + ("sqlite", "DECIMAL(18, 6)"), + ("postgres", "INTEGER"), + ("duckdb", "DOUBLE"), + ("duckdb", "NUMERIC"), + ("duckdb", "DECIMAL(38, 0)"), + ("duckdb", "HUGEINT"), + ], +) +def test_unproven_histogram_strategies_preserve_the_literal_ladder(adapter_type, column_type): + """Catch fallback dispatch changing exact ordered-comparison SQL.""" + generated = histogram._generate_histogram_sql_for_adapter( + "customers", + "amount", + Decimal("0"), + Decimal("1"), + 5, + Decimal("0.2"), + adapter_type=adapter_type, + column_type=column_type, + ) + existing_sql, existing_width = histogram.generate_histogram_sql_numeric( + "customers", + "amount", + Decimal("0"), + Decimal("1"), + 5, + Decimal("0.2"), + ) + + assert generated.strategy == "literal_ordered_comparison" + assert generated.sql == existing_sql + assert generated.bin_size == existing_width + assert generated.sql.count("WHEN amount <") == 5 + + +def test_duckdb_compact_generator_is_bounded_at_ten_thousand_bins(): + """Catch compact dispatch retaining a hidden per-bin SQL fragment.""" + generated = histogram._generate_histogram_sql_for_adapter( + "customers", + "amount", + Decimal("0"), + Decimal("10000"), + 10000, + Decimal("1"), + adapter_type="duckdb", + column_type="INTEGER", + ) + + assert generated.strategy == "duckdb_scaled_integer" + assert len(generated.sql) < 5000 + assert generated.sql.count("WHEN amount <") == 0 + + class HistogramRows: def __init__(self, rows): self.rows = rows @@ -116,8 +256,10 @@ class NumericHistogramQueryTask: def __init__(self, base_rows, current_rows): self.base_rows = HistogramRows(base_rows) self.current_rows = HistogramRows(current_rows) + self.executed_sql = [] - def execute_sql(self, _sql, *, base): + def execute_sql(self, sql, *, base): + self.executed_sql.append(sql) return self.base_rows if base else self.current_rows def check_cancel(self): @@ -140,6 +282,48 @@ def check_cancel(self): return None +def create_typed_histogram_model( + dbt_test_helper, + model_name, + column_type, + values, + *, + current_column_type=None, + current_values=None, +): + """Create matching real DuckDB relations plus dbt manifest/catalog entries.""" + current_column_type = current_column_type or column_type + current_values = values if current_values is None else current_values + adapter = dbt_test_helper.adapter + with adapter.connection_named("create typed histogram model"): + for schema, physical_type, environment_values in ( + (dbt_test_helper.base_schema, column_type, values), + (dbt_test_helper.curr_schema, current_column_type, current_values), + ): + value_rows = ", ".join("(NULL)" if value is None else f"({value})" for value in environment_values) + adapter.execute(f"CREATE TABLE {schema}.{model_name} (amount {physical_type})") + adapter.execute(f"INSERT INTO {schema}.{model_name} VALUES {value_rows}") + + model_sql = f"select cast(amount as {column_type}) as amount from source_values" + current_model_sql = f"select cast(amount as {current_column_type}) as amount from source_values" + dbt_test_helper.create_model( + model_name, + base_sql=model_sql, + curr_sql=current_model_sql, + base_columns={"amount": column_type}, + curr_columns={"amount": current_column_type}, + ) + + +def execute_histogram_template(dbt_test_helper, sql_template): + """Compile and execute a histogram template through Recce's real dbt adapter.""" + adapter = dbt_test_helper.adapter + with adapter.connection_named("execute histogram strategy"): + sql = adapter.generate_sql(sql_template, base=True) + _, result = adapter.execute(sql, fetch=True, auto_begin=True) + return [(row[0], row[1]) for row in result.rows] + + def test_daily_datetime_histogram_edges_cross_months_monotonically(): """Catch absolute ``relativedelta(day=...)`` fields resetting edges within a month.""" task = DatetimeHistogramQueryTask( @@ -333,6 +517,401 @@ def test_histogram_integral_type_detection_respects_precision_and_scale(column_t assert histogram.is_integral_histogram_type(column_type) is expected +def test_duckdb_compact_decimal_execution_owns_exact_internal_and_terminal_boundaries(dbt_test_helper, monkeypatch): + """Catch fixed DECIMAL bucketing drifting at 0.6 or either side of a boundary.""" + create_typed_histogram_model( + dbt_test_helper, + "compact_decimal_boundaries", + "DECIMAL(18, 6)", + [ + None, + "0", + "0.199999", + "0.2", + "0.200001", + "0.399999", + "0.4", + "0.400001", + "0.599999", + "0.6", + "0.600001", + "0.799999", + "0.8", + "0.800001", + "1", + ], + ) + telemetry = MagicMock() + monkeypatch.setattr(histogram, "log_performance", telemetry, raising=False) + + result = HistogramDiffTask( + { + "model": "compact_decimal_boundaries", + "column_name": "amount", + "column_type": "DECIMAL(18, 6)", + "num_bins": 5, + } + ).execute() + + assert result["bin_edges"] == [0, 0.2, 0.4, 0.6, 0.8, 1] + assert result["base"] == {"counts": [2, 3, 3, 3, 3], "total": 14} + assert result["current"] == result["base"] + telemetry.assert_called_once() + feature_name, metrics = telemetry.call_args.args + assert feature_name == "histogram_sql" + assert metrics["adapter_type"] == "duckdb" + assert metrics["strategy"] == "duckdb_scaled_integer" + assert metrics["effective_bin_count"] == 5 + assert metrics["sql_length"] < 5000 + + +def test_duckdb_compact_decimal_execution_expands_beyond_the_declared_scale(dbt_test_helper, monkeypatch): + """Catch a generated width with extra decimal places being rounded before bucketing.""" + create_typed_histogram_model( + dbt_test_helper, + "compact_decimal_scale_expansion", + "DECIMAL(12, 2)", + ["0", "0.01", "0.02"], + ) + telemetry = MagicMock() + monkeypatch.setattr(histogram, "log_performance", telemetry, raising=False) + + result = HistogramDiffTask( + { + "model": "compact_decimal_scale_expansion", + "column_name": "amount", + "column_type": "DECIMAL(12, 2)", + "num_bins": 10, + } + ).execute() + + assert result["bin_edges"] == [0, 0.002, 0.004, 0.006, 0.008, 0.01, 0.012, 0.014, 0.016, 0.018, 0.02] + assert {index: count for index, count in enumerate(result["base"]["counts"]) if count} == { + 0: 1, + 5: 1, + 9: 1, + } + assert telemetry.call_args.args[1]["strategy"] == "duckdb_scaled_integer" + + +def test_duckdb_compact_decimal_executes_at_the_proven_precision_boundary(dbt_test_helper, monkeypatch): + """Catch the widest enabled DECIMAL multiply overflowing before its HUGEINT cast.""" + create_typed_histogram_model( + dbt_test_helper, + "compact_decimal_precision_boundary", + "DECIMAL(37, 1)", + ["0", "500000000000000000000000000000000000.0"], + ) + telemetry = MagicMock() + monkeypatch.setattr(histogram, "log_performance", telemetry) + + result = HistogramDiffTask( + { + "model": "compact_decimal_precision_boundary", + "column_name": "amount", + "column_type": "DECIMAL(37, 1)", + "num_bins": 1, + } + ).execute() + + assert result["base"] == {"counts": [2], "total": 2} + assert telemetry.call_args.args[1]["strategy"] == "duckdb_scaled_integer" + + +def test_duckdb_compact_scale_zero_decimal_executes_through_the_integer_cast_path(dbt_test_helper, monkeypatch): + """Catch DECIMAL(..., 0) being cast or multiplied with lossy semantics.""" + create_typed_histogram_model( + dbt_test_helper, + "compact_scale_zero_decimal", + "DECIMAL(18, 0)", + ["0", "2", "3", "5", "10"], + ) + telemetry = MagicMock() + monkeypatch.setattr(histogram, "log_performance", telemetry, raising=False) + + result = HistogramDiffTask( + { + "model": "compact_scale_zero_decimal", + "column_name": "amount", + "column_type": "DECIMAL(18, 0)", + "num_bins": 4, + } + ).execute() + + assert result["bin_edges"] == [0, 2.5, 5, 7.5, 10] + assert result["base"] == {"counts": [2, 1, 1, 1], "total": 5} + assert telemetry.call_args.args[1]["strategy"] == "duckdb_scaled_integer" + + +def test_duckdb_compact_large_decimal_geometry_executes_without_context_rounding(dbt_test_helper): + """Catch a 37-digit minimum being rounded up to the next boundary.""" + minimum = Decimal("1234567890123456789012345678900000000") + maximum = Decimal("1234567890123456789012345679900000000") + create_typed_histogram_model( + dbt_test_helper, + "compact_large_decimal_geometry", + "DECIMAL(37, 0)", + [ + "1234567890123456789012345678900000000", + "1234567890123456789012345679000000000", + "1234567890123456789012345679900000000", + ], + ) + generated = histogram._generate_histogram_sql_for_adapter( + "compact_large_decimal_geometry", + "amount", + minimum, + maximum, + 10, + Decimal("100000000"), + adapter_type="duckdb", + column_type="DECIMAL(37, 0)", + ) + + compiled = dbt_test_helper.adapter.generate_sql(generated.sql, base=True) + with dbt_test_helper.adapter.connection_named("execute compact large-decimal histogram"): + _, table = dbt_test_helper.adapter.execute(compiled, fetch=True, auto_begin=True) + + assert generated.strategy == "duckdb_scaled_integer" + assert {row[0]: row[1] for row in table.rows} == {0: 1, 1: 1, 9: 1} + + +def test_duckdb_falls_back_when_base_and_current_physical_types_differ(dbt_test_helper, monkeypatch): + """Catch one environment's integer type authorizing a lossy cast in the other.""" + create_typed_histogram_model( + dbt_test_helper, + "mixed_physical_histogram_types", + "INTEGER", + ["0", "2", "10"], + current_column_type="DOUBLE", + current_values=["0", "1.9", "2", "10"], + ) + telemetry = MagicMock() + monkeypatch.setattr(histogram, "log_performance", telemetry) + + result = HistogramDiffTask( + { + "model": "mixed_physical_histogram_types", + "column_name": "amount", + "column_type": "INTEGER", + "num_bins": 5, + } + ).execute() + + assert result["base"] == {"counts": [1, 1, 0, 0, 1], "total": 3} + assert result["current"] == {"counts": [2, 1, 0, 0, 1], "total": 4} + assert telemetry.call_args.args[1]["strategy"] == "literal_ordered_comparison" + + +def test_duckdb_falls_back_when_saved_type_is_stale_for_both_relations(dbt_test_helper, monkeypatch): + """Catch stale saved INTEGER metadata compact-casting two physical DOUBLE columns.""" + create_typed_histogram_model( + dbt_test_helper, + "stale_saved_histogram_type", + "DOUBLE", + ["0", "1.9", "2", "10"], + ) + telemetry = MagicMock() + monkeypatch.setattr(histogram, "log_performance", telemetry) + + result = HistogramDiffTask( + { + "model": "stale_saved_histogram_type", + "column_name": "amount", + "column_type": "INTEGER", + "num_bins": 5, + } + ).execute() + + assert result["base"] == {"counts": [2, 1, 0, 0, 1], "total": 4} + assert result["current"] == result["base"] + assert telemetry.call_args.args[1]["strategy"] == "literal_ordered_comparison" + + +def test_duckdb_decimal_38_executes_with_the_exact_literal_fallback(dbt_test_helper, monkeypatch): + """Catch overflow-risk DECIMAL input being compacted instead of executing the safe ladder.""" + create_typed_histogram_model( + dbt_test_helper, + "fallback_decimal_38", + "DECIMAL(38, 0)", + ["0", "2", "5"], + ) + telemetry = MagicMock() + monkeypatch.setattr(histogram, "log_performance", telemetry) + + result = HistogramDiffTask( + { + "model": "fallback_decimal_38", + "column_name": "amount", + "column_type": "DECIMAL(38, 0)", + "num_bins": 5, + } + ).execute() + + assert result["base"] == {"counts": [1, 0, 1, 0, 1], "total": 3} + assert telemetry.call_args.args[1]["strategy"] == "literal_ordered_comparison" + + +def test_duckdb_compact_integer_execution_preserves_negative_domain_boundaries(dbt_test_helper, monkeypatch): + """Catch negative values being bucketed with truncation rather than exact edge ownership.""" + create_typed_histogram_model( + dbt_test_helper, + "compact_negative_integer", + "INTEGER", + ["-10", "-9", "-8", "-7", "0", "9", "10"], + ) + telemetry = MagicMock() + monkeypatch.setattr(histogram, "log_performance", telemetry, raising=False) + + result = HistogramDiffTask( + { + "model": "compact_negative_integer", + "column_name": "amount", + "column_type": "INTEGER", + "num_bins": 10, + } + ).execute() + + assert result["bin_edges"] == [-10, -8, -6, -4, -2, 0, 2, 4, 6, 8, 10] + assert {index: count for index, count in enumerate(result["base"]["counts"]) if count} == { + 0: 2, + 1: 2, + 5: 1, + 9: 2, + } + assert telemetry.call_args.args[1]["strategy"] == "duckdb_scaled_integer" + + +def test_duckdb_compact_sql_leaves_concurrent_out_of_domain_values_out_of_range(dbt_test_helper): + """Catch compact arithmetic clamping below/above-domain rows into edge bins.""" + create_typed_histogram_model( + dbt_test_helper, + "compact_out_of_domain", + "DECIMAL(18, 6)", + [None, "-0.000001", "0", "0.199999", "0.2", "0.6", "1", "1.000001"], + ) + generated = histogram._generate_histogram_sql_for_adapter( + "compact_out_of_domain", + "amount", + Decimal("0"), + Decimal("1"), + 5, + Decimal("0.2"), + adapter_type="duckdb", + column_type="DECIMAL(18, 6)", + ) + + compiled = dbt_test_helper.adapter.generate_sql(generated.sql, base=True) + with dbt_test_helper.adapter.connection_named("execute compact out-of-domain histogram"): + _, table = dbt_test_helper.adapter.execute(compiled, fetch=True, auto_begin=True) + counts_by_bin = {row[0]: row[1] for row in table.rows} + + assert generated.strategy == "duckdb_scaled_integer" + assert counts_by_bin == {-1: 1, 0: 2, 1: 1, 3: 1, 4: 1, 5: 1, None: 1} + + +def test_duckdb_compact_ten_thousand_bin_request_executes_without_sql_growth(dbt_test_helper, monkeypatch): + """Catch application orchestration accepting 10,000 bins but emitting a 10,000-branch query.""" + create_typed_histogram_model( + dbt_test_helper, + "compact_ten_thousand_bins", + "INTEGER", + ["0", "1", "9999", "10000"], + ) + telemetry = MagicMock() + monkeypatch.setattr(histogram, "log_performance", telemetry, raising=False) + + task = HistogramDiffTask( + { + "model": "compact_ten_thousand_bins", + "column_name": "amount", + "column_type": "INTEGER", + "num_bins": 10000, + } + ) + result = task.execute() + + assert task.params.num_bins == 10000 + assert len(result["bin_edges"]) == 10001 + assert len(result["base"]["counts"]) == 10000 + assert {index: count for index, count in enumerate(result["base"]["counts"]) if count} == { + 0: 1, + 1: 1, + 9999: 2, + } + metrics = telemetry.call_args.args[1] + assert metrics["effective_bin_count"] == 10000 + assert metrics["strategy"] == "duckdb_scaled_integer" + assert metrics["sql_length"] < 5000 + + +def test_histogram_sql_telemetry_failure_does_not_change_results(monkeypatch): + """Catch optional telemetry turning a successful warehouse result into a failed run.""" + telemetry = MagicMock(side_effect=RuntimeError("telemetry unavailable")) + monkeypatch.setattr(histogram, "log_performance", telemetry, raising=False) + task = NumericHistogramQueryTask([(0, 2)], [(0, 3)]) + + base, current, bin_edges, labels = histogram.query_numeric_histogram( + task, + "numbers", + "amount", + "INTEGER", + Decimal("0"), + Decimal("1"), + 1, + adapter_type="duckdb", + ) + + assert base == {"counts": [2]} + assert current == {"counts": [3]} + assert bin_edges == [0, 1] + assert labels + telemetry.assert_called_once() + + +def test_literal_fallback_reports_its_actual_sql_length(monkeypatch): + """Catch fallback telemetry claiming a compact strategy or synthetic query size.""" + telemetry = MagicMock() + monkeypatch.setattr(histogram, "log_performance", telemetry) + task = NumericHistogramQueryTask([(0, 2)], [(0, 3)]) + + histogram.query_numeric_histogram( + task, + "numbers", + "amount", + "DOUBLE", + Decimal("0"), + Decimal("1"), + 5, + adapter_type="duckdb", + ) + + metrics = telemetry.call_args.args[1] + assert metrics["adapter_type"] == "duckdb" + assert metrics["strategy"] == "literal_ordered_comparison" + assert metrics["effective_bin_count"] == 5 + assert metrics["sql_length"] == len(task.executed_sql[0]) + assert task.executed_sql[0] == task.executed_sql[1] + + +def test_compact_out_of_domain_bucket_still_reaches_the_existing_python_guard(monkeypatch): + """Catch telemetry or compact dispatch swallowing the established range error.""" + monkeypatch.setattr(histogram, "log_performance", MagicMock(), raising=False) + task = NumericHistogramQueryTask([(-1, 1)], [(0, 1)]) + + with pytest.raises(ValueError, match="outside the computed edge domain"): + histogram.query_numeric_histogram( + task, + "numbers", + "amount", + "INTEGER", + Decimal("0"), + Decimal("1"), + 1, + adapter_type="duckdb", + ) + + def test_fractional_histogram_uses_decimal_edges_for_labels_and_real_sql_boundaries(dbt_test_helper): """Catch fractional label failures and floating internal/terminal misbucketing.""" base_csv = """ @@ -614,6 +1193,7 @@ def validate(params: dict = {}, view_options: dict = {}): "model": "customers", "column_name": "age", "column_type": "int", + "num_bins": 1_000_000, } ) diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 52281bf76..9308b39a5 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -1464,6 +1464,36 @@ async def test_tool_histogram_diff(self, mcp_server): assert "current" in result mock_context.get_model.assert_called_once_with("model.project.my_model", base=False) + @pytest.mark.asyncio + async def test_tool_histogram_diff_forwards_unbounded_num_bins(self, mcp_server): + """Catch the MCP surface imposing a cap absent from direct histogram requests.""" + server, mock_context = mcp_server + mock_context.build_name_to_unique_id_index.return_value = {"my_model": "model.project.my_model"} + mock_context.get_model.return_value = { + "columns": {"age": {"name": "age", "type": "INTEGER"}}, + } + mock_result = { + "base": {"counts": [], "total": 0}, + "current": {"counts": [], "total": 0}, + "min": None, + "max": None, + "bin_edges": [], + "labels": [], + } + + with patch("recce.mcp_server.HistogramDiffTask") as task_type: + task_type.return_value.execute.return_value = mock_result + await server._tool_histogram_diff({"model": "my_model", "column_name": "age", "num_bins": 1_000_000}) + + task_type.assert_called_once_with( + params={ + "model": "my_model", + "column_name": "age", + "num_bins": 1_000_000, + "column_type": "INTEGER", + } + ) + @pytest.mark.asyncio async def test_tool_histogram_diff_missing_model(self, mcp_server): """Test histogram_diff raises when model is missing"""