o ~ i($@sdZddlmZddlZddlZddlmZmZddlmZddl Z ddl m Z e e e ZdZgdZd d ZGd d d ZdS) uSupabase orchestrator for SFT shard encoding pipeline. Table: sft_encoding_shards — one row per audio shard to process. Uses the same atomic FOR UPDATE SKIP LOCKED claim pattern as the pretraining pipeline to allow hundreds of GPU workers. ) annotationsN)datetimetimezone)Any) load_dotenvsft_encoding_shards)PENDINGCLAIMED PROCESSINGDONEFAILEDcCs"tdd}|s tdt|S)N DATABASE_URLzDATABASE_URL not set)osgetenv RuntimeErrorpsycopg2connect)db_urlr=/home/ubuntu/bench-codecs/codecbench/pipeline/sft_supabase.py _get_pg_conns  rc@seZdZdZddZd2d3d d Zd2d4ddZd5ddZd5ddZ d6d7ddZ d8d9d d!Z d:d;d&d'Z  (dd0d1Z dS)?SFTOrchestratorzCManages sft_encoding_shards table for distributed shard processing.cCsdSNrselfrrr__init__'szSFTOrchestrator.__init__rsqlstrparamstuplereturnNonecCsht}z+d|_|}|||Wdn1swYW|dSW|dS|wNT)r autocommitcursorexecutecloserrrconncurrrr _exec_sql*s  zSFTOrchestrator._exec_sql list[dict]cst}z9|&}|||dd|jDfdd|DWdW|S1s1wYW|dS|w)NcSg|]}|dqSrr.0drrr 8z*SFTOrchestrator._query..csg|] }tt|qSr)dictzip)r0rowcolsrrr29s)rr%r& descriptionfetchallr'r(rr7r_query3s   zSFTOrchestrator._querycCsF|dtdtdtdtdtdtdtdtd tdS) Nz$ CREATE TABLE IF NOT EXISTS az ( shard_key text PRIMARY KEY, dataset text NOT NULL, language text NOT NULL, segment_count integer, status text DEFAULT 'PENDING', claimed_by text, claimed_at timestamptz, segments_encoded integer, segments_failed integer, total_audio_s real, total_encode_ms real, output_r2_key text, error_detail text, started_at timestamptz, finished_at timestamptz, updated_at timestamptz DEFAULT now() ); CREATE INDEX IF NOT EXISTS idx_z _status ON z2 (status); CREATE INDEX IF NOT EXISTS idx_z _dataset ON z3 (dataset); CREATE INDEX IF NOT EXISTS idx_z _language ON z (language); zSFT table %s verified/createdr+TABLEloggerinforrrr ensure_tables=s  zSFTOrchestrator.ensure_tablescCsD|dtdtdtdtdtdtdtdtd d S) z*Atomic claim using FOR UPDATE SKIP LOCKED.z CREATE OR REPLACE FUNCTION claim_next_sft_shard( p_worker_id text, p_dataset text DEFAULT NULL, p_language text DEFAULT NULL ) RETURNS SETOF z) AS $$ DECLARE v_row z~; BEGIN IF p_dataset IS NOT NULL AND p_language IS NOT NULL THEN SELECT * INTO v_row FROM z WHERE status = 'PENDING' AND dataset = p_dataset AND language = p_language LIMIT 1 FOR UPDATE SKIP LOCKED; ELSIF p_dataset IS NOT NULL THEN SELECT * INTO v_row FROM z WHERE status = 'PENDING' AND dataset = p_dataset LIMIT 1 FOR UPDATE SKIP LOCKED; ELSIF p_language IS NOT NULL THEN SELECT * INTO v_row FROM z WHERE status = 'PENDING' AND language = p_language LIMIT 1 FOR UPDATE SKIP LOCKED; ELSE SELECT * INTO v_row FROM z WHERE status = 'PENDING' LIMIT 1 FOR UPDATE SKIP LOCKED; END IF; IF v_row IS NULL THEN RETURN; END IF; UPDATE ao SET status = 'CLAIMED', claimed_by = p_worker_id, claimed_at = now(), updated_at = now() WHERE shard_key = v_row.shard_key; v_row.status := 'CLAIMED'; v_row.claimed_by := p_worker_id; RETURN NEXT v_row; END; $$ LANGUAGE plpgsql; z Created claim_next_sft_shard RPCNr<rrrrcreate_claim_rpcXs   +z SFTOrchestrator.create_claim_rpcN worker_iddataset str | Nonelanguage dict | Nonec Cs&t}zz`d|_|O}|d|||fdd|jD}|}||rLtt||}t d|d|d|d|Wd WW| S Wd WW| d S1s^wYWn"t y} z| t d | WYd } ~ W| d Sd } ~ wwW| d S| w) z(Atomically claim the next PENDING shard.Fz.SELECT * FROM claim_next_sft_shard(%s, %s, %s)cSr-r.rr/rrrr2r3z/SFTOrchestrator.claim_shard..zClaimed shard %s (%s/%s) shard_keyrCrENzclaim_shard failed: %s)rr$r%r&r9fetchonecommitr4r5r>r?r' Exceptionrollbackerror) rrBrCrEr)r*r8r6sharderrr claim_shards<       zSFTOrchestrator.claim_shardrGstatusextradict[str, Any] | Nonec Csttj}ddg}||g}|dkr|d||n|dvr-|d|||rE|D]\}}||d||q3|||dtd d |d t |dS) Nz status = %szupdated_at = %sr zstarted_at = %s)r r zfinished_at = %sz = %szUPDATE z SET , z WHERE shard_key = %s) rnowrutc isoformatappenditemsr+r=joinr ) rrGrPrQrTsetsvalskvrrrupdate_shard_statuss$      z#SFTOrchestrator.update_shard_statusrows batch_sizeintc Cst}d}zud|_|`}tdt||D]O}||||}g}|D]} ||d| d| d| d| dfq$dt d d |d } | | |t|7}|d dkret d |t|qWdn1spwYW|n|wt d||S)z:Bulk insert shard rows. Skip conflicts (already ingested).rTz(%s, %s, %s, %s, 'PENDING')rGrCrE segment_countz INSERT INTO z> (shard_key, dataset, language, segment_count, status) VALUES ,z# ON CONFLICT (shard_key) DO NOTHINGizIngested %d / %d shardsNzIngested %d shards total)rr$r%rangelenrWmogrifygetdecoder=rYr&r>r?r') rr`rar)totalr*ibatchargsrrrrr ingest_shardss@     zSFTOrchestrator.ingest_shardsTstatuseslist[str] | None clear_outputsboolc Cs|durgd}gd}|r|gdt}z3d|_|}|dtdd|d |f|jp6d }Wdn1sAwYW|n|wt d |d ||S) zReset selected statuses back to PENDING for a clean rerun. Keeps static shard metadata (dataset/language/segment_count) but clears worker/runtime fields so workers can reclaim from scratch. N)r r r r )zstatus = 'PENDING'zclaimed_by = NULLzclaimed_at = NULLzstarted_at = NULLzfinished_at = NULLzupdated_at = now()zerror_detail = NULL)zsegments_encoded = NULLzsegments_failed = NULLztotal_audio_s = NULLztotal_encode_ms = NULLzoutput_r2_key = NULLTz UPDATE z SET rSz@ WHERE status = ANY(%s) rz-Reset %d shards back to PENDING (statuses=%s)rd) extendrr$r%r&r=rYrowcountr'r>r?)rrprrrZr)r*countrrr reset_pendings.     zSFTOrchestrator.reset_pendingdict[str, Any]cCs|dtd}|S)Nz SELECT status, dataset, count(*) as cnt, coalesce(sum(total_audio_s), 0) as audio_s FROM zS GROUP BY status, dataset ORDER BY dataset, status )r;r=)rr`rrr get_statss zSFTOrchestrator.get_stats)r)rrrr r!r")rrrr r!r,)r!r")NN)rBrrCrDrErDr!rFr)rGrrPrrQrRr!r")r_)r`r,rarbr!rbr#)rprqrrrsr!rb)r!rx)__name__ __module__ __qualname____doc__rr+r;r@rArOr^rorwryrrrrr$s  0   2r)r} __future__rloggingrrrtypingrrdotenvr getLoggerrzr>r= SHARD_STATESrrrrrrs