Coverage for chalicelib/utils/exp_ch_helper.py: 16%

1import logging 

2import math 

3import re 

4import struct 

5from decimal import Decimal 

6from typing import Union, Any 

7 

8import schemas 

9from chalicelib.utils import sql_helper as sh 

10from chalicelib.utils.TimeUTC import TimeUTC 

11from schemas import SearchEventOperator 

12 

13logger = logging.getLogger(__name__) 

14 

15 

16def get_main_events_table(timestamp=0, platform="web"): 

17 if platform == "web": 17 ↛ 20line 17 didn't jump to line 20 because the condition on line 17 was always true

18 return "product_analytics.events" 

19 else: 

20 return "experimental.ios_events" 

21 

22 

23def get_main_sessions_table(timestamp=0): 

24 return "experimental.sessions" 

25 

26 

27def get_user_favorite_sessions_table(timestamp=0): 

28 return "experimental.user_favorite_sessions" 

29 

30 

31def get_user_viewed_sessions_table(timestamp=0): 

32 return "experimental.user_viewed_sessions" 

33 

34 

35def get_user_viewed_errors_table(timestamp=0): 

36 return "experimental.user_viewed_errors" 

37 

38 

39def get_main_js_errors_sessions_table(timestamp=0): 

40 return get_main_events_table(timestamp=timestamp) 

41 

42 

43def get_event_type(event_type: Union[schemas.EventType, schemas.PerformanceEventType], platform="web"): 

44 defs = { 

45 schemas.EventType.CLICK: "CLICK", 

46 schemas.EventType.INPUT: "INPUT", 

47 schemas.EventType.LOCATION: "LOCATION", 

48 schemas.PerformanceEventType.LOCATION_DOM_COMPLETE: "LOCATION", 

49 schemas.PerformanceEventType.LOCATION_LARGEST_CONTENTFUL_PAINT_TIME: "LOCATION", 

50 schemas.PerformanceEventType.LOCATION_TTFB: "LOCATION", 

51 schemas.EventType.CUSTOM: "CUSTOM", 

52 schemas.EventType.REQUEST: "REQUEST", 

53 schemas.EventType.REQUEST_DETAILS: "REQUEST", 

54 schemas.PerformanceEventType.FETCH_FAILED: "REQUEST", 

55 schemas.GraphqlFilterType.GRAPHQL_NAME: "GRAPHQL", 

56 schemas.EventType.STATE_ACTION: "STATEACTION", 

57 schemas.EventType.ERROR: "ERROR", 

58 schemas.PerformanceEventType.LOCATION_AVG_CPU_LOAD: 'PERFORMANCE', 

59 schemas.PerformanceEventType.LOCATION_AVG_MEMORY_USAGE: 'PERFORMANCE', 

60 schemas.FetchFilterType.FETCH_URL: 'REQUEST', 

61 schemas.EventType.INCIDENT: "INCIDENT", 

62 } 

63 defs_mobile = { 

64 schemas.EventType.CLICK_MOBILE: "TAP", 

65 schemas.EventType.INPUT_MOBILE: "INPUT", 

66 schemas.EventType.CUSTOM_MOBILE: "CUSTOM", 

67 schemas.EventType.REQUEST_MOBILE: "REQUEST", 

68 schemas.EventType.ERROR_MOBILE: "CRASH", 

69 schemas.EventType.VIEW_MOBILE: "VIEW", 

70 schemas.EventType.SWIPE_MOBILE: "SWIPE", 

71 schemas.EventType.INCIDENT: "INCIDENT" 

72 } 

73 if platform != "web" and event_type in defs_mobile: 

74 return defs_mobile.get(event_type) 

75 if event_type.lower() not in defs: 

76 raise Exception(f"unsupported EventType:{event_type}") 

77 return defs.get(event_type.lower()) 

78 

79 

80# AI generated 

81def simplify_clickhouse_type(ch_type: str) -> str: 

82 """ 

83 Simplify a ClickHouse data type name to a broader category like: 

84 int, float, decimal, datetime, string, uuid, enum, array, tuple, map, nested, etc. 

85 """ 

86 

87 # 1) Strip out common wrappers like Nullable(...) or LowCardinality(...) 

88 # Possibly multiple wrappers: e.g. "LowCardinality(Nullable(Int32))" 

89 pattern_wrappers = re.compile(r'(Nullable|LowCardinality)\((.*)\)') 

90 while True: 

91 match = pattern_wrappers.match(ch_type) 

92 if match: 

93 ch_type = match.group(2) 

94 else: 

95 break 

96 

97 # 2) Normalize (lowercase) for easier checks 

98 normalized_type = ch_type.lower() 

99 

100 # 3) Use pattern matching or direct checks for known categories 

101 # (You can adapt this as you see fit for your environment.) 

102 

103 # Integers: Int8, Int16, Int32, Int64, Int128, Int256, UInt8, UInt16, ... 

104 if re.match(r'^(u?int)(8|16|32|64|128|256)$', normalized_type): 

105 return "int" 

106 

107 # Floats: Float32, Float64 

