chronaxis.execution.record_scope_calculations
源码基线：8904c2f744951857659abd40aaa3535d57faf53c
1"""只对明确的事件指标保存必要计算值，并按事件真实范围回读。23同表达式、条件和驱动共享一份紧凑历史；每次查询复用已验证的记录积分/统计核。4窗口尚未确定前只接事件，默认会话统计不创建这个所有者。5"""6import heapq7from ..core.moments import SAMPLE_MOMENTS8from ..core.quantiles import SAMPLE_QUANTILES9from functools import partial as bind_consumer10from ..core.types import QualityFlag as Q11from dataclasses import replace12from ..core.models import Interval13from ..core.calculations import CalculationDefinition14from ..results.errors import ResultClosedError15from ..results.spool import _reserve_owned_ram,_release_owned_ram,_cleanup_owner16from .history import _metadata_bytes17from .driver_expression import DriverExpression18from .record_value_history import RecordValueHistory19from .record_integral import RecordIntegral20from .record_time_calculation import RecordTimeCalculation21from .record_selection import RecordSelection222324class RecordScopeCalculations:25 """拥有必要表达式历史、等待右端的查询及当前统计器；计划/状态为借用。"""2627 def __init__(self,plan,state,budget,path,steps,requests,ends,consumer,*,selection_consumer=None):28 """建立已请求的有限表达式分组。参数: 原计划/状态/预算/路径/周期/请求/EOF/回调。返回: owner。异常: 原构造错。"""29 self._plan,self._budget,self._path,self._consumer=plan,budget,path,consumer30 self._selection_consumer=selection_consumer31 self._requests,self._steps=requests,steps32 self._held_cost=self._fixed_cost=self._queue_cost=033 self._streams,self._request_stream={},{}34 self._queue=[]35 self._serial=036 self._pending=self._calculation=None37 self._calculators=[]38 self._closed=self._failed=False39 try:40 _reserve_owned_ram(self,131072+64*_metadata_bytes(requests),'_fixed_cost')41 choices=dict(plan.intervals.resolved)42 from ..planning.intervals import IntervalChoice43 self._hosted=plan.extensions.reductions if plan.extensions is not None else {}44 sample_ops=('sample_count','sample_minimum','sample_maximum','sample_mean','select', *SAMPLE_MOMENTS, *SAMPLE_QUANTILES,*self._hosted)45 for request in requests:46 if request.operation in sample_ops:choices[request.calculation_id]=IntervalChoice('actual')47 sample_keys={(r.signal.identity,None if r.condition is None else r.condition.identity,None)48 for r in requests if r.operation in sample_ops}49 for request in requests:50 expression=request.signal if request.signal is not None else request.condition51 condition=request.condition if request.signal is not None else None52 driver=choices[request.calculation_id].driver53 key=(expression.identity,None if condition is None else condition.identity,driver)54 self._request_stream[request.calculation_id]=key55 if key in self._streams:56 continue57 self._streams[key]=[None,None,None,expression,condition,58 driver is None and not expression.dependencies and (condition is None or not condition.dependencies)]59 stream=self._streams[key]60 for index,static in enumerate((expression,condition)):61 if static is not None:62 stream[index]=self._new(DriverExpression,'driver_expression_cleanup_owner',static,state,driver,63 backend=plan.execution.backend,derived=tuple(item.definition for item in plan.derived),observation_ends=ends)64 stream[2]=self._new(RecordValueHistory,'record_value_history_cleanup_owner',65 budget,path,expression.root.dtype,condition is not None,sample_updates=key in sample_keys,66 labels=expression.root.labels)67 except BaseException as error:68 _cleanup_owner(self,error,'record_scope_calculations_cleanup_owner','事件指标构造')69 raise7071 def _new(self,factory,recovery,*args,**kwargs):72 """接住未返回的实际子owner。参数: 工厂/恢复字段/构造参数。返回: owner。异常: 原错。"""73 try:74 return factory(*args,**kwargs)75 except BaseException as error:76 self._pending=getattr(error,recovery,None)77 raise7879 def _open(self):80 """拒绝已失败/关闭的执行。参数: 无。返回: None。异常: ResultClosedError。"""81 if self._closed or self._failed:82 raise ResultClosedError('事件指标已关闭或失败')8384 def observe_group(self):85 """完整组已原子提交，更新各必要表达式一次。参数: 无。返回: None。异常: 求值/写入/预算原错。"""86 self._open()87 for value,condition,history,_,__,constant in self._streams.values():88 if constant:89 continue90 row=value.evaluate()91 selected=None if condition is None else condition.evaluate()92 if value.invalidated_at is not None:93 history.expire_at(value.invalidated_at)94 if condition is not None and condition.invalidated_at is not None:95 history.expire_at(condition.invalidated_at,condition=True)96 if row is not None:97 if condition is not None and (selected is None or selected[0]!=row[0]):98 raise ValueError('事件值和条件必须共用驱动')99 actual=history._sample_updates and value.actual_update_at(row[0])100 history.append(row,selected,actual_update=actual)101 self._flush()102103 def observe_event(self,event):104 """仅保等待计算需要的父ID和真实端点，不复制事件全文。参数: 事件。返回: None。异常: 预算/计算原错。"""105 self._open()106 requests=tuple(request for request in self._requests if request.scope=='event:'+event.rule.rule_id)107 if requests:108 end=event.start_ns if event.end_ns is None else event.end_ns109 # 选择起点只裁统计范围；事件自身仍保留原活动开始、确认和结束。110 rule=next(item for item in self._plan.rules if item.rule_id==event.rule.rule_id)111 start=event.confirmed_ns if rule.selection_origin=='confirmed' else event.start_ns112 cost=8192+16*_metadata_bytes(event.event_id)+128*len(requests)113 _reserve_owned_ram(self,cost,'_queue_cost')114 heapq.heappush(self._queue,(end,self._serial,requests,event.event_id,start,cost,115 bool(event.dependency_quality & Q.ESTIMATED)))116 self._serial+=1117 self._flush()118119 def observe_segment(self,segment):120 """EOF后按实际片段范围计算，缺失扩展不伪装完整。121122 参数: 精简核心片段。返回: None。异常: 原查询错；空片段只返回无观测值。123 """124 self._open()125 requests=tuple(request for request in self._requests if request.scope=='segment:'+segment.window_id)126 if not requests:127 return128 actual=segment.actual129 start=self._plan.dataset.domain.start_ns if actual is None else actual.start_ns130 end=start if actual is None else actual.end_ns131 self._calculate(requests,segment.segment_id,start,end,partial=segment.status!='complete',132 parent_coverage=segment.coverage,estimated=bool(segment.quality & Q.ESTIMATED))133134 def _calculate(self,requests,parent,start_ns,end_ns,*,partial=False,parent_coverage=None,estimated=False):135 """同范围、同表达式的相邻指标共读一次；预算余量不足时缩为单项。136137 参数: 原声明顺序的请求tuple/父ID/实际范围及覆盖。返回: None。异常: 原查询、预算或输出错误。138 不缓存整个范围；每次借一行同时交给有限计算器，仍按声明顺序发布。139 """140 from itertools import groupby141 for key,group in groupby(requests,key=lambda request:self._request_stream[request.calculation_id]):142 items=tuple(group)143 # 留出每项计算器和批数组的余量；所有真实分配仍由共享预算检查。144 available=self._budget.memory_budget_bytes-self._budget.ram_bytes145 width=max(1,min(8,available//(32*1024*1024)))146 for offset in range(0,len(items),width):147 chunk=items[offset:offset+width]148 self._calculate_chunk(key,chunk,parent,start_ns,end_ns,partial,parent_coverage,estimated)149150 def _calculate_chunk(self,key,requests,parent,start_ns,end_ns,partial,parent_coverage,estimated):151 """建立有限计算器组并共享一次回读。参数: 流key/请求/范围/质量。返回: None。异常: 原错误保留owner。152 """153 history,expression,condition=self._streams[key][2:5]154 for request in requests:155 descriptor=self._hosted.get(request.operation)156 spec=CalculationDefinition(request.calculation_id,'sample_minimum' if descriptor else request.operation,Interval(start_ns,end_ns),request.partial_policy,157 request.unit,request.scope.split(':',1)[0],request.minimum_coverage,config_version=self._plan.config_version, ddof=request.ddof, q=request.q)158 factory=RecordIntegral if request.operation=='integral' else RecordTimeCalculation159 recovery='record_integral_cleanup_owner' if request.operation=='integral' else 'record_time_cleanup_owner'160 extra={'work_dir':self._path} if request.operation=='time_weighted_mean' else {}161 if descriptor:162 extra={'hosted':descriptor}163 if request.operation=='select':164 factory,recovery=RecordSelection,'record_selection_cleanup_owner'165 extra={'consumer':bind_consumer(self._selection_consumer,request,parent)}166 if estimated:167 extra['consumer']=bind_consumer(self._estimated_selection,request,parent)168 self._calculation=self._new(factory,recovery,self._budget,spec,expression.root.unit,expression.root.dtype,169 self._steps[request.calculation_id],backend=self._plan.execution.backend,170 batch_intervals=min(self._plan.execution.batch_rows,4096),conditioned=condition is not None,171 **extra)172 self._calculators.append(self._calculation);self._calculation=None173 for point in history.rows(start_ns,end_ns):174 time,value,deadline,selected,selected_end=point[:5]175 for calculation in self._calculators:176 calculation.consume(((time,value,deadline),),**({} if condition is None else177 {'conditions':(selected,),'condition_valid_until_ns':(selected_end,)}),178 **({'actual_value_updates':(point[5],)} if getattr(calculation,'_samples',False) else {}))179 for request,calculation in zip(requests,self._calculators):180 self._finish_observation_tail(key,calculation,end_ns)181 result=calculation.finish()182 if request.operation=='select':183 calculation.close()184 continue185 if estimated:186 from .transform_partial import partial_result187 result=partial_result(result,request.partial_policy)188 elif partial and result.value is not None:189 result=replace(result,value=None if request.partial_policy=='strict' else result.value,190 status='indeterminate' if request.partial_policy=='strict' else 'partial',reason='partial_input')191 if request.scope.startswith('segment:') and request.minimum_coverage>0 and (192 parent_coverage is None or parent_coverage<request.minimum_coverage):193 result=replace(result,coverage_status='indeterminate')194 self._consumer(request,parent,result)195 calculation.close()196 self._calculators.clear()197198 def _estimated_selection(self,request,parent,row):199 """给部分父区间中的选中行标明估算。参数: 请求、父ID和行。返回: None。异常: 发布错误。"""200 from ..core.selection import partial_selection201 value,quality=partial_selection(row.value,row.quality,request)202 self._selection_consumer(request,parent,replace(row,value=value,quality=quality))203204 def _finish_observation_tail(self,key,calculation,end_ns):205 """用真实来源终点补查询尾段，既不补固定周期也不增加真实样本。206207 参数: key为共享表达式流，calculation为本次范围统计器，end_ns为查询右端。208 返回: None。异常: 表达式、预算或累计错误沿查询清理路径传播。209 只处理整个历史的最后一点；旧范围已有右邻点时绝不借用当前状态重新求值。210 """211 samples=getattr(calculation,'_samples',False)212 value,condition,history=self._streams[key][:3]213 if (not calculation._rows or not samples and calculation._step_ns is not None214 or calculation._last_time!=history.last_time or calculation._last_time>=end_ns):215 return216 sources={definition.source for expression in (value,condition) if expression is not None217 for _,definition in expression._definitions}218 ends=value._observation_ends219 if not sources or ends is None or any(source not in ends for source in sources):220 return221 start=calculation._last_time222 end=min(end_ns,max(ends[source] for source in sources))223 if end<=start:224 return225 # 此点是整个流的最新相关更新；EOF已知后重新核对右侧保持，保留TTL及短路条件含义。226 current,deadline=value.evaluate_at(start)227 selected,condition_deadline=(None,None) if condition is None else condition.evaluate_at(start)228 if samples:229 previous=calculation._rows[-1]230 value_ends=tuple(t for t in (previous[2],deadline,start if not current.valid else None) if t is not None)231 condition_ends=tuple(t for t in (previous[4],condition_deadline,232 start if selected is not None and not selected.valid else None) if t is not None)233 calculation._rows[-1]=(start,previous[1],min(value_ends) if value_ends else None,234 previous[3],min(condition_ends) if condition_ends else None)+previous[5:]235 else:236 calculation._rows[-1]=(start,current,deadline,selected,condition_deadline)237 calculation.consume(((end,current,deadline),),238 **({'actual_value_updates':(False,)} if samples else {}),239 **({} if condition is None else240 {'conditions':(selected,),'condition_valid_until_ns':(condition_deadline,)}))241242 def _flush(self,final=False):243 """右侧实际驱动到达才结算；EOF直接按已有点计算，不补周期。244245 参数: final只在真实EOF时为True。返回: None。异常: 原查询错，未退费用由close清理。246 """247 pending=[]248 try:249 while self._queue:250 item=heapq.heappop(self._queue)251 end,_,requests,parent,start,cost,estimated=item252 histories=(self._streams[self._request_stream[request.calculation_id]][2] for request in requests)253 if not final and any(history.last_time is None or history.last_time<end for history in histories):254 pending.append(item)255 continue256 if estimated:257 self._calculate(requests,parent,start,end,estimated=True)258 else:259 self._calculate(requests,parent,start,end)260 item=None261 _release_owned_ram(self,cost,'_queue_cost')262 except BaseException:263 self._failed=True264 raise265 finally:266 for item in pending:267 heapq.heappush(self._queue,item)268269 def finish(self,cursors):270 """来源全部完成后处理常量和剩余事件。参数: 实际来源游标。返回: None。异常: 原求值/计算错。"""271 self._open()272 starts=[c.observed_start for c in cursors if c.observed_start is not None]273 ends=[c.observed_end for c in cursors if c.observed_start is not None and c.observed_end is not None]274 if starts and ends:275 first,last=min(starts),max(ends)276 for value,condition,history,_,__,constant in self._streams.values():277 if constant:278 result,_=value.evaluate_at(last)279 selected=None if condition is None else condition.evaluate_at(last)[0]280 for time in dict.fromkeys((first,last)):281 history.append((time,result,None),None if condition is None else (time,selected,None))282 self._flush(final=True)283284 @property285 def closed(self):286 """返回关闭事实。参数: 无。返回: bool。异常: 无。"""287 return self._closed288289 def close(self):290 """按责任关闭子资源，失败时保槽供重试。参数: 无。返回: None。异常: 原关闭错。"""291 if self._closed:292 return293 for slot in ('_pending','_calculation'):294 owner=getattr(self,slot)295 if owner is not None:296 owner.close()297 setattr(self,slot,None)298 for owner in self._calculators:299 owner.close()300 self._calculators.clear()301 for stream in self._streams.values():302 for index in range(3):303 if stream[index] is not None:304 stream[index].close()305 stream[index]=None306 self._streams.clear()307 self._queue.clear()308 self._requests=self._plan=self._consumer=self._request_stream=None309 _release_owned_ram(self,self._queue_cost,'_queue_cost')310 _release_owned_ram(self,self._fixed_cost,'_fixed_cost')311 self._closed=True312