perf(selfop): build versioned warehouse daily sales summaries

This commit is contained in:
2026-10-13 09:40:55 +08:00
parent ca4e7086e9
commit 25db631ff7
24 changed files with 320 additions and 31 deletions
@@ -13,6 +13,7 @@ import java.time.LocalDateTime;
public class SelfopSyncRun {
private String executionPurpose;
private LocalDateTime buildStartedAtUtc;
private Boolean warehouseDailyReady;
private LocalDateTime supersededAtUtc;
private String progressSnapshotJson;
private LocalDateTime heartbeatAtUtc;
@@ -99,6 +99,30 @@ public interface OutboundAggregateMapper {
""")
int insertDaily(@Param("rows") List<SelfopSkuDailySales> rows);
@Insert("""
<script>
INSERT INTO ll_selfop_warehouse_daily_sales
(sales_run_id,source_account,warehouse_code,business_date,observed_quantity,invalid_row_count,source_row_count,created_at,updated_at)
SELECT sales_run_id,source_account,warehouse_code,business_date,
SUM(CASE WHEN quantity_unit='PCS' AND outbound_quantity &gt;= 0 THEN outbound_quantity END),
SUM(CASE WHEN quantity_unit='PCS' AND outbound_quantity &gt;= 0 THEN 0 ELSE 1 END),COUNT(*),MIN(created_at),MAX(updated_at)
FROM ll_selfop_sku_daily_sales WHERE id IN
<foreach collection='rows' item='r' open='(' separator=',' close=')'>#{r.id}</foreach>
GROUP BY sales_run_id,source_account,warehouse_code,business_date
ON DUPLICATE KEY UPDATE
observed_quantity=COALESCE(observed_quantity+VALUES(observed_quantity),observed_quantity,VALUES(observed_quantity)),
invalid_row_count=invalid_row_count+VALUES(invalid_row_count),
source_row_count=source_row_count+VALUES(source_row_count),updated_at=VALUES(updated_at)
</script>
""")
int insertWarehouseDaily(@Param("rows") List<SelfopSkuDailySales> rows);
@Select("SELECT COALESCE(SUM(source_row_count),0) FROM ll_selfop_warehouse_daily_sales WHERE sales_run_id=#{runId} AND source_account=#{account} AND warehouse_code=#{warehouse}")
long warehouseDailySourceRows(@Param("runId") Long runId,@Param("account") String account,@Param("warehouse") String warehouse);
@Delete("DELETE FROM ll_selfop_warehouse_daily_sales WHERE sales_run_id=#{runId}")
int deleteWarehouseDaily(@Param("runId") Long runId);
@Insert("""
<script>
INSERT INTO ll_selfop_sku_profile
@@ -35,7 +35,7 @@ public class OutboundCandidateTransaction {
SelfopSupport.require(!Set.of("PUBLISHED","REUSED","FAILED","DEAD").contains(run.getStatus()),"CANDIDATE_RUN_INVALID");
SelfopSupport.require("APPLIED".equals(held.task().getApplyState()) || "REBUILD".equals(held.task().getTaskType()),"TASK_NOT_APPLIED");
facts.assertNoBarrier(scope); validateFacts(scope);
if("READY".equals(run.getStatus())) return new Checkpoint(runId,"DONE",null,null,null,0,0,0);
if("READY".equals(run.getStatus())) {SelfopSupport.require(Boolean.TRUE.equals(run.getWarehouseDailyReady()),"WAREHOUSE_DAILY_NOT_READY");return new Checkpoint(runId,"DONE",null,null,null,0,0,0);}
Checkpoint existing=checkpoint(held.task());
if(run.getBuildStartedAtUtc()!=null && existing!=null && Objects.equals(existing.runId(),runId)) {
SelfopSupport.require("BUILDING".equals(run.getStatus()) && Objects.equals(scope.getFactRevision(),run.getCalculatedFactRevision()),"CANDIDATE_FACT_REVISION_CHANGED");
@@ -44,6 +44,7 @@ public class OutboundCandidateTransaction {
}
SelfopSupport.require(run.getBuildStartedAtUtc()==null,"CANDIDATE_CHECKPOINT_MISSING");
// A new attempt has a new run ID; failed attempts are never overwritten or emptied.
aggregates.deleteWarehouseDaily(runId);run.setWarehouseDailyReady(false);
daily.delete(new QueryWrapper<SelfopSkuDailySales>().eq("sales_run_id",runId));
profiles.delete(new QueryWrapper<SelfopSkuProfile>().eq("profile_run_id",runId));
OutboundCandidateBuilder.Coverage coverage=coverage(scope);
@@ -66,7 +67,7 @@ public class OutboundCandidateTransaction {
if(rows.isEmpty()) next=new Checkpoint(runId,"PROFILE",null,null,null,cursor.dailyRows(),0,cursor.committedBatches());
else {
for(var row:rows){row.setId(IdWorker.getId());row.setSalesRunId(runId);row.setSourceAccount(held.scope().getSourceAccount());row.setWarehouseCode(held.scope().getWarehouseCode());row.setCreatedAt(SelfopSupport.utcNow());row.setUpdatedAt(row.getCreatedAt());}
aggregates.insertDaily(rows);var last=rows.get(rows.size()-1);
aggregates.insertDaily(rows);aggregates.insertWarehouseDaily(rows);var last=rows.get(rows.size()-1);
SelfopSupport.require(!Objects.equals(cursor.customer(),last.getCustomerCode()) || !Objects.equals(cursor.sku(),last.getSkuCode()) || !Objects.equals(cursor.date(),last.getBusinessDate()),"CANDIDATE_CURSOR_STALLED");
next=new Checkpoint(runId,"DAILY",last.getCustomerCode(),last.getSkuCode(),last.getBusinessDate(),cursor.dailyRows()+rows.size(),0,cursor.committedBatches()+1);
}
@@ -90,7 +91,8 @@ public class OutboundCandidateTransaction {
SelfopSupport.require(cursor!=null && Objects.equals(cursor.runId(),runId) && "FINISH".equals(cursor.phase()),"CANDIDATE_NOT_COMPLETE");
SelfopSupport.require(daily.selectCount(new QueryWrapper<SelfopSkuDailySales>().eq("sales_run_id",runId))==cursor.dailyRows()
&& profiles.selectCount(new QueryWrapper<SelfopSkuProfile>().eq("profile_run_id",runId))==cursor.profileRows(),"CANDIDATE_ROWS_INCOMPLETE");
facts.assertNoBarrier(held.scope());held.run().setStatus("READY");save(held,new Checkpoint(runId,"DONE",null,null,null,cursor.dailyRows(),cursor.profileRows(),cursor.committedBatches()));
SelfopSupport.require(aggregates.warehouseDailySourceRows(runId,held.scope().getSourceAccount(),held.scope().getWarehouseCode())==cursor.dailyRows(),"WAREHOUSE_DAILY_ROWS_INCOMPLETE");
facts.assertNoBarrier(held.scope());held.run().setWarehouseDailyReady(true);held.run().setStatus("READY");save(held,new Checkpoint(runId,"DONE",null,null,null,cursor.dailyRows(),cursor.profileRows(),cursor.committedBatches()));
}
@Transactional(isolation=Isolation.REPEATABLE_READ,rollbackFor=Exception.class)
@@ -101,7 +103,7 @@ public class OutboundCandidateTransaction {
facts.assertNoBarrier(scope);validateFacts(scope);
if(scope.getPublishedRunId()==null)return null;
SelfopSyncRun published=runs.selectById(scope.getPublishedRunId());
if(published==null || !"PUBLISHED".equals(published.getStatus()) || !"BUSINESS".equals(published.getExecutionPurpose())
if(published==null || !Boolean.TRUE.equals(published.getWarehouseDailyReady()) || !"PUBLISHED".equals(published.getStatus()) || !"BUSINESS".equals(published.getExecutionPurpose())
|| published.getBuildStartedAtUtc()==null || !Objects.equals(scopeId,published.getScopeId()) || !Objects.equals(scope.getScopeHash(),published.getScopeHash())
|| !Objects.equals(scope.getFactRevision(),published.getCalculatedFactRevision()) || !Objects.equals(scope.getFactRevision(),scope.getPublishedFactRevision())
|| !Objects.equals(SelfopSupport.read(held.run().getScopeSnapshotJson(),Map.class),SelfopSupport.read(published.getScopeSnapshotJson(),Map.class)))return null;
@@ -8,7 +8,7 @@ import org.apache.ibatis.annotations.*;
public interface AnalyticsReadMapper {
String RUN_COLUMNS="id,task_id,scope_id,source_account,warehouse_code,run_role,scope_hash,scope_snapshot_json,status,"
+"coverage_status,covered_from_utc,covered_until_utc,complete_day_from,complete_day_until,"
+"fetch_started_at_utc,fetch_finished_at_utc,published_at_utc,calculated_fact_revision,error_code,error_message,superseded_at_utc,execution_purpose";
+"fetch_started_at_utc,fetch_finished_at_utc,published_at_utc,calculated_fact_revision,error_code,error_message,superseded_at_utc,execution_purpose,warehouse_daily_ready";
@Select("SELECT id,source_account,warehouse_code,data_family,scope_hash,business_timezone,source_timezone,"
+"stale_after_seconds,published_run_id,fact_revision,published_fact_revision,rebuild_required,"
+"known_order_round_finished_at_utc,history_round_finished_at_utc,review_target_days "
@@ -19,7 +19,7 @@ public final class AnalyticsReadRows {
LocalDate completeDayFrom, LocalDate completeDayUntil,
LocalDateTime fetchStartedAtUtc, LocalDateTime fetchFinishedAtUtc,
LocalDateTime publishedAtUtc, Long calculatedFactRevision,
String errorCode, String errorMessage, LocalDateTime supersededAtUtc, String executionPurpose) {}
String errorCode, String errorMessage, LocalDateTime supersededAtUtc, String executionPurpose, Boolean warehouseDailyReady) {}
public record Profile(String customerCode, String skuCode, String productName, String quantityUnit,
LocalDateTime firstSeenAtUtc, LocalDateTime earliestEvidenceAtUtc,
String evidenceType, Long evidenceRunId, Long lastSeenRunId) {}
@@ -18,6 +18,8 @@ public class DashboardQueryService {
public DashboardResponses.Dashboard dashboard(AnalyticsQuery input){
AnalyticsQuery q=DashboardRequest.normalize(input);
var snapshot=contextService.context(q,true,false,false);var c=snapshot.context();
if(snapshot.sales()!=null&&!Boolean.TRUE.equals(snapshot.sales().warehouseDailyReady()))
throw new com.baiyue.common.core.exception.BizException("该销量版本尚未生成仓库每日汇总,请重新发布统计后再查询");
var base=contextService.request(snapshot,"DASHBOARD");LocalDate fixedFrom=c.anchorDate().minusDays(30),fixedUntil=c.anchorDate();
var p=new AnalyticsSqlRequest(base.account(),base.warehouse(),null,null,base.salesRun(),base.inventoryRun(),fixedFrom,fixedUntil,base.anchor(),fixedFrom,
base.completeFrom(),base.completeUntil(),base.coveredFrom(),base.coveredUntil(),base.salesKnown(),base.formalSales(),base.complete30(),base.complete30(),base.stockFresh(),0,10,"DASHBOARD");
@@ -12,12 +12,11 @@ public interface DashboardReadMapper {
record Ranked(String rankCode,String customerCode,String skuCode,String productName,BigDecimal mainValue,
BigDecimal ownerOutbound,Long orderCount,BigDecimal physicalQuantity,BigDecimal availableQuantity,
BigDecimal frozenQuantity,BigDecimal defectivePhysicalQuantity,BigDecimal outboundQuantity30d,BigDecimal coverageDays){}
// Dense 366-day EXPLAIN: the natural unique-index plan performs non-covering reads;
// the date-ordered covering index also matches warehouse daily GROUP BY/ORDER BY.
@Select("SELECT /*+ INDEX(ll_selfop_sku_daily_sales idx_selfop_daily_window_cover) */ business_date,SUM(CASE WHEN quantity_unit='PCS' AND outbound_quantity>=0 THEN outbound_quantity END) observed_quantity,"
+"COUNT(CASE WHEN quantity_unit IS NULL OR quantity_unit<>'PCS' OR outbound_quantity IS NULL OR outbound_quantity<0 THEN 1 END) invalid_row_count "
+"FROM ll_selfop_sku_daily_sales WHERE sales_run_id=#{p.salesRun} AND source_account=#{p.account} AND warehouse_code=#{p.warehouse} "
+"AND ((business_date>=#{chartFrom} AND business_date<#{chartUntil}) OR (business_date>=#{p.from} AND business_date<#{p.until})) GROUP BY business_date ORDER BY business_date")
// Published-version warehouse facts retain valid subtotal and invalid-row evidence.
// Coverage and formal-zero proof still come from the pinned run, never row presence.
@Select("SELECT business_date,observed_quantity,invalid_row_count "
+"FROM ll_selfop_warehouse_daily_sales WHERE sales_run_id=#{p.salesRun} AND source_account=#{p.account} AND warehouse_code=#{p.warehouse} "
+"AND ((business_date>=#{chartFrom} AND business_date<#{chartUntil}) OR (business_date>=#{p.from} AND business_date<#{p.until})) ORDER BY business_date")
List<Day> daily(@Param("p") AnalyticsSqlRequest request,@Param("chartFrom") LocalDate chartFrom,@Param("chartUntil") LocalDate chartUntil);
@SelectProvider(type=DashboardReadSql.class,method="report")
List<Map<String,Object>> report(@Param("p") AnalyticsSqlRequest request,@Param("o") AnalyticsInventoryOptions options);
@@ -20,6 +20,7 @@ public interface SalesRetentionMapper {
AND r.inventory_daily_sample_date IS NULL AND r.inventory_sample_rule_version IS NULL
AND r.id != #{current} AND (#{previous} IS NULL OR r.id != #{previous})
AND (EXISTS (SELECT 1 FROM ll_selfop_sku_daily_sales d WHERE d.sales_run_id=r.id)
OR EXISTS (SELECT 1 FROM ll_selfop_warehouse_daily_sales w WHERE w.sales_run_id=r.id)
OR EXISTS (SELECT 1 FROM ll_selfop_sku_profile p WHERE p.profile_run_id=r.id))
AND NOT EXISTS (
SELECT 1 FROM ll_selfop_sync_run input JOIN ll_selfop_sync_task t ON t.id=input.task_id
@@ -39,6 +40,8 @@ public interface SalesRetentionMapper {
long activeReferences(@Param("account") String account,@Param("warehouse") String warehouse,@Param("id") Long id);
@Delete("DELETE FROM ll_selfop_sku_daily_sales WHERE sales_run_id=#{id} LIMIT #{batchSize}")
int deleteDaily(@Param("id") Long id,@Param("batchSize") int batchSize);
@Delete("DELETE FROM ll_selfop_warehouse_daily_sales WHERE sales_run_id=#{id} LIMIT #{batchSize}")
int deleteWarehouseDaily(@Param("id") Long id,@Param("batchSize") int batchSize);
@Delete("DELETE FROM ll_selfop_sku_profile WHERE profile_run_id=#{id} LIMIT #{batchSize}")
int deleteProfiles(@Param("id") Long id,@Param("batchSize") int batchSize);
}
@@ -16,7 +16,7 @@ public class SalesRetentionService {
private final SelfopAnalyticsProperties config;
public record Preview(Long scopeId,Long current,Long previous,List<Long> eligibleRunIds,LocalDateTime cutoff) { }
public record Deleted(Long runId,int dailyRows,int profileRows) { }
public record Deleted(Long runId,int dailyRows,int profileRows,int warehouseDailyRows) { }
@Transactional(isolation=Isolation.READ_COMMITTED,rollbackFor=Exception.class)
public Preview preview(Long scopeId,LocalDateTime now) {
@@ -46,7 +46,7 @@ public class SalesRetentionService {
&& !Objects.equals(runId,previous),"SALES_VERSION_PROTECTED");
SelfopSupport.require(mapper.activeReferences(scope.getSourceAccount(),scope.getWarehouseCode(),runId)==0,"SALES_INPUT_REFERENCED");
int size=Math.max(1,Math.min(500,config.getPersistenceBatchSize()));
return new Deleted(runId,mapper.deleteDaily(runId,size),mapper.deleteProfiles(runId,size));
return new Deleted(runId,mapper.deleteDaily(runId,size),mapper.deleteProfiles(runId,size),mapper.deleteWarehouseDaily(runId,size));
}
private void validateScope(SelfopSyncScope scope) {
SelfopSupport.require(scope!=null && "OUTBOUND".equals(scope.getDataFamily()),"SALES_RETENTION_SCOPE_INVALID");
@@ -25,6 +25,7 @@ public class SelfopSalesPublisher {
SelfopCollectionPolicy.requireExecution(config,"BUSINESS");SelfopCollectionPolicy.requireBusiness(run);
require(scope!=null&&run!=null&&"OUTBOUND".equals(scope.getDataFamily()),"SELFOP_SALES_SCOPE_INVALID");
SelfopLeaseService.check(new SelfopLeaseService.Lease(scopeId,owner,fence,false),scope.getLeaseOwner(),scope.getFenceToken(),scope.getLeaseUntilUtc(),now);
require(Boolean.TRUE.equals(run.getWarehouseDailyReady()),"SELFOP_WAREHOUSE_DAILY_NOT_READY");
if(Objects.equals(scope.getPublishedRunId(),runId)&&"PUBLISHED".equals(run.getStatus()))return;
require(scope.getSourceTimezone()!=null&&!scope.getSourceTimezone().isBlank(),"SOURCE_TIMEZONE_UNCONFIRMED");
require(config.getQuantityUnit()!=null&&!config.getQuantityUnit().isBlank(),"QUANTITY_UNIT_UNCONFIRMED");
@@ -52,6 +53,7 @@ public class SelfopSalesPublisher {
SelfopCollectionPolicy.requireExecution(config,"BUSINESS");require(config.isSalesPublicationEnabled(),"SELFOP_SALES_PUBLICATION_DISABLED");
var scope=locks.lockScope(scopeId);var run=locks.lockRun(runId);var task=locks.lockTask(run.getTaskId());var published=runs.selectById(publishedId);
SelfopCollectionPolicy.requireBusiness(task);SelfopCollectionPolicy.requireBusiness(run);SelfopCollectionPolicy.requireBusiness(published);
require(published!=null&&Boolean.TRUE.equals(published.getWarehouseDailyReady()),"SELFOP_WAREHOUSE_DAILY_NOT_READY");
SelfopLeaseService.check(new SelfopLeaseService.Lease(scopeId,owner,fence,false),scope.getLeaseOwner(),scope.getFenceToken(),scope.getLeaseUntilUtc(),utcNow());
require(Objects.equals(run.getScopeId(),scopeId)&&Objects.equals(run.getFenceToken(),fence)&&Objects.equals(run.getScopeHash(),scope.getScopeHash()),"SELFOP_REUSE_SCOPE_INVALID");
require(Objects.equals(scope.getPublishedRunId(),publishedId)&&"PUBLISHED".equals(published.getStatus())&&Objects.equals(published.getCalculatedFactRevision(),scope.getFactRevision())