fix(selfop): retain turnover precision and scope reused publication runs
This commit is contained in:
+13
-2
@@ -23,8 +23,19 @@ public interface SelfopCoordinationMapper {
|
||||
long barriers(@Param("account") String account, @Param("warehouse") String warehouse);
|
||||
@Update("UPDATE ll_selfop_sync_task SET status='SUCCEEDED',result_published_run_id=#{published},next_retry_at_utc=NULL,updated_at=#{now} WHERE scope_id=#{scope} AND scope_hash=#{hash} AND execution_purpose='BUSINESS' AND apply_state='APPLIED' AND (query_from_utc IS NULL OR (query_from_utc>=#{from} AND query_until_utc<=#{until})) AND status IN ('READY','RUNNING','RETRY_WAIT','DEAD')")
|
||||
int completeCoveredTasks(@Param("scope") long scope,@Param("hash") String hash,@Param("published") long published,@Param("from") LocalDateTime from,@Param("until") LocalDateTime until,@Param("now") LocalDateTime now);
|
||||
@Update("UPDATE ll_selfop_sync_run r JOIN ll_selfop_sync_task t ON t.id=r.task_id SET r.status='REUSED',r.finished_at_utc=#{now},r.updated_at=#{now},r.progress_snapshot_json=COALESCE(r.progress_snapshot_json,CASE WHEN JSON_EXTRACT(t.progress_json,'$.attemptNo')=r.attempt_no THEN t.progress_json ELSE NULL END) WHERE t.result_published_run_id=#{published} AND r.id<>#{published} AND r.attempt_no=t.attempt_count AND r.execution_purpose='BUSINESS' AND r.status IN ('FETCHING','FACTS_APPLIED','BUILDING','READY')")
|
||||
int completeReusedRuns(@Param("published") long published,@Param("now") LocalDateTime now);
|
||||
@Update("""
|
||||
UPDATE ll_selfop_sync_run r JOIN ll_selfop_sync_task t ON t.id=r.task_id
|
||||
SET r.status='REUSED',r.finished_at_utc=#{now},r.updated_at=#{now},r.progress_snapshot_json=COALESCE(r.progress_snapshot_json,CASE WHEN JSON_EXTRACT(t.progress_json,'$.attemptNo')=r.attempt_no THEN t.progress_json ELSE NULL END)
|
||||
WHERE r.scope_id=#{scope} AND t.scope_id=#{scope}
|
||||
AND r.source_account=#{account} AND t.source_account=#{account}
|
||||
AND r.warehouse_code=#{warehouse} AND t.warehouse_code=#{warehouse}
|
||||
AND r.scope_hash=#{hash} AND t.scope_hash=#{hash}
|
||||
AND t.result_published_run_id=#{published} AND r.id<>#{published}
|
||||
AND r.attempt_no=t.attempt_count AND r.execution_purpose='BUSINESS' AND t.execution_purpose='BUSINESS'
|
||||
AND r.status IN ('FETCHING','FACTS_APPLIED','BUILDING','READY')
|
||||
""")
|
||||
int completeReusedRuns(@Param("scope") long scope,@Param("account") String account,@Param("warehouse") String warehouse,
|
||||
@Param("hash") String hash,@Param("published") long published,@Param("now") LocalDateTime now);
|
||||
@Update("UPDATE ll_selfop_sync_task SET status='SUCCEEDED',result_published_run_id=#{published},next_retry_at_utc=NULL,updated_at=#{now} WHERE scope_id=#{scope} AND scope_hash=#{hash} AND execution_purpose='BUSINESS' AND task_type='SCHEDULED' AND apply_state='NOT_APPLICABLE' AND status='WAITING_SOURCE' AND planned_at_utc<=#{until}")
|
||||
int completeCoveredMarkers(@Param("scope") long scope,@Param("hash") String hash,@Param("published") long published,@Param("until") LocalDateTime until,@Param("now") LocalDateTime now);
|
||||
@Update("<script>UPDATE ll_selfop_sync_run r JOIN ll_selfop_sync_task t ON t.id=r.task_id SET r.heartbeat_at_utc=#{now},r.updated_at=#{now} WHERE r.fence_token=#{fence} AND t.status='RUNNING' AND r.status IN ('FETCHING','VALIDATED','FACTS_APPLIED','BUILDING','READY') <choose><when test='account'>AND t.discovery_scope_id=#{scope}</when><otherwise>AND r.scope_id=#{scope}</otherwise></choose></script>")
|
||||
|
||||
+4
-2
@@ -49,11 +49,13 @@ public final class InventoryTurnoverPolicy {
|
||||
var r=normalized(raw);boolean owner="OWNER".equals(q.group());
|
||||
Total physical=total(r,"physical",owner),available=total(r,"available",owner);
|
||||
var windows=windows(r,context.endDate().plusDays(1));var base=windows.stream().filter(w->w.days()==q.days()).findFirst().orElseThrow();
|
||||
var coverage=owner?null:scale(decimal(r,"coverage_days"));var average=scale(decimal(r,"average_physical"));
|
||||
var coverage=owner?null:scale(decimal(r,"coverage_days"));var rawAverage=decimal(r,"average_physical");var average=scale(rawAverage);
|
||||
var turnover=scale(decimal(r,"turnover"));int sampleCount=count(r,"sample_count");
|
||||
String coverageStatus=owner?"NOT_APPLICABLE":available.value()==null?"INVENTORY_UNKNOWN_OR_STALE":available.value().signum()==0?"NO_AVAILABLE_STOCK":
|
||||
base.outboundQuantity()==null||count(r,"unassessable")>0?"UNASSESSABLE":base.outboundQuantity().signum()==0?"NO_OUTBOUND_IN_WINDOW":"VALID";
|
||||
String sampleStatus=sampleCount<q.days()?"INSUFFICIENT":average!=null&&average.signum()==0?"ZERO_AVERAGE_STOCK":turnover==null?"SALES_WINDOW_UNAVAILABLE":"VALID";
|
||||
// Qualification precedes output rounding: a positive daily mean can round to six-place zero.
|
||||
String sampleStatus=sampleCount<q.days()||rawAverage==null?"INSUFFICIENT":rawAverage.signum()==0?"ZERO_AVERAGE_STOCK":turnover==null?"SALES_WINDOW_UNAVAILABLE":"VALID";
|
||||
if(!"VALID".equals(sampleStatus))turnover=null;
|
||||
List<LocalDate> missing=new ArrayList<>();String dates=text(r,"missing_dates");
|
||||
if(dates!=null&&!dates.isBlank())for(String date:dates.split(","))missing.add(LocalDate.parse(date.substring(0,10)));
|
||||
List<String> attention=new ArrayList<>();
|
||||
|
||||
+7
-4
@@ -69,12 +69,12 @@ public final class InventoryTurnoverSql {
|
||||
.append(scope("")).append(" AND sales_run_id=#{p.salesRun} AND business_date<CAST(#{p.until} AS DATE) AND outbound_quantity>0 AND quantity_unit='PCS' GROUP BY customer_code,sku_code)");
|
||||
sql.append(", sku_base AS (SELECT c.customer_code,c.sku_code,c.product_name,c.evidence,c.unit_valid");
|
||||
for(int n:List.of(7,15,30)) for(String suffix:List.of("outbound","observed","complete","partial")) sql.append(",w.w").append(n).append('_').append(suffix);
|
||||
sql.append(",w.sample_count,w.average_physical,w.sample_mask,ls.last_outbound");
|
||||
sql.append(",w.sample_count,w.sample_total,w.average_physical,w.sample_mask,ls.last_outbound");
|
||||
for(String field:List.of("physical","available","frozen","defective")) sql.append(",CASE WHEN #{p.stockFresh}=1 AND ip.quantity_unit='PCS' THEN st.").append(field).append(" END ").append(field);
|
||||
sql.append(" FROM profiles c JOIN sku_windows w ON w.customer_code=c.customer_code AND w.sku_code=c.sku_code LEFT JOIN last_sales ls ON ls.customer_code=c.customer_code AND ls.sku_code=c.sku_code LEFT JOIN stocks st ON st.snapshot_run_id=#{p.inventoryRun} AND st.customer_code=c.customer_code AND st.sku_code=c.sku_code ")
|
||||
.append("LEFT JOIN fixed_profiles ip ON ip.profile_run_id=#{p.inventoryRun} AND ip.customer_code=c.customer_code AND ip.sku_code=c.sku_code), sku_calculated AS (SELECT b.*,")
|
||||
.append("CASE WHEN available=0 THEN 0 WHEN available>0 AND w").append(p.query.days()).append("_outbound>0 AND evidence<=#{p.baseFromUtc} THEN available*#{p.query.days}/w").append(p.query.days()).append("_outbound END coverage_days,")
|
||||
.append("CASE WHEN average_physical>0 AND evidence<=#{p.baseFromUtc} THEN w").append(p.query.days()).append("_outbound/average_physical END turnover,")
|
||||
.append("CASE WHEN sample_total>0 AND evidence<=#{p.baseFromUtc} THEN w").append(p.query.days()).append("_outbound*#{p.query.days}/sample_total END turnover,")
|
||||
.append("CASE WHEN physical=0 THEN 0 WHEN physical>0 AND w30_outbound IS NOT NULL AND evidence<=#{p.fromUtc} THEN CASE WHEN w30_outbound=0 THEN 1 ELSE 0 END END no_outbound,")
|
||||
.append("CASE WHEN physical IS NULL OR available IS NULL OR w").append(p.query.days()).append("_outbound IS NULL OR evidence IS NULL OR evidence>#{p.baseFromUtc} THEN 1 ELSE 0 END unassessable FROM sku_base b), sku_result AS (")
|
||||
.append("SELECT b.*,CASE WHEN #{p.query.coverageDaysThreshold,jdbcType=INTEGER} IS NULL THEN NULL WHEN physical=0 OR available=0 THEN 0 WHEN coverage_days IS NOT NULL AND available>0 THEN CASE WHEN coverage_days>#{p.query.coverageDaysThreshold,jdbcType=INTEGER} THEN 1 ELSE 0 END WHEN available>0 AND w")
|
||||
@@ -115,8 +115,10 @@ public final class InventoryTurnoverSql {
|
||||
.append("JOIN fixed_profiles ip ON ip.profile_run_id=d.sample_run AND ip.customer_code=st.customer_code AND ip.sku_code=st.sku_code ")
|
||||
.append("WHERE ip.quantity_unit='PCS' AND st.physical IS NOT NULL), sku_sample_counts AS (SELECT customer_code,sku_code,")
|
||||
.append("COUNT(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END) sample_count,")
|
||||
// Keep the qualified sum exact; do not divide again by a rounded daily mean.
|
||||
.append("CASE WHEN COUNT(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END)=#{p.query.days} THEN SUM(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END) END sample_total,")
|
||||
.append("CASE WHEN COUNT(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END)=#{p.query.days} THEN SUM(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END)/#{p.query.days} END average_physical,")
|
||||
.append(sampleMask()).append(" sample_mask FROM sku_sample_days GROUP BY customer_code,sku_code), sku_window_counts AS (SELECT z.customer_code,z.sku_code,COALESCE(s.sample_count,0) sample_count,s.average_physical,COALESCE(s.sample_mask,0) sample_mask");
|
||||
.append(sampleMask()).append(" sample_mask FROM sku_sample_days GROUP BY customer_code,sku_code), sku_window_counts AS (SELECT z.customer_code,z.sku_code,COALESCE(s.sample_count,0) sample_count,s.sample_total,s.average_physical,COALESCE(s.sample_mask,0) sample_mask");
|
||||
for(int n:List.of(7,15,30)) sql.append(",CASE WHEN z.w").append(n).append("_known+COALESCE(a.w").append(n).append("_known,0)>0 THEN COALESCE(a.w").append(n).append("_sum,0) END w").append(n).append("_observed")
|
||||
.append(",z.w").append(n).append("_complete+COALESCE(a.w").append(n).append("_complete,0) w").append(n).append("_complete")
|
||||
.append(",z.w").append(n).append("_partial+COALESCE(a.w").append(n).append("_partial,0) w").append(n).append("_partial");
|
||||
@@ -167,10 +169,11 @@ public final class InventoryTurnoverSql {
|
||||
+"s.sample_physical FROM owner_candidates c CROSS JOIN calendar d LEFT JOIN owner_actual_days a ON a.customer_code=c.customer_code AND a.business_date=d.business_date "
|
||||
+"LEFT JOIN owner_sample_days s ON s.customer_code=c.customer_code AND s.business_date=d.business_date), owner_window_counts AS (SELECT customer_code"
|
||||
+windowAggregates()+",COUNT(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END) sample_count,"
|
||||
+"CASE WHEN COUNT(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END)=#{p.query.days} THEN SUM(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END) END sample_total,"
|
||||
+"CASE WHEN COUNT(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END)=#{p.query.days} THEN SUM(CASE WHEN business_date>=CAST(#{p.baseFrom} AS DATE) THEN sample_physical END)/#{p.query.days} END average_physical,"
|
||||
+sampleMask()+" sample_mask FROM owner_days GROUP BY customer_code), owner_windows AS (SELECT w.*"+formalWindows()+" FROM owner_window_counts w), owner_totals AS ("
|
||||
+totals("customer_code",true)+" FROM sku_result GROUP BY customer_code), owner_result AS (SELECT t.*,w.sample_count,w.average_physical,w.sample_mask,"
|
||||
+"CASE WHEN w.average_physical>0 THEN w.w"+p.query.days()+"_outbound/w.average_physical END turnover,"
|
||||
+"CASE WHEN w.sample_total>0 THEN w.w"+p.query.days()+"_outbound*#{p.query.days}/w.sample_total END turnover,"
|
||||
+String.join(",",List.of(7,15,30).stream().flatMap(n->List.of("outbound","observed","complete","partial").stream().map(s->"w.w"+n+"_"+s)).toList())
|
||||
+" FROM owner_totals t JOIN owner_windows w ON w.customer_code=t.customer_code) ";
|
||||
}
|
||||
|
||||
+1
-1
@@ -45,7 +45,7 @@ public class SelfopSalesPublisher {
|
||||
run.setStatus("PUBLISHED");run.setPublishedAtUtc(now);run.setFinishedAtUtc(now);run.setUpdatedAt(now);runs.updateById(run);
|
||||
task.setResultPublishedRunId(runId);task.setStatus("SUCCEEDED");task.setUpdatedAt(now);tasks.updateById(task);
|
||||
locks.completeCoveredTasks(scopeId,scope.getScopeHash(),runId,run.getCoveredFromUtc(),run.getCoveredUntilUtc(),now);
|
||||
locks.completeReusedRuns(runId,now);
|
||||
locks.completeReusedRuns(scopeId,scope.getSourceAccount(),scope.getWarehouseCode(),scope.getScopeHash(),runId,now);
|
||||
locks.completeCoveredMarkers(scopeId,scope.getScopeHash(),runId,run.getCoveredUntilUtc(),now);
|
||||
}
|
||||
@Transactional
|
||||
|
||||
+38
@@ -0,0 +1,38 @@
|
||||
package com.baiyue.lastleg.selfoperated.analytics.query;
|
||||
|
||||
import java.math.BigDecimal;
|
||||
import java.util.*;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import static org.junit.jupiter.api.Assertions.*;
|
||||
import static com.baiyue.lastleg.selfoperated.analytics.query.InventoryTurnoverResponses.*;
|
||||
|
||||
class InventoryTurnoverPolicyTest {
|
||||
static Context context() {
|
||||
return new Context(null,"demo","W",InventoryTurnoverSqlTest.END,30,null,null,null,null,null,null,List.of(),List.of(),null);
|
||||
}
|
||||
private Row mapped(String average,String turnover,int samples) {
|
||||
var raw=new HashMap<String,Object>();raw.put("average_physical",average);raw.put("turnover",turnover);raw.put("sample_count",samples);
|
||||
raw.put("w30_outbound",BigDecimal.ONE);raw.put("w30_complete",30);
|
||||
return InventoryTurnoverPolicy.row(raw,InventoryTurnoverSqlTest.request("SKU","ALL","TURNOVER_DESC",null,null,1,20),context());
|
||||
}
|
||||
@Test void tinyPositiveAverageQualifiesBeforeSixPlaceOutputRounding() {
|
||||
var row=mapped("0.0000000333","30000000",30);
|
||||
assertEquals("VALID",row.turnoverStatus());assertTrue(row.sampleReasonCodes().isEmpty());
|
||||
assertEquals(new BigDecimal("0.000000"),row.averagePhysicalQuantity());
|
||||
assertEquals(new BigDecimal("30000000.000000"),row.quantityTurnoverCount());
|
||||
}
|
||||
@Test void zeroUnknownAndIncompleteSamplesCannotLeakATurnoverValue() {
|
||||
for(var input:List.of(new String[]{"0","ZERO_AVERAGE_STOCK"},new String[]{null,"INSUFFICIENT"})) {
|
||||
var row=mapped(input[0],"30000000",30);assertNull(row.quantityTurnoverCount());
|
||||
assertEquals(input[1],row.turnoverStatus());assertEquals(List.of(input[1]),row.sampleReasonCodes());
|
||||
}
|
||||
var row=mapped("1","30000000",29);assertNull(row.quantityTurnoverCount());
|
||||
assertEquals("INSUFFICIENT",row.turnoverStatus());assertEquals(List.of("INSUFFICIENT"),row.sampleReasonCodes());
|
||||
}
|
||||
@Test void positiveAverageSeparatesMissingSalesFromKnownZeroSales() {
|
||||
var missing=mapped("0.0000000333",null,30);assertNull(missing.quantityTurnoverCount());
|
||||
assertEquals("SALES_WINDOW_UNAVAILABLE",missing.turnoverStatus());assertEquals(List.of("SALES_WINDOW_UNAVAILABLE"),missing.sampleReasonCodes());
|
||||
var zero=mapped("0.0000000333","0",30);assertEquals("VALID",zero.turnoverStatus());
|
||||
assertEquals(new BigDecimal("0.000000"),zero.quantityTurnoverCount());
|
||||
}
|
||||
}
|
||||
+59
-1
@@ -362,12 +362,70 @@ class InventoryTurnoverSqlTest {
|
||||
assertEquals("SKU".equals(group)?0:3,report.stream().filter(r->"ROW".equals(r.get("section"))).count());
|
||||
}
|
||||
}
|
||||
@Test void tinyQualifiedDailySumUsesOneDivisionForSkuAndOwnerThroughProductionCtes() throws Exception {
|
||||
seed("A","ONE",30);seed("A","TWO",30);
|
||||
execute("UPDATE ll_selfop_inventory_snapshot SET physical_quantity=CASE WHEN snapshot_run_id=100 AND product_quality=0 THEN CASE WHEN sku_code='ONE' THEN 0.000001 ELSE 0.000002 END ELSE 0 END WHERE snapshot_run_id>=100");
|
||||
execute("UPDATE ll_selfop_sku_daily_sales SET outbound_quantity=CASE WHEN business_date=DATE '"+END+"' THEN CASE WHEN sku_code='ONE' THEN 1 ELSE 2 END ELSE 0 END WHERE warehouse_code='W'");
|
||||
for(String group:List.of("SKU","OWNER")) {
|
||||
var q=request(group,"ALL","TURNOVER_DESC",null,null,1,20);var p=input(q,30,30,true);
|
||||
for(var row:reportRows(p)) {
|
||||
assertTrue(new BigDecimal(row.get("average_physical").toString()).signum()>0);
|
||||
decimal("30000000",row.get("turnover"));decimal("30",row.get("sample_count"));
|
||||
var mapped=InventoryTurnoverPolicy.row(row,q,InventoryTurnoverPolicyTest.context());
|
||||
assertEquals("VALID",mapped.turnoverStatus());assertTrue(mapped.sampleReasonCodes().isEmpty());
|
||||
assertEquals(new BigDecimal("0.000000"),mapped.averagePhysicalQuantity());
|
||||
assertEquals(new BigDecimal("30000000.000000"),mapped.quantityTurnoverCount());
|
||||
}
|
||||
for(var row:query("page",p))decimal("30000000",row.get("turnover"));
|
||||
}
|
||||
execute("UPDATE ll_selfop_inventory_snapshot SET physical_quantity=0 WHERE snapshot_run_id>=100");
|
||||
for(String group:List.of("SKU","OWNER")) {
|
||||
var q=request(group,"ALL","TURNOVER_DESC",null,null,1,20);
|
||||
for(var row:reportRows(input(q,30,30,true))) {
|
||||
decimal("0",row.get("average_physical"));assertNull(row.get("turnover"));
|
||||
var mapped=InventoryTurnoverPolicy.row(row,q,InventoryTurnoverPolicyTest.context());
|
||||
assertEquals("ZERO_AVERAGE_STOCK",mapped.turnoverStatus());assertNull(mapped.quantityTurnoverCount());
|
||||
assertEquals(List.of("ZERO_AVERAGE_STOCK"),mapped.sampleReasonCodes());
|
||||
}
|
||||
for(var row:reportRows(input(q,30,29,true))) {
|
||||
assertNull(row.get("average_physical"));assertNull(row.get("turnover"));
|
||||
var mapped=InventoryTurnoverPolicy.row(row,q,InventoryTurnoverPolicyTest.context());
|
||||
assertEquals("INSUFFICIENT",mapped.turnoverStatus());assertNull(mapped.quantityTurnoverCount());
|
||||
}
|
||||
}
|
||||
}
|
||||
@Test void unroundedThresholdAndStableTiesSurviveActualReportPagination() throws Exception {
|
||||
seed("OA","HIGH",30);seed("OA","A",30);seed("OZ","A",30);seed("OA","LOW",30);
|
||||
execute("UPDATE ll_selfop_inventory_snapshot SET available_quantity=CASE sku_code WHEN 'HIGH' THEN 200.000001 WHEN 'LOW' THEN 199.999999 ELSE 200 END WHERE snapshot_run_id=22 AND product_quality=0");
|
||||
var gt=request("SKU","COVERAGE_GT","COVERAGE_DESC",null,20,1,20);
|
||||
var greater=reportRows(input(gt,30,30,true));assertEquals(List.of("HIGH"),greater.stream().map(r->r.get("sku_code")).toList());
|
||||
assertEquals(new BigDecimal("20.000000"),InventoryTurnoverPolicy.row(greater.get(0),gt,InventoryTurnoverPolicyTest.context()).inventoryCoverageDays());
|
||||
var le=request("SKU","COVERAGE_LE","COVERAGE_DESC",null,20,1,20);
|
||||
assertEquals(List.of("A/OA","A/OZ","LOW/OA"),reportRows(input(le,30,30,true)).stream().map(r->r.get("sku_code")+"/"+r.get("customer_code")).toList());
|
||||
var expected=List.of("HIGH/OA","A/OA","A/OZ","LOW/OA");
|
||||
for(int page=1;page<=4;page++) {
|
||||
var q=request("SKU","ALL","COVERAGE_DESC",null,20,page,1);var result=reportRows(input(q,30,30,true));
|
||||
assertEquals(1,result.size());assertEquals(expected.get(page-1),result.get(0).get("sku_code")+"/"+result.get(0).get("customer_code"));
|
||||
}
|
||||
}
|
||||
@Test void turnoverOrderingUsesUnroundedRatioBeforeStableSkuAndOwnerPages() throws Exception {
|
||||
seed("Z","HIGH",30);seed("A","TIE",30);seed("B","TIE",30);seed("L","LOW",30);
|
||||
execute("UPDATE ll_selfop_inventory_snapshot SET physical_quantity=CASE WHEN product_quality=0 THEN 10 ELSE 0 END WHERE snapshot_run_id>=100");
|
||||
execute("UPDATE ll_selfop_sku_daily_sales SET outbound_quantity=CASE WHEN business_date=DATE '"+END+"' THEN CASE sku_code WHEN 'HIGH' THEN 10.000001 WHEN 'LOW' THEN 9.999999 ELSE 10 END ELSE 0 END WHERE warehouse_code='W'");
|
||||
for(String group:List.of("SKU","OWNER")) for(int page=1;page<=4;page++) {
|
||||
var q=request(group,"ALL","TURNOVER_DESC",null,null,page,1);
|
||||
var result=reportRows(input(q,30,30,true));assertEquals(1,result.size());
|
||||
assertEquals(List.of("Z","A","B","L").get(page-1),result.get(0).get("customer_code"));
|
||||
var mapped=InventoryTurnoverPolicy.row(result.get(0),q,InventoryTurnoverPolicyTest.context());
|
||||
assertEquals("VALID",mapped.turnoverStatus());assertEquals(new BigDecimal("1.000000"),mapped.quantityTurnoverCount());
|
||||
}
|
||||
}
|
||||
private static com.fasterxml.jackson.databind.JsonNode payload(Map<String,Object> row) throws Exception {
|
||||
var json=new com.fasterxml.jackson.databind.ObjectMapper().enable(com.fasterxml.jackson.databind.DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS);Object raw=row.get("payload");
|
||||
return raw instanceof byte[] bytes?json.readTree(bytes):json.readTree(raw.toString());
|
||||
}
|
||||
private List<Map<String,Object>> reportRows(InventoryTurnoverSql.Input p) throws Exception {
|
||||
var result=new ArrayList<Map<String,Object>>();var json=new com.fasterxml.jackson.databind.ObjectMapper();
|
||||
var result=new ArrayList<Map<String,Object>>();var json=new com.fasterxml.jackson.databind.ObjectMapper().enable(com.fasterxml.jackson.databind.DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS);
|
||||
for(var row:query("report",p)) if("ROW".equals(row.get("section")))
|
||||
result.add(json.convertValue(payload(row),new com.fasterxml.jackson.core.type.TypeReference<Map<String,Object>>(){}));
|
||||
return result;
|
||||
|
||||
+102
@@ -0,0 +1,102 @@
|
||||
package com.baiyue.lastleg.selfoperated.analytics.sync;
|
||||
|
||||
import com.baiyue.lastleg.selfoperated.analytics.mapper.SelfopCoordinationMapper;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import org.apache.ibatis.mapping.BoundSql;
|
||||
import org.apache.ibatis.mapping.MappedStatement;
|
||||
import org.apache.ibatis.scripting.defaults.DefaultParameterHandler;
|
||||
import org.apache.ibatis.session.Configuration;
|
||||
import org.junit.jupiter.api.*;
|
||||
import org.junit.jupiter.params.ParameterizedTest;
|
||||
import org.junit.jupiter.params.provider.ValueSource;
|
||||
import java.sql.*;
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.*;
|
||||
import static org.junit.jupiter.api.Assertions.*;
|
||||
|
||||
/** Executes production JOIN/predicates/assignments offline. Only MySQL UPDATE JOIN syntax is
|
||||
* adapted to H2 MERGE; this does not prove MySQL optimizer choices or lock isolation. */
|
||||
public class SelfopReusedRunScopeSqlTest {
|
||||
Connection db; MappedStatement statement; BoundSql bound;
|
||||
final Map<String,Object> params=Map.of("scope",7L,"account","SYNTH-ACCOUNT","warehouse","SYNTH-A",
|
||||
"hash","scope-hash","published",900L,"now",LocalDateTime.of(2026,10,14,0,0));
|
||||
|
||||
@BeforeEach void setup() throws Exception {
|
||||
db=DriverManager.getConnection("jdbc:h2:mem:"+UUID.randomUUID()+";MODE=MySQL;DATABASE_TO_LOWER=TRUE","sa","");
|
||||
execute("CREATE TABLE ll_selfop_sync_task(id BIGINT PRIMARY KEY,scope_id BIGINT,source_account VARCHAR,warehouse_code VARCHAR,scope_hash VARCHAR,result_published_run_id BIGINT,attempt_count INT,execution_purpose VARCHAR,progress_json VARCHAR)");
|
||||
execute("CREATE TABLE ll_selfop_sync_run(id BIGINT PRIMARY KEY,task_id BIGINT,scope_id BIGINT,source_account VARCHAR,warehouse_code VARCHAR,scope_hash VARCHAR,attempt_no INT,execution_purpose VARCHAR,status VARCHAR,finished_at_utc TIMESTAMP,updated_at TIMESTAMP,progress_snapshot_json VARCHAR)");
|
||||
execute("CREATE ALIAS JSON_EXTRACT FOR '"+getClass().getName()+".jsonExtract'");
|
||||
var config=new Configuration();config.addMapper(SelfopCoordinationMapper.class);
|
||||
statement=config.getMappedStatement(SelfopCoordinationMapper.class.getName()+".completeReusedRuns");
|
||||
bound=statement.getBoundSql(params);
|
||||
}
|
||||
@AfterEach void close() throws Exception {db.close();}
|
||||
public static String jsonExtract(String body,String path) throws Exception {
|
||||
if(body==null)return null;assertEquals("$.attemptNo",path);
|
||||
var value=new ObjectMapper().readTree(body).get("attemptNo");return value==null?null:value.asText();
|
||||
}
|
||||
void execute(String sql,Object...args) throws Exception {
|
||||
try(var p=db.prepareStatement(sql)){for(int n=0;n<args.length;n++)p.setObject(n+1,args[n]);p.executeUpdate();}
|
||||
}
|
||||
void pair(long id,String status) throws Exception {
|
||||
execute("INSERT INTO ll_selfop_sync_task VALUES (?,7,'SYNTH-ACCOUNT','SYNTH-A','scope-hash',900,2,'BUSINESS',?)",id+1000,"{\"attemptNo\":2,\"checkpoint\":\"kept\"}");
|
||||
execute("INSERT INTO ll_selfop_sync_run VALUES (?,?,7,'SYNTH-ACCOUNT','SYNTH-A','scope-hash',2,'BUSINESS',?,NULL,NULL,NULL)",id,id+1000,status);
|
||||
}
|
||||
Map<String,Object> row(long id) throws Exception {
|
||||
try(var p=db.prepareStatement("SELECT * FROM ll_selfop_sync_run WHERE id=?")){p.setLong(1,id);try(var r=p.executeQuery()){assertTrue(r.next());var out=new TreeMap<String,Object>();for(int n=1;n<=r.getMetaData().getColumnCount();n++)out.put(r.getMetaData().getColumnLabel(n),r.getObject(n));return out;}}
|
||||
}
|
||||
int update() throws Exception {
|
||||
String sql=bound.getSql().replaceAll("\\s+"," ").trim();
|
||||
int set=sql.indexOf(" SET "),where=sql.indexOf(" WHERE ");assertTrue(sql.startsWith("UPDATE ")&&set>0&&where>set);
|
||||
String from=sql.substring(7,set),predicate=sql.substring(where+7);
|
||||
// Assignment order preserves MyBatis parameter order. No test-owned selection predicate.
|
||||
var projection=new ArrayList<String>();var assignments=new ArrayList<String>();
|
||||
for(String assignment:sql.substring(set+5,where).split(",(?=r\\.)")){
|
||||
int equal=assignment.indexOf('=');String column=assignment.substring(0,equal).replace("r.","");
|
||||
String expression=assignment.substring(equal+1);
|
||||
// H2 cannot infer a bare parameter's type in a derived table; production columns are timestamps.
|
||||
if(expression.equals("?")){assertTrue(column.equals("finished_at_utc")||column.equals("updated_at"));expression="CAST(? AS TIMESTAMP)";}
|
||||
projection.add(expression+" AS "+column);assignments.add("destination."+column+"=source."+column);
|
||||
}
|
||||
String adapted="MERGE INTO ll_selfop_sync_run destination USING (SELECT r.id,"+String.join(",",projection)+" FROM "+from+" WHERE "+predicate+") source ON destination.id=source.id WHEN MATCHED THEN UPDATE SET "+String.join(",",assignments);
|
||||
try(var p=db.prepareStatement(adapted)){new DefaultParameterHandler(statement,params,bound).setParameters(p);return p.executeUpdate();}
|
||||
}
|
||||
@Test void nonemptyMatchingAttemptsCompleteAllFourEligibleStatusesAndKeepProgress() throws Exception {
|
||||
String[] statuses={"FETCHING","FACTS_APPLIED","BUILDING","READY"};
|
||||
for(int n=0;n<statuses.length;n++)pair(11+n,statuses[n]);
|
||||
execute("UPDATE ll_selfop_sync_run SET progress_snapshot_json=? WHERE id=11","{\"frozen\":true}");
|
||||
pair(900,"READY");var publishedBefore=row(900);
|
||||
assertEquals(4,update());assertEquals(publishedBefore,row(900),"publishing run must not become REUSED");
|
||||
for(int n=0;n<4;n++){var r=row(11+n);assertEquals("REUSED",r.get("status"));assertEquals(Timestamp.valueOf((LocalDateTime)params.get("now")),r.get("finished_at_utc"));assertEquals(r.get("finished_at_utc"),r.get("updated_at"));assertEquals(n==0?"{\"frozen\":true}":"{\"attemptNo\":2,\"checkpoint\":\"kept\"}",r.get("progress_snapshot_json"));}
|
||||
assertEquals(0,update(),"already completed attempts are untouched");
|
||||
}
|
||||
@ParameterizedTest @ValueSource(strings={"scope_id","source_account","warehouse_code","scope_hash"})
|
||||
void eitherSideOutsidePublishedScopeIsUntouchedWhileValidSiblingCompletes(String column) throws Exception {
|
||||
pair(11,"READY");pair(12,"READY");pair(13,"READY");pair(14,"READY");
|
||||
Object wrong=column.equals("scope_id")?8L:"different";
|
||||
execute("UPDATE ll_selfop_sync_run SET "+column+"=? WHERE id=12",wrong);
|
||||
execute("UPDATE ll_selfop_sync_task SET "+column+"=? WHERE id=1013",wrong);
|
||||
execute("UPDATE ll_selfop_sync_run SET "+column+"=? WHERE id=14",wrong);
|
||||
execute("UPDATE ll_selfop_sync_task SET "+column+"=? WHERE id=1014",wrong);
|
||||
var before=Map.of(12L,row(12),13L,row(13),14L,row(14));
|
||||
assertEquals(1,update());assertEquals("REUSED",row(11).get("status"));
|
||||
for(var item:before.entrySet())assertEquals(item.getValue(),row(item.getKey()),"foreign association must not be changed: "+column);
|
||||
}
|
||||
@ParameterizedTest @ValueSource(strings={"old-attempt","wrong-result","run-validation","task-validation","PUBLISHED","FAILED","DEAD","REUSED","null-scope"})
|
||||
void existingPurposeAttemptPointerAndStatusGatesRemain(String fault) throws Exception {
|
||||
pair(11,"READY");pair(12,"READY");
|
||||
switch(fault){
|
||||
case "old-attempt" -> execute("UPDATE ll_selfop_sync_run SET attempt_no=1 WHERE id=12");
|
||||
case "wrong-result" -> execute("UPDATE ll_selfop_sync_task SET result_published_run_id=901 WHERE id=1012");
|
||||
case "run-validation" -> execute("UPDATE ll_selfop_sync_run SET execution_purpose='VALIDATION' WHERE id=12");
|
||||
case "task-validation" -> execute("UPDATE ll_selfop_sync_task SET execution_purpose='VALIDATION' WHERE id=1012");
|
||||
case "null-scope" -> execute("UPDATE ll_selfop_sync_task SET scope_id=NULL WHERE id=1012");
|
||||
default -> execute("UPDATE ll_selfop_sync_run SET status=? WHERE id=12",fault);
|
||||
}
|
||||
var before=row(12);assertEquals(1,update());assertEquals("REUSED",row(11).get("status"));assertEquals(before,row(12));
|
||||
}
|
||||
@Test void mismatchedProgressAttemptIsNotCopiedIntoCompletedRun() throws Exception {
|
||||
pair(11,"READY");execute("UPDATE ll_selfop_sync_task SET progress_json=? WHERE id=1011","{\"attemptNo\":1}");
|
||||
assertEquals(1,update());assertEquals("REUSED",row(11).get("status"));assertNull(row(11).get("progress_snapshot_json"));
|
||||
}
|
||||
}
|
||||
+3
@@ -122,6 +122,9 @@ class SelfopSalesPublisherTest {
|
||||
ordered.verify(runs).update(isNull(),any());
|
||||
ordered.verify(scopes).updateById(any(SelfopSyncScope.class));
|
||||
ordered.verify(runs).updateById(any(SelfopSyncRun.class)); ordered.verify(tasks).updateById(any(SelfopSyncTask.class));
|
||||
ordered.verify(locks).completeCoveredTasks(eq(1L),eq(scope.getScopeHash()),eq(10L),eq(run.getCoveredFromUtc()),eq(run.getCoveredUntilUtc()),any());
|
||||
ordered.verify(locks).completeReusedRuns(eq(1L),eq(scope.getSourceAccount()),eq(scope.getWarehouseCode()),eq(scope.getScopeHash()),eq(10L),any());
|
||||
ordered.verify(locks).completeCoveredMarkers(eq(1L),eq(scope.getScopeHash()),eq(10L),eq(run.getCoveredUntilUtc()),any());
|
||||
assertNotNull(SelfopSalesPublisher.class.getMethod("publish",long.class,long.class,String.class,long.class).getAnnotation(Transactional.class));
|
||||
assertEquals(10L,task.getResultPublishedRunId());assertNotNull(run.getProgressSnapshotJson());
|
||||
// Publisher constructor has no inventory repositories or inventory publication prerequisite.
|
||||
|
||||
Reference in New Issue
Block a user