diff --git a/app/Console/Commands/BackfillBusinessDataColumnar.php b/app/Console/Commands/BackfillBusinessDataColumnar.php
new file mode 100644
index 00000000..29e4a7fc
--- /dev/null
+++ b/app/Console/Commands/BackfillBusinessDataColumnar.php
@@ -0,0 +1,257 @@
+<?php
+
+namespace App\Console\Commands;
+
+use GuzzleHttp\Psr7\StreamWrapper;
+use Illuminate\Console\Command;
+use Illuminate\Support\Facades\Schema;
+use Illuminate\Support\Facades\Storage;
+use Illuminate\Support\Str;
+use Modules\EcommerceFinance\Repositories\ColumnarBusinessMirrorRepository;
+use Modules\EcommerceFinance\Contracts\BusinessMirrorDataset;
+use Modules\EcommerceFinance\Services\BusinessDataPlacement;
+use Modules\EcommerceFinance\Services\BusinessMirrorDatasetRegistry;
+use Modules\EcommerceFinance\Services\ClickHouseClient;
+use Modules\EcommerceFinance\Services\ClickHouseSchemaManager;
+use RuntimeException;
+use Throwable;
+
+class BackfillBusinessDataColumnar extends Command
+{
+    protected $signature = 'business-data:backfill-columnar
+        {dataset : config/upstream_mirrors.php 中的数据集键}
+        {--batch=5000 : 每批从现有 MariaDB 镜像读取的行数}
+        {--max-batches=10 : 本轮最多批次数，0 表示直到完成}
+        {--restart : 使用新代次从头重建，不删除旧代次}
+        {--archive : 对账通过后将 32 个哈希分区分别写入 Parquet 归档}';
+    protected $description = '从现有 MariaDB 镜像断点复制到 ClickHouse，不访问上游、不删除原表';
+
+    public function handle(
+        BusinessMirrorDatasetRegistry $datasets,
+        ClickHouseSchemaManager $clickHouseSchemas,
+        ColumnarBusinessMirrorRepository $columnar,
+        BusinessDataPlacement $placements,
+        ClickHouseClient $clickHouse,
+    ): int {
+        $datasetKey = (string) $this->argument('dataset');
+
+        try {
+            $dataset = $datasets->get($datasetKey);
+            $targetStore = $placements->targetStore($datasetKey);
+            if (! in_array($targetStore, ['clickhouse', 'parquet'], true)) {
+                throw new RuntimeException('business_data_dataset_not_columnar');
+            }
+            if (! $clickHouse->enabled() || ! $clickHouse->ping()) {
+                throw new RuntimeException('business_clickhouse_unavailable');
+            }
+
+            $connection = $placements->connection();
+            $schema = Schema::connection($connection->getName());
+            if (! $schema->hasTable('business_data_placements') || ! $schema->hasTable($dataset->targetTable())) {
+                throw new RuntimeException('business_data_source_or_state_table_missing');
+            }
+
+            $columns = $clickHouseSchemas->ensure($dataset);
+            $columnNames = array_map(static fn (array $column): string => (string) $column['column_name'], $columns);
+            $state = $placements->state($datasetKey);
+            $restart = (bool) $this->option('restart');
+            $verified = ! $restart
+                && is_array($state)
+                && in_array((string) ($state['phase'] ?? ''), ['verified', 'active'], true)
+                && ! empty($state['backfill_generation'])
+                && ! empty($state['verified_at']);
+            if ($verified) {
+                $generation = (string) $state['backfill_generation'];
+                $this->info(sprintf(
+                    '%s：复用已对账代次 %s（%d 行）；MariaDB 原表仍保留。',
+                    $datasetKey,
+                    $generation,
+                    (int) ($state['target_rows'] ?? 0),
+                ));
+                if ((bool) $this->option('archive')) {
+                    $this->archive($datasetKey, $dataset, $generation, $columnar, $placements);
+                }
+
+                return self::SUCCESS;
+            }
+            $resume = ! $restart
+                && is_array($state)
+                && in_array((string) ($state['phase'] ?? ''), ['backfilling', 'failed'], true)
+                && ! empty($state['backfill_generation']);
+
+            $generation = $resume ? (string) $state['backfill_generation'] : (string) Str::uuid();
+            $cursor = $resume ? (string) ($state['backfill_cursor'] ?? '') : '';
+            $cutoff = $resume && ! empty($state['source_cutoff_at'])
+                ? (string) $state['source_cutoff_at']
+                : now()->format('Y-m-d H:i:s.u');
+            $sourceQuery = $connection->table($dataset->targetTable())
+                ->where('_mirror_synced_at', '<=', $cutoff);
+            $sourceRows = $resume ? (int) $state['source_rows'] : (int) (clone $sourceQuery)->count();
+            $backfilledRows = $resume ? (int) $state['backfilled_rows'] : 0;
+
+            $placements->save($datasetKey, [
+                'target_store' => $targetStore,
+                'phase' => 'backfilling',
+                'backfill_generation' => $generation,
+                'backfill_cursor' => $cursor === '' ? null : $cursor,
+                'source_cutoff_at' => $cutoff,
+                'source_rows' => $sourceRows,
+                'backfilled_rows' => $backfilledRows,
+                'target_rows' => 0,
+                'verified_at' => null,
+                'last_error_code' => null,
+            ]);
+
+            $batchSize = min(max((int) $this->option('batch'), 100), 10000);
+            $maxBatches = max((int) $this->option('max-batches'), 0);
+            $batches = 0;
+
+            while ($maxBatches === 0 || $batches < $maxBatches) {
+                $query = $connection->table($dataset->targetTable())
+                    ->where('_mirror_synced_at', '<=', $cutoff)
+                    ->orderBy('_mirror_key')
+                    ->limit($batchSize);
+                if ($cursor !== '') {
+                    $query->where('_mirror_key', '>', $cursor);
+                }
+                $rows = $query->get(['_mirror_key', ...$columnNames, '_mirror_synced_at'])
+                    ->map(static fn (object $row): array => (array) $row)
+                    ->all();
+                if ($rows === []) {
+                    break;
+                }
+
+                $columnar->storeBackfillBatch($dataset, $columns, $rows, $generation);
+                $cursor = (string) $rows[array_key_last($rows)]['_mirror_key'];
+                $backfilledRows += count($rows);
+                $batches++;
+                $placements->save($datasetKey, [
+                    'phase' => 'backfilling',
+                    'backfill_generation' => $generation,
+                    'backfill_cursor' => $cursor,
+                    'source_cutoff_at' => $cutoff,
+                    'source_rows' => $sourceRows,
+                    'backfilled_rows' => $backfilledRows,
+                    'last_error_code' => null,
+                ]);
+
+                if (count($rows) < $batchSize) {
+                    break;
+                }
+            }
+
+            if ($backfilledRows < $sourceRows) {
+                $this->info(sprintf('%s：本轮 %d 批，已复制 %d/%d 行；可重复执行续传。', $datasetKey, $batches, $backfilledRows, $sourceRows));
+
+                return self::SUCCESS;
+            }
+
+            $targetRows = $columnar->count($dataset, $generation);
+            $distinctKeys = $columnar->distinctKeys($dataset, $generation);
+            if ($targetRows !== $sourceRows || $distinctKeys !== $sourceRows) {
+                $mismatch = sprintf(
+                    'business_data_backfill_count_mismatch:%d:%d:%d',
+                    $sourceRows,
+                    $targetRows,
+                    $distinctKeys,
+                );
+                $placements->save($datasetKey, [
+                    'phase' => 'failed',
+                    'target_rows' => $targetRows,
+                    'last_error_code' => $mismatch,
+                ]);
+                throw new RuntimeException($mismatch);
+            }
+
+            $placements->save($datasetKey, [
+                'phase' => 'verified',
+                'target_rows' => $targetRows,
+                'verified_at' => now(),
+                'last_error_code' => null,
+            ]);
+            $this->info(sprintf('%s：MariaDB 与 ClickHouse 已对账，代次 %s 共 %d 行；尚未切换页面或删除原表。', $datasetKey, $generation, $targetRows));
+
+            if ((bool) $this->option('archive')) {
+                $this->archive($datasetKey, $dataset, $generation, $columnar, $placements);
+            }
+
+            return self::SUCCESS;
+        } catch (Throwable $exception) {
+            $errorCode = preg_match('/^(business_|clickhouse_)[A-Za-z0-9_:.-]+$/', $exception->getMessage()) === 1
+                ? $exception->getMessage()
+                : $exception::class;
+            if (isset($placements) && Schema::connection($placements->connection()->getName())->hasTable('business_data_placements')) {
+                try {
+                    $placements->save($datasetKey, [
+                        'phase' => 'failed',
+                        'last_error_code' => mb_substr($errorCode, 0, 120),
+                    ]);
+                } catch (Throwable) {
+                    // Preserve the original failure when the control store is unavailable.
+                }
+            }
+            report($exception);
+            $this->error($datasetKey . '：列式回填失败（' . $errorCode . '），现有 MariaDB 数据未删除、页面未切换。');
+
+            return self::FAILURE;
+        }
+    }
+
+    private function archive(
+        string $datasetKey,
+        BusinessMirrorDataset $dataset,
+        string $generation,
+        ColumnarBusinessMirrorRepository $columnar,
+        BusinessDataPlacement $placements,
+    ): void {
+        $diskName = (string) config('business_data.archive.disk', 'business_uploads_qiniu');
+        $prefix = trim((string) config('business_data.archive.prefix', 'business-data-archive'), '/');
+        $slug = trim((string) preg_replace('/[^A-Za-z0-9_.-]+/', '-', $datasetKey), '-');
+        $disk = Storage::disk($diskName);
+
+        for ($partition = 0; $partition < 32; $partition++) {
+            $partitionKey = sprintf('hash-%02d', $partition);
+            $objectKey = sprintf('%s/%s/%s/%s.parquet', $prefix, $slug, $generation, $partitionKey);
+            $existing = $placements->connection()->table('business_data_archives')
+                ->where('dataset_key', $datasetKey)
+                ->where('generation', $generation)
+                ->where('partition_key', $partitionKey)
+                ->first();
+            if ($existing !== null
+                && (string) ($existing->status ?? '') === 'verified'
+                && $disk->exists($objectKey)
+                && (int) $disk->size($objectKey) === (int) ($existing->byte_size ?? 0)) {
+                continue;
+            }
+            $stream = $columnar->parquetPartition($dataset, $generation, $partition);
+            $resource = StreamWrapper::getResource($stream);
+            try {
+                $disk->writeStream($objectKey, $resource);
+            } finally {
+                fclose($resource);
+            }
+
+            $bytes = (int) $disk->size($objectKey);
+            $stats = $columnar->partitionStats($dataset, $generation, $partition);
+            $rowCount = $stats['row_count'];
+            if ($bytes <= 0 || ($rowCount > 0 && $bytes < 8)) {
+                throw new RuntimeException('business_data_archive_verification_failed');
+            }
+            $placements->connection()->table('business_data_archives')->updateOrInsert(
+                ['dataset_key' => $datasetKey, 'generation' => $generation, 'partition_key' => $partitionKey],
+                [
+                    'object_key' => $objectKey,
+                    'row_count' => $rowCount,
+                    'byte_size' => $bytes,
+                    'logical_checksum' => $stats['logical_checksum'],
+                    'status' => 'verified',
+                    'verified_at' => now(),
+                    'created_at' => now(),
+                    'updated_at' => now(),
+                ],
+            );
+        }
+
+        $this->info($datasetKey . '：32 个 Parquet 哈希分区已写入对象存储并登记清单；MariaDB 原表仍保留。');
+    }
+}
diff --git a/app/Console/Commands/BusinessDataPlan.php b/app/Console/Commands/BusinessDataPlan.php
new file mode 100644
index 00000000..b29a27af
--- /dev/null
+++ b/app/Console/Commands/BusinessDataPlan.php
@@ -0,0 +1,89 @@
+<?php
+
+namespace App\Console\Commands;
+
+use Illuminate\Console\Command;
+use Modules\EcommerceFinance\Services\BusinessDataPlacement;
+use Modules\EcommerceFinance\Services\BusinessMirrorDatasetRegistry;
+use Throwable;
+
+class BusinessDataPlan extends Command
+{
+    protected $signature = 'business-data:plan {--json : 输出机器可读 JSON}';
+    protected $description = '只读核对业务镜像数据集的页面用途、当前体量和目标存储层';
+
+    public function handle(BusinessMirrorDatasetRegistry $datasets, BusinessDataPlacement $placements): int
+    {
+        try {
+            $registered = $datasets->all();
+            $policies = $placements->policies();
+            $codeOwned = array_keys((array) config('upstream_mirrors.datasets', []));
+            $missing = array_values(array_diff($codeOwned, array_keys($policies)));
+            $unknown = array_values(array_diff(array_keys($policies), $codeOwned));
+            $sizes = $this->tableSizes($placements);
+            $rows = [];
+
+            foreach ($registered as $key => $dataset) {
+                $policy = in_array($key, $missing, true)
+                    ? ['role' => 'unclassified', 'consumer' => '', 'target_store' => 'unclassified', 'sync' => 'unknown']
+                    : $placements->policy($key);
+                $size = $sizes[$dataset->targetTable()] ?? ['rows' => null, 'bytes' => null];
+                $state = $placements->state($key);
+                $rows[] = [
+                    'dataset' => $key,
+                    'table' => $dataset->targetTable(),
+                    'role' => $policy['role'] ?? 'unclassified',
+                    'consumer' => $policy['consumer'] ?? '',
+                    'target_store' => $policy['target_store'] ?? 'unclassified',
+                    'sync' => $policy['sync'] ?? 'always',
+                    'estimated_rows' => $size['rows'],
+                    'size_mib' => $size['bytes'] === null ? null : round($size['bytes'] / 1024 / 1024, 1),
+                    'migration_phase' => $state['phase'] ?? 'not_started',
+                ];
+            }
+
+            $payload = ['datasets' => $rows, 'missing_policies' => $missing, 'unknown_policies' => $unknown];
+            if ($this->option('json')) {
+                $this->line(json_encode($payload, JSON_UNESCAPED_UNICODE | JSON_THROW_ON_ERROR));
+            } else {
+                $this->table(
+                    ['数据集', '当前表', '角色', '页面/用途', '目标层', '同步', '估算行数', 'MiB', '迁移阶段'],
+                    array_map(static fn (array $row): array => array_values($row), $rows),
+                );
+                if ($missing !== []) {
+                    $this->error('缺少存储策略：' . implode(', ', $missing));
+                }
+                if ($unknown !== []) {
+                    $this->error('存在无对应数据集的存储策略：' . implode(', ', $unknown));
+                }
+            }
+
+            return $missing === [] && $unknown === [] ? self::SUCCESS : self::FAILURE;
+        } catch (Throwable $exception) {
+            report($exception);
+            $this->error('无法读取业务数据分层计划，请确认共享系统库和业务镜像库连接可用。');
+
+            return self::FAILURE;
+        }
+    }
+
+    /** @return array<string,array{rows:?int,bytes:?int}> */
+    private function tableSizes(BusinessDataPlacement $placements): array
+    {
+        $connection = $placements->connection();
+        if ($connection->getDriverName() !== 'mysql') {
+            return [];
+        }
+
+        $database = $connection->getDatabaseName();
+        return $connection->table('information_schema.TABLES')
+            ->where('TABLE_SCHEMA', $database)
+            ->get(['TABLE_NAME', 'TABLE_ROWS', 'DATA_LENGTH', 'INDEX_LENGTH'])
+            ->mapWithKeys(static fn (object $row): array => [
+                (string) $row->TABLE_NAME => [[
+                    'rows' => $row->TABLE_ROWS === null ? null : (int) $row->TABLE_ROWS,
+                    'bytes' => (int) $row->DATA_LENGTH + (int) $row->INDEX_LENGTH,
+                ]],
+            ])->all();
+    }
+}
diff --git a/compose.jiangling.yml b/compose.jiangling.yml
index 883a5160..f3d9f8ac 100644
--- a/compose.jiangling.yml
+++ b/compose.jiangling.yml
@@ -49,6 +49,40 @@ services:
       retries: 12
       start_period: 30s
 
