fix: spool parquet archives before upload

This commit is contained in:
WuYongJie 2026-09-07 12:56:00 +08:00
parent 7c6bf6757d
commit 5e2ccc58f7
2 changed files with 56 additions and 1 deletions

View File

@ -13,6 +13,7 @@ use Modules\EcommerceFinance\Services\BusinessDataPlacement;
use Modules\EcommerceFinance\Services\BusinessMirrorDatasetRegistry;
use Modules\EcommerceFinance\Services\ClickHouseClient;
use Modules\EcommerceFinance\Services\ClickHouseSchemaManager;
use Psr\Http\Message\StreamInterface;
use RuntimeException;
use Throwable;
@ -224,7 +225,7 @@ class BackfillBusinessDataColumnar extends Command
continue;
}
$stream = $columnar->parquetPartition($dataset, $generation, $partition);
$resource = StreamWrapper::getResource($stream);
$resource = $this->seekableResource($stream);
try {
$disk->writeStream($objectKey, $resource);
} finally {
@ -254,4 +255,28 @@ class BackfillBusinessDataColumnar extends Command
$this->info($datasetKey . ':32 个 Parquet 哈希分区已写入对象存储并登记清单;MariaDB 原表仍保留。');
}
/** @return resource */
private function seekableResource(StreamInterface $stream): mixed
{
$source = StreamWrapper::getResource($stream);
$target = tmpfile();
if ($target === false) {
fclose($source);
throw new RuntimeException('business_data_archive_temp_file_unavailable');
}
try {
if (stream_copy_to_stream($source, $target) === false || ! rewind($target)) {
throw new RuntimeException('business_data_archive_spool_failed');
}
} catch (Throwable $exception) {
fclose($target);
throw $exception;
} finally {
fclose($source);
}
return $target;
}
}

View File

@ -0,0 +1,30 @@
<?php
namespace Tests\Unit\Console;
use App\Console\Commands\BackfillBusinessDataColumnar;
use GuzzleHttp\Psr7\PumpStream;
use PHPUnit\Framework\TestCase;
use ReflectionMethod;
class BusinessDataArchiveStreamTest extends TestCase
{
public function test_non_seekable_parquet_stream_is_spooled_to_a_seekable_resource(): void
{
$chunks = ['PAR1-', 'fixture'];
$stream = new PumpStream(static function () use (&$chunks): string|false {
return array_shift($chunks) ?? false;
});
$this->assertFalse($stream->isSeekable());
$method = new ReflectionMethod(BackfillBusinessDataColumnar::class, 'seekableResource');
$resource = $method->invoke(new BackfillBusinessDataColumnar(), $stream);
try {
$this->assertTrue(stream_get_meta_data($resource)['seekable']);
$this->assertSame('PAR1-fixture', stream_get_contents($resource));
} finally {
fclose($resource);
}
}
}