Add dummy snapshots to race with the sink's commits.
()
| 178 | stats = {"modifications": 0, "self_conflicts": 0, "errors": 0} |
| 179 | |
| 180 | def modify_table_loop() -> None: |
| 181 | """Add dummy snapshots to race with the sink's commits.""" |
| 182 | while not stop_event.is_set(): |
| 183 | try: |
| 184 | data = _polaris_get(table_url, access_token) |
| 185 | metadata = data["metadata"] |
| 186 | snap_id = metadata["current-snapshot-id"] |
| 187 | |
| 188 | snap = None |
| 189 | for s in metadata.get("snapshots", []): |
| 190 | if s["snapshot-id"] == snap_id: |
| 191 | snap = s |
| 192 | break |
| 193 | if snap is None: |
| 194 | continue |
| 195 | |
| 196 | # Dummy snapshot: same manifest-list and summary (including |
| 197 | # mz-frontier) so the sink's fencing check passes. |
| 198 | dummy_id = snap_id + 10_000_000 + stats["modifications"] |
| 199 | dummy = { |
| 200 | "snapshot-id": dummy_id, |
| 201 | "parent-snapshot-id": snap_id, |
| 202 | "timestamp-ms": int(time.time() * 1000), |
| 203 | "sequence-number": snap.get("sequence-number", 0) + 1, |
| 204 | "summary": snap["summary"], |
| 205 | "manifest-list": snap["manifest-list"], |
| 206 | "schema-id": snap.get("schema-id", 0), |
| 207 | } |
| 208 | |
| 209 | payload = json.dumps( |
| 210 | { |
| 211 | "requirements": [ |
| 212 | { |
| 213 | "type": "assert-ref-snapshot-id", |
| 214 | "ref": "main", |
| 215 | "snapshot-id": snap_id, |
| 216 | } |
| 217 | ], |
| 218 | "updates": [ |
| 219 | {"action": "add-snapshot", "snapshot": dummy}, |
| 220 | { |
| 221 | "action": "set-snapshot-ref", |
| 222 | "ref-name": "main", |
| 223 | "type": "branch", |
| 224 | "snapshot-id": dummy_id, |
| 225 | }, |
| 226 | ], |
| 227 | } |
| 228 | ) |
| 229 | |
| 230 | post_req = urllib.request.Request( |
| 231 | table_url, |
| 232 | data=payload.encode(), |
| 233 | headers={ |
| 234 | "Authorization": f"Bearer {access_token}", |
| 235 | "Content-Type": "application/json", |
| 236 | }, |
| 237 | method="POST", |
nothing calls this directly
no test coverage detected