+  # 大事实表的列式存储。当前发布只提供受控回填能力；页面仍读 MariaDB，
+  # 直到对应数据集完成行数对账、Parquet 归档并显式切换。
+  clickhouse:
+    image: m.daocloud.io/docker.io/clickhouse/clickhouse-server:26.8.2.7-alpine
+    pull_policy: missing
+    container_name: caiwu_clickhouse
+    restart: unless-stopped
+    env_file:
+      - ./runtime/clickhouse.env
+    environment:
+      TZ: Asia/Shanghai
+    volumes:
+      - caiwu_clickhouse_database:/var/lib/clickhouse
+      - caiwu_clickhouse_logs:/var/log/clickhouse-server
+    mem_limit: 4g
+    cpus: "2.00"
+    ulimits:
+      nofile:
+        soft: 262144
+        hard: 262144
+    logging:
+      driver: json-file
+      options:
+        max-size: "20m"
+        max-file: "5"
+    healthcheck:
+      test:
+        - "CMD-SHELL"
+        - "clickhouse-client --user \"$$CLICKHOUSE_USER\" --password \"$$CLICKHOUSE_PASSWORD\" --query 'SELECT 1' >/dev/null"
+      interval: 10s
+      timeout: 5s
+      retries: 12
+      start_period: 30s
+
   app:
     image: caiwu_akaunting:${IMAGE_TAG:?IMAGE_TAG is required}
     pull_policy: never
@@ -63,6 +97,7 @@ services:
       RUN_MIGRATIONS: ${RUN_MIGRATIONS:-0}
     env_file:
       - ./runtime/qiniu-storage.env
+      - ./runtime/clickhouse.env
     ports:
       - "127.0.0.1:18088:18088"
     volumes:
@@ -84,6 +119,10 @@ volumes:
     name: caiwu_storage
   caiwu_config_database:
     name: caiwu_config_database
+  caiwu_clickhouse_database:
+    name: caiwu_clickhouse_database
+  caiwu_clickhouse_logs:
+    name: caiwu_clickhouse_logs
 
 networks:
   config_database:
diff --git a/config/business_data.php b/config/business_data.php
new file mode 100644
index 00000000..b5fae3c5
--- /dev/null
+++ b/config/business_data.php
@@ -0,0 +1,143 @@
+<?php
+
+/*
+|--------------------------------------------------------------------------
+| Shared business-data placement
+|--------------------------------------------------------------------------
+|
+| MariaDB remains the control plane and serving store for dimensions,
+| operational records and compact aggregates. Large analytical facts are
+| copied to ClickHouse from the existing mirror before any reader cutover.
+| Immutable Parquet archives are written only after reconciliation succeeds.
+|
+| `target_store` is a migration destination, not an activation switch. The
+| application continues to read MariaDB until a verified placement record is
+| explicitly activated. No policy below deletes an existing mirror table.
+|
+*/
+
+return [
+    'clickhouse' => [
+        'enabled' => filter_var(env('BUSINESS_CLICKHOUSE_ENABLED', false), FILTER_VALIDATE_BOOL),
+        'url' => env('BUSINESS_CLICKHOUSE_URL', 'http://clickhouse:8123'),
+        'database' => env('BUSINESS_CLICKHOUSE_DATABASE', 'caiwu_business_analytics'),
+        'username' => env('BUSINESS_CLICKHOUSE_USERNAME', 'caiwu_app'),
+        'password' => env('BUSINESS_CLICKHOUSE_PASSWORD', ''),
+        'connect_timeout_seconds' => (float) env('BUSINESS_CLICKHOUSE_CONNECT_TIMEOUT', 3),
+        'query_timeout_seconds' => (float) env('BUSINESS_CLICKHOUSE_QUERY_TIMEOUT', 30),
+        'stream_timeout_seconds' => (float) env('BUSINESS_CLICKHOUSE_STREAM_TIMEOUT', 1800),
+    ],
+
+    'archive' => [
+        'disk' => env('BUSINESS_DATA_ARCHIVE_DISK', 'business_uploads_qiniu'),
+        'prefix' => trim((string) env('BUSINESS_DATA_ARCHIVE_PREFIX', 'business-data-archive'), '/'),
+    ],
+
+    /*
+     * Every code-owned dataset in config/upstream_mirrors.php is classified.
+     * `sync` values:
+     * - always: keep the existing central incremental sync running;
+     * - after_columnar_activation: do not grow MariaDB before a columnar
+     *   generation has been verified and activated;
+     * - disabled: no current page consumes the dataset; retain/archive the
+     *   existing table, but do not keep querying the upstream source.
+     */
+    'datasets' => [
+        // Small organization and catalog dimensions stay in MariaDB.
+        'operations_replica.rpt.fys_companyInfo' => ['role' => 'dimension', 'target_store' => 'mariadb', 'consumer' => '公司资料'],
+        'operations_replica.rpt.fys_shopInfo' => ['role' => 'dimension', 'target_store' => 'mariadb', 'consumer' => '店铺资料/店铺报表'],
+        'operations_replica.rpt.fys_TrademarkInfo' => ['role' => 'dimension', 'target_store' => 'mariadb', 'consumer' => '商标资料'],
+        'operations_replica.rpt.jst_sku' => ['role' => 'dimension', 'target_store' => 'mariadb', 'consumer' => '商品资料/售后费用'],
+        'operations_replica.rpt.jst_warehouseInfo' => ['role' => 'dimension', 'target_store' => 'mariadb', 'consumer' => '售后费用'],
+        'operations_replica.rpt.jst_shopInfo' => ['role' => 'dimension', 'target_store' => 'mariadb', 'consumer' => '预留聚水潭店铺维表'],
+
+        // Large analytical facts move to ClickHouse after a MariaDB backfill.
+        'operations_replica.rpt.jst_Orders' => [
+            'role' => 'fact', 'target_store' => 'clickhouse', 'consumer' => '售后费用',
+            'partition_column' => 'order_date', 'activation_group' => 'order-analysis',
+        ],
+        'operations_replica.rpt.jst_OrderItem' => [
+            'role' => 'fact', 'target_store' => 'clickhouse', 'consumer' => '售后费用',
+            'activation_group' => 'order-analysis',
+        ],
+        'operations_replica.rpt.ads_itemid_sales_cost_ratio_v2_di' => [
+            'role' => 'fact', 'target_store' => 'clickhouse', 'consumer' => '推广费用',
+            'partition_column' => 'stat_date', 'activation_group' => 'promotion-analysis',
+        ],
+        'operations_replica.rpt.jst_returnAndRefundInfo' => [
+            'role' => 'fact', 'target_store' => 'clickhouse', 'consumer' => '售后费用/非京喜退款',
+            'partition_column' => 'as_date', 'activation_group' => 'order-analysis',
+        ],
+        'operations_replica.rpt.jst_returnAndRefundItem' => [
+            'role' => 'fact', 'target_store' => 'clickhouse', 'consumer' => '售后费用/非京喜退款',
+            'activation_group' => 'order-analysis',
+        ],
+        'operations_replica.rpt.jst_originalBillInfo' => [
+            'role' => 'fact', 'target_store' => 'clickhouse', 'consumer' => '通用原始账单（当前无页面直读）',
+            'partition_column' => 'fee_time', 'sync' => 'after_columnar_activation',
+        ],
+
+        // This 20M+ row mapping is not queried by the current promotion page.
+        // Keep/archive its existing data, but stop upstream refresh immediately.
+        'operations_replica.rpt.jst_skumap' => [
+            'role' => 'archive_only', 'target_store' => 'parquet', 'consumer' => '当前无页面读取',
+            'sync' => 'disabled',
+        ],
+
+        // Work orders are current-state operational data and remain row-store.
+        'operations_replica.rpt.work_order_main' => ['role' => 'operational', 'target_store' => 'mariadb', 'consumer' => '协同工单'],
+        'operations_replica.rpt.work_order_sub' => ['role' => 'operational', 'target_store' => 'mariadb', 'consumer' => '协同工单'],
+        'operations_replica.rpt.work_order_form_config' => ['role' => 'dimension', 'target_store' => 'mariadb', 'consumer' => '协同工单'],
+        'operations_replica.rpt.work_order_form_value' => ['role' => 'operational', 'target_store' => 'mariadb', 'consumer' => '协同工单'],
+        'jiansun.dbo.jst_workOrderInfo' => ['role' => 'operational', 'target_store' => 'mariadb', 'consumer' => '小额打款工单'],
+        'jiansun.dbo.jst_small_payment_order_map' => ['role' => 'operational', 'target_store' => 'mariadb', 'consumer' => '小额打款工单'],
+
+        // Account bills and supplier pages are bounded serving datasets today.
+        'jiansun.dbo.tmall_spread_bills' => ['role' => 'serving_fact', 'target_store' => 'mariadb', 'consumer' => '淘宝推广账单'],
+        'jiansun.dbo.pdd_spread_bills' => ['role' => 'serving_fact', 'target_store' => 'mariadb', 'consumer' => '拼多多推广账单'],
+        'jiansun.dbo.alibaba_bill' => ['role' => 'serving_fact', 'target_store' => 'mariadb', 'consumer' => '阿里巴巴账单'],
+        'jiansun.dbo.jd_spreadBills' => ['role' => 'serving_fact', 'target_store' => 'mariadb', 'consumer' => '京东推广账单'],
+        'jiansun.dbo.jd_costBills' => ['role' => 'serving_fact', 'target_store' => 'mariadb', 'consumer' => '京东钱包账单'],
+        'jiansun.dbo.fys_Ticketing_install_view' => ['role' => 'serving_fact', 'target_store' => 'mariadb', 'consumer' => '安装明细'],
+        'jiansun.dbo.fys_Ticketing_install_item' => ['role' => 'serving_fact', 'target_store' => 'mariadb', 'consumer' => '安装明细详情'],
+        'jiansun.dbo.finance_purchaseReturn' => ['role' => 'serving_fact', 'target_store' => 'mariadb', 'consumer' => '供应商退货对账'],
+        'jiansun.dbo.finance_invoiceDeduction' => ['role' => 'serving_fact', 'target_store' => 'mariadb', 'consumer' => '供应商发票扣款'],
+        'jiansun.dbo.fys_supplierPaymentSummary' => ['role' => 'serving_fact', 'target_store' => 'mariadb', 'consumer' => '供应商应付'],
+        'jiansun.dbo.finance_advanceLoans' => ['role' => 'serving_fact', 'target_store' => 'mariadb', 'consumer' => '借支明细'],
+        'jiansun.dbo.pm_unOrderDetails' => ['role' => 'serving_fact', 'target_store' => 'mariadb', 'consumer' => '未定价明细'],
+        'jiansun.dbo.finance_sampleDeduction' => ['role' => 'serving_fact', 'target_store' => 'mariadb', 'consumer' => '样品扣款'],
+        'jiansun.dbo.fys_express_courier_settlement' => ['role' => 'serving_fact', 'target_store' => 'mariadb', 'consumer' => '代发快递'],
+        'jiansun.dbo.finance_boardDeduction' => ['role' => 'serving_fact', 'target_store' => 'mariadb', 'consumer' => '板材扣款'],
+        'jiansun.dbo.pm_plainPlateData' => ['role' => 'dimension', 'target_store' => 'mariadb', 'consumer' => '素板资料'],
+        'jiansun.dbo.pm_boardPurchasePrice' => ['role' => 'dimension', 'target_store' => 'mariadb', 'consumer' => '板材采购价'],
+        'jiansun.dbo.pm_boardPriceChangeImport' => ['role' => 'serving_fact', 'target_store' => 'mariadb', 'consumer' => '板材调价'],
+        'jiansun.dbo.pm_boardApplyDetail' => ['role' => 'serving_fact', 'target_store' => 'mariadb', 'consumer' => '板材申请明细'],
+        'jiansun.dbo.pm_edgeBandData' => ['role' => 'dimension', 'target_store' => 'mariadb', 'consumer' => '封边条资料'],
+        'jiansun.dbo.pm_edgeBandApplyDetail' => ['role' => 'serving_fact', 'target_store' => 'mariadb', 'consumer' => '封边条明细'],
+        'jiansun.dbo.pm_edgeBandProfitInfo' => ['role' => 'serving_fact', 'target_store' => 'mariadb', 'consumer' => '封边条利润'],
+        'jiansun.dbo.finance_edgeBandDeductio' => ['role' => 'serving_fact', 'target_store' => 'mariadb', 'consumer' => '封边条扣款'],
+        'jiansun.dbo.pm_averageUnitPrice' => ['role' => 'dimension', 'target_store' => 'mariadb', 'consumer' => '平均单价'],
+        'jiansun.dbo.pm_boardInventoryDetail_snapshot' => ['role' => 'aggregate', 'target_store' => 'mariadb', 'consumer' => '板材库存'],
+        'jiansun.dbo.fys_supplierGoods_sttlement_purchase' => ['role' => 'serving_fact', 'target_store' => 'mariadb', 'consumer' => '货款采购对账'],
+        'jiansun.dbo.fys_supplierGoods_sttlement_sale' => ['role' => 'serving_fact', 'target_store' => 'mariadb', 'consumer' => '货款销售对账'],
+        'jiansun.dbo.pm_purchaseIn' => ['role' => 'serving_fact', 'target_store' => 'mariadb', 'consumer' => '采购入库'],
+        'jiansun.dbo.wp_express' => ['role' => 'dimension', 'target_store' => 'mariadb', 'consumer' => '快递资料'],
+        'jiansun.dbo.wp_express_expNet' => ['role' => 'dimension', 'target_store' => 'mariadb', 'consumer' => '快递网点'],
+        'jiansun.dbo.wp_express_account' => ['role' => 'dimension', 'target_store' => 'mariadb', 'consumer' => '快递打款账户'],
+        'jiansun.dbo.wp_express_extension' => ['role' => 'dimension', 'target_store' => 'mariadb', 'consumer' => '快递扩展资料'],
+    ],
+
+    'special_tables' => [
+        'store_report_bill_rows' => [
+            'role' => 'fact',
+            'target_store' => 'clickhouse',
+            'consumer' => '店铺经营报表的可恢复原始层；页面只读日汇总',
+            'partition_column' => 'occurred_at',
+        ],
+        'store_report_daily_rows' => [
+            'role' => 'aggregate',
+            'target_store' => 'mariadb',
+            'consumer' => '店铺经营报表',
+        ],
+    ],
+];
diff --git a/database/migrations/2026_09_07_120000_create_business_data_placement_tables.php b/database/migrations/2026_09_07_120000_create_business_data_placement_tables.php
new file mode 100644
index 00000000..fe3c2329
--- /dev/null
+++ b/database/migrations/2026_09_07_120000_create_business_data_placement_tables.php
@@ -0,0 +1,68 @@
+<?php
+
+use Illuminate\Database\Migrations\Migration;
+use Illuminate\Database\Schema\Blueprint;
+use Illuminate\Support\Facades\Schema;
+
+return new class extends Migration
+{
+    private function connection(): string
+    {
+        return (string) config('upstream_mirrors.connection', 'business_mirror');
+    }
+
+    public function up(): void
+    {
+        $schema = Schema::connection($this->connection());
+
+        if (! $schema->hasTable('business_data_placements')) {
+            $schema->create('business_data_placements', function (Blueprint $table): void {
+                $table->string('dataset_key', 191)->primary();
+                $table->string('target_store', 24);
+                $table->string('phase', 24)->default('planned');
+                $table->uuid('backfill_generation')->nullable();
+                $table->string('backfill_cursor', 64)->nullable();
+                $table->dateTime('source_cutoff_at', 6)->nullable();
+                $table->unsignedBigInteger('source_rows')->default(0);
+                $table->unsignedBigInteger('backfilled_rows')->default(0);
+                $table->unsignedBigInteger('target_rows')->default(0);
+                $table->uuid('active_generation')->nullable();
+                $table->dateTime('verified_at')->nullable();
+                $table->dateTime('activated_at')->nullable();
+                $table->string('last_error_code', 120)->nullable();
+                $table->timestamps();
+
+                $table->index(['target_store', 'phase'], 'business_data_placement_store_phase');
+            });
+        }
+
+        if (! $schema->hasTable('business_data_archives')) {
+            $schema->create('business_data_archives', function (Blueprint $table): void {
+                $table->bigIncrements('id');
+                $table->string('dataset_key', 191);
+                $table->uuid('generation');
+                $table->string('partition_key', 64);
+                $table->string('object_key', 512);
+                $table->unsignedBigInteger('row_count')->default(0);
+                $table->unsignedBigInteger('byte_size')->default(0);
+                $table->string('logical_checksum', 64)->nullable();
+                $table->string('status', 24)->default('pending');
+                $table->dateTime('verified_at')->nullable();
+                $table->timestamps();
+
+                $table->unique(
+                    ['dataset_key', 'generation', 'partition_key'],
+                    'business_data_archive_dataset_generation_partition',
+                );
+                $table->index(['dataset_key', 'status'], 'business_data_archive_dataset_status');
+            });
+        }
+    }
+
+    public function down(): void
+    {
+        $schema = Schema::connection($this->connection());
+        $schema->dropIfExists('business_data_archives');
+        $schema->dropIfExists('business_data_placements');
+    }
+};
diff --git a/deploy/activate-incremental-release.sh b/deploy/activate-incremental-release.sh
index add13565..5306007e 100644
--- a/deploy/activate-incremental-release.sh
+++ b/deploy/activate-incremental-release.sh
@@ -22,6 +22,31 @@ if [ ! -f "$REMOTE_DIR/runtime/.env" ]; then
     exit 1
 fi
 
