-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathanalyzer.py
More file actions
158 lines (122 loc) · 5.27 KB
/
Copy pathanalyzer.py
File metadata and controls
158 lines (122 loc) · 5.27 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
"""SQL analyzer component - resolves field types from schema."""
from __future__ import annotations
from dataclasses import dataclass, field
from sql_redis.parser import (
AggregationSpec,
ComputedField,
Condition,
DateFunctionSpec,
ParsedQuery,
)
@dataclass
class VectorSearchAnalysis:
"""Analyzed vector search details."""
field: str
k: int
alias: str
@dataclass
class AnalyzedQuery:
"""Result of analyzing a parsed SQL query with schema context."""
parsed: ParsedQuery = field(default_factory=ParsedQuery)
field_types: dict[str, str] = field(default_factory=dict)
aggregations: list[AggregationSpec] = field(default_factory=list)
computed_fields: list[ComputedField] = field(default_factory=list)
date_functions: list[DateFunctionSpec] = field(default_factory=list)
groupby_fields: list[str] = field(default_factory=list)
is_global_aggregation: bool = False
vector_search: VectorSearchAnalysis | None = None
has_prefilter: bool = False
def get_field_type(self, field_name: str) -> str | None:
"""Get the type of a field."""
return self.field_types.get(field_name)
def get_conditions_by_type(self, field_type: str) -> list[Condition]:
"""Get conditions for fields of a specific type."""
return [
c
for c in self.parsed.conditions
if self.field_types.get(c.field) == field_type
]
class Analyzer:
"""Analyzes parsed SQL queries with schema context."""
def __init__(self, schemas: dict[str, dict[str, str]]):
"""Initialize analyzer with schema registry data.
Args:
schemas: Dict mapping index names to field->type dicts.
"""
self._schemas = schemas
def analyze(self, parsed: ParsedQuery) -> AnalyzedQuery:
"""Analyze a parsed query, resolving field types.
Args:
parsed: The parsed SQL query.
Returns:
An AnalyzedQuery with field types resolved.
Raises:
ValueError: If the index or a field is unknown.
"""
# Validate index exists
if parsed.index not in self._schemas:
raise ValueError(f"Unknown index: {parsed.index}")
schema = self._schemas[parsed.index]
result = AnalyzedQuery(parsed=parsed)
# Collect all fields referenced in the query
referenced_fields: set[str] = set()
# Fields from SELECT
for field_name in parsed.fields:
if field_name != "*":
referenced_fields.add(field_name)
# Fields from conditions
for condition in parsed.conditions:
referenced_fields.add(condition.field)
# Fields from geo_conditions
for geo_cond in parsed.geo_conditions:
referenced_fields.add(geo_cond.field)
# Fields from geo_distance selects
for geo_select in parsed.geo_distance_selects:
referenced_fields.add(geo_select.field)
# Fields from aggregations
for agg in parsed.aggregations:
if agg.field:
referenced_fields.add(agg.field)
# Fields from computed fields (extract field references from expressions)
for computed in parsed.computed_fields:
# Simple extraction - look for field names in the expression
for field_name in schema.keys():
if field_name in computed.expression:
referenced_fields.add(field_name)
# Fields from vector search
if parsed.vector_search:
referenced_fields.add(parsed.vector_search.field)
# Fields from date functions (YEAR, MONTH, etc.)
for date_func in parsed.date_functions:
referenced_fields.add(date_func.field)
# Collect aliases from date functions and computed fields (for GROUP BY)
alias_names = {df.alias for df in parsed.date_functions}
alias_names.update(cf.alias for cf in parsed.computed_fields)
# Fields from GROUP BY (exclude aliases since they're computed)
for field_name in parsed.groupby_fields:
if field_name not in alias_names:
referenced_fields.add(field_name)
# Resolve field types
for field_name in referenced_fields:
if field_name not in schema:
raise ValueError(f"Unknown field: {field_name}")
result.field_types[field_name] = schema[field_name]
# Copy aggregations, computed fields, and date functions
result.aggregations = parsed.aggregations
result.computed_fields = parsed.computed_fields
result.date_functions = parsed.date_functions
result.groupby_fields = parsed.groupby_fields
# Determine if this is a global aggregation
result.is_global_aggregation = (
len(parsed.aggregations) > 0 and len(parsed.groupby_fields) == 0
)
# Analyze vector search
if parsed.vector_search:
result.vector_search = VectorSearchAnalysis(
field=parsed.vector_search.field,
k=parsed.limit or parsed.vector_search.k or 10,
alias=parsed.vector_search.alias,
)
# Has prefilter if there are conditions
result.has_prefilter = len(parsed.conditions) > 0
return result