Ingestion, late events and replay

"""Original bounded teaching model. No Beam runner, network or durable state."""
import json
import re
import sys


def integer(value, low, high):
 if type(value) is not int or not low <= value <= high:
 raise ValueError('integer outside the declared contract')
 return value


def exact(value, keys):
 if type(value) is not dict or set(value)!= set(keys):
 raise ValueError('unexpected or missing fields')


def evaluate(document):
 exact(document, ['windowSeconds', 'allowedLatenessSeconds', 'events'])
 width = integer(document['windowSeconds'], 1, 86400)
 grace = integer(document['allowedLatenessSeconds'], 0, 86400)
 events = document['events']
 if type(events) is not list or len(events) > 10000:
 raise ValueError('events must be a list of at most 10000 operations')
 # Validate the whole input before producing a result.
 last_watermark = 0
 for event in events:
 if type(event) is dict and set(event) == {'watermark'}:
 last_watermark = integer(event['watermark'], last_watermark, 10**9)
 else:
 exact(event, ['id', 'eventTime', 'value'])
 if type(event['id']) is not str or not re.fullmatch(r'[A-Za-z0-9_-]{1,64}', event['id']):
 raise ValueError('invalid logical event id')
 integer(event['eventTime'], 0, 10**9)
 integer(event['value'], -10**6, 10**6)
 watermark = 0
 accepted = {}
 decisions = []
 for index, event in enumerate(events):
 if 'watermark' in event:
 watermark = event['watermark']
 decisions.append({'index': index, 'outcome': 'watermark', 'value': watermark})
 continue
 identifier = event['id']
 payload = (event['eventTime'], event['value'])
 end = (event['eventTime'] // width + 1) * width
 if identifier in accepted:
 outcome = 'duplicate' if accepted[identifier] == payload else 'conflict'
 elif watermark >= end + grace:
 outcome = 'too-late'
 else:
 accepted[identifier] = payload
 outcome = 'accepted-late' if watermark >= end else 'accepted'
 decisions.append({'index': index, 'id': identifier, 'outcome': outcome})
 windows = {}
 for event_time, value in accepted.values:
 start = event_time // width * width
 row = windows.setdefault(start, {'start': start, 'end': start + width, 'count': 0, 'sum': 0})
 row['count'] += 1
 row['sum'] += value
 for row in windows.values:
 row['closedByLocalPolicy'] = watermark >= row['end'] + grace
 return {'accepted': [{'id': key, 'eventTime': value[0], 'value': value[1]} for key, value in sorted(accepted.items)],
 'windows': [windows[key] for key in sorted(windows)], 'decisions': decisions,
 'watermark': watermark, 'stateLifetime': 'this invocation only',
 'sourceCompletenessProven': False, 'cloudSemanticsValidated': False,
 'productionReplayApproved': False}


def unique_object(pairs):
 result = {}
 for key, value in pairs:
 if key in result:
 raise ValueError('duplicate JSON field')
 result[key] = value
 return result


if __name__ == '__main__':
 try:
 raw = sys.stdin.read(2_000_001)
 if len(raw) > 2_000_000:
 raise ValueError('input exceeds local limit')
 result = evaluate(json.loads(raw, object_pairs_hook=unique_object))
 print(json.dumps(result, sort_keys=True))
 except (ValueError, TypeError, RecursionError) as error:
 print(json.dumps({'error': str(error)}), file=sys.stderr)
 sys.exit(2)

Related topics: Migration reconciliation · Data modeling · Observability and SLOs

Reference: Professional Data Engineer standard exam guide · Current linked standard guide (document title v4.2); edition date unconfirmed (2026-09-30 inspection)