dialect = DbSource::dialectOf($db); } public function warnings(): array { return array_values(array_unique($this->warnings)); } private function q(string $n): string { return Convert::quote($this->dialect, $n); } public function begin(bool $drop): void { $this->drop = $drop; $this->existing = array_map('strtolower', DbSource::tableNames($this->db)); if ($this->dialect === 'mysql') $this->db->query('SET FOREIGN_KEY_CHECKS = 0'); } private bool $drop = false; private array $existing = []; public function createTable(array $t): void { if (in_array(strtolower($t['name']), $this->existing, true)) { if (!$this->drop) throw new DbException("テーブル{$t['name']}は既に有ります(消す場合は「drop」)", 0, 'HY000'); $this->db->query('DROP TABLE '.$this->q($t['name']).($this->dialect === 'pgsql' ? ' CASCADE' : '')); } $this->db->query(Convert::createTable($this->dialect, $t, $this->warnings)); } public function insertRows(array $t, array $rows): void { $nc = count($t['columns']); $per = max(1, min(500, intdiv(30000, max(1, $nc)))); $drv = $this->db->driver(); $drv->connect(); if (!$this->inTx) { $drv->beginTransaction(); $this->inTx = true; } foreach (array_chunk($rows, $per) as $chunk) { $n = count($chunk); $key = $t['name']."\0".$n; if (!isset($this->stmts[$key])) { $ph = '('.implode(', ', array_fill(0, $nc, '?')).')'; $sql = 'INSERT INTO '.$this->q($t['name']).' ('.implode(', ', array_map(fn($c) => $this->q($c['name']), $t['columns'])).') VALUES ' .implode(', ', array_fill(0, $n, $ph)); if (count($this->stmts) > 20) { foreach ($this->stmts as $id) $drv->demolish($id); $this->stmts = []; } $this->stmts[$key] = $drv->prepare($sql); } $params = []; foreach ($chunk as $r) { foreach ($t['columns'] as $i => $c) $params[] = $this->param($r[$i], $c, $t); } $drv->execute($this->stmts[$key], $params); $this->pending += $n; if ($this->pending >= 20000) { $drv->commit(); $drv->beginTransaction(); $this->pending = 0; } } } private function param(mixed $v, array $c, array $t): mixed { if ($v === null) return null; if ($v instanceof Blob) return $this->dialect === 'sqlite' ? $v : $v->data; if (is_bool($v)) return $this->dialect === 'pgsql' ? $v : (int)$v; if (is_float($v) && (is_nan($v) || is_infinite($v)) && $this->dialect !== 'pgsql') { if ($this->dialect === 'mysql' || is_nan($v)) { $this->warnings[] = "NaN・無限はNULLにしました:{$t['name']}.{$c['name']}"; return null; } } if (is_string($v) && $this->dialect === 'pgsql' && str_contains($v, "\0")) { $this->warnings[] = "PostgreSQLの文字列に0のバイトは入れられないので、消しました:{$t['name']}.{$c['name']}"; return str_replace("\0", '', $v); } return $v; } public function finishTable(array $t): void { $drv = $this->db->driver(); if ($this->inTx) { $drv->commit(); $this->inTx = false; $this->pending = 0; } foreach ($t['indexes'] as $idx) $this->db->query(Convert::createIndex($this->dialect, $t, $idx)); if ($this->dialect === 'pgsql') { foreach ($t['columns'] as $c) { if (!$c['autoinc'] || count($t['pk']) !== 1 || strcasecmp($t['pk'][0], $c['name']) !== 0) continue; $this->db->query("SELECT setval(pg_get_serial_sequence(?, ?), COALESCE(MAX({$this->q($c['name'])}), 1), MAX({$this->q($c['name'])}) IS NOT NULL) FROM {$this->q($t['name'])}", [$this->q($t['name']), $c['name']]); } } } public function abort(): void { $drv = $this->db->driver(); try { if ($this->inTx && $drv->inTransaction()) $drv->rollback(); } catch (\Throwable $e) { // 接続が切れた等 } $this->inTx = false; $this->stmts = []; } public function end(): void { $drv = $this->db->driver(); foreach ($this->stmts as $id) $drv->demolish($id); $this->stmts = []; if ($this->inTx) { $drv->commit(); $this->inTx = false; } if ($this->dialect === 'mysql') $this->db->query('SET FOREIGN_KEY_CHECKS = 1'); } }