+clickhouse_env="$REMOTE_DIR/runtime/clickhouse.env"
+if [ ! -f "$clickhouse_env" ]; then
+    clickhouse_password="$(openssl rand -hex 24)"
+    umask 077
+    {
+        printf '%s\n' \
+            'CLICKHOUSE_DB=caiwu_business_analytics' \
+            'CLICKHOUSE_USER=caiwu_app' \
+            "CLICKHOUSE_PASSWORD=$clickhouse_password" \
+            'CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT=1' \
+            'BUSINESS_CLICKHOUSE_ENABLED=true' \
+            'BUSINESS_CLICKHOUSE_URL=http://clickhouse:8123' \
+            'BUSINESS_CLICKHOUSE_DATABASE=caiwu_business_analytics' \
+            'BUSINESS_CLICKHOUSE_USERNAME=caiwu_app' \
+            "BUSINESS_CLICKHOUSE_PASSWORD=$clickhouse_password"
+    } > "$clickhouse_env"
+    chmod 600 "$clickhouse_env"
+fi
+
+if ! grep -q '^CLICKHOUSE_PASSWORD=.' "$clickhouse_env" \
+    || ! grep -q '^BUSINESS_CLICKHOUSE_PASSWORD=.' "$clickhouse_env"; then
+    echo "Incomplete $clickhouse_env; refusing to overwrite private ClickHouse credentials" >&2
+    exit 1
+fi
+
 if [ ! -f "$REMOTE_DIR/runtime/config-db.env" ]; then
     echo "Missing $REMOTE_DIR/runtime/config-db.env; run bootstrap-server.sh first" >&2
     exit 1
@@ -66,6 +91,10 @@ fi
 cd "$REMOTE_DIR"
 export IMAGE_TAG RUN_MIGRATIONS
 
+# Pull the independent columnar runtime before migrations or app switching.
+# A registry/network failure therefore leaves both the database and current app untouched.
+docker compose -f compose.jiangling.yml pull clickhouse
+
 if [ "$RUN_MIGRATIONS" = "1" ]; then
     # 先创建独立业务镜像库并授权应用账号，再启动会执行迁移的新应用。
     # 该脚本幂等；已有数据库只会补齐授权，不会覆盖业务数据。
@@ -89,6 +118,25 @@ fi
 
 docker compose -f compose.jiangling.yml up -d --remove-orphans
 
+clickhouse_healthy=0
+clickhouse_attempt=1
+while [ "$clickhouse_attempt" -le 24 ]; do
+    if [ "$(docker inspect caiwu_clickhouse --format '{{.State.Health.Status}}' 2>/dev/null || true)" = "healthy" ]; then
+        clickhouse_healthy=1
+        break
+    fi
+
+    sleep 5
+    clickhouse_attempt=$((clickhouse_attempt + 1))
+done
+
+if [ "$clickhouse_healthy" -ne 1 ]; then
+    docker compose -f compose.jiangling.yml logs --tail 120 clickhouse >&2 || true
+    restore_previous_image
+    echo "ClickHouse did not become healthy; previous application image was restored when available" >&2
+    exit 1
+fi
+
 healthy=0
 attempt=1
 while [ "$attempt" -le 30 ]; do
diff --git a/deploy/activate-release.sh b/deploy/activate-release.sh
index 3376abc5..17d90254 100644
--- a/deploy/activate-release.sh
+++ b/deploy/activate-release.sh
@@ -26,6 +26,31 @@ if [ ! -f "$REMOTE_DIR/runtime/.env" ]; then
     exit 1
 fi
 
+clickhouse_env="$REMOTE_DIR/runtime/clickhouse.env"
+if [ ! -f "$clickhouse_env" ]; then
+    clickhouse_password="$(openssl rand -hex 24)"
+    umask 077
+    {
+        printf '%s\n' \
+            'CLICKHOUSE_DB=caiwu_business_analytics' \
+            'CLICKHOUSE_USER=caiwu_app' \
+            "CLICKHOUSE_PASSWORD=$clickhouse_password" \
+            'CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT=1' \
+            'BUSINESS_CLICKHOUSE_ENABLED=true' \
+            'BUSINESS_CLICKHOUSE_URL=http://clickhouse:8123' \
+            'BUSINESS_CLICKHOUSE_DATABASE=caiwu_business_analytics' \
+            'BUSINESS_CLICKHOUSE_USERNAME=caiwu_app' \
+            "BUSINESS_CLICKHOUSE_PASSWORD=$clickhouse_password"
+    } > "$clickhouse_env"
+    chmod 600 "$clickhouse_env"
+fi
+
+if ! grep -q '^CLICKHOUSE_PASSWORD=.' "$clickhouse_env" \
+    || ! grep -q '^BUSINESS_CLICKHOUSE_PASSWORD=.' "$clickhouse_env"; then
+    echo "Incomplete $clickhouse_env; refusing to overwrite private ClickHouse credentials" >&2
+    exit 1
+fi
+
 if [ ! -f "$REMOTE_DIR/runtime/config-db.env" ]; then
     echo "Missing $REMOTE_DIR/runtime/config-db.env; run bootstrap-server.sh first" >&2
     exit 1
@@ -70,6 +95,10 @@ fi
 cd "$REMOTE_DIR"
 export IMAGE_TAG RUN_MIGRATIONS
 
+# Pull the independent columnar runtime before migrations or app switching.
+# A registry/network failure therefore leaves both the database and current app untouched.
+docker compose -f compose.jiangling.yml pull clickhouse
+
 if [ "$RUN_MIGRATIONS" = "1" ]; then
     # 先创建独立业务镜像库并授权应用账号，再启动会执行迁移的新应用。
     # 该脚本幂等；已有数据库只会补齐授权，不会覆盖业务数据。
@@ -93,6 +122,25 @@ fi
 
 docker compose -f compose.jiangling.yml up -d --remove-orphans
 
+clickhouse_healthy=0
+clickhouse_attempt=1
+while [ "$clickhouse_attempt" -le 24 ]; do
+    if [ "$(docker inspect caiwu_clickhouse --format '{{.State.Health.Status}}' 2>/dev/null || true)" = "healthy" ]; then
+        clickhouse_healthy=1
+        break
+    fi
+
+    sleep 5
+    clickhouse_attempt=$((clickhouse_attempt + 1))
+done
+
+if [ "$clickhouse_healthy" -ne 1 ]; then
+    docker compose -f compose.jiangling.yml logs --tail 120 clickhouse >&2 || true
+    restore_previous_image
+    echo "ClickHouse did not become healthy; previous application image was restored when available" >&2
+    exit 1
+fi
+
 healthy=0
 attempt=1
 while [ "$attempt" -le 30 ]; do
diff --git a/deploy/bootstrap-server.sh b/deploy/bootstrap-server.sh
index d924d294..7cc49b6e 100644
--- a/deploy/bootstrap-server.sh
+++ b/deploy/bootstrap-server.sh
@@ -36,6 +36,25 @@ if [ ! -f "$REMOTE_DIR/runtime/qiniu-storage.env" ]; then
     chmod 600 "$REMOTE_DIR/runtime/qiniu-storage.env"
 fi
 
+clickhouse_env="$REMOTE_DIR/runtime/clickhouse.env"
+if [ ! -f "$clickhouse_env" ]; then
+    clickhouse_password="$(openssl rand -hex 24)"
+    umask 077
+    {
+        printf '%s\n' \
+            'CLICKHOUSE_DB=caiwu_business_analytics' \
+            'CLICKHOUSE_USER=caiwu_app' \
+            "CLICKHOUSE_PASSWORD=$clickhouse_password" \
+            'CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT=1' \
+            'BUSINESS_CLICKHOUSE_ENABLED=true' \
+            'BUSINESS_CLICKHOUSE_URL=http://clickhouse:8123' \
+            'BUSINESS_CLICKHOUSE_DATABASE=caiwu_business_analytics' \
+            'BUSINESS_CLICKHOUSE_USERNAME=caiwu_app' \
+            "BUSINESS_CLICKHOUSE_PASSWORD=$clickhouse_password"
+    } > "$clickhouse_env"
+    chmod 600 "$clickhouse_env"
+fi
+
 if [ ! -f "$REMOTE_DIR/runtime/config-db.env" ]; then
     CONFIG_DB_PASSWORD="$(openssl rand -hex 32)"
     CONFIG_DB_ROOT_PASSWORD="$(openssl rand -hex 32)"
diff --git a/docker-compose.local.yml b/docker-compose.local.yml
index 3cf5eb60..45276cf0 100644
--- a/docker-compose.local.yml
+++ b/docker-compose.local.yml
@@ -71,11 +71,43 @@ services:
       retries: 20
       start_period: 30s
 
+  # 按需启动：docker compose -f docker-compose.local.yml --profile analytics up -d clickhouse
+  # 普通页面开发不需要运行列式库。
+  clickhouse:
+    image: m.daocloud.io/docker.io/clickhouse/clickhouse-server:26.8.2.7-alpine
+    container_name: jiangling_clickhouse
+    profiles: ["analytics"]
+    restart: unless-stopped
+    environment:
+      CLICKHOUSE_DB: caiwu_business_analytics
+      CLICKHOUSE_USER: caiwu_app
+      CLICKHOUSE_PASSWORD: local-clickhouse-only
+      CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT: 1
+      TZ: Asia/Shanghai
+    ports:
+      - "127.0.0.1:18123:8123"
+    volumes:
+      - jiangling_clickhouse:/var/lib/clickhouse
+    networks:
+      - jiangling_net
+    ulimits:
+      nofile:
+        soft: 262144
+        hard: 262144
+    healthcheck:
+      test: ["CMD-SHELL", "clickhouse-client --user \"$$CLICKHOUSE_USER\" --password \"$$CLICKHOUSE_PASSWORD\" --query 'SELECT 1' >/dev/null"]
+      interval: 10s
+      timeout: 5s
+      retries: 12
+      start_period: 30s
+
 volumes:
   jiangling_storage:
     name: jiangling_storage
   jiangling_mysql:
     name: jiangling_mysql
+  jiangling_clickhouse:
+    name: jiangling_clickhouse
 
 networks:
   jiangling_net:
