|
|
@@ -0,0 +1,348 @@
|
|
|
+-- A transactionally maintained projection index, validated against CURRENT source
|
|
|
+-- dependencies at every read. It never defines row existence or financial authority.
|
|
|
+-- Writes are best effort and never require a company-wide lock. Missing/stale rows
|
|
|
+-- are rebuilt in the reader's snapshot; extra rows cannot become payable sources.
|
|
|
+BEGIN;
|
|
|
+CREATE TABLE IF NOT EXISTS "PayrollSourceIndex" (
|
|
|
+ kind text NOT NULL,company text NOT NULL,"objectId" text NOT NULL,
|
|
|
+ row_data jsonb NOT NULL,row_text text NOT NULL,input_signature text,
|
|
|
+ generation text NOT NULL,selection jsonb NOT NULL DEFAULT '{}'::jsonb,PRIMARY KEY(kind,company,"objectId")
|
|
|
+);
|
|
|
+ALTER TABLE "PayrollSourceIndex" ADD COLUMN IF NOT EXISTS selection jsonb NOT NULL DEFAULT '{}'::jsonb;
|
|
|
+CREATE TABLE IF NOT EXISTS "PayrollSourceIndexMeta" (
|
|
|
+ singleton boolean PRIMARY KEY DEFAULT true CHECK(singleton),generation text NOT NULL,
|
|
|
+ signature text NOT NULL,valid boolean NOT NULL DEFAULT false
|
|
|
+);
|
|
|
+REVOKE ALL ON "PayrollSourceIndex","PayrollSourceIndexMeta" FROM PUBLIC;
|
|
|
+-- Only model 59 participates in canonical source selection; avoid scanning the
|
|
|
+-- much larger combined CommonModel digest for models 54 and 60.
|
|
|
+CREATE INDEX IF NOT EXISTS payroll_source_index_common59_digest
|
|
|
+ ON "PayrollReadDigest" (company,"objectId") INCLUDE(hash) WHERE kind='common' AND model=59;
|
|
|
+
|
|
|
+-- Conservative selectors reproduce the gateway transport rule. Unknown dates
|
|
|
+-- remain candidates. Identity keys retain the existing SQL COALESCE text semantics.
|
|
|
+CREATE OR REPLACE FUNCTION xs_payroll_source_index_metadata(data jsonb) RETURNS jsonb LANGUAGE sql STABLE AS $$
|
|
|
+ SELECT jsonb_build_object('standard_key',data->>'standardLessonKey',
|
|
|
+ 'month_candidate_key',CASE WHEN data->>'lessonAt' ~ '^\d{4}-[0-9]{2}-[0-9]{2}T([01][0-9]|2[0-3]):[0-5][0-9]:[0-5][0-9](\.[0-9]{1,3})?(Z|[+-][0-9]{2}:[0-9]{2})$'
|
|
|
+ AND xs_payroll_time(data->>'lessonAt') IS NOT NULL THEN to_char(xs_payroll_time(data->>'lessonAt') AT TIME ZONE 'Asia/Shanghai','YYYY-MM') END,
|
|
|
+ 'scope_data',jsonb_build_array(data->'coachId',data->'jsmz',data->'storeId',data->'szmdid'),
|
|
|
+ 'coach_key',COALESCE(data->>'coachId',data->>'jsmz'),
|
|
|
+ 'appointment_key',COALESCE(data->>'appointmentGeneralId',data->>'yyds'));
|
|
|
+$$;
|
|
|
+
|
|
|
+CREATE OR REPLACE FUNCTION xs_payroll_source_index_signature() RETURNS text LANGUAGE sql STABLE AS $$
|
|
|
+ SELECT md5(COALESCE((SELECT string_agg(c.oid||':'||c.relfilenode||':'||a.attnum||':'||a.attname||':'||a.atttypid||':'||a.atttypmod,'|' ORDER BY c.oid,a.attnum)
|
|
|
+ FROM pg_class c JOIN pg_namespace n ON n.oid=c.relnamespace JOIN pg_attribute a ON a.attrelid=c.oid
|
|
|
+ WHERE n.nspname='public' AND a.attnum>0 AND NOT a.attisdropped
|
|
|
+ AND c.relname IN ('CommonModel','CourseAppointment','LessonRecord','LessonRecordFact','PayrollSourceIndex','PayrollSourceIndexMeta')),'')||'|'||
|
|
|
+ COALESCE((SELECT string_agg(t.tgrelid||':'||t.tgname||':'||t.xmin||':'||t.ctid||':'||t.tgenabled::text||':'||pg_get_triggerdef(t.oid),'|' ORDER BY t.tgrelid,t.tgname)
|
|
|
+ FROM pg_trigger t WHERE t.tgrelid IN ('"CommonModel"'::regclass,'"CourseAppointment"'::regclass,'"LessonRecord"'::regclass,'"LessonRecordFact"'::regclass,'"PayrollSourceIndexMeta"'::regclass) AND t.tgname IN ('xs_payroll_source_index_row','xs_payroll_source_index_truncate','xs_payroll_source_index_meta_guard')),'')||'|'||
|
|
|
+ COALESCE((SELECT string_agg(pg_get_functiondef(p.oid),'|' ORDER BY p.oid) FROM pg_proc p JOIN pg_namespace n ON n.oid=p.pronamespace
|
|
|
+ WHERE n.nspname='public' AND p.prokind='f' AND (p.proname LIKE 'xs_payroll_source_index_%' OR p.proname IN ('xs_payroll_source_entries','xs_payroll_source_entries_scoped','xs_payroll_sources_indexed','xs_payroll_source_version_indexed','xs_payroll_time'))),''));
|
|
|
+$$;
|
|
|
+CREATE OR REPLACE FUNCTION xs_payroll_source_index_ready() RETURNS boolean LANGUAGE sql STABLE AS $$
|
|
|
+ SELECT xs_payroll_authority_ready() AND xs_payroll_digest_ready()
|
|
|
+ AND COALESCE((SELECT valid AND signature=xs_payroll_source_index_signature() FROM "PayrollSourceIndexMeta" WHERE singleton),false)
|
|
|
+ AND (SELECT count(*) FROM pg_trigger WHERE tgrelid IN ('"CommonModel"'::regclass,'"CourseAppointment"'::regclass,'"LessonRecord"'::regclass,'"LessonRecordFact"'::regclass)
|
|
|
+ AND tgname IN ('xs_payroll_source_index_row','xs_payroll_source_index_truncate') AND tgenabled='A')=8;
|
|
|
+$$;
|
|
|
+CREATE OR REPLACE FUNCTION xs_payroll_source_index_meta_guard() RETURNS trigger LANGUAGE plpgsql AS $$
|
|
|
+BEGIN
|
|
|
+ IF NEW.valid AND (TG_OP='INSERT' OR NEW IS DISTINCT FROM OLD) AND current_setting('xiaoshu.source_index_reseed',true) IS DISTINCT FROM pg_current_xact_id()::text THEN
|
|
|
+ RAISE EXCEPTION 'source index requires a full reseed and new generation';
|
|
|
+ END IF;
|
|
|
+ RETURN NEW;
|
|
|
+END $$;
|
|
|
+DROP TRIGGER IF EXISTS xs_payroll_source_index_meta_guard ON "PayrollSourceIndexMeta";
|
|
|
+CREATE TRIGGER xs_payroll_source_index_meta_guard BEFORE INSERT OR UPDATE ON "PayrollSourceIndexMeta" FOR EACH ROW EXECUTE FUNCTION xs_payroll_source_index_meta_guard();
|
|
|
+ALTER TABLE "PayrollSourceIndexMeta" ENABLE ALWAYS TRIGGER xs_payroll_source_index_meta_guard;
|
|
|
+
|
|
|
+-- Only a single target source is projected. Hashes and output are produced by the
|
|
|
+-- same SELECT snapshot, never by a digest row which another trigger may update later.
|
|
|
+CREATE OR REPLACE FUNCTION xs_payroll_source_index_target(co text,source_kind text,source_id text)
|
|
|
+RETURNS TABLE(row_data jsonb,row_text text,input_signature text) LANGUAGE plpgsql VOLATILE SET jit=off AS $$
|
|
|
+BEGIN
|
|
|
+ IF source_kind='lesson' THEN
|
|
|
+ RETURN QUERY WITH projected AS MATERIALIZED (SELECT
|
|
|
+(to_jsonb(l)-'kzsj')||jsonb_build_object(
|
|
|
+ 'lessonObjectId',l."objectId",'legacyGeneralId',COALESCE(c."generalId",l."legacyGeneralId",0),
|
|
|
+ 'standardLessonKey',CASE WHEN COALESCE(c."generalId",l."legacyGeneralId",0)>0 THEN 'legacy:'||COALESCE(c."generalId",l."legacyGeneralId")::text ELSE 'lesson:'||l."objectId" END,
|
|
|
+ 'source','canonical','commonStatus',COALESCE(c."status",99),
|
|
|
+ 'lessonAt',COALESCE(l."lessonAt",xs_payroll_time(l."plpjsj"),xs_payroll_time(a."jssj"),l."sourceCreatedAt",c."createdAt"),
|
|
|
+ 'courseCategoryKey',COALESCE(NULLIF(l."courseCategoryKey",''),a."courseCategoryKey"),
|
|
|
+ 'courseCategoryName',COALESCE(NULLIF(l."courseCategoryName",''),a."courseCategoryName"),
|
|
|
+ 'teacherPaySnapshot',COALESCE(l."teacherPaySnapshot",a."teacherPaySnapshot"),
|
|
|
+ 'durationMinutes',a."durationMinutes",'deliveryMode',COALESCE(NULLIF(l."jffs",''),a."deliveryMode",a."jffs"),
|
|
|
+ 'teachingRuleId',a."teachingRuleId",'studentName',COALESCE(NULLIF(l."studentName",''),c."title"),
|
|
|
+ 'courseName',COALESCE(NULLIF(l."courseName",''),NULLIF(l."kcmc",''),c."subtitle")
|
|
|
+ ) AS data,
|
|
|
+ md5(jsonb_build_array(md5(l::text),CASE WHEN c."objectId" IS NOT NULL THEN md5(c::text) END,CASE WHEN a."objectId" IS NOT NULL THEN md5(a::text) END)::text) AS signature
|
|
|
+
|
|
|
+ FROM "LessonRecord" l
|
|
|
+ LEFT JOIN LATERAL (SELECT * FROM "CommonModel" c WHERE c."company"=co AND c."modelId"=59 AND c."itemId"=l."id" ORDER BY c."objectId" LIMIT 1) c ON true
|
|
|
+ LEFT JOIN LATERAL (SELECT a.* FROM "CourseAppointment" a JOIN "CommonModel" ac ON ac."company"=co AND ac."modelId"=54 AND ac."itemId"=a."id" WHERE a."company"=co AND ac."generalId"::text=l."yyds"::text ORDER BY ac."objectId" LIMIT 1) a ON true
|
|
|
+ WHERE l."company"=co AND l."objectId"=source_id) SELECT p.data,p.data::text,p.signature FROM projected p;
|
|
|
+ ELSIF source_kind='fact' THEN
|
|
|
+ RETURN QUERY WITH projected AS MATERIALIZED (SELECT
|
|
|
+(to_jsonb(f)-'kzsj')||jsonb_build_object('lessonObjectId',f."objectId",'source','historical-fact','commonStatus',99,'courseCategoryKey',a."courseCategoryKey",'courseCategoryName',a."courseCategoryName",'teacherPaySnapshot',a."teacherPaySnapshot",'durationMinutes',a."durationMinutes",'deliveryMode',COALESCE(NULLIF(f."jffs",''),a."deliveryMode",a."jffs"),'teachingRuleId',a."teachingRuleId",
|
|
|
+ 'standardLessonKey',CASE WHEN COALESCE(f."legacyGeneralId",0)>0 THEN 'legacy:'||f."legacyGeneralId"::text ELSE 'fact:'||f."objectId" END) AS data,
|
|
|
+ md5(jsonb_build_array(md5(f::text),CASE WHEN a."objectId" IS NOT NULL THEN md5(a::text) END)::text) AS signature
|
|
|
+
|
|
|
+ FROM "LessonRecordFact" f LEFT JOIN LATERAL (SELECT a.* FROM "CourseAppointment" a JOIN "CommonModel" ac ON ac."company"=co AND ac."modelId"=54 AND ac."itemId"=a."id" WHERE a."company"=co AND ac."generalId"::text=f."yyds"::text ORDER BY ac."objectId" LIMIT 1) a ON true WHERE f."company"=co AND f."objectId"=source_id) SELECT p.data,p.data::text,p.signature FROM projected p;
|
|
|
+ END IF;
|
|
|
+END $$;
|
|
|
+
|
|
|
+CREATE OR REPLACE FUNCTION xs_payroll_source_index_put(co text,source_kind text,source_id text) RETURNS text LANGUAGE plpgsql VOLATILE SET jit=off AS $$
|
|
|
+DECLARE projected record; current_generation text;
|
|
|
+BEGIN
|
|
|
+ IF co IS NULL OR source_id IS NULL OR NOT pg_try_advisory_xact_lock(hashtext('payroll-source-index'),hashtext(co||':'||source_kind||':'||source_id)) THEN RETURN NULL;END IF;
|
|
|
+ SELECT generation INTO current_generation FROM "PayrollSourceIndexMeta" WHERE singleton;
|
|
|
+ SELECT * INTO projected FROM xs_payroll_source_index_target(co,source_kind,source_id);
|
|
|
+ IF NOT FOUND THEN DELETE FROM "PayrollSourceIndex" WHERE company=co AND kind=source_kind AND "objectId"=source_id;RETURN NULL;END IF;
|
|
|
+ INSERT INTO "PayrollSourceIndex"(kind,company,"objectId",row_data,row_text,input_signature,generation,selection) VALUES(source_kind,co,source_id,projected.row_data,projected.row_text,projected.input_signature,current_generation,xs_payroll_source_index_metadata(projected.row_data))
|
|
|
+ ON CONFLICT(kind,company,"objectId") DO UPDATE SET row_data=EXCLUDED.row_data,row_text=EXCLUDED.row_text,input_signature=EXCLUDED.input_signature,generation=EXCLUDED.generation,selection=EXCLUDED.selection;
|
|
|
+ RETURN projected.row_data->>'legacyGeneralId';
|
|
|
+END $$;
|
|
|
+-- Canonical changes alter fact visibility, not a fact's projection. The current
|
|
|
+-- reader decides membership; keeping hidden fact entries is safe and avoids
|
|
|
+-- rewriting historical facts for every lesson/comment update.
|
|
|
+CREATE OR REPLACE FUNCTION xs_payroll_source_index_lesson(co text,source_id text) RETURNS void LANGUAGE plpgsql VOLATILE SET jit=off AS $$
|
|
|
+BEGIN PERFORM xs_payroll_source_index_put(co,'lesson',source_id);END $$;
|
|
|
+CREATE OR REPLACE FUNCTION xs_payroll_source_index_appointment(co text,appointment_key text) RETURNS void LANGUAGE plpgsql VOLATILE SET jit=off AS $$
|
|
|
+DECLARE r record;
|
|
|
+BEGIN
|
|
|
+ FOR r IN SELECT 'lesson' AS kind,"objectId" FROM "LessonRecord" WHERE company=co AND yyds::text=appointment_key
|
|
|
+ UNION ALL SELECT 'fact',"objectId" FROM "LessonRecordFact" WHERE company=co AND yyds::text=appointment_key ORDER BY kind,"objectId" LOOP
|
|
|
+ PERFORM xs_payroll_source_index_put(co,r.kind,r."objectId");
|
|
|
+ END LOOP;
|
|
|
+END $$;
|
|
|
+CREATE OR REPLACE FUNCTION xs_payroll_source_index_row() RETURNS trigger LANGUAGE plpgsql
|
|
|
+ SET TimeZone='Etc/GMT-8' SET DateStyle='ISO, MDY' SET IntervalStyle='postgres' SET extra_float_digits=1 SET bytea_output='hex' AS $$
|
|
|
+DECLARE item jsonb; r record; changes jsonb[]:=ARRAY[]::jsonb[]; seen text[]:=ARRAY[]::text[]; changed_key text;
|
|
|
+BEGIN
|
|
|
+ -- Review/model-60 synchronization can touch hundreds of thousands of rows.
|
|
|
+ -- It cannot affect lesson projections: skip before catalog/meta reads or JSON.
|
|
|
+ IF TG_TABLE_NAME='CommonModel' THEN
|
|
|
+ IF TG_OP='INSERT' AND NOT COALESCE(NEW."modelId" IN (54,59),false) THEN RETURN NULL;END IF;
|
|
|
+ IF TG_OP='DELETE' AND NOT COALESCE(OLD."modelId" IN (54,59),false) THEN RETURN NULL;END IF;
|
|
|
+ IF TG_OP='UPDATE' AND NOT COALESCE(OLD."modelId" IN (54,59),false) AND NOT COALESCE(NEW."modelId" IN (54,59),false) THEN RETURN NULL;END IF;
|
|
|
+ END IF;
|
|
|
+ IF NOT EXISTS(SELECT 1 FROM "PayrollSourceIndexMeta" WHERE singleton AND valid) THEN RETURN NULL;END IF;
|
|
|
+ IF TG_OP<>'INSERT' THEN changes:=array_append(changes,to_jsonb(OLD));END IF;
|
|
|
+ IF TG_OP<>'DELETE' THEN changes:=array_append(changes,to_jsonb(NEW));END IF;
|
|
|
+ FOREACH item IN ARRAY changes LOOP
|
|
|
+ IF item->>'company' IS NULL THEN CONTINUE;END IF;
|
|
|
+ changed_key:=CASE WHEN TG_TABLE_NAME IN ('LessonRecord','LessonRecordFact') THEN jsonb_build_array(item->>'company',item->>'objectId')::text
|
|
|
+ WHEN TG_TABLE_NAME='CourseAppointment' THEN jsonb_build_array(item->>'company',item->>'id')::text
|
|
|
+ WHEN (item->>'modelId')::numeric=59 THEN jsonb_build_array(item->>'company','59',item->>'itemId')::text
|
|
|
+ ELSE jsonb_build_array(item->>'company',item->>'modelId',item->>'generalId')::text END;
|
|
|
+ IF changed_key=ANY(seen) THEN CONTINUE;END IF;
|
|
|
+ seen:=array_append(seen,changed_key);
|
|
|
+ IF TG_TABLE_NAME='LessonRecord' THEN PERFORM xs_payroll_source_index_lesson(item->>'company',item->>'objectId');
|
|
|
+ ELSIF TG_TABLE_NAME='LessonRecordFact' THEN PERFORM xs_payroll_source_index_put(item->>'company','fact',item->>'objectId');
|
|
|
+ ELSIF TG_TABLE_NAME='CommonModel' AND (item->>'modelId')::numeric=59 THEN
|
|
|
+ FOR r IN SELECT "objectId" FROM "LessonRecord" WHERE company=item->>'company' AND id=(item->>'itemId')::numeric ORDER BY "objectId" LOOP
|
|
|
+ PERFORM xs_payroll_source_index_lesson(item->>'company',r."objectId");
|
|
|
+ END LOOP;
|
|
|
+ ELSIF TG_TABLE_NAME='CommonModel' AND (item->>'modelId')::numeric=54 THEN
|
|
|
+ PERFORM xs_payroll_source_index_appointment(item->>'company',item->>'generalId');
|
|
|
+ ELSIF TG_TABLE_NAME='CourseAppointment' THEN
|
|
|
+ FOR r IN SELECT DISTINCT "generalId"::text AS key FROM "CommonModel" WHERE company=item->>'company' AND "modelId"=54 AND "itemId"=(item->>'id')::numeric ORDER BY key LOOP
|
|
|
+ PERFORM xs_payroll_source_index_appointment(item->>'company',r.key);
|
|
|
+ END LOOP;
|
|
|
+ END IF;
|
|
|
+ END LOOP;
|
|
|
+ RETURN NULL;
|
|
|
+ -- Projection upkeep must not prevent a real source write. The reader verifies
|
|
|
+ -- every signature and recomputes missing/stale rows in its own current snapshot.
|
|
|
+EXCEPTION WHEN OTHERS THEN RETURN NULL;
|
|
|
+END $$;
|
|
|
+CREATE OR REPLACE FUNCTION xs_payroll_source_index_truncate() RETURNS trigger LANGUAGE plpgsql AS $$
|
|
|
+BEGIN UPDATE "PayrollSourceIndexMeta" SET valid=false WHERE singleton;RETURN NULL;END $$;
|
|
|
+
|
|
|
+CREATE OR REPLACE FUNCTION xs_payroll_source_index_scan(co text,reuse_index boolean,include_data boolean,mon text,coach_id text,appointment_ids text[],with_text boolean)
|
|
|
+RETURNS TABLE(kind text,"objectId" text,row_data jsonb,row_text text,input_signature text,ordinal bigint,standard_key text,month_candidate_key text,scope_data jsonb)
|
|
|
+LANGUAGE sql STABLE AS $$
|
|
|
+ WITH common59 AS MATERIALIZED (
|
|
|
+ SELECT DISTINCT ON (c."itemId") c.* FROM "CommonModel" c
|
|
|
+ WHERE c.company=co AND c."modelId"=59
|
|
|
+ ORDER BY c."itemId",c."objectId"
|
|
|
+ ), appointment_map AS MATERIALIZED (
|
|
|
+ -- Join before choosing: the earliest CommonModel can refer to a missing appointment.
|
|
|
+ SELECT DISTINCT ON (ac."generalId"::text) ac."generalId"::text AS join_key,a AS appointment
|
|
|
+ FROM "CommonModel" ac JOIN "CourseAppointment" a ON a.company=co AND a.id=ac."itemId"
|
|
|
+ WHERE ac.company=co AND ac."modelId"=54
|
|
|
+ ORDER BY ac."generalId"::text,ac."objectId"
|
|
|
+ ), lessons AS MATERIALIZED (
|
|
|
+ SELECT l AS lesson,row_number() OVER () AS ord FROM "LessonRecord" l WHERE l.company=co
|
|
|
+ ), canonical_info AS MATERIALIZED (
|
|
|
+ SELECT l.lesson,l.ord,c AS common,
|
|
|
+ to_jsonb(COALESCE(c."generalId",(l.lesson)."legacyGeneralId",0))#>>'{}' AS legacy_text,
|
|
|
+ CASE WHEN COALESCE(c."generalId",(l.lesson)."legacyGeneralId",0)>0
|
|
|
+ THEN 'legacy:'||COALESCE(c."generalId",(l.lesson)."legacyGeneralId")::text
|
|
|
+ ELSE 'lesson:'||(l.lesson)."objectId" END AS standard_key
|
|
|
+ FROM lessons l LEFT JOIN common59 c ON c."itemId"=(l.lesson).id
|
|
|
+ ), canonical_keys AS MATERIALIZED (
|
|
|
+ SELECT legacy_text FROM canonical_info
|
|
|
+ ), facts AS MATERIALIZED (
|
|
|
+ SELECT f AS fact,row_number() OVER () AS ord,
|
|
|
+ CASE WHEN COALESCE(f."legacyGeneralId",0)>0 THEN 'legacy:'||f."legacyGeneralId"::text ELSE 'fact:'||f."objectId" END AS standard_key
|
|
|
+ FROM "LessonRecordFact" f WHERE f.company=co AND NOT EXISTS (
|
|
|
+ SELECT 1 FROM canonical_keys k WHERE k.legacy_text=f."legacyGeneralId"::text AND COALESCE(f."legacyGeneralId",0)>0
|
|
|
+ )
|
|
|
+ ), ambiguous AS MATERIALIZED (
|
|
|
+ SELECT EXISTS (SELECT 1 FROM (
|
|
|
+ SELECT standard_key FROM canonical_info UNION ALL SELECT standard_key FROM facts
|
|
|
+ ) k GROUP BY standard_key HAVING count(*)>1) AS value
|
|
|
+ ), canonical_dependencies AS MATERIALIZED (
|
|
|
+ SELECT x.*,am.appointment,
|
|
|
+ md5(jsonb_build_array(COALESCE(ld.hash,md5((x.lesson)::text)),CASE WHEN (x.common)."objectId" IS NOT NULL THEN COALESCE(cd.hash,md5((x.common)::text)) END,CASE WHEN (am.appointment)."objectId" IS NOT NULL THEN COALESCE(ad.hash,md5((am.appointment)::text)) END)::text) AS signature
|
|
|
+ FROM canonical_info x LEFT JOIN appointment_map am ON am.join_key=(x.lesson).yyds::text
|
|
|
+ LEFT JOIN "PayrollReadDigest" ld ON ld.kind='lesson' AND ld.company=co AND ld."objectId"=(x.lesson)."objectId"
|
|
|
+ LEFT JOIN "PayrollReadDigest" cd ON cd.kind='common' AND cd.model=59 AND cd.company=co AND cd."objectId"=(x.common)."objectId"
|
|
|
+ LEFT JOIN "PayrollReadDigest" ad ON ad.kind='appointment' AND ad.company=co AND ad."objectId"=(am.appointment)."objectId"
|
|
|
+ ), fact_dependencies AS MATERIALIZED (
|
|
|
+ SELECT x.*,am.appointment,
|
|
|
+ md5(jsonb_build_array(md5((x.fact)::text),CASE WHEN (am.appointment)."objectId" IS NOT NULL THEN COALESCE(ad.hash,md5((am.appointment)::text)) END)::text) AS signature
|
|
|
+ FROM facts x LEFT JOIN appointment_map am ON am.join_key=(x.fact).yyds::text
|
|
|
+ LEFT JOIN "PayrollReadDigest" ad ON ad.kind='appointment' AND ad.company=co AND ad."objectId"=(am.appointment)."objectId"
|
|
|
+ ), projected AS (
|
|
|
+ SELECT 'lesson'::text AS kind,l."objectId",0 AS source_group,x.ord,x.signature,
|
|
|
+ CASE WHEN include_data AND (mon IS NULL OR m.selection->>'month_candidate_key' IS NULL OR m.selection->>'month_candidate_key'=mon) AND (coach_id IS NULL OR m.selection->>'coach_key' IS NULL OR m.selection->>'coach_key'=coach_id) AND (appointment_ids IS NULL OR m.selection->>'appointment_key' IS NULL OR m.selection->>'appointment_key'=ANY(appointment_ids)) THEN CASE WHEN i.input_signature=x.signature THEN i.row_data ELSE
|
|
|
+(to_jsonb(l)-'kzsj')||jsonb_build_object(
|
|
|
+ 'lessonObjectId',l."objectId",'legacyGeneralId',COALESCE(c."generalId",l."legacyGeneralId",0),
|
|
|
+ 'standardLessonKey',CASE WHEN COALESCE(c."generalId",l."legacyGeneralId",0)>0 THEN 'legacy:'||COALESCE(c."generalId",l."legacyGeneralId")::text ELSE 'lesson:'||l."objectId" END,
|
|
|
+ 'source','canonical','commonStatus',COALESCE(c."status",99),
|
|
|
+ 'lessonAt',COALESCE(l."lessonAt",xs_payroll_time(l."plpjsj"),xs_payroll_time(a."jssj"),l."sourceCreatedAt",c."createdAt"),
|
|
|
+ 'courseCategoryKey',COALESCE(NULLIF(l."courseCategoryKey",''),a."courseCategoryKey"),
|
|
|
+ 'courseCategoryName',COALESCE(NULLIF(l."courseCategoryName",''),a."courseCategoryName"),
|
|
|
+ 'teacherPaySnapshot',COALESCE(l."teacherPaySnapshot",a."teacherPaySnapshot"),
|
|
|
+ 'durationMinutes',a."durationMinutes",'deliveryMode',COALESCE(NULLIF(l."jffs",''),a."deliveryMode",a."jffs"),
|
|
|
+ 'teachingRuleId',a."teachingRuleId",'studentName',COALESCE(NULLIF(l."studentName",''),c."title"),
|
|
|
+ 'courseName',COALESCE(NULLIF(l."courseName",''),NULLIF(l."kcmc",''),c."subtitle")
|
|
|
+ ) END END AS data,
|
|
|
+ CASE WHEN with_text THEN CASE WHEN i.input_signature=x.signature THEN i.row_text||'' ELSE ((to_jsonb(l)-'kzsj')||jsonb_build_object(
|
|
|
+ 'lessonObjectId',l."objectId",'legacyGeneralId',COALESCE(c."generalId",l."legacyGeneralId",0),
|
|
|
+ 'standardLessonKey',CASE WHEN COALESCE(c."generalId",l."legacyGeneralId",0)>0 THEN 'legacy:'||COALESCE(c."generalId",l."legacyGeneralId")::text ELSE 'lesson:'||l."objectId" END,
|
|
|
+ 'source','canonical','commonStatus',COALESCE(c."status",99),
|
|
|
+ 'lessonAt',COALESCE(l."lessonAt",xs_payroll_time(l."plpjsj"),xs_payroll_time(a."jssj"),l."sourceCreatedAt",c."createdAt"),
|
|
|
+ 'courseCategoryKey',COALESCE(NULLIF(l."courseCategoryKey",''),a."courseCategoryKey"),
|
|
|
+ 'courseCategoryName',COALESCE(NULLIF(l."courseCategoryName",''),a."courseCategoryName"),
|
|
|
+ 'teacherPaySnapshot',COALESCE(l."teacherPaySnapshot",a."teacherPaySnapshot"),
|
|
|
+ 'durationMinutes',a."durationMinutes",'deliveryMode',COALESCE(NULLIF(l."jffs",''),a."deliveryMode",a."jffs"),
|
|
|
+ 'teachingRuleId',a."teachingRuleId",'studentName',COALESCE(NULLIF(l."studentName",''),c."title"),
|
|
|
+ 'courseName',COALESCE(NULLIF(l."courseName",''),NULLIF(l."kcmc",''),c."subtitle")
|
|
|
+ ))::text END END AS text_value,m.selection
|
|
|
+ FROM canonical_dependencies x CROSS JOIN LATERAL (SELECT (x.lesson).*) l
|
|
|
+ CROSS JOIN LATERAL (SELECT (x.common).*) c CROSS JOIN LATERAL (SELECT (x.appointment).*) a
|
|
|
+ LEFT JOIN "PayrollSourceIndex" i ON reuse_index AND i.kind='lesson' AND i.company=co AND i."objectId"=l."objectId" AND i.generation=(SELECT generation FROM "PayrollSourceIndexMeta" WHERE singleton)
|
|
|
+ CROSS JOIN LATERAL (SELECT CASE WHEN i.input_signature=x.signature THEN i.selection ELSE xs_payroll_source_index_metadata((to_jsonb(l)-'kzsj')||jsonb_build_object(
|
|
|
+ 'lessonObjectId',l."objectId",'legacyGeneralId',COALESCE(c."generalId",l."legacyGeneralId",0),
|
|
|
+ 'standardLessonKey',CASE WHEN COALESCE(c."generalId",l."legacyGeneralId",0)>0 THEN 'legacy:'||COALESCE(c."generalId",l."legacyGeneralId")::text ELSE 'lesson:'||l."objectId" END,
|
|
|
+ 'source','canonical','commonStatus',COALESCE(c."status",99),
|
|
|
+ 'lessonAt',COALESCE(l."lessonAt",xs_payroll_time(l."plpjsj"),xs_payroll_time(a."jssj"),l."sourceCreatedAt",c."createdAt"),
|
|
|
+ 'courseCategoryKey',COALESCE(NULLIF(l."courseCategoryKey",''),a."courseCategoryKey"),
|
|
|
+ 'courseCategoryName',COALESCE(NULLIF(l."courseCategoryName",''),a."courseCategoryName"),
|
|
|
+ 'teacherPaySnapshot',COALESCE(l."teacherPaySnapshot",a."teacherPaySnapshot"),
|
|
|
+ 'durationMinutes',a."durationMinutes",'deliveryMode',COALESCE(NULLIF(l."jffs",''),a."deliveryMode",a."jffs"),
|
|
|
+ 'teachingRuleId',a."teachingRuleId",'studentName',COALESCE(NULLIF(l."studentName",''),c."title"),
|
|
|
+ 'courseName',COALESCE(NULLIF(l."courseName",''),NULLIF(l."kcmc",''),c."subtitle")
|
|
|
+ )) END AS selection OFFSET 0) m
|
|
|
+ WHERE NOT (SELECT value FROM ambiguous)
|
|
|
+ UNION ALL
|
|
|
+ SELECT 'fact',f."objectId",1,x.ord,x.signature,
|
|
|
+ CASE WHEN include_data AND (mon IS NULL OR m.selection->>'month_candidate_key' IS NULL OR m.selection->>'month_candidate_key'=mon) AND (coach_id IS NULL OR m.selection->>'coach_key' IS NULL OR m.selection->>'coach_key'=coach_id) AND (appointment_ids IS NULL OR m.selection->>'appointment_key' IS NULL OR m.selection->>'appointment_key'=ANY(appointment_ids)) THEN CASE WHEN i.input_signature=x.signature THEN i.row_data ELSE
|
|
|
+(to_jsonb(f)-'kzsj')||jsonb_build_object('lessonObjectId',f."objectId",'source','historical-fact','commonStatus',99,'courseCategoryKey',a."courseCategoryKey",'courseCategoryName',a."courseCategoryName",'teacherPaySnapshot',a."teacherPaySnapshot",'durationMinutes',a."durationMinutes",'deliveryMode',COALESCE(NULLIF(f."jffs",''),a."deliveryMode",a."jffs"),'teachingRuleId',a."teachingRuleId",
|
|
|
+ 'standardLessonKey',CASE WHEN COALESCE(f."legacyGeneralId",0)>0 THEN 'legacy:'||f."legacyGeneralId"::text ELSE 'fact:'||f."objectId" END) END END,
|
|
|
+ CASE WHEN with_text THEN CASE WHEN i.input_signature=x.signature THEN i.row_text||'' ELSE ((to_jsonb(f)-'kzsj')||jsonb_build_object('lessonObjectId',f."objectId",'source','historical-fact','commonStatus',99,'courseCategoryKey',a."courseCategoryKey",'courseCategoryName',a."courseCategoryName",'teacherPaySnapshot',a."teacherPaySnapshot",'durationMinutes',a."durationMinutes",'deliveryMode',COALESCE(NULLIF(f."jffs",''),a."deliveryMode",a."jffs"),'teachingRuleId',a."teachingRuleId",
|
|
|
+ 'standardLessonKey',CASE WHEN COALESCE(f."legacyGeneralId",0)>0 THEN 'legacy:'||f."legacyGeneralId"::text ELSE 'fact:'||f."objectId" END))::text END END,m.selection
|
|
|
+ FROM fact_dependencies x CROSS JOIN LATERAL (SELECT (x.fact).*) f CROSS JOIN LATERAL (SELECT (x.appointment).*) a
|
|
|
+ LEFT JOIN "PayrollSourceIndex" i ON reuse_index AND i.kind='fact' AND i.company=co AND i."objectId"=f."objectId" AND i.generation=(SELECT generation FROM "PayrollSourceIndexMeta" WHERE singleton)
|
|
|
+ CROSS JOIN LATERAL (SELECT CASE WHEN i.input_signature=x.signature THEN i.selection ELSE xs_payroll_source_index_metadata((to_jsonb(f)-'kzsj')||jsonb_build_object('lessonObjectId',f."objectId",'source','historical-fact','commonStatus',99,'courseCategoryKey',a."courseCategoryKey",'courseCategoryName',a."courseCategoryName",'teacherPaySnapshot',a."teacherPaySnapshot",'durationMinutes',a."durationMinutes",'deliveryMode',COALESCE(NULLIF(f."jffs",''),a."deliveryMode",a."jffs"),'teachingRuleId',a."teachingRuleId",
|
|
|
+ 'standardLessonKey',CASE WHEN COALESCE(f."legacyGeneralId",0)>0 THEN 'legacy:'||f."legacyGeneralId"::text ELSE 'fact:'||f."objectId" END)) END AS selection OFFSET 0) m
|
|
|
+ WHERE NOT (SELECT value FROM ambiguous)
|
|
|
+ UNION ALL
|
|
|
+ SELECT CASE WHEN s.row_data->>'source'='canonical' THEN 'lesson' ELSE 'fact' END,s.row_data->>'lessonObjectId',0,s.ord,NULL,CASE WHEN include_data THEN s.row_data END,CASE WHEN with_text THEN s.row_data::text END,xs_payroll_source_index_metadata(s.row_data)
|
|
|
+ FROM ambiguous guard CROSS JOIN LATERAL xs_payroll_sources(co) WITH ORDINALITY AS s(row_data,ord) WHERE guard.value
|
|
|
+ ) SELECT p.kind,p."objectId",p.data,p.text_value,p.signature,CASE WHEN include_data THEN CASE WHEN p.source_group=0 THEN p.ord ELSE (SELECT count(*) FROM lessons)+p.ord END END,p.selection->>'standard_key',p.selection->>'month_candidate_key',p.selection->'scope_data'
|
|
|
+ FROM projected p;
|
|
|
+$$;
|
|
|
+CREATE OR REPLACE FUNCTION xs_payroll_source_index_entries(co text,reuse_index boolean,include_data boolean)
|
|
|
+RETURNS TABLE(kind text,"objectId" text,row_data jsonb,row_text text,input_signature text,ordinal bigint)
|
|
|
+LANGUAGE sql STABLE AS $$
|
|
|
+ SELECT e.kind,e."objectId",e.row_data,e.row_text,e.input_signature,e.ordinal
|
|
|
+ FROM xs_payroll_source_index_scan(co,reuse_index,include_data,NULL,NULL,NULL,true) e
|
|
|
+ ORDER BY CASE WHEN include_data THEN e.ordinal END;
|
|
|
+$$;
|
|
|
+-- Compatibility path keeps complete ordered rows. The three-argument function
|
|
|
+-- can inline at a version-only call, eliminating JSON and ordinal-sort branches.
|
|
|
+CREATE OR REPLACE FUNCTION xs_payroll_source_index_entries(co text,reuse_index boolean DEFAULT true)
|
|
|
+RETURNS TABLE(kind text,"objectId" text,row_data jsonb,row_text text,input_signature text,ordinal bigint)
|
|
|
+LANGUAGE sql STABLE AS $$ SELECT * FROM xs_payroll_source_index_entries(co,reuse_index,true) $$;
|
|
|
+CREATE OR REPLACE FUNCTION xs_payroll_source_entries(co text)
|
|
|
+RETURNS TABLE(row_data jsonb,row_text text,ordinal bigint) LANGUAGE plpgsql STABLE SET jit=off SET work_mem='64MB' AS $$
|
|
|
+BEGIN
|
|
|
+ IF NOT xs_payroll_source_index_ready() OR EXISTS(SELECT 1 FROM "CourseAppointment" WHERE company=co AND id IS NOT NULL GROUP BY id HAVING count(*)>1) THEN
|
|
|
+ RETURN QUERY SELECT s.data,s.data::text,s.ord FROM xs_payroll_sources(co) WITH ORDINALITY s(data,ord);RETURN;
|
|
|
+ END IF;
|
|
|
+ RETURN QUERY SELECT e.row_data,e.row_text,e.ordinal FROM xs_payroll_source_index_entries(co,true,true) e;
|
|
|
+END $$;
|
|
|
+CREATE OR REPLACE FUNCTION xs_payroll_sources_indexed(co text) RETURNS SETOF jsonb LANGUAGE sql STABLE AS $$
|
|
|
+ SELECT row_data FROM xs_payroll_source_entries(co) ORDER BY ordinal;
|
|
|
+$$;
|
|
|
+-- Every current source contributes narrow metadata and (optionally) exact text.
|
|
|
+-- Only selected candidates fetch their full JSON. Callers apply their final exact
|
|
|
+-- financial filters; unknown metadata and ambiguous source identities fail open
|
|
|
+-- for transport inclusion, while authority/schema failures use original sources.
|
|
|
+CREATE OR REPLACE FUNCTION xs_payroll_source_entries_scoped(co text,mon text DEFAULT NULL,coach_id text DEFAULT NULL,appointment_ids text[] DEFAULT NULL,with_text boolean DEFAULT true)
|
|
|
+RETURNS TABLE(row_data jsonb,row_text text,ordinal bigint,standard_key text,month_candidate_key text,scope_data jsonb)
|
|
|
+LANGUAGE plpgsql STABLE SET jit=off SET work_mem='64MB' AS $$
|
|
|
+BEGIN
|
|
|
+ IF NOT xs_payroll_source_index_ready() OR EXISTS(SELECT 1 FROM "CourseAppointment" WHERE company=co AND id IS NOT NULL GROUP BY id HAVING count(*)>1) THEN
|
|
|
+ RETURN QUERY SELECT s.data,CASE WHEN with_text THEN s.data::text END,s.ord,s.data->>'standardLessonKey',m.selection->>'month_candidate_key',m.selection->'scope_data'
|
|
|
+ FROM xs_payroll_sources(co) WITH ORDINALITY s(data,ord)
|
|
|
+ CROSS JOIN LATERAL (SELECT xs_payroll_source_index_metadata(s.data) AS selection) m;
|
|
|
+ RETURN;
|
|
|
+ END IF;
|
|
|
+ RETURN QUERY SELECT e.row_data,e.row_text,e.ordinal,e.standard_key,e.month_candidate_key,e.scope_data
|
|
|
+ FROM xs_payroll_source_index_scan(co,true,true,mon,coach_id,appointment_ids,with_text) e;
|
|
|
+END $$;
|
|
|
+
|
|
|
+CREATE OR REPLACE FUNCTION xs_payroll_source_version_indexed(co text) RETURNS text LANGUAGE plpgsql STABLE SET jit=off SET work_mem='64MB' AS $$
|
|
|
+BEGIN
|
|
|
+ IF NOT xs_payroll_source_index_ready() OR EXISTS(SELECT 1 FROM "CourseAppointment" WHERE company=co AND id IS NOT NULL GROUP BY id HAVING count(*)>1) THEN RETURN xs_payroll_source_version(co);END IF;
|
|
|
+ RETURN (
|
|
|
+ SELECT md5(COALESCE((SELECT string_agg(row_text,'|' ORDER BY row_text) FROM xs_payroll_source_index_entries(co,true,false)),'')||
|
|
|
+ COALESCE((SELECT string_agg(to_jsonb(r)::text,'|' ORDER BY r."objectId") FROM "TeachingCompensationRule" r WHERE r."company"=co),'')||
|
|
|
+ COALESCE((SELECT string_agg(to_jsonb(p)::text,'|' ORDER BY p."objectId") FROM "CourseTeachingProfile" p WHERE p."company"=co),'')||
|
|
|
+ COALESCE((SELECT string_agg(to_jsonb(p)::text,'|' ORDER BY p."objectId") FROM "CoachPayRate" p WHERE p."company"=co),'')||
|
|
|
+ COALESCE((SELECT string_agg(to_jsonb(r)::text,'|' ORDER BY r."objectId") FROM "ReviewRewardIncome" r WHERE r."company"=co AND r."status"='earned'),'')));
|
|
|
+END $$;
|
|
|
+
|
|
|
+CREATE OR REPLACE FUNCTION xs_payroll_source_index_rebuild() RETURNS void LANGUAGE plpgsql SET jit=off SET work_mem='64MB' AS $$
|
|
|
+DECLARE new_generation text:=md5(clock_timestamp()::text||random()::text); co record; prior_setting text:=current_setting('xiaoshu.source_index_reseed',true);
|
|
|
+BEGIN
|
|
|
+ IF NOT xs_payroll_authority_ready() OR NOT xs_payroll_digest_ready() THEN RAISE EXCEPTION 'authoritative source and digest prerequisites required for source index rebuild';END IF;
|
|
|
+ LOCK TABLE "CommonModel","CourseAppointment","LessonRecord","LessonRecordFact" IN SHARE ROW EXCLUSIVE MODE;
|
|
|
+ PERFORM set_config('xiaoshu.source_index_reseed',pg_current_xact_id()::text,true);
|
|
|
+ INSERT INTO "PayrollSourceIndexMeta" VALUES(true,new_generation,'',false) ON CONFLICT(singleton) DO UPDATE SET generation=EXCLUDED.generation,signature='',valid=false;
|
|
|
+ DELETE FROM "PayrollSourceIndex";
|
|
|
+ FOR co IN SELECT DISTINCT company FROM "LessonRecord" WHERE company IS NOT NULL UNION SELECT DISTINCT company FROM "LessonRecordFact" WHERE company IS NOT NULL LOOP
|
|
|
+ INSERT INTO "PayrollSourceIndex"(kind,company,"objectId",row_data,row_text,input_signature,generation,selection) SELECT e.kind,co.company,e."objectId",e.row_data,e.row_text,e.input_signature,new_generation,xs_payroll_source_index_metadata(e.row_data) FROM xs_payroll_source_index_entries(co.company,false,true) e;
|
|
|
+ END LOOP;
|
|
|
+ UPDATE "PayrollSourceIndexMeta" SET signature=xs_payroll_source_index_signature(),valid=true WHERE singleton;
|
|
|
+ PERFORM set_config('xiaoshu.source_index_reseed',COALESCE(prior_setting,''),true);
|
|
|
+END $$;
|
|
|
+DO $$ DECLARE tab text;BEGIN
|
|
|
+ FOREACH tab IN ARRAY ARRAY['CommonModel','CourseAppointment','LessonRecord','LessonRecordFact'] LOOP
|
|
|
+ EXECUTE format('LOCK TABLE %I IN SHARE ROW EXCLUSIVE MODE',tab);
|
|
|
+ EXECUTE format('DROP TRIGGER IF EXISTS xs_payroll_source_index_row ON %I',tab);
|
|
|
+ EXECUTE format('DROP TRIGGER IF EXISTS xs_payroll_source_index_truncate ON %I',tab);
|
|
|
+ EXECUTE format('CREATE TRIGGER xs_payroll_source_index_row AFTER INSERT OR UPDATE OR DELETE ON %I FOR EACH ROW EXECUTE FUNCTION xs_payroll_source_index_row()',tab);
|
|
|
+ EXECUTE format('CREATE TRIGGER xs_payroll_source_index_truncate AFTER TRUNCATE ON %I FOR EACH STATEMENT EXECUTE FUNCTION xs_payroll_source_index_truncate()',tab);
|
|
|
+ EXECUTE format('ALTER TABLE %I ENABLE ALWAYS TRIGGER xs_payroll_source_index_row',tab);
|
|
|
+ EXECUTE format('ALTER TABLE %I ENABLE ALWAYS TRIGGER xs_payroll_source_index_truncate',tab);
|
|
|
+ END LOOP;
|
|
|
+END $$;
|
|
|
+SELECT xs_payroll_source_index_rebuild();
|
|
|
+COMMIT;
|