|
|
@@ -68,7 +68,7 @@ public class S1MdpSyncTransformService : ITransient
|
|
|
await ReportAsync(reportProgress, new S1MdpProgressUpdate(S1MdpRebuildStage.Standard, 2, 50, "标准化数据完成", result.StandardRows, S1MdpRebuildStage.Standard));
|
|
|
|
|
|
await ReportAsync(reportProgress, new S1MdpProgressUpdate(S1MdpRebuildStage.Dwd, 3, 50, "正在生成 DWD 明细"));
|
|
|
- result.DwdRows = await BuildDwdAsync(scope, batchId, now, cancellationToken);
|
|
|
+ result.DwdRows = await BuildDwdAsync(scope, batchId, now, result, cancellationToken);
|
|
|
await ReportAsync(reportProgress, new S1MdpProgressUpdate(S1MdpRebuildStage.Dwd, 3, 65, "生成 DWD 明细完成", result.DwdRows, S1MdpRebuildStage.Dwd));
|
|
|
|
|
|
await ReportAsync(reportProgress, new S1MdpProgressUpdate(S1MdpRebuildStage.Kpi, 4, 65, "正在重算 KPI"));
|
|
|
@@ -122,7 +122,7 @@ public class S1MdpSyncTransformService : ITransient
|
|
|
{
|
|
|
await EnsureS1RuntimeObjectsAsync();
|
|
|
result.StandardRows = await TransformStandardAsync(scope, batchId, now, cancellationToken);
|
|
|
- result.DwdRows = await BuildDwdAsync(scope, batchId, now, cancellationToken);
|
|
|
+ result.DwdRows = await BuildDwdAsync(scope, batchId, now, result, cancellationToken);
|
|
|
result.KpiRows = await BuildS1KpiValuesAsync(scope, batchId, now, cancellationToken);
|
|
|
result.AtomicRows = await _atomicBuild.BuildOrderDeliveryDomainForAllDatesAsync(
|
|
|
scope.TenantId, scope.FactoryId, batchId, cancellationToken);
|
|
|
@@ -282,7 +282,7 @@ public class S1MdpSyncTransformService : ITransient
|
|
|
codes, batchId, scope.TenantId, fullRefresh, "S1_MDP_INBOUND", cancellationToken,
|
|
|
scope.FactoryId, requireMatchingSourceTenant: true);
|
|
|
result.StandardRows = await TransformStandardAsync(scope, batchId, now, cancellationToken);
|
|
|
- result.DwdRows = await BuildDwdAsync(scope, batchId, now, cancellationToken);
|
|
|
+ result.DwdRows = await BuildDwdAsync(scope, batchId, now, result, cancellationToken);
|
|
|
result.KpiRows = await BuildS1KpiValuesAsync(scope, batchId, now, cancellationToken);
|
|
|
result.AtomicRows = await _atomicBuild.BuildOrderDeliveryDomainForAllDatesAsync(
|
|
|
scope.TenantId, scope.FactoryId, batchId, cancellationToken);
|
|
|
@@ -386,7 +386,9 @@ public class S1MdpSyncTransformService : ITransient
|
|
|
return total;
|
|
|
}
|
|
|
|
|
|
- private async Task<int> BuildDwdAsync(S1MdpRunScope scope, string batchId, DateTime now, CancellationToken cancellationToken)
|
|
|
+ private async Task<int> BuildDwdAsync(
|
|
|
+ S1MdpRunScope scope, string batchId, DateTime now,
|
|
|
+ S1MdpSyncTransformResult result, CancellationToken cancellationToken)
|
|
|
{
|
|
|
using var db = _db.CopyNew();
|
|
|
db.Ado.CommandTimeOut = Math.Max(db.Ado.CommandTimeOut, 180);
|
|
|
@@ -401,7 +403,18 @@ public class S1MdpSyncTransformService : ITransient
|
|
|
// 必须在所有 DWD 写入完成之后,否则读者会看到一个还没灌完的批次被标成 current。
|
|
|
// 不计入 total —— total 语义是「本轮构建的 DWD 行数」,翻牌影响的行数属于发布动作,不是新建行。
|
|
|
cancellationToken.ThrowIfCancellationRequested();
|
|
|
- await PublishCurrentSnapshotAsync(db, scope, batchId);
|
|
|
+ var publishedRows = await PublishCurrentSnapshotAsync(db, scope, batchId);
|
|
|
+
|
|
|
+ // 发布证据随运行日志一起落盘(见 MarkTransformRunSuccessAsync)。
|
|
|
+ // 走到这里就意味着两条 DWD 写入与发布语句都已成功;若中途抛错,
|
|
|
+ // 调用方的 catch 会写 FAILED,本字段保持 null —— 失败方向是保守的。
|
|
|
+ result.Publication = new S1MdpSnapshotPublication
|
|
|
+ {
|
|
|
+ Table = "dwd_requirement_examine_detail",
|
|
|
+ BatchId = batchId,
|
|
|
+ FactoryId = scope.FactoryId,
|
|
|
+ CurrentRows = publishedRows
|
|
|
+ };
|
|
|
|
|
|
return total;
|
|
|
}
|
|
|
@@ -431,7 +444,12 @@ public class S1MdpSyncTransformService : ITransient
|
|
|
/// 结果就是任务里点名要避免的「两个当前批次」永久共存。
|
|
|
/// 故必须用 COALESCE(NULLIF(factory_id,0),1) 归一 —— 这也是本仓库通用的 factory 作用域写法。
|
|
|
/// </summary>
|
|
|
- private static async Task PublishCurrentSnapshotAsync(ISqlSugarClient db, S1MdpRunScope scope, string batchId)
|
|
|
+ /// <summary>
|
|
|
+ /// 发布当前快照,并返回本批次发布后的当前行数。
|
|
|
+ /// <para>返回 0 是<b>合法结果</b>——源侧本轮没有数据时,成功发布的就是一个空快照。
|
|
|
+ /// 调用方把这个数字记进运行日志,下游才能把「正常的空」与「没发布成」区分开。</para>
|
|
|
+ /// </summary>
|
|
|
+ private static async Task<int> PublishCurrentSnapshotAsync(ISqlSugarClient db, S1MdpRunScope scope, string batchId)
|
|
|
{
|
|
|
await db.Ado.ExecuteCommandAsync(
|
|
|
"""
|
|
|
@@ -444,6 +462,19 @@ public class S1MdpSyncTransformService : ITransient
|
|
|
new SugarParameter("@BatchId", batchId),
|
|
|
new SugarParameter("@TenantId", scope.TenantId),
|
|
|
new SugarParameter("@FactoryId", scope.FactoryId));
|
|
|
+
|
|
|
+ // 复核发布结果而不是相信影响行数:ODKU 与 CASE 的影响行数都不等于「当前有几行」。
|
|
|
+ return await db.Ado.GetIntAsync(
|
|
|
+ """
|
|
|
+ SELECT COUNT(*) FROM dwd_requirement_examine_detail
|
|
|
+ WHERE tenant_id=@TenantId
|
|
|
+ AND COALESCE(NULLIF(factory_id, 0), 1)=@FactoryId
|
|
|
+ AND calc_batch_id=@BatchId
|
|
|
+ AND is_current_flag=1
|
|
|
+ """,
|
|
|
+ new SugarParameter("@BatchId", batchId),
|
|
|
+ new SugarParameter("@TenantId", scope.TenantId),
|
|
|
+ new SugarParameter("@FactoryId", scope.FactoryId));
|
|
|
}
|
|
|
|
|
|
private async Task<int> BuildS1KpiValuesAsync(S1MdpRunScope scope, string batchId, DateTime now, CancellationToken cancellationToken)
|
|
|
@@ -1712,9 +1743,29 @@ public class S1MdpSyncTransformService : ITransient
|
|
|
return string.IsNullOrWhiteSpace(triggerType) ? "AUTO" : triggerType.Trim().ToUpperInvariant();
|
|
|
}
|
|
|
|
|
|
+ /// <summary>
|
|
|
+ /// 运行摘要。改用 JsonSerializer 而非字符串插值:batchId 今天不可能含引号,
|
|
|
+ /// 但拼 JSON 是会被后来者继承的脆弱写法,且本次要加的 publish 是嵌套对象。
|
|
|
+ /// </summary>
|
|
|
private static string BuildRunSummaryJson(S1MdpSyncTransformResult result)
|
|
|
{
|
|
|
- return $$"""{"batchId":"{{result.BatchId}}","stageRows":{{result.StageRows}},"standardRows":{{result.StandardRows}},"dwdRows":{{result.DwdRows}},"kpiRows":{{result.KpiRows}}}""";
|
|
|
+ return System.Text.Json.JsonSerializer.Serialize(new
|
|
|
+ {
|
|
|
+ batchId = result.BatchId,
|
|
|
+ stageRows = result.StageRows,
|
|
|
+ standardRows = result.StandardRows,
|
|
|
+ dwdRows = result.DwdRows,
|
|
|
+ kpiRows = result.KpiRows,
|
|
|
+ publish = result.Publication is null
|
|
|
+ ? null
|
|
|
+ : new
|
|
|
+ {
|
|
|
+ table = result.Publication.Table,
|
|
|
+ batchId = result.Publication.BatchId,
|
|
|
+ factoryId = result.Publication.FactoryId,
|
|
|
+ currentRows = result.Publication.CurrentRows
|
|
|
+ }
|
|
|
+ });
|
|
|
}
|
|
|
|
|
|
private static string ResolveKpiValueTable(int metricLevel)
|
|
|
@@ -1806,6 +1857,38 @@ public sealed class S1MdpSyncTransformResult
|
|
|
public int DwdRows { get; set; }
|
|
|
public int KpiRows { get; set; }
|
|
|
public int AtomicRows { get; set; }
|
|
|
+
|
|
|
+ /// <summary>
|
|
|
+ /// 当前快照发布证据。为 null 表示本轮没走到发布这一步。
|
|
|
+ ///
|
|
|
+ /// <para><b>为什么需要它</b>:发布语句在本轮零行时命中 0 行,在库里不留任何痕迹,
|
|
|
+ /// 于是「源侧本轮确实没有数据」与「发布压根没跑成」在 DWD 里字面等同。
|
|
|
+ /// 下游的 Authority 健康判定因此只能保守地判 UNKNOWN、拦下恢复 ——
|
|
|
+ /// 哪怕这是一个完全正常的空快照。</para>
|
|
|
+ ///
|
|
|
+ /// <para>把发布结果记进运行日志,就把这个隐式事实变成了可观测事实:
|
|
|
+ /// 「本轮成功发布了批次 X,其中当前行数为 N(N 可以是 0)」。</para>
|
|
|
+ /// </summary>
|
|
|
+ public S1MdpSnapshotPublication? Publication { get; set; }
|
|
|
+}
|
|
|
+
|
|
|
+/// <summary>
|
|
|
+/// 一次当前快照发布的结果。与 <c>status='SUCCESS'</c> 写在同一条 UPDATE 里,
|
|
|
+/// 因此不存在「成功了但没有发布证据」的中间态。
|
|
|
+/// </summary>
|
|
|
+public sealed class S1MdpSnapshotPublication
|
|
|
+{
|
|
|
+ /// <summary>被发布为当前快照的表。</summary>
|
|
|
+ public string Table { get; set; } = string.Empty;
|
|
|
+
|
|
|
+ /// <summary>本轮批次号。读者据此确认「当前快照」出自哪一次运行。</summary>
|
|
|
+ public string BatchId { get; set; } = string.Empty;
|
|
|
+
|
|
|
+ /// <summary>发布作用域的工厂号。运行日志表本身没有工厂列,只能由这里承载。</summary>
|
|
|
+ public long FactoryId { get; set; }
|
|
|
+
|
|
|
+ /// <summary>发布后该批次的当前行数。<b>0 是合法值</b>,表示成功发布了一个空快照。</summary>
|
|
|
+ public int CurrentRows { get; set; }
|
|
|
}
|
|
|
|
|
|
internal sealed record S1MdpSqlCommand(string Sql, SugarParameter[] Parameters);
|