diff --git a/docs/业务数据分层存储.md b/docs/业务数据分层存储.md
new file mode 100644
index 00000000..a5c53ab1
--- /dev/null
+++ b/docs/业务数据分层存储.md
@@ -0,0 +1,107 @@
+# 业务数据分层存储与迁移清单
+
+## 结论
+
+`caiwu_business_mirror` 继续作为唯一的中央同步入口，但不再要求所有数据都以 MariaDB 行存表长期保存：
+
+- MariaDB 保存同步状态、权限/配置、小型维表、当前态业务表和页面直接使用的汇总表；
+- ClickHouse 保存订单、退款、推广费用、原始账单等大事实表，承担筛选、聚合和历史明细查询；
+- 七牛 S3 兼容对象存储保存按稳定哈希分区生成的不可变 Parquet，作为重建和灾难恢复层；
+- 页面只能经过固定 Repository 读取服务层，不能直接访问上游，也不能自行选择数据库。
+
+列式迁移采用 `MariaDB 旧表 -> ClickHouse 新代次 -> 行数/唯一键对账 -> Parquet 分区归档 -> 页面切换 -> 观察 -> 旧表清理`。任何一步失败都保留 MariaDB 原表和当前页面链路。
+
+## 2026-09-07 生产只读盘点
+
+本次通过生产 `information_schema` 与当前代码引用交叉核对。生产部署 SHA 为 `f0150f1c4971c807239e216398df32ec1751d121`；服务器 500 GiB 系统盘当时约有 385 GiB 可用。`caiwu_business_mirror` 有 65 张物理表，InnoDB 估算约 5172 万行、51.4 GiB（数据 32.2 GiB、索引 19.2 GiB）。InnoDB `TABLE_ROWS` 是估算值，迁移验收必须使用精确 `COUNT(*)` 与唯一键计数。
+
+占用最大的表如下：
+
+| 表 | 估算行数 | 总占用 | 当前真实用途 | 处理 |
+| --- | ---: | ---: | --- | --- |
+| `ops_jst_sku_map` | 2492 万 | 28.1 GiB | 当前页面没有查询，只曾被推广费用页误当作就绪条件 | 立即停止后续同步；现有表保留，完成 Parquet 归档后再申请删除 |
+| `store_report_bill_rows` | 1385 万 | 7.8 GiB | 只用于生成 `store_report_daily_rows`，页面不应扫描原始行 | 页面改读日汇总；原始层迁入 ClickHouse 并归档后，仅保留必要增量窗口 |
+| `ops_item_sales_cost_ratio` | 289 万 | 3.6 GiB | 推广费用页面 | 迁入 ClickHouse；页面查询验收后切换 |
+| `ops_jst_original_bill_info` | 184 万估算 / 已处理 208 万 | 2.3 GiB | 新登记的通用原始账单，当前无页面直读，首次同步已暂停 | 不再向 MariaDB 续传；先用现有 208 万行验证列式迁移，不重新读取上游 |
+| `ops_jst_orders` | 112 万 | 1.7 GiB | 售后费用 | 与订单明细、退款表作为同一激活组迁入 ClickHouse |
+| `ops_jst_order_item` | 158 万 | 1.7 GiB | 售后费用 | 同上 |
+| `ops_jst_return_refund_info` | 113 万 | 1.6 GiB | 售后费用、非京喜退款 | 同上 |
+| `ops_jst_return_refund_item` | 135 万 | 1.3 GiB | 售后费用、非京喜退款 | 同上 |
+| `ops_jst_sku` | 50 万 | 778 MiB | 商品资料及订单行补充 | 作为高频维表留在 MariaDB |
+| `ops_work_order_form_value` | 41 万 | 436 MiB | 协同工单当前态 | 留在 MariaDB，继续观察增速 |
+| `js_fys_ticketing_install_view` | 24 万 | 400 MiB | 安装明细 | 留在 MariaDB；达到阈值后再单独迁移 |
+
+前三个可清理的 MariaDB 原始层合计约 38.2 GiB，但只能在归档对象存在、对象大小/逻辑校验通过、页面或同步器不再读取且获得生产删除授权后清理。当前迁移和部署不会自动删除这些表。
+
+## 页面与数据层映射
+
+代码当前登记 51 个固定上游数据集，另有 PostgreSQL 连接器动态登记的店铺账号维表。完整逐数据集策略由 `config/business_data.php` 强制登记；新增固定数据集若未登记，`business-data:plan` 返回失败。
+
+| 页面/业务域 | 数据集 | 目标层 | 说明 |
+| --- | --- | --- | --- |
+| 公司、店铺、商标 | `fys_companyInfo`、`fys_shopInfo`、`fys_TrademarkInfo` | MariaDB 维表 | 行数小、点查与关联频繁 |
+| 店铺经营报表 | `store_report_lifecycle_rows` + `store_report_daily_rows` | MariaDB 维表/汇总 | 页面已改为共享日汇总，不依赖 `runtime/bills.sqlite`，不扫描原始账单 |
+| 普通商品资料 | `jst_sku` + 采购价镜像 | MariaDB 维表/服务表 | 约 55 万商品仍适合行存筛选；保留现有完整导出校准 |
+| 推广费用 | `ads_itemid_sales_cost_ratio_v2_di` | ClickHouse 事实表 | `jst_skumap` 不再作为页面就绪条件 |
+| 售后费用、非京喜退款 | 订单、订单明细、退款主表、退款明细 | ClickHouse 事实表 + MariaDB SKU/仓库/店铺维表 | 四张事实表必须成组对账和切换，不能逐张切换造成跨库半成品 |
+| 原始账单 | `jst_originalBillInfo` | ClickHouse + Parquet | 当前无页面直读；先迁移已保留部分验证压缩率和恢复能力 |
+| 营销账户账单 | 淘宝、拼多多、阿里巴巴、京东账单 | MariaDB 服务表 | 当前单表约 2 万至 21 万行，先保留；按增速和查询成本复评 |
+| 供应商应付、板材、封边条、库存 | 对应 `finance_*`、`pm_*`、结算表 | MariaDB 服务表/汇总 | 当前规模可控；不做无收益的跨库改写 |
+| 安装明细 | 安装视图与明细 | MariaDB 服务表 | 当前约 408 MiB，页面需要稳定分页与详情点查 |
+| 协同工单、小额打款 | 工单主/子/表单/映射 | MariaDB 当前态 | 更新频繁且需要当前态精确读取 |
+| 快递资料 | 快递、网点、账户、扩展 | MariaDB 维表 | 小型主数据 |
+| 动态店铺账号数据集 | `database_4` 动态登记表 | MariaDB 外部维表 | 未经单独评审不得自动进入列式层 |
+
+## 已实现的安全边界
+
+1. `business-data:plan` 只读输出固定数据集、页面用途、目标层、同步策略、当前表体量与迁移阶段；固定数据集漏配或多配即失败。
+2. `business-data:backfill-columnar <dataset>` 只从现有 MariaDB 镜像按 `_mirror_key` 断点读取，不连接上游；读取范围固定在任务启动时的 `_mirror_synced_at` 截止点。
+3. ClickHouse 表使用 `ReplacingMergeTree`，以 `_mirror_key` 为排序/去重键，并按其哈希固定分为 32 个分区；同一业务键的所有版本必定落在同一分区。
+4. 回填结束同时核对截止点精确行数、ClickHouse `FINAL` 行数和唯一键数；不一致只记录失败，不切换页面。
+5. `--archive` 在对账后把 32 个哈希分区分别输出为 Parquet，经现有私有七牛 S3 磁盘流式上传；每个对象记录路径、行数、字节数和逻辑校验值。
+6. 本阶段没有“激活”或“删除”命令。页面切换与 MariaDB 清理必须在各 Repository 的 ClickHouse 查询完成真实页面验收后另行提交，避免有人只执行一条命令就绕过保护门。
+7. ClickHouse 凭据只生成在服务器权限为 `600` 的 `runtime/clickhouse.env`；Git、PR、日志和命令参数中不出现密码。
+
+## 执行顺序
+
+以下是上线后的运维顺序，不是普通开发机操作。生产迁移、回填、归档和删除分别需要对应授权。
+
+```bash
+# 1. 只读核对配置与现状
+docker exec -u www-data caiwu_app php artisan business-data:plan
+
+# 2. 用已经暂停的原始账单部分数据验证 ClickHouse，不访问上游
+docker exec -u www-data caiwu_app php artisan business-data:backfill-columnar \
+  operations_replica.rpt.jst_originalBillInfo --batch=5000 --max-batches=20
+
+# 3. 对账完成后分区归档；仍不会删除 MariaDB 原表
+docker exec -u www-data caiwu_app php artisan business-data:backfill-columnar \
+  operations_replica.rpt.jst_originalBillInfo --batch=5000 --max-batches=0 --archive
+```
+
+随后按以下顺序推进：
+
+1. 记录 ClickHouse 实际压缩率、回填吞吐、CPU/内存峰值和对象存储归档大小；
+2. 为推广费用 Repository 增加列式读取，在相同筛选下对比首页、第二页、末页、金额汇总与无数据；
+3. 将订单/退款四张事实表作为一个激活组迁移并完成同口径页面验收；
+4. 将店铺报表原始层迁入 ClickHouse，确认日汇总可从归档或列式原始层重建；
+5. 至少观察一个完整增量周期，再提出精确旧表清理清单、备份位置和回滚方案；
+6. 获得删除授权后才逐表清理 MariaDB，清理后再次执行页面与归档恢复抽查。
+
+## 容量与告警建议
+
+- MariaDB 镜像总量达到 80 GiB 或单表达到 5 GiB 时告警，不再自动扩大索引；
+- ClickHouse 数据卷达到 70% 时暂停新回填，优先确认归档与旧代次清理；
+- 对象存储按 `数据集/代次/哈希分区` 保留不可变对象，清单写入 `business_data_archives`；
+- 同步状态的 `running` 只代表任务游标，不代表页面已经切换；页面切换必须另有 `active_generation` 和验收记录；
+- 上游断线时保留 MariaDB/ClickHouse 上一代数据，不由 HTTP 请求回退直连上游。
+
+## 本地协议验证
+
+2026-09-07 使用与部署配置相同的 ClickHouse `26.8.2.7-alpine` 完成两级验证：应用 HTTP 客户端健康检查成功；同一 `_mirror_key` 写入两个版本后，`FINAL` 只返回版本号较大的 1 行；金额为最新值；`SELECT ... FORMAT Parquet` 返回标准 `PAR1` 文件头。随后使用 2 行隔离 SQLite 镜像走通完整命令，结果为 `exit=0 / phase=verified / source=2 / target=2 / archives=32 / files=32 / parquet_magic=50415231`。验证结束已删除 ClickHouse 临时表和临时归档，并停止本地容器；未连接生产数据。
+
+ClickHouse 官方说明中，MergeTree 数据以不可变数据片段写入并在后台合并；`ReplacingMergeTree` 的查询在需要即时去重时必须使用 `FINAL`。S3 表函数与 Parquet 格式支持从列式表导出和后续重建，本方案据此把服务存储与不可变归档分开：
+
+- https://clickhouse.com/resources/engineering/clickhouse-optimize-table-final
+- https://clickhouse.com/integrations/amazon_s3
+- https://clickhouse.com/resources/engineering/what-is-clickhouse-local
diff --git a/modules/EcommerceFinance/Repositories/ColumnarBusinessMirrorRepository.php b/modules/EcommerceFinance/Repositories/ColumnarBusinessMirrorRepository.php
new file mode 100644
index 00000000..dce06900
--- /dev/null
+++ b/modules/EcommerceFinance/Repositories/ColumnarBusinessMirrorRepository.php
@@ -0,0 +1,148 @@
+<?php
+
+namespace Modules\EcommerceFinance\Repositories;
+
+use Modules\EcommerceFinance\Contracts\BusinessMirrorDataset;
+use Modules\EcommerceFinance\Services\ClickHouseClient;
+use Psr\Http\Message\StreamInterface;
+
+class ColumnarBusinessMirrorRepository
+{
+    public function __construct(private readonly ClickHouseClient $clickHouse)
+    {
+    }
+
+    /**
+     * @param list<array<string,mixed>> $columns
+     * @param list<array<string,mixed>> $rows
+     */
+    public function storeBackfillBatch(
+        BusinessMirrorDataset $dataset,
+        array $columns,
+        array $rows,
+        string $generation,
+    ): void {
+        if ($rows === []) {
+            return;
+        }
+
+        $names = array_map(static fn (array $column): string => (string) $column['column_name'], $columns);
+        $types = [];
+        foreach ($columns as $column) {
+            $types[strtolower((string) $column['column_name'])] = strtolower((string) ($column['data_type'] ?? ''));
+        }
+
+        $version = (int) floor(microtime(true) * 1_000_000);
+        $mapped = [];
+        foreach (array_values($rows) as $index => $row) {
+            $values = ['_mirror_key' => (string) $row['_mirror_key']];
+            foreach ($names as $name) {
+                $values[$name] = $this->value($row[$name] ?? null, $types[strtolower($name)] ?? '');
+            }
+            $values['_mirror_generation'] = $generation;
+            $values['_mirror_synced_at'] = $this->dateTime($row['_mirror_synced_at'] ?? null);
+            $values['_mirror_version'] = $version + $index;
+            $mapped[] = $values;
+        }
+
+        $this->clickHouse->insertJsonEachRow(
+            $dataset->targetTable(),
+            ['_mirror_key', ...$names, '_mirror_generation', '_mirror_synced_at', '_mirror_version'],
+            $mapped,
+        );
+    }
+
+    public function count(BusinessMirrorDataset $dataset, string $generation): int
+    {
+        return (int) $this->clickHouse->scalar(
+            sprintf(
+                'SELECT count() FROM %s FINAL WHERE _mirror_generation = {generation:UUID}',
+                $this->clickHouse->qualifiedTable($dataset->targetTable()),
+            ),
+            ['generation' => $generation],
+        );
+    }
+
+    public function distinctKeys(BusinessMirrorDataset $dataset, string $generation): int
+    {
+        return (int) $this->clickHouse->scalar(
+            sprintf(
+                'SELECT uniqExact(_mirror_key) FROM %s FINAL WHERE _mirror_generation = {generation:UUID}',
+                $this->clickHouse->qualifiedTable($dataset->targetTable()),
+            ),
+            ['generation' => $generation],
+        );
+    }
+
+    /** @return array{row_count:int,logical_checksum:string} */
+    public function partitionStats(BusinessMirrorDataset $dataset, string $generation, int $partition): array
+    {
+        $rows = $this->clickHouse->rows(
+            sprintf(
+                'SELECT count() AS row_count, hex(groupBitXor(cityHash64(_mirror_key))) AS logical_checksum '
+                . 'FROM %s FINAL WHERE _mirror_generation = {generation:UUID} '
+                . 'AND modulo(cityHash64(_mirror_key), 32) = {partition:UInt8}',
+                $this->clickHouse->qualifiedTable($dataset->targetTable()),
+            ),
+            ['generation' => $generation, 'partition' => $this->partition($partition)],
+        );
+        $row = $rows[0] ?? [];
+
+        return [
+            'row_count' => (int) ($row['row_count'] ?? 0),
+            'logical_checksum' => (string) ($row['logical_checksum'] ?? ''),
+        ];
+    }
+
+    public function parquetPartition(
+        BusinessMirrorDataset $dataset,
+        string $generation,
+        int $partition,
+    ): StreamInterface
+    {
+        return $this->clickHouse->stream(
+            sprintf(
+                'SELECT * EXCEPT (_mirror_version) FROM %s FINAL '
+                . 'WHERE _mirror_generation = {generation:UUID} '
+                . 'AND modulo(cityHash64(_mirror_key), 32) = {partition:UInt8} ORDER BY _mirror_key',
+                $this->clickHouse->qualifiedTable($dataset->targetTable()),
+            ),
+            'Parquet',
+            ['generation' => $generation, 'partition' => $this->partition($partition)],
+        );
+    }
+
+    private function partition(int $partition): int
+    {
+        if ($partition < 0 || $partition > 31) {
+            throw new \InvalidArgumentException('business_data_invalid_archive_partition');
+        }
+
+        return $partition;
+    }
+
+    private function value(mixed $value, string $type): mixed
+    {
+        if ($value === null) {
+            return null;
+        }
+        if (is_resource($value)) {
+            $value = stream_get_contents($value);
+        }
+        if (is_string($value) && in_array($type, ['binary', 'varbinary', 'image', 'timestamp', 'rowversion', 'bytea'], true)) {
+            return base64_encode($value);
+        }
+        if (is_string($value) && ! mb_check_encoding($value, 'UTF-8')) {
+            return mb_scrub($value, 'UTF-8');
+        }
+
+        return $value;
+    }
+
+    private function dateTime(mixed $value): string
+    {
+        $value = trim((string) ($value ?? ''));
+
+        return $value === '' ? now()->format('Y-m-d H:i:s.u') : str_replace('T', ' ', $value);
+    }
+}
diff --git a/modules/EcommerceFinance/Repositories/OrderDataMirrorRepository.php b/modules/EcommerceFinance/Repositories/OrderDataMirrorRepository.php
index faa5d911..b3b0abdc 100644
--- a/modules/EcommerceFinance/Repositories/OrderDataMirrorRepository.php
+++ b/modules/EcommerceFinance/Repositories/OrderDataMirrorRepository.php
@@ -19,8 +19,6 @@ class OrderDataMirrorRepository
 
     private const PROMOTION_DATASETS = [
         'operations_replica.rpt.ads_itemid_sales_cost_ratio_v2_di',
-        'operations_replica.rpt.jst_skumap',
-        'operations_replica.rpt.jst_shopInfo',
     ];
 
     private readonly ConnectionInterface $connection;
diff --git a/modules/EcommerceFinance/Repositories/StoreReportRepository.php b/modules/EcommerceFinance/Repositories/StoreReportRepository.php
index 26954bc9..398f342a 100644
--- a/modules/EcommerceFinance/Repositories/StoreReportRepository.php
+++ b/modules/EcommerceFinance/Repositories/StoreReportRepository.php
@@ -2,19 +2,20 @@
 
 namespace Modules\EcommerceFinance\Repositories;
 
+use Illuminate\Database\ConnectionInterface;
 use Illuminate\Database\DatabaseManager;
