packages/ztd-query-pdo-adapter/tests/Integration/MySql/LoadDataTest.php
1<?php
2
3declare(strict_types=1);
4
5namespace Tests\Integration\MySql;
6
7use Container\MySql80Container;
8use Container\MySql84Container;
9use PDO;
10use PHPUnit\Framework\Attributes\CoversNothing;
11use PHPUnit\Framework\Attributes\Large;
12use PHPUnit\Framework\TestCase;
13use ZtdQuery\Adapter\Pdo\ZtdPdo;
14
15/**
16 * @requires extension pdo_mysql
17 * @group integration
18 * @group mysql
19 */
20#[CoversNothing]
21#[Large]
22final class LoadDataTest extends TestCase
23{
24 public function testLoadDataVariantsMutateOnlyTheShadowTable(): void
25 {
26 $containerInstance = \Testcontainers\Testcontainers::run(getenv('MYSQL_VERSION') === '8.4.7' ? MySql84Container::class : MySql80Container::class);
27 /** @var PDO $pdo */
28 $pdo = new PDO(
29 sprintf('mysql:host=%s;port=%d;dbname=test;charset=utf8mb4', str_replace('localhost', '127.0.0.1', $containerInstance->getHost()), $containerInstance->getMappedPort(3306)),
30 'root',
31 'root',
32 [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION, PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC],
33 );
34
35 $databaseName = 'ztd_' . bin2hex(random_bytes(8));
36 $pdo->exec(sprintf('CREATE DATABASE `%s` CHARACTER SET utf8mb4', $databaseName));
37 $pdo->exec(sprintf('USE `%s`', $databaseName));
38
39 $table = 'load_' . bin2hex(random_bytes(8));
40
41 try {
42 $pdo->exec(sprintf(
43 'CREATE TABLE `%s` (id INT PRIMARY KEY, name VARCHAR(100) DEFAULT \'unknown\')',
44 $table,
45 ));
46 $ztdPdo = ZtdPdo::fromPdo($pdo);
47
48 $initial = tmpfile();
49 self::assertIsResource($initial);
50 $initialData = "id,name\r\n>1,\"Alice, A\"\r\n>2,\\N\r\n";
51 self::assertSame(strlen($initialData), fwrite($initial, $initialData));
52 $initialMetadata = stream_get_meta_data($initial);
53 $initialPath = $initialMetadata['uri'] ?? null;
54 self::assertIsString($initialPath);
55 $initialSql = sprintf(
56 "LOAD DATA LOCAL INFILE '%s' INTO TABLE `%s` "
57 . "FIELDS TERMINATED BY ',' OPTIONALLY ENCLOSED BY '\"' ESCAPED BY '\\\\' "
58 . "LINES STARTING BY '>' TERMINATED BY '\\r\\n' IGNORE 1 LINES (id, @raw) "
59 . "SET name = COALESCE(@raw, 'unknown')",
60 str_replace("'", "''", $initialPath),
61 $table,
62 );
63 self::assertSame(2, $ztdPdo->exec($initialSql));
64
65 $replacement = tmpfile();
66 self::assertIsResource($replacement);
67 self::assertSame(14, fwrite($replacement, "2\tBob\n3\tCarol\n"));
68 $replacementMetadata = stream_get_meta_data($replacement);
69 $replacementPath = $replacementMetadata['uri'] ?? null;
70 self::assertIsString($replacementPath);
71 self::assertSame(2, $ztdPdo->exec(sprintf(
72 "LOAD DATA INFILE '%s' REPLACE INTO TABLE `%s`",
73 str_replace("'", "''", $replacementPath),
74 $table,
75 )));
76
77 $ignored = tmpfile();
78 self::assertIsResource($ignored);
79 self::assertSame(17, fwrite($ignored, "3\tIgnored\n4\tDave\n"));
80 $ignoredMetadata = stream_get_meta_data($ignored);
81 $ignoredPath = $ignoredMetadata['uri'] ?? null;
82 self::assertIsString($ignoredPath);
83 self::assertSame(1, $ztdPdo->exec(sprintf(
84 "LOAD DATA INFILE '%s' IGNORE INTO TABLE `%s`",
85 str_replace("'", "''", $ignoredPath),
86 $table,
87 )));
88
89 $rows = $ztdPdo->query(sprintf('SELECT id, name FROM `%s` ORDER BY id', $table));
90 self::assertNotFalse($rows);
91 self::assertSame([
92 ['id' => 1, 'name' => 'Alice, A'],
93 ['id' => 2, 'name' => 'Bob'],
94 ['id' => 3, 'name' => 'Carol'],
95 ['id' => 4, 'name' => 'Dave'],
96 ], $rows->fetchAll());
97
98 $physical = $pdo->query(sprintf('SELECT COUNT(*) FROM `%s`', $table));
99 self::assertNotFalse($physical);
100 self::assertSame(0, (int) $physical->fetchColumn());
101 } finally {
102 $pdo->exec(sprintf('DROP DATABASE IF EXISTS `%s`', $databaseName));
103 }
104 }
105}
106