108 if re.match(r'^float(32|64)|double$', normalized_type): 

109 return "float" 

110 

111 # Decimal: Decimal(P, S) 

112 if normalized_type.startswith("decimal"): 

113 # return "decimal" 

114 return "float" 

115 

116 # Date/DateTime 

117 if normalized_type.startswith("date"): 

118 return "datetime" 

119 if normalized_type.startswith("datetime"): 

120 return "datetime" 

121 

122 # Strings: String, FixedString(N) 

123 if normalized_type.startswith("string"): 

124 return "string" 

125 if normalized_type.startswith("fixedstring"): 

126 return "string" 

127 

128 # UUID 

129 if normalized_type.startswith("uuid"): 

130 # return "uuid" 

131 return "string" 

132 

133 # Enums: Enum8(...) or Enum16(...) 

134 if normalized_type.startswith("enum8") or normalized_type.startswith("enum16"): 

135 # return "enum" 

136 return "string" 

137 

138 # Arrays: Array(T) 

139 if normalized_type.startswith("array"): 

140 return "array" 

141 

142 # Tuples: Tuple(T1, T2, ...) 

143 if normalized_type.startswith("tuple"): 

144 return "tuple" 

145 

146 # Map(K, V) 

147 if normalized_type.startswith("map"): 

148 return "map" 

149 

150 # Nested(...) 

151 if normalized_type.startswith("nested"): 

152 return "nested" 

153 

154 # If we didn't match above, just return the original type in lowercase 

155 return normalized_type 

156 

157 

158def simplify_clickhouse_types(ch_types: list[str]) -> list[str]: 

159 """ 

160 Takes a list of ClickHouse types and returns a list of simplified types 

161 by calling `simplify_clickhouse_type` on each. 

162 """ 

163 return list(set([simplify_clickhouse_type(t) for t in ch_types])) 

164 

165 

166def get_sub_condition(col_name: str, val_name: str, 

167 operator: Union[schemas.SearchEventOperator, schemas.MathOperator]) -> str: 

168 if operator == SearchEventOperator.PATTERN: 

169 return f"match({col_name}, %({val_name})s)" 

170 op = sh.get_sql_operator(operator) 

171 return f"{col_name} {op} %({val_name})s" 

172 

173 

174def get_col_cast(data_type: schemas.PropertyType, value: Any) -> str: 

175 if value is None or len(value) == 0: 

176 return "" 

177 if isinstance(value, list): 

178 value = value[0] 

179 if data_type in (schemas.PropertyType.INT, schemas.PropertyType.FLOAT): 

180 return best_clickhouse_type(value) 

181 return data_type.capitalize() 

182 

183 

184# (type_name, minimum, maximum) – ordered by increasing size 

185_INT_RANGES = [ 

186 ("Int8", -128, 127), 

187 ("UInt8", 0, 255), 

188 ("Int16", -32_768, 32_767), 

189 ("UInt16", 0, 65_535), 

190 ("Int32", -2_147_483_648, 2_147_483_647), 

191 ("UInt32", 0, 4_294_967_295), 

192 ("Int64", -9_223_372_036_854_775_808, 9_223_372_036_854_775_807), 

193 ("UInt64", 0, 18_446_744_073_709_551_615), 

194] 

195 

196 

197def best_clickhouse_type(value): 

198 """ 

199 Return the most compact ClickHouse numeric type that can store *value* loss-lessly. 

200 

201 """ 

202 # Treat bool like tiny int 

203 if isinstance(value, bool): 

204 value = int(value) 

205 

206 # --- Integers --- 

207 if isinstance(value, int): 

208 for name, lo, hi in _INT_RANGES: 

209 if lo <= value <= hi: 

210 return name 

211 # Beyond UInt64: ClickHouse offers Int128 / Int256 or Decimal 

212 return "Int128" 

213 

214 # --- Decimal.Decimal (exact) --- 

215 if isinstance(value, Decimal): 

216 # ClickHouse Decimal32/64/128 have 9 / 18 / 38 significant digits. 

217 digits = len(value.as_tuple().digits) 

218 if digits <= 9: 

219 return "Decimal32" 

220 elif digits <= 18: 

221 return "Decimal64" 

222 else: 

223 return "Decimal128" 

224 

225 # --- Floats --- 

226 if isinstance(value, float): 

227 if not math.isfinite(value): 

228 return "Float64" # inf / nan → always Float64 

229 

230 # Check if a round-trip through 32-bit float preserves the bit pattern 

231 packed = struct.pack("f", value) 

232 if struct.unpack("f", packed)[0] == value: 

233 return "Float32" 

234 return "Float64" 

235 

236 raise TypeError(f"Unsupported type: {type(value).__name__}") 

237 

238 

239def explode_dproperties(rows): 

240 for i in range(len(rows)): 

241 rows[i] = {**rows[i], **rows[i]["$properties"]} 

242 rows[i].pop("$properties") 

243 return rows 

244 

245 

246def add_timestamp(rows): 

247 for row in rows: 

248 row["timestamp"] = TimeUTC.datetime_to_timestamp(row["createdAt"]) 

249 return rows