添加试运行模式，修改生产环境数据重建规则

MesETL.App/HostedServices/FileInputService.cs

@@ -31,6 +31,8 @@ public class FileInputService : IInputService
     private readonly ProcessContext _context;
     private readonly DataReaderFactory _dataReaderFactory;
     private readonly long _memoryThreshold;
+    private readonly bool _dryRun;
+    private readonly int _dryRunCount;
 
     public FileInputService(ILogger<FileInputService> logger,
             IOptions<DataInputOptions> dataInputOptions,
@@ -45,12 +47,16 @@ public class FileInputService : IInputService
         _producerQueue = producerQueue;
         _dataReaderFactory = dataReaderFactory;
         _memoryThreshold = (long)(configuration.GetValue<double>("MemoryThreshold", 8) * 1024 * 1024 * 1024);
+        _dryRun = configuration.GetValue("DryRun", false);
+        _dryRunCount = configuration.GetValue("DryRunCount", 100_000);
     }
 
     public async Task ExecuteAsync(CancellationToken cancellationToken)
     {
         var inputDir = _dataInputOptions.Value.InputDir ?? throw new ApplicationException("未配置文件输入目录");
         _logger.LogInformation("***** 输入服务已启动，工作目录为：{InputDir} *****", inputDir);
+        if (_dryRun)
+            _logger.LogInformation("***** 试运行模式已开启，只读取前 {Count} 行数据 *****", _dryRunCount);
 
         var orderedInfo = GetOrderedInputInfo(inputDir);
         
@@ -74,6 +80,8 @@ public class FileInputService : IInputService
                 await _producerQueue.EnqueueAsync(record);
                 count++;
                 _context.AddInput();
+                if (_dryRun && count >= _dryRunCount)
+                    break;
             }
 
             _context.AddTableInput(info.TableName, count);
@@ -82,7 +90,7 @@ public class FileInputService : IInputService
         }
 
         _context.CompleteInput();
-        _logger.LogInformation("***** 输入服务已执行完毕 *****");
+        _logger.LogInformation("***** 输入服务{DryRun}已执行完毕 *****", _dryRun ? " (试运行)" : "");
     }
 
     public IEnumerable<FileInputInfo> GetOrderedInputInfo(string dir)

MesETL.App/HostedServices/MainHostedService.cs

@@ -163,9 +163,16 @@ public class MainHostedService : BackgroundService
     private async Task ExportResultAsync()
     {
         var sb = new StringBuilder();
-        if (_context.HasException)
+
-            sb.AppendLine("# 程序执行完毕，**但中途发生了异常**");
+        var title = (_config.GetValue("DryRun", false), _context.HasException) switch
-        else sb.AppendLine("# 程序执行完毕，没有发生错误");
+        {
+            (true, true) => "# 试运行结束，**请注意处理异常**",
+            (true, false) => "# 试运行结束，没有发生异常",
+            (false, true) => "# 程序执行完毕，**但中途发生了异常**",
+            (false, false) => "# 程序执行完毕，没有发生错误"
+        };
+        sb.AppendLine(title);
+          
         sb.AppendLine("## 处理计数");
         var processCount = new[]
         {

MesETL.App/Program.cs

@@ -210,15 +210,6 @@ async Task RunProgram()
 
                     break;
                 }
-#if !FIX_PLAN_ITEM
-                // 删除PlanID和PackageID列
-                case TableNames.OrderItem:
-                {
-                    record.RemoveField("PlanID");
-                    record.RemoveField("PackageID");
-                    break;
-                }
-#endif
                 // 将JsonStr列转换为Data列，添加CompressionType列
                 case TableNames.OrderModuleExtra:
                 {
@@ -327,6 +318,9 @@ async Task RunProgram()
                         [ "ItemID", "ShardKey", "PackageID", "CompanyID" ]
                     ));
                 }
+                
+                record.RemoveField("PlanID");
+                record.RemoveField("PackageID");
 
                 return resultList;
             }

MesETL.App/appsettings.json

@@ -2,6 +2,7 @@
   "MemoryThreshold": 6,
   "GCIntervalMilliseconds": -1,
   "UnsafeVariable": true,
+  "DryRun": false, // 试运行，仅输入每张表的前100000条数据
   "Logging": {
     "LogLevel": {
       "Default": "Trace"
@@ -11,7 +12,7 @@
     "InputDir": "D:\\Data\\DatabaseDump\\Prod_Mock_CSV_2024-12-31", // Csv数据输入目录
     "UseMock": false, // 使用模拟数据进行测试
     "MockCountMultiplier": 1, // 模拟数据量级的乘数
-    "TableOrder": ["order_item"], // 按顺序输入的表
+//    "TableOrder": ["order_item"], // 按顺序输入的表
     "TableIgnoreList": [] // 忽略输入的表
   },
   "Transform":{
@@ -44,7 +45,7 @@
   "TenantDb": // 分库配置
   {
     "TenantKey" : "CompanyID",
-    "UseDbGroup": "mock",
+    "UseDbGroup": "prod",
     "DbGroups": {
       "test": {
         "cferp_test_1": 1000,
@@ -59,13 +60,6 @@
         "mesdb_5": 20000,
         "mesdb_6": 2147483647
       },
-      "mock":{
-        "mesdb_1": 5000,
-        "mesdb_2": 10000,
-        "mesdb_3": 15000,
-        "mesdb_4": 20000,
-        "mesdb_5": 2147483647
-      },
       "mock_void":{
         "mesdb_1_void": 5000,
         "mesdb_2_void": 10000,

MesETL.Test/DatabaseToolBox.cs

@@ -12,7 +12,7 @@ namespace TestProject1;
 public class DatabaseToolBox
 {
     private readonly ITestOutputHelper _output;
-    public const string ConnStr = "Server=192.168.1.245;Port=3306;UserId=root;Password=ruixinjie!@#123;";
+    public const string ConnStr = "Server=localhost;Port=3306;UserId=root;Password=123456;";
 
     public DatabaseToolBox(ITestOutputHelper output)
     {