diff --git a/app/Console/Commands/BackfillBusinessDataColumnar.php b/app/Console/Commands/BackfillBusinessDataColumnar.php index 29e4a7fc..3b56e80d 100644 --- a/app/Console/Commands/BackfillBusinessDataColumnar.php +++ b/app/Console/Commands/BackfillBusinessDataColumnar.php @@ -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; + } } diff --git a/tests/Unit/Console/BusinessDataArchiveStreamTest.php b/tests/Unit/Console/BusinessDataArchiveStreamTest.php new file mode 100644 index 00000000..0f4945f1 --- /dev/null +++ b/tests/Unit/Console/BusinessDataArchiveStreamTest.php @@ -0,0 +1,30 @@ +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); + } + } +}