-use PDO;
+use Illuminate\Support\Facades\Schema;
 use RuntimeException;
 
 /**
- * 将经营分析副本库中的店铺资料与本地账单 SQLite 组合为店铺维度报表。
+ * 将经营分析副本库中的店铺资料与共享日汇总组合为店铺维度报表。
  *
- * SQLite 只读保存店铺生命周期和平台账单；生命周期区间统一使用
- * [effective_at, expires_at)，避免主体切换日的账单被重复统计。
+ * 页面不扫描千万级原始账单，也不依赖开发机或镜像中不存在的 SQLite。
+ * 生命周期区间统一使用 [effective_at, expires_at)，避免主体切换日重复统计。
  */
 class StoreReportRepository
 {
-    // 保留导入命令使用的数据集键；报表页面本身读取本地 SQLite。
+    // 保留同步命令使用的数据集键；页面只读生命周期与日汇总。
     public const BILLS_DATASET = 'store-report.platform-bills';
     public const LIFECYCLES_DATASET = 'store-report.store-lifecycles';
     /** @deprecated Kept for existing synchronizer/test contracts. */
@@ -24,9 +25,14 @@ class StoreReportRepository
     private const DEFAULT_EXPIRES_AT = '2999-01-01';
     /** 主表 rpt.fys_shopinfo 中这些平台不支持变更主体（能否变更=否），其余平台均为是。 */
     public const UNCHANGEABLE_PLATFORMS = ['唯品会MP', '淘工厂'];
-    private ?PDO $connection = null;
+    private readonly ConnectionInterface $connection;
 
-    public function __construct(private readonly DatabaseManager $database) {}
+    public function __construct(DatabaseManager $database)
+    {
+        $this->connection = $database->connection(
+            (string) config('upstream_mirrors.connection', 'business_mirror'),
+        );
+    }
 
     /**
      * 店铺维度报表的查询条件；日期统一左闭右开 [from, to)。
@@ -63,15 +69,18 @@ class StoreReportRepository
             return [];
         }
 
-        $this->assertDatabaseFile();
-        $this->connection ??= $this->readOnlyConnection();
+        $dailyGeneration = $this->activeGeneration(self::DAILY_DATASET);
+        $lifecycleGeneration = $this->activeGeneration(self::LIFECYCLES_DATASET);
         $names = array_values(array_unique(array_filter(array_map(
             static fn (array $row): string => trim((string) ($row['erp_store_name'] ?? '')),
             $stores,
         ))));
-        $lifecycles = $this->lifecycles($names);
-        $lifecycleSales = $this->lifecycleSales($names);
-        $fallbackSales = $this->fallbackSales(array_values(array_diff($names, array_keys($lifecycles))));
+        $lifecycles = $this->lifecycles($names, $lifecycleGeneration);
+        $lifecycleSales = $this->lifecycleSales($names, $lifecycleGeneration, $dailyGeneration);
+        $fallbackSales = $this->fallbackSales(
+            array_values(array_diff($names, array_keys($lifecycles))),
+            $dailyGeneration,
+        );
         $reportYear = $reportYear ?? (int) now()->format('Y');
         $currentMonth = now()->startOfMonth();
         $rollingMonths = collect(range(12, 1))->map(
@@ -141,30 +150,18 @@ class StoreReportRepository
         return $rows;
     }
 
-    private function assertDatabaseFile(): void
-    {
-        $path = (string) config('database.connections.bills_sqlite.database');
-        if ($path === '' || ! is_file($path)) {
-            throw new RuntimeException('店铺账单 SQLite 尚未挂载：' . $path);
-        }
-    }
-
     /** @return array<string,array<int,array<string,mixed>>> */
-    private function lifecycles(array $names): array
+    private function lifecycles(array $names, string $generation): array
     {
-        if (! $this->hasTable('store_lifecycles')) {
-            return [];
-        }
-
         $result = [];
-        foreach (array_chunk($names, 100) as $chunk) {
+        foreach ($this->nameChunks($names) as $chunk) {
             $placeholders = implode(',', array_fill(0, count($chunk), '?'));
             $rows = $this->select(
                 "SELECT id, erp_store_name, company_name, effective_at, expires_at
-                 FROM store_lifecycles
-                 WHERE erp_store_name IN ($placeholders)
+                 FROM store_report_lifecycle_rows
+                 WHERE generation = ? AND erp_store_name IN ($placeholders)
                  ORDER BY erp_store_name, effective_at, id",
-                $chunk,
+                [$generation, ...$chunk],
             );
 
             foreach ($rows as $row) {
@@ -181,27 +178,42 @@ class StoreReportRepository
     }
 
     /** @return array<string,array{period:float,months:array<string,float>}> */
-    private function lifecycleSales(array $names): array
+    private function lifecycleSales(array $names, string $lifecycleGeneration, string $dailyGeneration): array
     {
-        if ($names === [] || ! $this->hasTable('store_lifecycles') || ! $this->hasTable('bills')) {
+        if ($names === []) {
             return [];
         }
 
+        $month = $this->connection->getDriverName() === 'sqlite'
+            ? "substr(d.occurred_on, 1, 7)"
+            : "DATE_FORMAT(d.occurred_on, '%Y-%m')";
+        $lowerBound = $this->connection->getDriverName() === 'sqlite'
+            ? 'date(d.occurred_on) >= date(COALESCE(l.effective_at, ?))'
+            : 'd.occurred_on >= COALESCE(l.effective_at, ?)';
+        $upperBound = $this->connection->getDriverName() === 'sqlite'
+            ? 'date(d.occurred_on) < date(COALESCE(l.expires_at, ?))'
+            : 'd.occurred_on < COALESCE(l.expires_at, ?)';
         $result = [];
-        foreach (array_chunk($names, 100) as $chunk) {
+        foreach ($this->nameChunks($names) as $chunk) {
             $placeholders = implode(',', array_fill(0, count($chunk), '?'));
             $rows = $this->select(
-                "SELECT l.id AS lifecycle_id, substr(b.occurred_at, 1, 7) AS bill_month,
-                        SUM(COALESCE(b.income, 0)) AS income_total
-                 FROM store_lifecycles AS l
-                 INNER JOIN bills AS b
-                   ON b.store_name = l.erp_store_name
-                  AND b.occurred_at IS NOT NULL AND b.occurred_at <> ''
-                  AND b.occurred_at >= COALESCE(NULLIF(l.effective_at, ''), ?)
-                  AND b.occurred_at < COALESCE(NULLIF(l.expires_at, ''), ?)
-                 WHERE l.erp_store_name IN ($placeholders)
-                 GROUP BY l.id, substr(b.occurred_at, 1, 7)",
-                [self::DEFAULT_EFFECTIVE_AT, self::DEFAULT_EXPIRES_AT, ...$chunk],
+                "SELECT l.id AS lifecycle_id, {$month} AS bill_month,
+                        SUM(COALESCE(d.income, 0)) AS income_total
+                 FROM store_report_lifecycle_rows AS l
+                 INNER JOIN store_report_daily_rows AS d
+                  ON d.store_name = l.erp_store_name
+                  AND d.generation = ?
+                  AND {$lowerBound}
+                  AND {$upperBound}
+                 WHERE l.generation = ? AND l.erp_store_name IN ($placeholders)
+                 GROUP BY l.id, {$month}",
+                [
+                    $dailyGeneration,
+                    self::DEFAULT_EFFECTIVE_AT,
+                    self::DEFAULT_EXPIRES_AT,
+                    $lifecycleGeneration,
+                    ...$chunk,
+                ],
             );
             $this->collectSales($result, $rows, 'lifecycle_id');
         }
@@ -210,24 +222,26 @@ class StoreReportRepository
     }
 
     /** @return array<string,array{period:float,months:array<string,float>}> */
-    private function fallbackSales(array $names): array
+    private function fallbackSales(array $names, string $dailyGeneration): array
     {
-        if ($names === [] || ! $this->hasTable('bills')) {
+        if ($names === []) {
             return [];
         }
 
+        $month = $this->connection->getDriverName() === 'sqlite'
+            ? "substr(occurred_on, 1, 7)"
+            : "DATE_FORMAT(occurred_on, '%Y-%m')";
         $result = [];
-        foreach (array_chunk($names, 100) as $chunk) {
+        foreach ($this->nameChunks($names) as $chunk) {
             $placeholders = implode(',', array_fill(0, count($chunk), '?'));
             $rows = $this->select(
-                "SELECT store_name, substr(occurred_at, 1, 7) AS bill_month,
+                "SELECT store_name, {$month} AS bill_month,
                         SUM(COALESCE(income, 0)) AS income_total
-                 FROM bills
-                 WHERE store_name IN ($placeholders)
-                   AND occurred_at IS NOT NULL AND occurred_at <> ''
-                   AND occurred_at >= ? AND occurred_at < ?
-                 GROUP BY store_name, substr(occurred_at, 1, 7)",
-                [...$chunk, self::DEFAULT_EFFECTIVE_AT, self::DEFAULT_EXPIRES_AT],
+                 FROM store_report_daily_rows
+                 WHERE generation = ? AND store_name IN ($placeholders)
+                   AND occurred_on >= ? AND occurred_on < ?
+                 GROUP BY store_name, {$month}",
+                [$dailyGeneration, ...$chunk, self::DEFAULT_EFFECTIVE_AT, self::DEFAULT_EXPIRES_AT],
             );
             $this->collectSales($result, $rows, 'store_name');
         }
@@ -248,32 +262,37 @@ class StoreReportRepository
     /** @return array<int,array<string,mixed>> */
     private function select(string $query, array $bindings = []): array
     {
-        $statement = $this->connection->prepare($query);
-        $statement->execute($bindings);
-
-        return $statement->fetchAll(PDO::FETCH_ASSOC);
-    }
-
-    private function readOnlyConnection(): PDO
-    {
-        $path = realpath((string) config('database.connections.bills_sqlite.database'));
-        if ($path === false) {
-            throw new RuntimeException('店铺账单 SQLite 尚未挂载。');
-        }
-
-        return new PDO(
-            'sqlite:file:' . $path . '?mode=ro&immutable=1',
-            null,
-            null,
-            [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION],
+        return array_map(
+            static fn (object $row): array => (array) $row,
+            $this->connection->select($query, $bindings),
         );
     }
 
-    /** SQLite Schema Builder 在只读/WAL 挂载下可能尝试写入，因此使用普通查询。 */
-    private function hasTable(string $table): bool
+    /** @return list<list<string>> */
+    private function nameChunks(array $names): array
     {
-        // 本地账单文件由导入脚本一次性创建，避免 Laravel 在只读挂载上执行元数据写检查。
-        return in_array($table, ['bills', 'store_lifecycles'], true);
+        // SQLite fixtures keep the conservative variable limit. Production
+        // MariaDB uses larger chunks to avoid turning one report page into
+        // dozens of otherwise identical round trips.
+        return array_chunk($names, $this->connection->getDriverName() === 'sqlite' ? 100 : 500);
+    }
+
+    private function activeGeneration(string $dataset): string
+    {
+        $schema = Schema::connection($this->connection->getName());
+        if (! $schema->hasTable('store_report_import_states')) {
+            throw new RuntimeException('store_report_state_missing');
+        }
+
+        $state = $this->connection->table('store_report_import_states')
+            ->where('dataset_key', $dataset)
+            ->first();
+        $generation = trim((string) ($state->active_generation ?? ''));
+        if ($generation === '' || empty($state->initial_completed_at)) {
+            throw new RuntimeException('store_report_dataset_not_ready:' . $dataset);
+        }
+
+        return $generation;
     }
 
     private function date(mixed $value, string $fallback): string
diff --git a/modules/EcommerceFinance/Services/BusinessDataPlacement.php b/modules/EcommerceFinance/Services/BusinessDataPlacement.php
new file mode 100644
index 00000000..08563e34
--- /dev/null
+++ b/modules/EcommerceFinance/Services/BusinessDataPlacement.php
@@ -0,0 +1,108 @@
+<?php
+
+namespace Modules\EcommerceFinance\Services;
+
+use Illuminate\Database\ConnectionInterface;
+use Illuminate\Database\DatabaseManager;
+use Illuminate\Support\Facades\Schema;
+use InvalidArgumentException;
+
+class BusinessDataPlacement
+{
+    private readonly ConnectionInterface $connection;
+
+    public function __construct(DatabaseManager $database)
+    {
+        $this->connection = $database->connection(
+            (string) config('upstream_mirrors.connection', 'business_mirror'),
+        );
+    }
+
+    /** @return array<string,array<string,mixed>> */
+    public function policies(): array
+    {
+        return (array) config('business_data.datasets', []);
+    }
+
+    /** @return array<string,mixed> */
+    public function policy(string $datasetKey): array
+    {
+        $policy = (array) ($this->policies()[$datasetKey] ?? []);
+        if ($policy === []) {
+            // Missing and runtime-registered connector datasets fail closed to
+            // MariaDB. `business-data:plan` still fails CI/operations for a
+            // missing code-owned policy, while a typo can never silently route
+            // production data into ClickHouse.
+            return [
+                'role' => 'external_dimension',
+                'target_store' => 'mariadb',
+                'consumer' => '外部登记数据集',
+                'sync' => 'always',
+            ];
+        }
+
+        return $policy + ['sync' => 'always'];
+    }
+
+    public function targetStore(string $datasetKey): string
+    {
+        return (string) ($this->policy($datasetKey)['target_store'] ?? 'mariadb');
+    }
+
+    public function shouldSchedule(string $datasetKey): bool
+    {
+        $sync = (string) ($this->policy($datasetKey)['sync'] ?? 'always');
+
+        return match ($sync) {
+            'always' => true,
+            'disabled' => false,
+            'after_columnar_activation' => $this->isActive($datasetKey, 'clickhouse'),
+            default => throw new InvalidArgumentException('业务数据集同步策略无效：' . $datasetKey),
+        };
+    }
+
+    public function isActive(string $datasetKey, ?string $store = null): bool
+    {
+        $row = $this->state($datasetKey);
+        if ($row === null || ($row['phase'] ?? null) !== 'active') {
+            return false;
+        }
+
+        return $store === null || ($row['target_store'] ?? null) === $store;
+    }
+
+    /** @return array<string,mixed>|null */
+    public function state(string $datasetKey): ?array
+    {
+        if (! Schema::connection($this->connection->getName())->hasTable('business_data_placements')) {
+            return null;
+        }
+
+        $row = $this->connection->table('business_data_placements')
+            ->where('dataset_key', $datasetKey)
+            ->first();
+
+        return $row === null ? null : (array) $row;
+    }
+
+    /** @param array<string,mixed> $values */
+    public function save(string $datasetKey, array $values): void
+    {
+        $now = now();
+        $this->connection->table('business_data_placements')->insertOrIgnore([
+            'dataset_key' => $datasetKey,
+            'target_store' => $this->targetStore($datasetKey),
+            'phase' => 'planned',
+            'created_at' => $now,
+            'updated_at' => $now,
+        ]);
+        $this->connection->table('business_data_placements')
+            ->where('dataset_key', $datasetKey)
+            ->update(['updated_at' => $now] + $values);
+    }
+
+    public function connection(): ConnectionInterface
+    {
+        return $this->connection;
+    }
+}
diff --git a/modules/EcommerceFinance/Services/BusinessMirrorDatasetRegistry.php b/modules/EcommerceFinance/Services/BusinessMirrorDatasetRegistry.php
index 52029d6d..911af9a2 100644
--- a/modules/EcommerceFinance/Services/BusinessMirrorDatasetRegistry.php
+++ b/modules/EcommerceFinance/Services/BusinessMirrorDatasetRegistry.php
@@ -55,6 +55,7 @@ class BusinessMirrorDatasetRegistry
     {
         $definition = $this->definitions()[$key] ?? null;
         if (! $definition || ! app(UpstreamSourceRegistry::class)->enabled($definition['source_key'])) return false;
+        if (! app(BusinessDataPlacement::class)->shouldSchedule($key)) return false;
         if (array_key_exists($key, (array) config('upstream_mirrors.datasets', []))) return true;
         return Schema::hasTable('business_mirror_datasets') && DB::table('business_mirror_datasets')
             ->where('dataset_key', $key)->where('sync_enabled', true)->exists();
diff --git a/modules/EcommerceFinance/Services/ClickHouseClient.php b/modules/EcommerceFinance/Services/ClickHouseClient.php
new file mode 100644
index 00000000..3336b1c1
--- /dev/null
+++ b/modules/EcommerceFinance/Services/ClickHouseClient.php
@@ -0,0 +1,205 @@
+<?php
+
+namespace Modules\EcommerceFinance\Services;
+
+use GuzzleHttp\Client;
+use GuzzleHttp\ClientInterface;
+use GuzzleHttp\Exception\GuzzleException;
+use Psr\Http\Message\StreamInterface;
+use RuntimeException;
+
+class ClickHouseClient
+{
+    private readonly ClientInterface $http;
+
+    /** @param array<string,mixed>|null $configuration */
+    public function __construct(?ClientInterface $http = null, private readonly ?array $configuration = null)
+    {
+        $config = $this->config();
+        $this->http = $http ?? new Client([
+            'base_uri' => rtrim((string) $config['url'], '/') . '/',
+            'connect_timeout' => (float) $config['connect_timeout_seconds'],
+            'timeout' => (float) $config['query_timeout_seconds'],
+        ]);
+    }
+
+    public function enabled(): bool
+    {
+        return (bool) $this->config()['enabled'];
+    }
+
+    public function ping(): bool
+    {
+        if (! $this->enabled()) {
+            return false;
+        }
+
+        try {
+            $response = $this->http->request('GET', 'ping', $this->authOptions());
+
+            return $response->getStatusCode() === 200
+                && trim((string) $response->getBody()) === 'Ok.';
+        } catch (GuzzleException) {
+            return false;
+        }
+    }
+
+    /** @param array<string,scalar|null> $parameters */
+    public function command(string $sql, array $parameters = []): void
+    {
+        $this->request($sql, $parameters);
+    }
+
+    /** @param array<string,scalar|null> $parameters @return list<array<string,mixed>> */
+    public function rows(string $sql, array $parameters = []): array
+    {
+        $body = (string) $this->request(rtrim($sql, "; \t\n\r") . ' FORMAT JSONEachRow', $parameters);
+        if (trim($body) === '') {
+            return [];
+        }
+
+        $rows = [];
+        foreach (preg_split('/\r?\n/', trim($body)) ?: [] as $line) {
+            $decoded = json_decode($line, true, 64, JSON_THROW_ON_ERROR);
+            if (! is_array($decoded)) {
+                throw new RuntimeException('clickhouse_invalid_json_row');
+            }
+            $rows[] = $decoded;
+        }
+
+        return $rows;
+    }
+
+    /** @param array<string,scalar|null> $parameters */
+    public function scalar(string $sql, array $parameters = []): mixed
+    {
+        $body = (string) $this->request(rtrim($sql, "; \t\n\r") . ' FORMAT TabSeparatedRaw', $parameters);
+
+        return trim($body);
+    }
+
+    /**
+     * @param list<string> $columns
+     * @param list<array<string,mixed>> $rows
+     */
+    public function insertJsonEachRow(string $table, array $columns, array $rows): void
+    {
+        if ($rows === []) {
+            return;
+        }
+
+        $table = $this->qualifiedTable($table);
+        $columns = array_map([$this, 'identifier'], $columns);
+        $payload = '';
+        foreach ($rows as $row) {
+            $payload .= json_encode($row, JSON_UNESCAPED_UNICODE | JSON_PRESERVE_ZERO_FRACTION | JSON_THROW_ON_ERROR) . "\n";
+        }
+
+        $sql = sprintf('INSERT INTO %s (%s) FORMAT JSONEachRow', $table, implode(', ', $columns));
+        $this->request($sql, [], $payload, 'application/x-ndjson');
+    }
+
+    /** @param array<string,scalar|null> $parameters */
+    public function stream(string $sql, string $format, array $parameters = []): StreamInterface
+    {
+        if (preg_match('/^[A-Za-z][A-Za-z0-9]*$/', $format) !== 1) {
+            throw new RuntimeException('clickhouse_invalid_format');
+        }
+
+        return $this->request(
+            rtrim($sql, "; \t\n\r") . ' FORMAT ' . $format,
+            $parameters,
+            null,
+            null,
+            true,
+        );
+    }
+
+    public function qualifiedTable(string $table): string
+    {
+        return $this->identifier((string) $this->config()['database']) . '.' . $this->identifier($table);
+    }
+
+    public function identifier(string $value): string
+    {
+        if (preg_match('/^[A-Za-z_][A-Za-z0-9_]*$/', $value) !== 1) {
+            throw new RuntimeException('clickhouse_invalid_identifier');
+        }
+
+        return '`' . $value . '`';
+    }
+
+    /**
+     * @param array<string,scalar|null> $parameters
+     * @return StreamInterface|string
+     */
+    private function request(
+        string $sql,
+        array $parameters = [],
+        ?string $body = null,
+        ?string $contentType = null,
+        bool $stream = false,
+    ): StreamInterface|string {
+        if (! $this->enabled()) {
+            throw new RuntimeException('clickhouse_disabled');
+        }
+
+        $query = [
+            'database' => (string) $this->config()['database'],
+            'wait_end_of_query' => 1,
+            'query' => $sql,
+        ];
+        foreach ($parameters as $name => $value) {
+            if (preg_match('/^[A-Za-z_][A-Za-z0-9_]*$/', $name) !== 1 || (! is_scalar($value) && $value !== null)) {
+                throw new RuntimeException('clickhouse_invalid_parameter');
+            }
+            $query['param_' . $name] = $value;
+        }
+
+        $options = $this->authOptions() + ['query' => $query, 'stream' => $stream];
+        if ($stream) {
+            $options['timeout'] = (float) $this->config()['stream_timeout_seconds'];
+        }
+        if ($body !== null) {
+            $options['body'] = $body;
+        }
+        if ($contentType !== null) {
+            $options['headers'] = ['Content-Type' => $contentType];
+        }
+
+        try {
+            $response = $this->http->request('POST', '', $options);
+        } catch (GuzzleException $exception) {
+            throw new RuntimeException('clickhouse_request_failed', 0, $exception);
+        }
+
+        if ($response->getStatusCode() < 200 || $response->getStatusCode() >= 300) {
+            throw new RuntimeException('clickhouse_unexpected_status');
+        }
+
+        return $stream ? $response->getBody() : (string) $response->getBody();
+    }
+
+    /** @return array<string,mixed> */
+    private function authOptions(): array
+    {
+        $username = (string) $this->config()['username'];
+
+        return $username === '' ? [] : ['auth' => [$username, (string) $this->config()['password']]];
+    }
+
+    /** @return array<string,mixed> */
+    private function config(): array
+    {
+        return ($this->configuration ?? (array) config('business_data.clickhouse', [])) + [
+            'enabled' => false,
+            'url' => 'http://clickhouse:8123',
+            'database' => 'caiwu_business_analytics',
+            'username' => 'caiwu_app',
+            'password' => '',
+            'connect_timeout_seconds' => 3,
+            'query_timeout_seconds' => 30,
+            'stream_timeout_seconds' => 1800,
+        ];
+    }
+}
diff --git a/modules/EcommerceFinance/Services/ClickHouseSchemaManager.php b/modules/EcommerceFinance/Services/ClickHouseSchemaManager.php
new file mode 100644
index 00000000..19cedaea
--- /dev/null
+++ b/modules/EcommerceFinance/Services/ClickHouseSchemaManager.php
@@ -0,0 +1,83 @@
+<?php
+
+namespace Modules\EcommerceFinance\Services;
+
+use Modules\EcommerceFinance\Contracts\BusinessMirrorDataset;
+use RuntimeException;
+
+class ClickHouseSchemaManager
+{
+    public function __construct(
+        private readonly BusinessMirrorSchemaManager $sourceSchemas,
+        private readonly ClickHouseClient $clickHouse,
+        private readonly BusinessDataPlacement $placements,
+    ) {
+    }
+
+    /** @return list<array<string,mixed>> */
+    public function ensure(BusinessMirrorDataset $dataset): array
+    {
+        if (! in_array($this->placements->targetStore($dataset->key()), ['clickhouse', 'parquet'], true)) {
+            throw new RuntimeException('business_data_dataset_not_columnar');
+        }
+
+        $columns = $this->sourceSchemas->columns($dataset);
+        $definitions = ['`_mirror_key` String'];
+        foreach ($columns as $column) {
+            $definitions[] = $this->clickHouse->identifier((string) $column['column_name'])
+                . ' Nullable(' . $this->type($column) . ')';
+        }
+        $definitions[] = '`_mirror_generation` UUID';
+        $definitions[] = "`_mirror_synced_at` DateTime64(6, 'Asia/Shanghai')";
+        $definitions[] = '`_mirror_version` UInt64';
+
+        // Hash partitioning keeps every version of one business key in the same
+        // partition, so ReplacingMergeTree can reconcile updates without relying
+        // on a business date that might later be corrected upstream.
+        $sql = sprintf(
+            'CREATE TABLE IF NOT EXISTS %s (%s) '
+            . 'ENGINE = ReplacingMergeTree(_mirror_version) '
+            . 'PARTITION BY modulo(cityHash64(_mirror_key), 32) '
+            . 'ORDER BY (_mirror_key) SETTINGS index_granularity = 8192',
+            $this->clickHouse->qualifiedTable($dataset->targetTable()),
+            implode(', ', $definitions),
+        );
+        $this->clickHouse->command($sql);
+
+        return $columns;
+    }
+
+    public function tableExists(BusinessMirrorDataset $dataset): bool
+    {
+        return (int) $this->clickHouse->scalar(
+            'SELECT count() FROM system.tables WHERE database = {database:String} AND name = {table:String}',
+            [
+                'database' => (string) config('business_data.clickhouse.database', 'caiwu_business_analytics'),
+                'table' => $dataset->targetTable(),
+            ],
+        ) === 1;
+    }
+
+    /** @param array<string,mixed> $column */
+    private function type(array $column): string
+    {
+        $type = strtolower((string) ($column['data_type'] ?? ''));
+        $precision = min(max((int) ($column['numeric_precision'] ?? 20), 1), 76);
+        $scale = min(max((int) ($column['numeric_scale'] ?? 0), 0), $precision);
+
+        return match ($type) {
+            'bit', 'boolean' => 'UInt8',
+            // SQL Server tinyint is unsigned (0..255), unlike MySQL tinyint.
+            'tinyint' => 'UInt8',
+            'smallint' => 'Int16',
+            'int', 'integer' => 'Int32',
+            'bigint' => 'Int64',
+            'decimal', 'numeric', 'money', 'smallmoney' => sprintf('Decimal(%d, %d)', $precision, $scale),
+            'float', 'real', 'double', 'double precision' => 'Float64',
+            'date' => 'Date32',
+            'datetime', 'datetime2', 'smalldatetime', 'timestamp without time zone', 'timestamp with time zone' => "DateTime64(6, 'Asia/Shanghai')",
+            'uniqueidentifier', 'uuid' => 'UUID',
+            default => 'String',
+        };
+    }
+}
diff --git a/tests/Feature/Console/BusinessDataColumnarBackfillTest.php b/tests/Feature/Console/BusinessDataColumnarBackfillTest.php
new file mode 100644
index 00000000..17e580b8
--- /dev/null
+++ b/tests/Feature/Console/BusinessDataColumnarBackfillTest.php
@@ -0,0 +1,140 @@
+<?php
+
+namespace Tests\Feature\Console;
+
+use Illuminate\Database\Schema\Blueprint;
+use Illuminate\Support\Facades\DB;
+use Illuminate\Support\Facades\Schema;
+use Illuminate\Support\Facades\Storage;
+use GuzzleHttp\Psr7\Utils;
+use Mockery;
+use Modules\EcommerceFinance\Repositories\ColumnarBusinessMirrorRepository;
+use Modules\EcommerceFinance\Services\ClickHouseClient;
+use Modules\EcommerceFinance\Services\ClickHouseSchemaManager;
+use Tests\TestCase;
+
+class BusinessDataColumnarBackfillTest extends TestCase
+{
+    protected function setUp(): void
+    {
+        parent::setUp();
+        config([
+            'database.connections.business_mirror' => ['driver' => 'sqlite', 'database' => ':memory:', 'prefix' => ''],
+            'upstream_mirrors.connection' => 'business_mirror',
+            'upstream_mirrors.datasets' => [
+                'source.schema.fact' => [
+                    'source_key' => 'source', 'schema' => 'schema', 'object' => 'fact',
+                    'target' => 'fact_rows', 'keys' => ['Id'], 'watermark' => 'updated_at',
+                ],
+            ],
+            'business_data.datasets' => [
+                'source.schema.fact' => ['role' => 'fact', 'target_store' => 'clickhouse'],
+            ],
+        ]);
+        DB::purge('business_mirror');
+
+        $migration = require base_path('database/migrations/2026_09_07_120000_create_business_data_placement_tables.php');
+        $migration->up();
+        Schema::connection('business_mirror')->create('fact_rows', function (Blueprint $table): void {
+            $table->string('_mirror_key')->primary();
+            $table->integer('Id');
+            $table->string('name')->nullable();
+            $table->dateTime('updated_at')->nullable();
+            $table->dateTime('_mirror_synced_at', 6);
+        });
+        DB::connection('business_mirror')->table('fact_rows')->insert([
+            ['_mirror_key' => 'key-1', 'Id' => 1, 'name' => '一', 'updated_at' => '2026-09-01 00:00:00', '_mirror_synced_at' => '2026-09-07 00:00:00'],
+            ['_mirror_key' => 'key-2', 'Id' => 2, 'name' => '二', 'updated_at' => '2026-09-02 00:00:00', '_mirror_synced_at' => '2026-09-07 00:00:00'],
+        ]);
+    }
+
+    protected function tearDown(): void
+    {
+        DB::purge('business_mirror');
+        parent::tearDown();
+    }
+
+    public function test_it_backfills_from_existing_mariadb_and_stops_at_verified_without_cutover_or_delete(): void
+    {
+        $columns = [
+            ['column_name' => 'Id', 'data_type' => 'int'],
+            ['column_name' => 'name', 'data_type' => 'nvarchar'],
+            ['column_name' => 'updated_at', 'data_type' => 'datetime2'],
+        ];
+        $schema = Mockery::mock(ClickHouseSchemaManager::class);
+        $schema->shouldReceive('ensure')->once()->andReturn($columns);
+        $columnar = Mockery::mock(ColumnarBusinessMirrorRepository::class);
+        $columnar->shouldReceive('storeBackfillBatch')->once()->withArgs(
+            static fn ($dataset, array $actualColumns, array $rows, string $generation): bool =>
+                $dataset->key() === 'source.schema.fact'
+                && $actualColumns === $columns
+                && count($rows) === 2
+                && $generation !== '',
+        );
+        $columnar->shouldReceive('count')->once()->andReturn(2);
+        $columnar->shouldReceive('distinctKeys')->once()->andReturn(2);
+        $clickHouse = Mockery::mock(ClickHouseClient::class);
+        $clickHouse->shouldReceive('enabled')->once()->andReturnTrue();
+        $clickHouse->shouldReceive('ping')->once()->andReturnTrue();
+        $this->app->instance(ClickHouseSchemaManager::class, $schema);
+        $this->app->instance(ColumnarBusinessMirrorRepository::class, $columnar);
+        $this->app->instance(ClickHouseClient::class, $clickHouse);
+
+        $exitCode = $this->artisan('business-data:backfill-columnar', [
+            'dataset' => 'source.schema.fact', '--batch' => 100, '--max-batches' => 1,
+        ])->run();
+
+        $state = DB::connection('business_mirror')->table('business_data_placements')
+            ->where('dataset_key', 'source.schema.fact')->first();
+        $this->assertSame('verified', $state->phase, (string) $state->last_error_code);
+        $this->assertSame(0, $exitCode);
+        $this->assertSame(2, (int) $state->source_rows);
+        $this->assertSame(2, (int) $state->target_rows);
+        $this->assertNull($state->activated_at);
+        $this->assertSame(2, DB::connection('business_mirror')->table('fact_rows')->count());
+    }
+
+    public function test_verified_backfill_can_archive_all_hash_partitions_without_deleting_source(): void
+    {
+        config([
+            'business_data.archive.disk' => 'business_data_archive_test',
+            'business_data.archive.prefix' => 'archive-root',
+        ]);
+        Storage::fake('business_data_archive_test');
+        $columns = [
+            ['column_name' => 'Id', 'data_type' => 'int'],
+            ['column_name' => 'name', 'data_type' => 'nvarchar'],
+            ['column_name' => 'updated_at', 'data_type' => 'datetime2'],
+        ];
+        $schema = Mockery::mock(ClickHouseSchemaManager::class);
+        $schema->shouldReceive('ensure')->twice()->andReturn($columns);
+        $columnar = Mockery::mock(ColumnarBusinessMirrorRepository::class);
+        $columnar->shouldReceive('storeBackfillBatch')->once();
+        $columnar->shouldReceive('count')->once()->andReturn(2);
+        $columnar->shouldReceive('distinctKeys')->once()->andReturn(2);
+        $columnar->shouldReceive('parquetPartition')->times(32)
+            ->andReturnUsing(static fn () => Utils::streamFor('PAR1-fixture'));
+        $columnar->shouldReceive('partitionStats')->times(32)->andReturn([
+            'row_count' => 0,
+            'logical_checksum' => '0000000000000000',
+        ]);
+        $clickHouse = Mockery::mock(ClickHouseClient::class);
+        $clickHouse->shouldReceive('enabled')->twice()->andReturnTrue();
+        $clickHouse->shouldReceive('ping')->twice()->andReturnTrue();
+        $this->app->instance(ClickHouseSchemaManager::class, $schema);
+        $this->app->instance(ColumnarBusinessMirrorRepository::class, $columnar);
+        $this->app->instance(ClickHouseClient::class, $clickHouse);
+
+        $this->artisan('business-data:backfill-columnar', [
+            'dataset' => 'source.schema.fact', '--batch' => 100, '--max-batches' => 1,
+        ])->expectsOutputToContain('已对账')->assertSuccessful();
+        $this->artisan('business-data:backfill-columnar', [
+            'dataset' => 'source.schema.fact', '--archive' => true,
+        ])->expectsOutputToContain('32 个 Parquet 哈希分区')->assertSuccessful();
+
+        $archives = DB::connection('business_mirror')->table('business_data_archives')->get();
+        $this->assertCount(32, $archives);
+        $this->assertSame(32, collect(Storage::disk('business_data_archive_test')->allFiles())->count());
+        $this->assertSame(2, DB::connection('business_mirror')->table('fact_rows')->count());
+    }
+}
diff --git a/tests/Feature/Console/BusinessDataPlanTest.php b/tests/Feature/Console/BusinessDataPlanTest.php
new file mode 100644
index 00000000..9f2c1898
--- /dev/null
+++ b/tests/Feature/Console/BusinessDataPlanTest.php
@@ -0,0 +1,34 @@
+<?php
+
+namespace Tests\Feature\Console;
+
+use Illuminate\Database\Schema\Blueprint;
+use Illuminate\Support\Facades\DB;
+use Illuminate\Support\Facades\Schema;
+use Tests\TestCase;
+
+class BusinessDataPlanTest extends TestCase
+{
+    public function test_every_code_owned_dataset_requires_an_explicit_policy(): void
+    {
+        config([
+            'database.connections.business_mirror' => ['driver' => 'sqlite', 'database' => ':memory:', 'prefix' => ''],
+            'upstream_mirrors.connection' => 'business_mirror',
+            'upstream_mirrors.datasets' => [
+                'source.schema.fact' => ['source_key' => 'source', 'schema' => 'schema', 'object' => 'fact', 'target' => 'fact_rows', 'keys' => ['id']],
+            ],
+            'business_data.datasets' => [],
+        ]);
+        DB::purge('business_mirror');
+        Schema::connection('business_mirror')->create('business_data_placements', function (Blueprint $table): void {
+            $table->string('dataset_key')->primary();
+            $table->string('phase');
+        });
+
+        $this->artisan('business-data:plan')
+            ->expectsOutputToContain('缺少存储策略：source.schema.fact')
+            ->assertFailed();
+
+        DB::purge('business_mirror');
+    }
+}
diff --git a/tests/Unit/Deployment/BusinessMirrorScheduleTest.php b/tests/Unit/Deployment/BusinessMirrorScheduleTest.php
index c67ab0ef..73e7075f 100644
--- a/tests/Unit/Deployment/BusinessMirrorScheduleTest.php
+++ b/tests/Unit/Deployment/BusinessMirrorScheduleTest.php
@@ -51,4 +51,32 @@ class BusinessMirrorScheduleTest extends TestCase
         $this->assertStringNotContainsString("['stat_date', 'push_fee', 'item_id']", $contents);
         $this->assertStringContainsString("['stat_date', 'push_fee']", $contents);
     }
+
+    public function test_clickhouse_is_internal_resource_bounded_and_uses_private_runtime_credentials(): void
+    {
+        $root = dirname(__DIR__, 3);
+        $compose = file_get_contents($root . '/compose.jiangling.yml');
+        $start = strpos($compose, "  clickhouse:\n");
+        $end = strpos($compose, "  app:\n", $start);
+        $service = substr($compose, $start, $end - $start);
+
+        $this->assertStringContainsString('clickhouse/clickhouse-server:26.8.2.7-alpine', $service);
+        $this->assertStringContainsString('mem_limit: 4g', $service);
+        $this->assertStringContainsString('cpus: "2.00"', $service);
+        $this->assertStringContainsString('max-size: "20m"', $service);
+        $this->assertStringNotContainsString('ports:', $service);
+        $this->assertStringContainsString('./runtime/clickhouse.env', $compose);
+
+        foreach (['activate-release.sh', 'activate-incremental-release.sh'] as $script) {
+            $contents = file_get_contents($root . '/deploy/' . $script);
+            $this->assertStringContainsString('openssl rand -hex 24', $contents, $script);
+            $this->assertStringContainsString('chmod 600 "$clickhouse_env"', $contents, $script);
+            $pull = strpos($contents, 'pull clickhouse');
+            $migration = strpos($contents, 'if [ "$RUN_MIGRATIONS" = "1" ]');
+            $this->assertNotFalse($pull, $script);
+            $this->assertLessThan($migration, $pull, $script);
+            $this->assertStringContainsString('docker inspect caiwu_clickhouse', $contents, $script);
+            $this->assertStringContainsString('ClickHouse did not become healthy', $contents, $script);
+        }
+    }
 }
diff --git a/tests/Unit/EcommerceFinance/BusinessDataPlacementTest.php b/tests/Unit/EcommerceFinance/BusinessDataPlacementTest.php
new file mode 100644
index 00000000..1caf66d1
--- /dev/null
+++ b/tests/Unit/EcommerceFinance/BusinessDataPlacementTest.php
@@ -0,0 +1,85 @@
+<?php
+
+namespace Tests\Unit\EcommerceFinance;
+
+use Illuminate\Database\Schema\Blueprint;
+use Illuminate\Support\Facades\DB;
+use Illuminate\Support\Facades\Schema;
+use Modules\EcommerceFinance\Services\BusinessDataPlacement;
+use Tests\TestCase;
+
+class BusinessDataPlacementTest extends TestCase
+{
+    protected function setUp(): void
+    {
+        parent::setUp();
+        config([
+            'database.connections.business_mirror' => ['driver' => 'sqlite', 'database' => ':memory:', 'prefix' => ''],
+            'upstream_mirrors.connection' => 'business_mirror',
+            'upstream_mirrors.datasets' => [
+                'source.schema.fact' => ['source_key' => 'source', 'schema' => 'schema', 'object' => 'fact', 'target' => 'fact_rows', 'keys' => ['id']],
+                'source.schema.unused' => ['source_key' => 'source', 'schema' => 'schema', 'object' => 'unused', 'target' => 'unused_rows', 'keys' => ['id']],
+            ],
+            'business_data.datasets' => [
+                'source.schema.fact' => ['role' => 'fact', 'target_store' => 'clickhouse', 'sync' => 'after_columnar_activation'],
+                'source.schema.unused' => ['role' => 'archive_only', 'target_store' => 'parquet', 'sync' => 'disabled'],
+            ],
+        ]);
+        DB::purge('business_mirror');
+        Schema::connection('business_mirror')->create('business_data_placements', function (Blueprint $table): void {
+            $table->string('dataset_key')->primary();
+            $table->string('target_store');
+            $table->string('phase')->default('planned');
+            $table->string('active_generation')->nullable();
+            $table->timestamps();
+        });
+    }
+
+    protected function tearDown(): void
+    {
+        DB::purge('business_mirror');
+        parent::tearDown();
+    }
+
+    public function test_it_stops_unused_data_and_enables_columnar_sync_only_after_activation(): void
+    {
+        $placement = new BusinessDataPlacement(app('db'));
+
+        $this->assertFalse($placement->shouldSchedule('source.schema.unused'));
+        $this->assertFalse($placement->shouldSchedule('source.schema.fact'));
+
+        DB::connection('business_mirror')->table('business_data_placements')->insert([
+            'dataset_key' => 'source.schema.fact',
+            'target_store' => 'clickhouse',
+            'phase' => 'active',
+            'active_generation' => '11111111-1111-4111-8111-111111111111',
+            'created_at' => now(),
+            'updated_at' => now(),
+        ]);
+
+        $this->assertTrue($placement->shouldSchedule('source.schema.fact'));
+        $this->assertTrue($placement->isActive('source.schema.fact', 'clickhouse'));
+    }
+
+    public function test_runtime_registered_dataset_defaults_to_a_mariadb_dimension(): void
+    {
+        $policy = (new BusinessDataPlacement(app('db')))->policy('database_4.accounts_v1');
+
+        $this->assertSame('external_dimension', $policy['role']);
+        $this->assertSame('mariadb', $policy['target_store']);
+        $this->assertSame('always', $policy['sync']);
+    }
+
+    public function test_every_fixed_upstream_dataset_has_exactly_one_storage_policy(): void
+    {
+        $upstream = require base_path('config/upstream_mirrors.php');
+        $businessData = require base_path('config/business_data.php');
+
+        $this->assertCount(51, $upstream['datasets']);
+        $upstreamKeys = array_keys($upstream['datasets']);
+        $policyKeys = array_keys($businessData['datasets']);
+        sort($upstreamKeys);
+        sort($policyKeys);
+        $this->assertSame($upstreamKeys, $policyKeys);
+    }
+}
diff --git a/tests/Unit/EcommerceFinance/ClickHouseClientTest.php b/tests/Unit/EcommerceFinance/ClickHouseClientTest.php
new file mode 100644
index 00000000..dac3d142
--- /dev/null
+++ b/tests/Unit/EcommerceFinance/ClickHouseClientTest.php
@@ -0,0 +1,71 @@
+<?php
+
+namespace Tests\Unit\EcommerceFinance;
+
+use GuzzleHttp\Client;
+use GuzzleHttp\Handler\MockHandler;
+use GuzzleHttp\HandlerStack;
+use GuzzleHttp\Middleware;
+use GuzzleHttp\Psr7\Response;
+use Modules\EcommerceFinance\Services\ClickHouseClient;
+use RuntimeException;
+use Tests\TestCase;
+
+class ClickHouseClientTest extends TestCase
+{
+    public function test_http_client_uses_parameters_and_json_each_row_without_putting_credentials_in_sql(): void
+    {
+        $history = [];
+        $handler = HandlerStack::create(new MockHandler([
+            new Response(200, [], "Ok.\n"),
+            new Response(200, [], "{\"value\":1}\n{\"value\":2}\n"),
+            new Response(200, [], ''),
+            new Response(200, [], 'PAR1'),
+        ]));
+        $handler->push(Middleware::history($history));
+        $http = new Client(['handler' => $handler]);
+        $client = new ClickHouseClient($http, $this->configuration());
+
+        $this->assertTrue($client->ping());
+        $this->assertSame([['value' => 1], ['value' => 2]], $client->rows(
+            'SELECT value FROM numbers(2) WHERE value >= {minimum:UInt8}',
+            ['minimum' => 0],
+        ));
+        $client->insertJsonEachRow('fact_rows', ['id', 'name'], [
+            ['id' => 1, 'name' => '测试'],
+        ]);
+
+        parse_str($history[1]['request']->getUri()->getQuery(), $query);
+        $this->assertSame('0', (string) $query['param_minimum']);
+        $this->assertStringContainsString('{minimum:UInt8}', $query['query']);
+        $this->assertStringNotContainsString('fixture-password', $query['query']);
+        parse_str($history[2]['request']->getUri()->getQuery(), $insertQuery);
+        $this->assertStringContainsString('FORMAT JSONEachRow', $insertQuery['query']);
+        $this->assertSame("{\"id\":1,\"name\":\"测试\"}\n", (string) $history[2]['request']->getBody());
+        $this->assertSame('PAR1', (string) $client->stream('SELECT 1', 'Parquet'));
+        $this->assertSame(1800.0, $history[3]['options']['timeout']);
+    }
+
+    public function test_invalid_identifier_is_rejected_before_a_query_is_sent(): void
+    {
+        $client = new ClickHouseClient(new Client(['handler' => HandlerStack::create(new MockHandler())]), $this->configuration());
+
+        $this->expectException(RuntimeException::class);
+        $this->expectExceptionMessage('clickhouse_invalid_identifier');
+        $client->qualifiedTable('fact_rows; DROP TABLE fact_rows');
+    }
+
+    /** @return array<string,mixed> */
+    private function configuration(): array
+    {
+        return [
+            'enabled' => true,
+            'url' => 'http://clickhouse:8123',
+            'database' => 'analytics',
+            'username' => 'fixture-user',
+            'password' => 'fixture-password',
+            'connect_timeout_seconds' => 1,
+            'query_timeout_seconds' => 1,
+        ];
+    }
+}
diff --git a/tests/Unit/EcommerceFinance/ColumnarBusinessMirrorRepositoryTest.php b/tests/Unit/EcommerceFinance/ColumnarBusinessMirrorRepositoryTest.php
new file mode 100644
index 00000000..f974a899
--- /dev/null
+++ b/tests/Unit/EcommerceFinance/ColumnarBusinessMirrorRepositoryTest.php
@@ -0,0 +1,86 @@
+<?php
+
+namespace Tests\Unit\EcommerceFinance;
+
+use Mockery;
+use Modules\EcommerceFinance\Contracts\BusinessMirrorDataset;
+use Modules\EcommerceFinance\Repositories\ColumnarBusinessMirrorRepository;
+use Modules\EcommerceFinance\Services\BusinessDataPlacement;
+use Modules\EcommerceFinance\Services\BusinessMirrorSchemaManager;
+use Modules\EcommerceFinance\Services\ClickHouseClient;
+use Modules\EcommerceFinance\Services\ClickHouseSchemaManager;
+use Tests\TestCase;
+
+class ColumnarBusinessMirrorRepositoryTest extends TestCase
+{
+    public function test_backfill_preserves_source_fields_and_adds_generation_metadata(): void
+    {
+        $dataset = $this->dataset();
+        $client = Mockery::mock(ClickHouseClient::class);
+        $client->shouldReceive('insertJsonEachRow')->once()->withArgs(
+            static function (string $table, array $columns, array $rows): bool {
+                return $table === 'fact_rows'
+                    && $columns === ['_mirror_key', 'Id', 'name', '_mirror_generation', '_mirror_synced_at', '_mirror_version']
+                    && $rows[0]['_mirror_key'] === 'key-1'
+                    && $rows[0]['Id'] === 1
+                    && $rows[0]['name'] === '记录一'
+                    && $rows[0]['_mirror_generation'] === '11111111-1111-4111-8111-111111111111'
+                    && is_int($rows[0]['_mirror_version']);
+            },
+        );
+        $repository = new ColumnarBusinessMirrorRepository($client);
+
+        $repository->storeBackfillBatch($dataset, [
+            ['column_name' => 'Id', 'data_type' => 'int'],
+            ['column_name' => 'name', 'data_type' => 'nvarchar'],
+        ], [[
+            '_mirror_key' => 'key-1',
+            'Id' => 1,
+            'name' => '记录一',
+            '_mirror_synced_at' => '2026-09-07 12:00:00.000000',
+        ]], '11111111-1111-4111-8111-111111111111');
+    }
+
+    public function test_columnar_schema_uses_replacing_merge_tree_and_sql_server_safe_types(): void
+    {
+        $dataset = $this->dataset();
+        $source = Mockery::mock(BusinessMirrorSchemaManager::class);
+        $source->shouldReceive('columns')->once()->andReturn([
+            ['column_name' => 'Id', 'data_type' => 'int'],
+            ['column_name' => 'flag', 'data_type' => 'tinyint'],
+            ['column_name' => 'amount', 'data_type' => 'decimal', 'numeric_precision' => 20, 'numeric_scale' => 4],
+            ['column_name' => 'updated_at', 'data_type' => 'datetime2'],
+        ]);
+        $placements = Mockery::mock(BusinessDataPlacement::class);
+        $placements->shouldReceive('targetStore')->once()->with('source.schema.fact')->andReturn('clickhouse');
+        $client = Mockery::mock(ClickHouseClient::class);
+        $client->shouldReceive('identifier')->times(4)->andReturnUsing(static fn (string $name): string => '`' . $name . '`');
+        $client->shouldReceive('qualifiedTable')->once()->with('fact_rows')->andReturn('`analytics`.`fact_rows`');
+        $client->shouldReceive('command')->once()->withArgs(static fn (string $sql): bool =>
+            str_contains($sql, '`flag` Nullable(UInt8)')
+            && str_contains($sql, '`amount` Nullable(Decimal(20, 4))')
+            && str_contains($sql, "`updated_at` Nullable(DateTime64(6, 'Asia/Shanghai'))")
+            && str_contains($sql, 'ENGINE = ReplacingMergeTree(_mirror_version)')
+            && str_contains($sql, 'PARTITION BY modulo(cityHash64(_mirror_key), 32)')
+            && str_contains($sql, 'ORDER BY (_mirror_key)')
+        );
+
+        $columns = (new ClickHouseSchemaManager($source, $client, $placements))->ensure($dataset);
+
+        $this->assertCount(4, $columns);
+    }
+
+    private function dataset(): BusinessMirrorDataset
+    {
+        return new class implements BusinessMirrorDataset {
+            public function key(): string { return 'source.schema.fact'; }
+            public function sourceKey(): string { return 'source'; }
+            public function sourceSchema(): string { return 'schema'; }
+            public function sourceObject(): string { return 'fact'; }
+            public function targetTable(): string { return 'fact_rows'; }
+            public function keyColumns(): array { return ['Id']; }
+            public function watermarkColumn(): ?string { return 'updated_at'; }
+            public function indexDefinitions(): array { return []; }
+        };
+    }
+}
diff --git a/tests/Unit/EcommerceFinance/OrderDataMirrorRepositoryTest.php b/tests/Unit/EcommerceFinance/OrderDataMirrorRepositoryTest.php
index 8ee0879b..f602b41e 100644
--- a/tests/Unit/EcommerceFinance/OrderDataMirrorRepositoryTest.php
+++ b/tests/Unit/EcommerceFinance/OrderDataMirrorRepositoryTest.php
@@ -102,6 +102,16 @@ class OrderDataMirrorRepositoryTest extends TestCase
         $this->assertSame(9.05, $result['rows'][0]['promotion_fee']);
     }
 
+    public function test_promotion_readiness_does_not_wait_for_the_unused_sku_map(): void
+    {
+        $datasets = app(BusinessMirrorDatasetRegistry::class);
+        $mirror = Mockery::mock(GenericBusinessMirrorRepository::class);
+        $mirror->shouldReceive('connection')->andReturn(DB::connection('business_mirror'));
+        $mirror->shouldReceive('ready')->once()->andReturnTrue();
+
+        $this->assertTrue((new OrderDataMirrorRepository($datasets, $mirror))->promotionReady());
+    }
+
     private function createTables(): void
     {
         $schema = Schema::connection('business_mirror');
diff --git a/tests/Unit/EcommerceFinance/StoreReportRepositoryFilterTest.php b/tests/Unit/EcommerceFinance/StoreReportRepositoryFilterTest.php
index 90302952..1d700dd5 100644
--- a/tests/Unit/EcommerceFinance/StoreReportRepositoryFilterTest.php
+++ b/tests/Unit/EcommerceFinance/StoreReportRepositoryFilterTest.php
@@ -2,15 +2,18 @@
 
 namespace Tests\Unit\EcommerceFinance;
 
+use Illuminate\Database\Schema\Blueprint;
+use Illuminate\Support\Facades\DB;
+use Illuminate\Support\Facades\Schema;
 use Modules\EcommerceFinance\Repositories\StoreReportRepository;
-use PDO;
 use Tests\TestCase;
 
 /** 店铺维度报表查询条件：生命周期字段按左闭右开区间过滤。 */
 class StoreReportRepositoryFilterTest extends TestCase
 {
-    private string $databasePath;
     private StoreReportRepository $repository;
+    private string $lifecycleGeneration = '11111111-1111-4111-8111-111111111111';
+    private string $dailyGeneration = '22222222-2222-4222-8222-222222222222';
 
     /** @var array<int,array<string,mixed>> */
     private array $stores = [
@@ -22,49 +25,61 @@ class StoreReportRepositoryFilterTest extends TestCase
     {
         parent::setUp();
 
-        $this->databasePath = sys_get_temp_dir() . '/store-report-filters-' . uniqid() . '.sqlite';
-        $pdo = new PDO('sqlite:' . $this->databasePath);
-        $pdo->exec('CREATE TABLE bills (
-            id INTEGER PRIMARY KEY AUTOINCREMENT,
-            platform TEXT NOT NULL,
-            type TEXT NOT NULL,
-            store_name TEXT NOT NULL,
-            occurred_at TEXT,
-            income NUMERIC NOT NULL DEFAULT 0,
-            expense NUMERIC NOT NULL DEFAULT 0,
-            bill_item TEXT,
-            record_code TEXT NOT NULL,
-            source_file TEXT NOT NULL,
-            raw_json TEXT NOT NULL,
-            imported_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
-            UNIQUE(platform, record_code)
-        )');
-        $pdo->exec('CREATE TABLE store_lifecycles (
-            id INTEGER PRIMARY KEY AUTOINCREMENT,
-            erp_store_name TEXT NOT NULL,
-            company_name TEXT,
-            effective_at TEXT,
-            expires_at TEXT,
-            source_file TEXT NOT NULL,
-            raw_json TEXT NOT NULL,
-            imported_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
-            UNIQUE(erp_store_name, effective_at, expires_at)
-        )');
-        $insert = $pdo->prepare(
-            'INSERT INTO store_lifecycles (erp_store_name, company_name, effective_at, expires_at, source_file, raw_json)
-             VALUES (?, ?, ?, ?, ?, ?)',
-        );
-        $insert->execute(['店铺A', '主体甲', '2026-07-15 00:00:00', '2026-08-01 00:00:00', 'test', '{}']);
-        $insert->execute(['店铺A', '主体乙', '2026-08-10 00:00:00', null, 'test', '{}']);
-        $pdo = null;
-
-        config(['database.connections.bills_sqlite.database' => $this->databasePath]);
+        config(['database.connections.business_mirror' => [
+            'driver' => 'sqlite', 'database' => ':memory:', 'prefix' => '',
+        ], 'upstream_mirrors.connection' => 'business_mirror']);
+        DB::purge('business_mirror');
+        $schema = Schema::connection('business_mirror');
+        $schema->create('store_report_import_states', function (Blueprint $table): void {
+            $table->string('dataset_key')->primary();
+            $table->string('active_generation')->nullable();
+            $table->dateTime('initial_completed_at')->nullable();
+        });
+        $schema->create('store_report_lifecycle_rows', function (Blueprint $table): void {
+            $table->increments('id');
+            $table->string('generation');
+            $table->string('business_key');
+            $table->string('erp_store_name');
+            $table->string('company_name')->nullable();
+            $table->dateTime('effective_at')->nullable();
+            $table->dateTime('expires_at')->nullable();
+        });
+        $schema->create('store_report_daily_rows', function (Blueprint $table): void {
+            $table->increments('id');
+            $table->string('generation');
+            $table->string('store_name');
+            $table->date('occurred_on');
+            $table->decimal('income', 20, 4)->default(0);
+            $table->decimal('expense', 20, 4)->default(0);
+        });
+        DB::connection('business_mirror')->table('store_report_import_states')->insert([
+            ['dataset_key' => StoreReportRepository::LIFECYCLES_DATASET, 'active_generation' => $this->lifecycleGeneration, 'initial_completed_at' => now()],
+            ['dataset_key' => StoreReportRepository::DAILY_DATASET, 'active_generation' => $this->dailyGeneration, 'initial_completed_at' => now()],
+        ]);
+        DB::connection('business_mirror')->table('store_report_lifecycle_rows')->insert([
+            [
+                'generation' => $this->lifecycleGeneration,
+                'business_key' => 'lifecycle-a-1',
+                'erp_store_name' => '店铺A',
+                'company_name' => '主体甲',
+                'effective_at' => '2026-07-15 00:00:00',
+                'expires_at' => '2026-08-01 00:00:00',
+            ],
+            [
+                'generation' => $this->lifecycleGeneration,
+                'business_key' => 'lifecycle-a-2',
+                'erp_store_name' => '店铺A',
+                'company_name' => '主体乙',
+                'effective_at' => '2026-08-10 00:00:00',
+                'expires_at' => null,
+            ],
+        ]);
         $this->repository = new StoreReportRepository(app('db'));
     }
 
     protected function tearDown(): void
     {
-        @unlink($this->databasePath);
+        DB::purge('business_mirror');
         parent::tearDown();
     }
 
diff --git a/tests/Unit/EcommerceFinance/StoreReportRepositoryTest.php b/tests/Unit/EcommerceFinance/StoreReportRepositoryTest.php
index 899591d6..148e47d3 100644
--- a/tests/Unit/EcommerceFinance/StoreReportRepositoryTest.php
+++ b/tests/Unit/EcommerceFinance/StoreReportRepositoryTest.php
@@ -3,68 +3,70 @@
 namespace Tests\Unit\EcommerceFinance;
 
 use App\Business\UpstreamBusinessInformation;
+use Illuminate\Database\Schema\Blueprint;
 use Illuminate\Support\Carbon;
+use Illuminate\Support\Facades\DB;
+use Illuminate\Support\Facades\Schema;
 use Modules\EcommerceFinance\Repositories\BusinessInformationMirrorRepository;
 use Modules\EcommerceFinance\Repositories\InstallationDetailMirrorRepository;
 use Modules\EcommerceFinance\Repositories\StoreAccountMirrorRepository;
 use Modules\EcommerceFinance\Repositories\StoreReportRepository;
 use Modules\EcommerceFinance\Services\UpstreamQueryCacheFactory;
 use Mockery;
-use PDO;
 use RuntimeException;
 use Tests\TestCase;
 
 /**
- * 店铺维度报表数据组合：本地账单 SQLite 的生命周期拆行与收入归属。
+ * 店铺维度报表数据组合：共享生命周期和日汇总的拆行与收入归属。
  */
 class StoreReportRepositoryTest extends TestCase
 {
-    private string $databasePath;
     private StoreReportRepository $repository;
-    private ?PDO $pdo = null;
+    private string $lifecycleGeneration = '11111111-1111-4111-8111-111111111111';
+    private string $dailyGeneration = '22222222-2222-4222-8222-222222222222';
 
     protected function setUp(): void
     {
         parent::setUp();
 
-        $this->databasePath = sys_get_temp_dir() . '/store-report-rows-' . uniqid() . '.sqlite';
-        $pdo = new PDO('sqlite:' . $this->databasePath);
-        $pdo->exec('CREATE TABLE bills (
-            id INTEGER PRIMARY KEY AUTOINCREMENT,
-            platform TEXT NOT NULL,
-            type TEXT NOT NULL,
-            store_name TEXT NOT NULL,
-            occurred_at TEXT,
-            income NUMERIC NOT NULL DEFAULT 0,
-            expense NUMERIC NOT NULL DEFAULT 0,
-            bill_item TEXT,
-            record_code TEXT NOT NULL,
-            source_file TEXT NOT NULL,
-            raw_json TEXT NOT NULL,
-            imported_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
-            UNIQUE(platform, record_code)
-        )');
-        $pdo->exec('CREATE TABLE store_lifecycles (
-            id INTEGER PRIMARY KEY AUTOINCREMENT,
-            erp_store_name TEXT NOT NULL,
-            company_name TEXT,
-            effective_at TEXT,
-            expires_at TEXT,
-            source_file TEXT NOT NULL,
-            raw_json TEXT NOT NULL,
-            imported_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
-            UNIQUE(erp_store_name, effective_at, expires_at)
-        )');
-        $this->pdo = $pdo;
-        config(['database.connections.bills_sqlite.database' => $this->databasePath]);
+        config(['database.connections.business_mirror' => [
+            'driver' => 'sqlite', 'database' => ':memory:', 'prefix' => '',
+        ], 'upstream_mirrors.connection' => 'business_mirror']);
+        DB::purge('business_mirror');
+        $schema = Schema::connection('business_mirror');
+        $schema->create('store_report_import_states', function (Blueprint $table): void {
+            $table->string('dataset_key')->primary();
+            $table->string('active_generation')->nullable();
+            $table->dateTime('initial_completed_at')->nullable();
+        });
+        $schema->create('store_report_lifecycle_rows', function (Blueprint $table): void {
+            $table->increments('id');
+            $table->string('generation');
+            $table->string('business_key');
+            $table->string('erp_store_name');
+            $table->string('company_name')->nullable();
+            $table->dateTime('effective_at')->nullable();
+            $table->dateTime('expires_at')->nullable();
+        });
+        $schema->create('store_report_daily_rows', function (Blueprint $table): void {
+            $table->increments('id');
+            $table->string('generation');
+            $table->string('store_name');
+            $table->date('occurred_on');
+            $table->decimal('income', 20, 4)->default(0);
+            $table->decimal('expense', 20, 4)->default(0);
+        });
+        DB::connection('business_mirror')->table('store_report_import_states')->insert([
+            ['dataset_key' => StoreReportRepository::LIFECYCLES_DATASET, 'active_generation' => $this->lifecycleGeneration, 'initial_completed_at' => now()],
+            ['dataset_key' => StoreReportRepository::DAILY_DATASET, 'active_generation' => $this->dailyGeneration, 'initial_completed_at' => now()],
+        ]);
         $this->repository = new StoreReportRepository(app('db'));
     }
 
     protected function tearDown(): void
     {
         Carbon::setTestNow();
-        $this->pdo = null;
-        @unlink($this->databasePath);
+        DB::purge('business_mirror');
         parent::tearDown();
     }
 
@@ -110,12 +112,14 @@ class StoreReportRepositoryTest extends TestCase
         $this->assertSame('存续中', $rows[0]['invalid_at']);
     }
 
-    public function test_missing_sqlite_file_is_reported_instead_of_returning_misleading_zeroes(): void
+    public function test_unfinished_shared_daily_summary_is_reported_instead_of_returning_misleading_zeroes(): void
     {
-        config(['database.connections.bills_sqlite.database' => sys_get_temp_dir() . '/store-report-missing-' . uniqid() . '.sqlite']);
+        DB::connection('business_mirror')->table('store_report_import_states')
+            ->where('dataset_key', StoreReportRepository::DAILY_DATASET)
+            ->update(['active_generation' => null, 'initial_completed_at' => null]);
 
         $this->expectException(RuntimeException::class);
-        $this->expectExceptionMessage('店铺账单 SQLite 尚未挂载');
+        $this->expectExceptionMessage('store_report_dataset_not_ready');
         $this->repository->rows([['erp_store_name' => '店铺A']], 2026);
     }
 
@@ -184,19 +188,24 @@ class StoreReportRepositoryTest extends TestCase
 
     private function insertBill(string $store, string $occurredAt, float $income): void
     {
-        $statement = $this->pdo->prepare(
-            'INSERT INTO bills (platform, type, store_name, occurred_at, income, expense, record_code, source_file, raw_json)
-             VALUES (?, ?, ?, ?, ?, 0, ?, ?, ?)',
-        );
-        $statement->execute(['test', 'income', $store, $occurredAt, $income, uniqid(), 'test', '{}']);
+        DB::connection('business_mirror')->table('store_report_daily_rows')->insert([
+            'generation' => $this->dailyGeneration,
+            'store_name' => $store,
+            'occurred_on' => substr($occurredAt, 0, 10),
+            'income' => $income,
+            'expense' => 0,
+        ]);
     }
 
     private function insertLifecycle(string $store, string $company, ?string $from, ?string $to): void
     {
-        $statement = $this->pdo->prepare(
-            'INSERT INTO store_lifecycles (erp_store_name, company_name, effective_at, expires_at, source_file, raw_json)
-             VALUES (?, ?, ?, ?, ?, ?)',
-        );
-        $statement->execute([$store, $company, $from, $to, 'test', '{}']);
+        DB::connection('business_mirror')->table('store_report_lifecycle_rows')->insert([
+            'generation' => $this->lifecycleGeneration,
+            'business_key' => hash('sha256', implode('|', [$store, (string) $from, (string) $to])),
+            'erp_store_name' => $store,
+            'company_name' => $company,
+            'effective_at' => $from,
+            'expires_at' => $to,
+        ]);
     }
 }