fix: 分批导出账单并加速渠道查询 (#144)
This commit is contained in:
parent
91bec8f386
commit
881796f13e
@ -246,6 +246,7 @@ return [
|
||||
['Id'],
|
||||
['shop_id', 'fee_time', 'Id'],
|
||||
['fee_time', 'Id'],
|
||||
['channel', 'fee_time', 'Id', 'inout_type', 'amount'],
|
||||
],
|
||||
],
|
||||
|
||||
|
||||
@ -0,0 +1,32 @@
|
||||
<?php
|
||||
|
||||
use Illuminate\Database\Migrations\Migration;
|
||||
use Illuminate\Database\Schema\Blueprint;
|
||||
use Illuminate\Support\Facades\Schema;
|
||||
|
||||
return new class extends Migration
|
||||
{
|
||||
private const TABLE = 'ops_jst_original_bill_info';
|
||||
private const INDEX = 'idx_original_bill_channel_time';
|
||||
|
||||
public function up(): void
|
||||
{
|
||||
if (config('system_database.mode', 'bootstrap') === 'bootstrap') return;
|
||||
$connection = (string) config('upstream_mirrors.connection', 'business_mirror');
|
||||
$schema = Schema::connection($connection);
|
||||
if (! $schema->hasTable(self::TABLE) || $schema->hasIndex(self::TABLE, self::INDEX)) return;
|
||||
$schema->table(self::TABLE, static function (Blueprint $table): void {
|
||||
$table->index(['channel', 'fee_time', 'Id', 'inout_type', 'amount'], self::INDEX);
|
||||
});
|
||||
}
|
||||
|
||||
public function down(): void
|
||||
{
|
||||
if (config('system_database.mode', 'bootstrap') === 'bootstrap') return;
|
||||
$schema = Schema::connection((string) config('upstream_mirrors.connection', 'business_mirror'));
|
||||
if (! $schema->hasTable(self::TABLE) || ! $schema->hasIndex(self::TABLE, self::INDEX)) return;
|
||||
$schema->table(self::TABLE, static function (Blueprint $table): void {
|
||||
$table->dropIndex(self::INDEX);
|
||||
});
|
||||
}
|
||||
};
|
||||
@ -139,9 +139,48 @@ class StoreReceivingAlipayBillMirrorRepository
|
||||
'balance' => 'b.amount',
|
||||
][$sort] ?? 'b.fee_time';
|
||||
|
||||
return $query->select('b.*')
|
||||
->orderBy($sortColumn, $direction)->orderBy('b.Id', $direction)->cursor()
|
||||
->map(fn (object $row): array => $this->mapRowFor($row, $shopNames[(string) $row->shop_id] ?? null, $mapping));
|
||||
return $this->exportBatches($query, $sortColumn, $direction, $shopNames, $mapping);
|
||||
}
|
||||
|
||||
/** Read bounded pages so a full export does not sort/buffer millions of rows at once. */
|
||||
private function exportBatches(Builder $query, string $sortColumn, string $direction, array $shopNames, string $mapping): iterable
|
||||
{
|
||||
$operator = $direction === 'asc' ? '>' : '<';
|
||||
$sortField = substr($sortColumn, 2);
|
||||
$last = null;
|
||||
$maximumId = $this->billQuery()->max('b.Id');
|
||||
if ($maximumId === null) return;
|
||||
|
||||
do {
|
||||
$batch = (clone $query)->where('b.Id', '<=', $maximumId);
|
||||
if ($last !== null) {
|
||||
if ($sortColumn === 'b.Id') {
|
||||
$batch->where('b.Id', $operator, $last->Id);
|
||||
} else {
|
||||
$value = $last->{$sortField};
|
||||
$batch->where(function (Builder $after) use ($sortColumn, $direction, $operator, $last, $value): void {
|
||||
if ($value === null) {
|
||||
$after->where(function (Builder $ties) use ($sortColumn, $operator, $last): void {
|
||||
$ties->whereNull($sortColumn)->where('b.Id', $operator, $last->Id);
|
||||
});
|
||||
if ($direction === 'asc') $after->orWhereNotNull($sortColumn);
|
||||
} else {
|
||||
$after->where($sortColumn, $operator, $value)
|
||||
->orWhere(function (Builder $ties) use ($sortColumn, $operator, $last, $value): void {
|
||||
$ties->where($sortColumn, $value)->where('b.Id', $operator, $last->Id);
|
||||
});
|
||||
if ($direction === 'desc') $after->orWhereNull($sortColumn);
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
$rows = $batch->select('b.*')->orderBy($sortColumn, $direction)
|
||||
->orderBy('b.Id', $direction)->limit(1000)->get();
|
||||
foreach ($rows as $row) {
|
||||
yield $this->mapRowFor($row, $shopNames[(string) $row->shop_id] ?? null, $mapping);
|
||||
}
|
||||
$last = $rows->last();
|
||||
} while ($rows->count() === 1000);
|
||||
}
|
||||
|
||||
private function billQuery(): Builder
|
||||
|
||||
@ -126,6 +126,68 @@ class StoreReceivingAlipayBillMirrorRepositoryTest extends TestCase
|
||||
$this->assertSame(20.10, $rows[0]['expense']);
|
||||
}
|
||||
|
||||
public function test_export_batches_preserve_sort_ties_nulls_and_channel_filter(): void
|
||||
{
|
||||
$this->markReady();
|
||||
$db = DB::connection('business_mirror');
|
||||
$table = $db->table('ops_jst_original_bill_info');
|
||||
$fixtures = [];
|
||||
for ($id = 1; $id <= 2105; $id++) {
|
||||
$row = $this->row($id, 7, '2026-08-01 09:00:00', '1', 10);
|
||||
$row['fee_time'] = $id <= 1005 ? null : '2026-08-01 09:00:00';
|
||||
$row['amount'] = $id <= 1005 ? null : 10;
|
||||
$fixtures[] = $row;
|
||||
}
|
||||
foreach (array_chunk($fixtures, 100) as $chunk) $table->insert($chunk);
|
||||
$table->insert($this->row(9999, 7, '2026-08-01 09:00:00', '1', 10, 'OtherChannel'));
|
||||
foreach (['occurred_at' => 'fee_time', 'income' => 'amount', 'record_code' => 'Id'] as $sort => $column) {
|
||||
foreach (['asc', 'desc'] as $direction) {
|
||||
$expected = (clone $table)->where('channel', 'CommonBillTGCPay')
|
||||
->orderBy($column, $direction)->orderBy('Id', $direction)->pluck('Id')->map('strval')->all();
|
||||
$db->flushQueryLog();
|
||||
$db->enableQueryLog();
|
||||
$rows = iterator_to_array($this->repository->exportChannelRows([], $sort, $direction, 'CommonBillTGCPay'));
|
||||
$this->assertSame($expected, array_column($rows, 'record_code'));
|
||||
$reads = array_filter($db->getQueryLog(), fn ($query) => str_contains($query['query'], 'select "b".*'));
|
||||
$this->assertCount(3, $reads);
|
||||
foreach ($reads as $read) $this->assertStringContainsString('limit 1000', $read['query']);
|
||||
$db->disableQueryLog();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public function test_export_does_not_chase_new_rows_inserted_after_it_starts(): void
|
||||
{
|
||||
$this->markReady();
|
||||
$table = DB::connection('business_mirror')->table('ops_jst_original_bill_info');
|
||||
for ($id = 1; $id <= 1001; $id++) {
|
||||
$table->insert($this->row($id, 7, '2026-08-01 09:00:00', '1', 10));
|
||||
}
|
||||
$ids = [];
|
||||
foreach ($this->repository->exportRows([], 'record_code', 'asc') as $row) {
|
||||
$ids[] = $row['record_code'];
|
||||
if (count($ids) === 1) $table->insert($this->row(9999, 7, '2026-08-01 09:00:00', '1', 10));
|
||||
}
|
||||
$this->assertCount(1001, $ids);
|
||||
$this->assertNotContains('9999', $ids);
|
||||
}
|
||||
|
||||
public function test_channel_index_migration_and_rollback_preserve_rows(): void
|
||||
{
|
||||
config(['system_database.mode' => 'connected']);
|
||||
$table = DB::connection('business_mirror')->table('ops_jst_original_bill_info');
|
||||
$table->insert($this->row(1, 7, '2026-08-01 09:00:00', '1', 10, str_repeat('A', 101)));
|
||||
$migration = require base_path('database/migrations/2026_09_09_170000_index_original_bill_channel_queries.php');
|
||||
$before = (array) $table->first();
|
||||
$migration->up();
|
||||
$migration->up();
|
||||
$this->assertTrue(Schema::connection('business_mirror')->hasIndex('ops_jst_original_bill_info', 'idx_original_bill_channel_time'));
|
||||
$this->assertEquals($before, (array) $table->first());
|
||||
$migration->down();
|
||||
$this->assertFalse(Schema::connection('business_mirror')->hasIndex('ops_jst_original_bill_info', 'idx_original_bill_channel_time'));
|
||||
$this->assertEquals($before, (array) $table->first());
|
||||
}
|
||||
|
||||
public function test_common_filters_use_the_existing_daily_summary_acceleration(): void
|
||||
{
|
||||
$this->markReady();
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user