neosqlite.collection.sql_tier_aggregator module

SQL Tier 1 Optimizer for Aggregation Pipelines.

This module implements SQL-based optimization for aggregation pipelines, providing 10-100x performance improvements over Python fallback (Tier 3).

The optimizer analyzes aggregation pipelines and generates optimized SQL queries using CTEs (Common Table Expressions) for multi-stage pipelines.

class neosqlite.collection.sql_tier_aggregator.SQLTierAggregator(collection, expr_evaluator: ExprEvaluator | None = None, translation_cache_size: int | None = 100)

Bases: StageBuildersMixin

SQL Tier 1 optimizer for aggregation pipelines.

SUPPORTED_STAGES = {'$addFields', '$bucket', '$bucketAuto', '$count', '$densify', '$facet', '$fill', '$graphLookup', '$group', '$limit', '$lookup', '$match', '$merge', '$project', '$redact', '$replaceRoot', '$replaceWith', '$sample', '$setWindowFields', '$skip', '$sort', '$unionWith', '$unset', '$unwind'}

UNSUPPORTED_STAGES = {'$indexStats', '$jsonSchema', '$out'}

UNSUPPORTED_EXPRESSIONS = {'$accumulator', '$function', '$jsonSchema', '$script'}

__init__(collection, expr_evaluator: ExprEvaluator | None = None, translation_cache_size: int | None = 100)

Initialize the SQL tier aggregator.

_get_json_extract(path: str | None = None) → str

Get JSON extract function with correct prefix.

_get_json_set() → str

Get JSON set function with correct prefix.

can_optimize_pipeline(pipeline: list[dict[str, Any]]) → bool

Check if pipeline can be optimized in SQL tier.

_can_optimize_stage_expressions(stage: dict[str, Any]) → bool

Check if all expressions in a stage can be optimized in SQL.

_check_expression_support(obj: Any) → bool

Recursively check if an object contains unsupported expressions.

build_pipeline_sql(pipeline: list[dict[str, Any]]) → tuple[str | None, list[Any]]

Build optimized SQL query for entire pipeline using CTEs.

_build_sql_template(pipeline: list[dict[str, Any]]) → tuple[str | None, list[Any]]

Build SQL template and return (template, params).

get_cache_stats() → dict[str, Any]

Get pipeline cache statistics.

clear_cache() → None

Clear the pipeline cache.

dump_cache() → list[dict]

Dump all cache entries for debugging.

cache_contains(pipeline: list[dict]) → bool

Check if pipeline is in cache.

evict_from_cache(pipeline: list[dict]) → bool

Evict a specific pipeline from cache.

cache_size() → int

Get current cache size.

is_cache_enabled() → bool

Check if cache is enabled.

resize_cache(new_size: int) → None

Resize the cache.

_pipeline_needs_root(pipeline: list[dict[str, Any]]) → bool

Check if pipeline uses $$ROOT variable.

_stage_uses_root(stage: dict[str, Any]) → bool

Check if a stage uses $$ROOT variable.

_expression_uses_root(obj: Any) → bool

Recursively check if expression uses $$ROOT.