class-mysql-dump-producer.php — Jetpack – WP Security, Backup, Speed, & Growth 16.1-beta.2

jetpack / vendor / wp-php-toolkit / reprint-exporter / src / class-mysql-dump-producer.php

jetpack / vendor / wp-php-toolkit / reprint-exporter / src Last commit date

class-file-tree-producer.php 3 weeks ago class-hmac-client.php 3 weeks ago class-hmac-server.php 3 weeks ago class-http-server.php 3 weeks ago class-mysql-dump-producer.php 3 weeks ago class-pdo-polyfill.php 3 weeks ago class-sqlite-driver-pdo.php 3 weeks ago class-staged-artifacts.php 3 weeks ago class-staged-endpoints.php 3 weeks ago class-staged-push-stream-protocol.php 3 weeks ago class-wpdb-driver-pdo.php 3 weeks ago export.php 3 weeks ago utils.php 3 weeks ago

class-mysql-dump-producer.php

1526 lines

1	<?php
2
3	namespace WordPress\DataLiberation;
4
5	use PDO;
6	use PDOStatement;
7
8	/**
9	* Generates a MySQL dump as a sequence of SQL fragments, one per call to next_sql_fragment().
10	*
11	* This class exists because shared hosting environments kill long-running PHP processes.
12	* A traditional mysqldump would time out on large databases. Instead, this producer
13	* yields one SQL fragment at a time — a CREATE TABLE, a batched INSERT, or an UPDATE —
14	* and exposes a JSON cursor that captures the full internal state. The caller can
15	* serialize that cursor, end the HTTP request, and resume from exactly where it left
16	* off in a subsequent request.
17	*
18	* The producer is a finite state machine that walks through tables sequentially:
19	*
20	* INIT → EMIT_HEADER → NEXT_TABLE → CREATE_TABLE → TABLE_HEADER →
21	* START_INSERT ⇄ EMIT_ROW → (EMIT_OVERSIZED_UPDATE) → … → EMIT_FOOTER → FINISHED
22	*
23	* All values are base64-encoded in the SQL output (via FROM_BASE64('...')). This avoids
24	* charset-related corruption: MySQL interprets string literals according to the
25	* connection charset, but base64 is pure ASCII and the decoded bytes are assigned
26	* directly to the column's declared charset. JSON columns are a special case — MySQL
27	* rejects binary charset input for JSON, so those get an extra CONVERT(... USING utf8mb4).
28	*
29	* Rows that would exceed MySQL's max_allowed_packet are handled by inserting the row
30	* with large columns set to empty strings, then appending the real data via a series
31	* of UPDATE ... SET col = CONCAT(col, chunk) statements.
32	*
33	* Known limitations:
34	*
35	* - Rows too large to be SELECTed. If a row is larger than max_allowed_packet or the
36	* PHP memory_limit, it won't be exported. The underlying assumption is that WordPress
37	* wouldn't be able to use that data anyway. If that turns out to be wrong, and there
38	* are plugins that use huge blobs with byte offset queries, we'll need to add measures
39	* to detect those situations and export that data in chunks.
40	* - Tables without a primary key can't use the oversized row handling as there's no
41	* stable row identifier for the UPDATE ... SET col = CONCAT(col, chunk) WHERE ... query.
42	*/
43	class MySQLDumpProducer
44	{
45	const STATE_INIT = "init";
46	const STATE_EMIT_HEADER = "emit_header";
47	const STATE_NEXT_TABLE = "next_table";
48	const STATE_CREATE_TABLE = "create_table";
49	const STATE_TABLE_HEADER = "table_header";
50	const STATE_START_INSERT = "start_insert";
51	const STATE_EMIT_ROW = "emit_row";
52	const STATE_EMIT_OVERSIZED_UPDATE = "emit_oversized_update";
53	const STATE_EMIT_FOOTER = "emit_footer";
54	const STATE_FINISHED = "finished";
55
56	/** @var PDO */
57	private $db;
58
59	/** @var string\|null */
60	private $current_sql_fragment = null;
61
62	/** @var array\|null */
63	private $current_pk_columns = null;
64
65	/**
66	* Cursor bookmark: the PK values of the last emitted row. The next SELECT
67	* uses a WHERE clause like `(pk1, pk2) > (last1, last2)` to resume without
68	* re-reading earlier rows. Null before the first row of a table.
69	*
70	* @var array\|null
71	*/
72	private $last_pk_values = null;
73
74	/**
75	* Fallback cursor for tables without a primary key. Unlike PK-based cursors,
76	* OFFSET pagination re-scans earlier rows on every query, so it's slower
77	* and vulnerable to drift if rows are inserted or deleted mid-export.
78	*
79	* @var int
80	*/
81	private $current_offset = 0;
82
83	/** @var string\|null */
84	private $current_table = null;
85
86	/** @var PDOStatement\|null */
87	private $current_result_set = null;
88
89	/**
90	* Distinguishes "query returned zero rows because the table is exhausted"
91	* from "query returned zero rows because we just opened a fresh cursor."
92	* Without this, the producer would stop after every batch_size rows.
93	*
94	* @var int
95	*/
96	private $rows_fetched_from_current_query = 0;
97
98	/** @var array */
99	private $tables_to_process;
100
101	/** @var string */
102	private $state = self::STATE_INIT;
103
104	/**
105	* INFORMATION_SCHEMA column metadata, cached per table to avoid repeated
106	* queries. Keyed by table name, then column name. Each entry contains
107	* 'data_type' (e.g. 'varchar') and 'column_type' (e.g. 'varchar(255)').
108	*
109	* @var array
110	*/
111	private $column_type_cache = [];
112
113	/** @var array\|null */
114	private $current_row = null;
115
116	/** @var int */
117	private $rows_in_batch = 0;
118
119	/** @var array\|null */
120	private $current_column_types = null;
121
122	/** @var array\|null */
123	private $current_column_names = null;
124
125	/** @var int */
126	private $batch_size;
127
128	/** @var bool */
129	private $emit_create_table;
130
131	/**
132	* Derived from MySQL's max_allowed_packet (at 80% to leave headroom for
133	* protocol framing). Rows whose formatted SQL exceeds this limit are split
134	* into an INSERT with empty placeholders followed by UPDATE ... CONCAT()
135	* statements that append the real data in chunks.
136	*
137	* @var int
138	*/
139	private $max_statement_size;
140
141	/** @var int\|null */
142	private $query_time_limit_ms = null;
143
144	/**
145	* Row exclusion rules keyed by table name. Each rule is an array with
146	* 'column' and 'value' keys.
147	*
148	* @var array<string, list<array{column: string, value: string}>>
149	*/
150	private $exclude_rows_by_table = [];
151
152	/**
153	* When a row is too large for a single INSERT, its big columns are split
154	* into chunks and queued here. Each entry tracks the column name, its
155	* data type, the current byte offset into the value, and the total value
156	* length. The actual data is re-fetched from the database on demand via
157	* SUBSTRING queries, keeping cursors small (a few hundred bytes rather
158	* than megabytes of raw data).
159	*
160	* @var array Array of {column: string, data_type: string, byte_offset: int, total_length: int}
161	*/
162	private $oversized_queue = [];
163
164	/** @var array\|null */
165	private $oversized_pk_values = null;
166
167	/** @var string\|null */
168	private $state_after_oversized = null;
169
170	/** @var int */
171	private $current_statement_size = 0;
172
173	/**
174	* @param PDO $db Database connection — either a real PDO (MySQL) or a
175	* PDO-compatible adapter (SQLite sites). No type hint because the
176	* adapter isn't a PDO subclass and PHP 7.4 lacks union types.
177	*/
178	public function __construct($db, $options = [])
179	{
180	$this->db = $db;
181	$this->tables_to_process = $options["tables_to_process"] ?? null;
182	$this->batch_size = max(1, (int)($options["batch_size"] ?? 250));
183	$this->emit_create_table = (bool)($options["create_table_query"] ?? true);
184
185	if (isset($options["max_statement_size"])) {
186	$this->max_statement_size = (int)$options["max_statement_size"];
187	} else {
188	$this->max_statement_size = $this->detect_max_statement_size();
189	}
190
191	if (isset($options["query_time_limit_ms"])) {
192	$limit = (int) $options["query_time_limit_ms"];
193	$this->query_time_limit_ms = $limit > 0 ? $limit : null;
194	}
195
196	if (isset($options["exclude_rows"]) && is_array($options["exclude_rows"])) {
197	foreach ($options["exclude_rows"] as $rule) {
198	if (
199	!is_array($rule) \|\|
200	!isset($rule["table"], $rule["column"], $rule["value"]) \|\|
201	!is_string($rule["table"]) \|\|
202	!is_string($rule["column"]) \|\|
203	!is_string($rule["value"])
204	) {
205	continue;
206	}
207	$this->exclude_rows_by_table[$rule["table"]][] = [
208	"column" => $rule["column"],
209	"value" => $rule["value"],
210	];
211	}
212	}
213
214	if (isset($options["cursor"])) {
215	$this->initialize_from_cursor($options["cursor"]);
216	}
217	}
218
219	public function get_sql_fragment(): ?string
220	{
221	return $this->current_sql_fragment;
222	}
223
224	public function is_finished(): bool
225	{
226	return self::STATE_FINISHED === $this->state;
227	}
228
229	/**
230	* Advances the state machine and populates the next SQL fragment.
231	*
232	* Call get_sql_fragment() after this returns true to retrieve the SQL.
233	* Returns false only when the dump is complete (state = FINISHED).
234	*/
235	public function next_sql_fragment()
236	{
237	if ($this->is_finished()) {
238	return false;
239	}
240
241	if (self::STATE_INIT === $this->state) {
242	if (null === $this->tables_to_process) {
243	$this->initialize_tables_to_process();
244	}
245	$this->state = self::STATE_EMIT_HEADER;
246	}
247
248	while (true) {
249	switch ($this->state) {
250	case self::STATE_EMIT_HEADER:
251	$this->emit_sql_header();
252	$this->state = self::STATE_NEXT_TABLE;
253	return true;
254
255	case self::STATE_NEXT_TABLE:
256	if ($this->move_to_next_table()) {
257	$this->state = $this->emit_create_table
258	? self::STATE_CREATE_TABLE
259	: self::STATE_TABLE_HEADER;
260	} else {
261	$this->state = self::STATE_EMIT_FOOTER;
262	}
263	break;
264
265	case self::STATE_EMIT_FOOTER:
266	$this->emit_sql_footer();
267	$this->state = self::STATE_FINISHED;
268	return true;
269
270	case self::STATE_CREATE_TABLE:
271	$this->emit_create_table_statement();
272	$this->state = self::STATE_TABLE_HEADER;
273	return true;
274
275	case self::STATE_TABLE_HEADER:
276	$this->emit_table_header_comment();
277	$this->state = self::STATE_START_INSERT;
278	return true;
279
280	case self::STATE_START_INSERT:
281	if ($this->emit_insert_header()) {
282	return true;
283	}
284	// Empty table — emit_insert_header set state to NEXT_TABLE
285	break;
286
287	case self::STATE_EMIT_ROW:
288	return $this->emit_row();
289
290	case self::STATE_EMIT_OVERSIZED_UPDATE:
291	if ($this->emit_oversized_update()) {
292	return true;
293	}
294	break;
295
296	case self::STATE_FINISHED:
297	return false;
298	}
299	}
300
301	return false;
302	}
303
304	/**
305	* Fetches the next row from the current result set into $this->current_row.
306	*
307	* When the result set is exhausted, checks whether the table has more rows
308	* by opening a new query from the current cursor position. Returns false
309	* only when a fresh query comes back empty, meaning the table is done.
310	*/
311	private function fetch_and_store_row()
312	{
313	if (!$this->current_result_set) {
314	$query = $this->build_select_query();
315	try {
316	$this->current_result_set = $this->db->query($query);
317	} catch (\PDOException $e) {
318	throw new \RuntimeException(
319	"Database query `{$query}` failed for table " . $this->quote_identifier($this->current_table) . ": " . $e->getMessage()
320	);
321	}
322	$this->rows_fetched_from_current_query = 0;
323	}
324
325	$record = $this->current_result_set->fetch(PDO::FETCH_ASSOC);
326	if (!$record) {
327	$this->current_result_set = null;
328
329	if ($this->rows_fetched_from_current_query === 0) {
330	return false;
331	}
332
333	// This batch is exhausted but returned rows earlier, so the table
334	// may have more. Open a new query starting after the last PK.
335	if ($this->last_pk_values !== null \|\| $this->current_offset > 0) {
336	return $this->fetch_and_store_row();
337	}
338
339	return false;
340	}
341
342	$this->rows_fetched_from_current_query++;
343
344	if ($this->current_column_names === null) {
345	$this->current_column_names = array_keys($record);
346	}
347
348	if ($this->current_pk_columns && count($this->current_pk_columns) > 0) {
349	$this->last_pk_values = [];
350	foreach ($this->current_pk_columns as $col) {
351	if (!array_key_exists($col, $record)) {
352	throw new \RuntimeException(
353	"Primary key column '{$col}' missing from SELECT result for table " .
354	$this->quote_identifier($this->current_table)
355	);
356	}
357	$this->last_pk_values[$col] = $record[$col];
358	}
359	} else {
360	$this->current_offset++;
361	}
362
363	$this->current_row = $record;
364	return true;
365	}
366	/**
367	* Emits "INSERT INTO ... VALUES (first_row)" as a single fragment.
368	*
369	* The first row is always bundled with the INSERT header to prevent
370	* emitting a dangling "INSERT INTO ... VALUES" with no rows — which
371	* would happen if the caller saves the cursor right after the header
372	* and the data changes before the next request.
373	*/
374	private function emit_insert_header()
375	{
376	if ($this->current_row === null) {
377	if (!$this->fetch_and_store_row()) {
378	$this->state = self::STATE_NEXT_TABLE;
379	return false;
380	}
381	}
382
383	$column_list = implode(
384	",",
385	array_map(function ($col) {
386	return $this->quote_identifier($col);
387	}, $this->current_column_names)
388	);
389
390	$header = "INSERT INTO " . $this->quote_identifier($this->current_table) . " ({$column_list}) VALUES\n";
391	$this->current_statement_size = strlen($header);
392
393	$first_row_sql = $this->format_row_for_insert($this->current_row);
394	$this->current_statement_size += strlen($first_row_sql) + 1;
395
396	$this->current_row = null;
397	$this->rows_in_batch = 1;
398
399	$has_next_row = $this->fetch_and_store_row();
400
401	// Oversized updates require closing this INSERT with a semicolon so the
402	// subsequent UPDATE statements are syntactically separate.
403	$has_oversized = $this->has_pending_oversized_updates();
404
405	if (!$has_next_row) {
406	$sql = $header . $first_row_sql . ";";
407	$this->current_sql_fragment = $sql;
408	$this->current_statement_size = 0;
409	if ($has_oversized) {
410	$this->state_after_oversized = self::STATE_NEXT_TABLE;
411	$this->state = self::STATE_EMIT_OVERSIZED_UPDATE;
412	} else {
413	$this->state = self::STATE_NEXT_TABLE;
414	}
415	} elseif ($this->rows_in_batch >= $this->batch_size \|\| $has_oversized) {
416	$sql = $header . $first_row_sql . ";";
417	$this->current_sql_fragment = $sql;
418	$this->current_statement_size = 0;
419	if ($has_oversized) {
420	$this->state_after_oversized = self::STATE_START_INSERT;
421	$this->state = self::STATE_EMIT_OVERSIZED_UPDATE;
422	} else {
423	$this->state = self::STATE_START_INSERT;
424	}
425	} else {
426	$sql = $header . $first_row_sql . ",";
427	$this->current_sql_fragment = $sql;
428	$this->state = self::STATE_EMIT_ROW;
429	}
430
431	return true;
432	}
433
434	/**
435	* Emits one row as a SQL fragment, terminated with "," (more rows follow)
436	* or ";" (INSERT statement complete).
437	*/
438	private function emit_row()
439	{
440	if ($this->current_row === null) {
441	$this->state = self::STATE_NEXT_TABLE;
442	return false;
443	}
444
445	$row_sql = $this->format_row_for_insert($this->current_row);
446	$this->current_statement_size += strlen($row_sql) + 2;
447	$this->current_row = null;
448	$this->rows_in_batch++;
449
450	$has_next_row = $this->fetch_and_store_row();
451	$has_oversized = $this->has_pending_oversized_updates();
452
453	if (!$has_next_row) {
454	$this->current_sql_fragment = $row_sql . ";";
455	$this->current_statement_size = 0;
456	if ($has_oversized) {
457	$this->state_after_oversized = self::STATE_NEXT_TABLE;
458	$this->state = self::STATE_EMIT_OVERSIZED_UPDATE;
459	} else {
460	$this->state = self::STATE_NEXT_TABLE;
461	}
462	} elseif ($this->rows_in_batch >= $this->batch_size \|\| $has_oversized) {
463	$this->current_sql_fragment = $row_sql . ";";
464	$this->current_statement_size = 0;
465	if ($has_oversized) {
466	$this->state_after_oversized = self::STATE_START_INSERT;
467	$this->state = self::STATE_EMIT_OVERSIZED_UPDATE;
468	} else {
469	$this->state = self::STATE_START_INSERT;
470	}
471	} else {
472	$this->current_sql_fragment = $row_sql . ",";
473	}
474
475	return true;
476	}
477
478	/**
479	* Emits DROP TABLE IF EXISTS followed by the CREATE TABLE from SHOW CREATE TABLE.
480	* Also handles views (SHOW CREATE TABLE returns 'Create View' for those).
481	*/
482	private function emit_create_table_statement()
483	{
484	$quoted_table = $this->quote_identifier($this->current_table);
485	try {
486	$query = "SHOW CREATE TABLE {$quoted_table}";
487	$result = $this->db->query($query);
488	$row = $result->fetch(PDO::FETCH_ASSOC);
489	} catch (\PDOException $e) {
490	throw new \RuntimeException(
491	"Failed to get CREATE TABLE for {$quoted_table}: " . $e->getMessage() . " Query: {$query}"
492	);
493	}
494
495	$sql = null;
496	if ($row) {
497	if (isset($row["Create Table"])) {
498	$sql = $row["Create Table"];
499	} elseif (isset($row["Create View"])) {
500	$sql = $row["Create View"];
501	}
502	}
503
504	if ($sql) {
505	// Prevent breaking the line by identifiers with a newline byte in them.
506	$header = "--\n-- Table structure for table ".str_replace("\n",'\n',$quoted_table)."\n--\n\n";
507	$drop = "DROP TABLE IF EXISTS {$quoted_table};\n";
508	$this->current_sql_fragment = $header . $drop . $sql . ";";
509	} else {
510	$keys = $row ? implode(", ", array_keys($row)) : "(no row returned)";
511	throw new \RuntimeException(
512	"SHOW CREATE TABLE {$quoted_table} returned no usable SQL. " .
513	"Available keys: {$keys}"
514	);
515	}
516	}
517
518	/**
519	* Emits SET statements that disable constraint checks and set a strict SQL mode.
520	* These are restored in emit_sql_footer(). Without disabling FK checks, tables
521	* that reference each other would need to be imported in dependency order.
522	*
523	* The SQL_MODE explicitly omits NO_ZERO_DATE and NO_ZERO_IN_DATE. This is
524	* intentional: many WordPress databases contain zero dates like '0000-00-00'
525	* or '0000-00-00 00:00:00' (e.g. in wp_posts.post_date for drafts). The
526	* source server may have been running without those restrictions, and the
527	* dump must be importable regardless of the target server's default sql_mode.
528	*
529	* From the MySQL 8.0 Reference Manual (§5.1.11 "Server SQL Modes"):
530	*
531	* NO_ZERO_DATE — [...] The server requires dates to have nonzero month
532	* and day values. If NO_ZERO_DATE is enabled and strict mode is enabled,
533	* '0000-00-00' is not permitted and inserts produce an error. [...]
534	* If NO_ZERO_DATE is disabled, '0000-00-00' is permitted and inserts
535	* produce no warning.
536	*
537	* NO_ZERO_IN_DATE — [...] Affects whether the server permits dates in
538	* which the year part is nonzero but the month or day part is 0.
539	* [...] If this mode is disabled, dates with zero parts are permitted
540	* and inserts produce no warning.
541	*
542	* By omitting both flags while keeping STRICT_TRANS_TABLES, the dump
543	* preserves MySQL's permissive behavior toward zero dates during import.
544	*
545	* @see https://dev.mysql.com/doc/refman/8.0/en/sql-mode.html#sqlmode_no_zero_date
546	* @see https://dev.mysql.com/doc/refman/8.0/en/sql-mode.html#sqlmode_no_zero_in_date
547	*/
548	private function emit_sql_header()
549	{
550	$header =
551	"SET @OLD_UNIQUE_CHECKS=@@UNIQUE_CHECKS, UNIQUE_CHECKS=0;\n" .
552	"SET @OLD_FOREIGN_KEY_CHECKS=@@FOREIGN_KEY_CHECKS, FOREIGN_KEY_CHECKS=0;\n" .
553	// @TODO: Restore STRICT_TRANS_TABLES
554	"SET @OLD_SQL_MODE=@@SQL_MODE, SQL_MODE='ONLY_FULL_GROUP_BY,ERROR_FOR_DIVISION_BY_ZERO,NO_ENGINE_SUBSTITUTION';\n" .
555	"SET AUTOCOMMIT=0;\n";
556	$this->current_sql_fragment = $header;
557	}
558
559	/** Emits COMMIT and restores the session variables saved in the header. */
560	private function emit_sql_footer()
561	{
562	$footer =
563	"\nCOMMIT;\n" .
564	"SET SQL_MODE=@OLD_SQL_MODE;\n" .
565	"SET FOREIGN_KEY_CHECKS=@OLD_FOREIGN_KEY_CHECKS;\n" .
566	"SET UNIQUE_CHECKS=@OLD_UNIQUE_CHECKS;\n";
567	$this->current_sql_fragment = $footer;
568	}
569
570	/** Emits a SQL comment marking the start of data for the current table. */
571	private function emit_table_header_comment()
572	{
573	$comment = "\n--\n-- Dumping data for table " . str_replace("\n",'\n',$this->quote_identifier($this->current_table)) . "\n--\n";
574	$this->current_sql_fragment = $comment;
575	}
576
577	/**
578	* Builds a SELECT query for the current table's next batch of rows.
579	*
580	* Non-numeric, non-binary columns are wrapped in CAST(... AS BINARY) so
581	* MySQL returns raw bytes rather than re-encoding through the connection
582	* charset. This is critical: without it, a latin1 column read over a utf8mb4
583	* connection would silently transcode the bytes, and our base64 encoding
584	* would capture the transcoded version instead of the original.
585	*/
586	private function build_select_query()
587	{
588	$table = $this->current_table;
589	$select = "SELECT";
590
591	if ($this->query_time_limit_ms !== null) {
592	// MySQL optimizer hint — caps this query's wall-clock time so a
593	// single slow table can't consume the entire PHP execution budget.
594	$select .= " /*+ MAX_EXECUTION_TIME(" .
595	$this->query_time_limit_ms .
596	") */";
597	}
598
599	if ($this->current_column_types) {
600	$select_parts = [];
601	foreach ($this->current_column_types as $col_name => $col_info) {
602	$quoted = $this->quote_identifier($col_name);
603	// Don't cast numeric or already-binary types
604	if (
605	$this->is_numeric_type($col_info["data_type"]) \|\|
606	$this->is_binary_type($col_info["data_type"])
607	) {
608	$select_parts[] = $quoted;
609	} else {
610	$select_parts[] = "CAST({$quoted} AS BINARY) AS {$quoted}";
611	}
612	}
613	$query =
614	$select .
615	" " .
616	implode(", ", $select_parts) .
617	" FROM " . $this->quote_identifier($table);
618	} else {
619	$query = $select . " * FROM " . $this->quote_identifier($table);
620	}
621
622	$where_conditions = $this->build_row_exclusion_where_conditions();
623	if ($this->current_pk_columns && count($this->current_pk_columns) > 0) {
624	if ($this->last_pk_values) {
625	$where_conditions[] = $this->build_pk_where_clause();
626	}
627
628	if ($where_conditions) {
629	$query .= " WHERE " . implode(" AND ", array_map(function ($condition) {
630	return "({$condition})";
631	}, $where_conditions));
632	}
633
634	$order_cols = array_map(function ($col) {
635	return $this->quote_identifier($col) . " ASC";
636	}, $this->current_pk_columns);
637	$query .= " ORDER BY " . implode(", ", $order_cols);
638	$query .= " LIMIT {$this->batch_size}";
639	} else {
640	if ($where_conditions) {
641	$query .= " WHERE " . implode(" AND ", array_map(function ($condition) {
642	return "({$condition})";
643	}, $where_conditions));
644	}
645
646	// Best effort pagination for tables without a primary key.
647	if ($this->current_offset > 0) {
648	$query .= " LIMIT {$this->batch_size} OFFSET {$this->current_offset}";
649	} else {
650	$query .= " LIMIT {$this->batch_size}";
651	}
652	}
653
654	return $query;
655	}
656
657	/** Builds WHERE fragments for configured row exclusions on the current table. */
658	private function build_row_exclusion_where_conditions(): array
659	{
660	if (!$this->current_table \|\| empty($this->exclude_rows_by_table[$this->current_table])) {
661	return [];
662	}
663
664	$conditions = [];
665	foreach ($this->exclude_rows_by_table[$this->current_table] as $rule) {
666	$column = $rule["column"];
667	if (!isset($this->current_column_types[$column])) {
668	continue;
669	}
670	$quoted_col = $this->quote_identifier($column);
671	$encoded_value = base64_encode($rule["value"]);
672	// Preserve rows with NULL in the filtered column. In SQL, NULL <> value
673	// evaluates to UNKNOWN, so without the explicit IS NULL branch those rows
674	// would be filtered out even though they do not match the excluded value.
675	$conditions[] = "{$quoted_col} IS NULL OR {$quoted_col} <> FROM_BASE64('{$encoded_value}')";
676	}
677	return $conditions;
678	}
679
680	/**
681	* Builds a WHERE clause that selects rows strictly after the last emitted PK.
682	*
683	* For composite primary keys (a, b, c), this produces the lexicographic
684	* "greater than" condition:
685	*
686	* (a > last_a) OR (a = last_a AND b > last_b) OR (a = last_a AND b = last_b AND c > last_c)
687	*
688	* This is equivalent to `(a, b, c) > (last_a, last_b, last_c)` but written
689	* in expanded form for compatibility with MySQL versions that don't optimize
690	* row-value comparisons well.
691	*/
692	private function build_pk_where_clause()
693	{
694	if (!$this->last_pk_values \|\| count($this->current_pk_columns) === 0) {
695	/**
696	* When we haven't seen any PK values yet, or when the table doesn't have a primary key,
697	* we return a dummy condition that will always be true.
698	*/
699	return "1=1";
700	}
701
702	$pk_cols = $this->current_pk_columns;
703
704	if (count($pk_cols) === 1) {
705	$col = $pk_cols[0];
706	$value = $this->last_pk_values[$col];
707	return $this->build_comparison($col, $value, ">");
708	}
709	$conditions = [];
710	$prefix_conditions = [];
711
712	foreach ($pk_cols as $col) {
713	$value = $this->last_pk_values[$col];
714
715	$current_condition_parts = $prefix_conditions;
716	$current_condition_parts[] = $this->build_comparison(
717	$col,
718	$value,
719	">"
720	);
721	$conditions[] =
722	"(" . implode(" AND ", $current_condition_parts) . ")";
723
724	$prefix_conditions[] = $this->build_comparison($col, $value, "=");
725	}
726
727	return "(" . implode(" OR ", $conditions) . ")";
728	}
729
730	/** Builds a single "column op value" SQL expression, handling NULL and quoting. */
731	private function build_comparison($column, $value, $operator)
732	{
733	$quoted_col = $this->quote_identifier($column);
734	if ($value === null) {
735	return $operator === "="
736	? "{$quoted_col} IS NULL"
737	: "{$quoted_col} IS NOT NULL";
738	}
739
740	if (is_numeric($value)) {
741	return "{$quoted_col} {$operator} {$value}";
742	} else {
743	$quoted = $this->db->quote($value);
744	return "{$quoted_col} {$operator} {$quoted}";
745	}
746	}
747
748	/** Returns primary key column names in ordinal order, or empty array if none. */
749	private function get_primary_key_columns($table)
750	{
751	$pk_columns = [];
752
753	$query = "SELECT COLUMN_NAME
754	FROM information_schema.KEY_COLUMN_USAGE
755	WHERE TABLE_SCHEMA = ?
756	AND TABLE_NAME = ?
757	AND CONSTRAINT_NAME = 'PRIMARY'
758	ORDER BY ORDINAL_POSITION";
759	try {
760	$db_name = $this->db->query("SELECT DATABASE()")->fetchColumn();
761	$stmt = $this->db->prepare($query);
762	$stmt->execute([$db_name, $table]);
763	} catch (\PDOException $e) {
764	throw new \RuntimeException(
765	"Failed to get primary key columns for " . $this->quote_identifier($table) . ": " . $e->getMessage() . " Query: {$query}"
766	);
767	}
768
769	while ($row = $stmt->fetch(PDO::FETCH_ASSOC)) {
770	$pk_columns[] = $row["COLUMN_NAME"];
771	}
772
773	return $pk_columns;
774	}
775
776	/** Advances to the next table and resets all per-table state. */
777	private function move_to_next_table()
778	{
779	if ($this->tables_to_process === null) {
780	return false;
781	}
782
783	if (!$this->current_table) {
784	$this->current_table = reset($this->tables_to_process) ?: null;
785	} else {
786	$this->current_table = next($this->tables_to_process) ?: null;
787	}
788
789	if ($this->current_table) {
790	$this->current_pk_columns = $this->get_primary_key_columns(
791	$this->current_table
792	);
793	$this->last_pk_values = null;
794	$this->current_offset = 0;
795	$this->current_column_types = $this->get_column_types(
796	$this->current_table
797	);
798	$this->current_column_names = null;
799	$this->current_row = null;
800	$this->rows_in_batch = 0;
801
802	$this->oversized_queue = [];
803	$this->oversized_pk_values = null;
804	$this->state_after_oversized = null;
805	$this->current_statement_size = 0;
806	}
807
808	return (bool) $this->current_table;
809	}
810
811	/**
812	* Discovers all BASE TABLEs in the current database (excludes views).
813	*
814	* @TODO: Use pagination or approach to support large databases with millions of tables.
815	*/
816	private function initialize_tables_to_process()
817	{
818	$this->tables_to_process = [];
819
820	$db_name = $this->db->query("SELECT DATABASE()")->fetchColumn();
821
822	$stmt = $this->db->prepare(
823	"SELECT TABLE_NAME
824	FROM INFORMATION_SCHEMA.TABLES
825	WHERE TABLE_SCHEMA = ?
826	AND TABLE_TYPE = 'BASE TABLE'
827	ORDER BY TABLE_NAME"
828	);
829	$stmt->execute([$db_name]);
830
831	while ($row = $stmt->fetch(PDO::FETCH_ASSOC)) {
832	$this->tables_to_process[] = $row["TABLE_NAME"];
833	}
834	}
835
836	/**
837	* Returns a JSON string that captures the producer's complete internal state.
838	*
839	* The caller can pass this string back as the "cursor" option to a new
840	* MySQLDumpProducer to resume exactly where this one left off. The JSON is
841	* NOT base64-encoded — that's the HTTP layer's concern (export.php).
842	*
843	* String and binary values in the in-flight row and oversized chunk queue
844	* are wrapped in {"__binary__": "<base64>"} markers because raw binary
845	* bytes can't survive JSON encoding.
846	*/
847	public function get_reentrancy_cursor()
848	{
849	$encoded_current_row = $this->encode_row_for_cursor($this->current_row);
850	$encoded_oversized_queue = $this->encode_oversized_queue_for_cursor($this->oversized_queue);
851
852	$json = json_encode([
853	"current_table" => $this->current_table,
854	"current_pk_columns" => $this->current_pk_columns,
855	"last_pk_values" => $this->last_pk_values,
856	"current_offset" => $this->current_offset,
857	"state" => $this->state,
858	"current_row" => $encoded_current_row,
859	"rows_in_batch" => $this->rows_in_batch,
860	"current_column_names" => $this->current_column_names,
861	/**
862	* Tracking for rows that are larger than max_allowed_packet or
863	* max_statement_size.
864	*/
865	"oversized_queue" => $encoded_oversized_queue,
866	"oversized_pk_values" => $this->oversized_pk_values,
867	"state_after_oversized" => $this->state_after_oversized,
868	"current_statement_size" => $this->current_statement_size,
869	]);
870	if ($json === false) {
871	throw new \RuntimeException(
872	"Failed to encode reentrancy cursor: " . json_last_error_msg()
873	);
874	}
875	return $json;
876	}
877
878	/**
879	* Wraps all string values in {"__binary__": base64} for JSON safety.
880	* JSON is UTF-8-encoded and cannot express arbitrary binary data. The
881	* "__binary__" "type brand" makes it easy to detect and decode binary data
882	* when restoring the cursor.
883	*/
884	private function encode_row_for_cursor($row)
885	{
886	if ($row === null) {
887	return null;
888	}
889
890	$encoded = [];
891	foreach ($row as $col => $value) {
892	if ($value !== null && is_string($value)) {
893	$encoded[$col] = ['__binary__' => base64_encode($value)];
894	} else {
895	$encoded[$col] = $value;
896	}
897	}
898	return $encoded;
899	}
900
901	/** Reverses encode_row_for_cursor(). */
902	private function decode_row_from_cursor($row)
903	{
904	if ($row === null) {
905	return null;
906	}
907
908	$decoded = [];
909	foreach ($row as $col => $value) {
910	if (is_array($value) && isset($value['__binary__'])) {
911	$decoded[$col] = base64_decode($value['__binary__']);
912	} else {
913	$decoded[$col] = $value;
914	}
915	}
916	return $decoded;
917	}
918
919	/** Base64-encodes all chunk payloads in the oversized queue for JSON safety. */
920	/**
921	* The oversized queue entries are already cursor-safe (just column names,
922	* data types, and integer offsets), so encoding is a no-op.
923	*/
924	private function encode_oversized_queue_for_cursor($queue)
925	{
926	return $queue;
927	}
928
929	/** Reverses encode_oversized_queue_for_cursor(). */
930	private function decode_oversized_queue_from_cursor($queue)
931	{
932	if (!is_array($queue)) {
933	return [];
934	}
935	$decoded = [];
936	foreach ($queue as $item) {
937	if (
938	!is_array($item) \|\|
939	!isset($item['column'], $item['data_type'], $item['byte_offset'], $item['total_length'])
940	) {
941	throw new \InvalidArgumentException(
942	"Invalid cursor: oversized_queue item must contain " .
943	"'column', 'data_type', 'byte_offset', and 'total_length' keys"
944	);
945	}
946	$decoded[] = [
947	'column' => $item['column'],
948	'data_type' => $item['data_type'],
949	'byte_offset' => (int) $item['byte_offset'],
950	'total_length' => (int) $item['total_length'],
951	];
952	}
953	return $decoded;
954	}
955
956	/**
957	* Restores internal state from a previously-serialized cursor.
958	*
959	* Re-queries INFORMATION_SCHEMA for column types (the cursor doesn't store
960	* them because schema can change between requests). If the current table
961	* was dropped between requests, resets to STATE_INIT so the producer
962	* gracefully skips forward rather than crashing.
963	*/
964	private function initialize_from_cursor($cursor)
965	{
966	$cursor_data = json_decode($cursor, true);
967	if ($cursor_data === null && json_last_error() !== JSON_ERROR_NONE) {
968	throw new \InvalidArgumentException(
969	'Invalid cursor format: cursor must be valid JSON. ' .
970	'JSON error: ' . json_last_error_msg() . '. ' .
971	'Received: ' . substr($cursor, 0, 100)
972	);
973	}
974	if (is_array($cursor_data)) {
975	$this->current_table = $cursor_data["current_table"] ?? null;
976	if ($this->current_table !== null && !is_string($this->current_table)) {
977	throw new \InvalidArgumentException(
978	"Invalid cursor: current_table must be string or null, got " . gettype($this->current_table)
979	);
980	}
981	$this->current_pk_columns =
982	$cursor_data["current_pk_columns"] ?? null;
983	$this->last_pk_values = $cursor_data["last_pk_values"] ?? null;
984	$this->current_offset = $cursor_data["current_offset"] ?? 0;
985	if (!is_int($this->current_offset) && !is_float($this->current_offset)) {
986	throw new \InvalidArgumentException(
987	"Invalid cursor: current_offset must be numeric, got " . gettype($this->current_offset)
988	);
989	}
990	$this->current_offset = (int) $this->current_offset;
991	$this->state = $cursor_data["state"] ?? self::STATE_INIT;
992	$encoded_row = $cursor_data["current_row"] ?? null;
993	$this->current_row = $this->decode_row_from_cursor($encoded_row);
994	$this->rows_in_batch = $cursor_data["rows_in_batch"] ?? 0;
995	if (!is_int($this->rows_in_batch) && !is_float($this->rows_in_batch)) {
996	throw new \InvalidArgumentException(
997	"Invalid cursor: rows_in_batch must be numeric, got " . gettype($this->rows_in_batch)
998	);
999	}
1000	$this->rows_in_batch = (int) $this->rows_in_batch;
1001	$this->current_column_names =
1002	$cursor_data["current_column_names"] ?? null;
1003
1004	$encoded_queue = $cursor_data["oversized_queue"] ?? [];
1005	$this->oversized_queue = $this->decode_oversized_queue_from_cursor($encoded_queue);
1006	$this->oversized_pk_values = $cursor_data["oversized_pk_values"] ?? null;
1007	$this->state_after_oversized = $cursor_data["state_after_oversized"] ?? null;
1008
1009	$this->current_statement_size = $cursor_data["current_statement_size"] ?? 0;
1010
1011	if ($this->tables_to_process === null) {
1012	$this->initialize_tables_to_process();
1013
1014	if ($this->current_table) {
1015	$found = false;
1016	reset($this->tables_to_process);
1017	while (
1018	($table = current($this->tables_to_process)) !== false
1019	) {
1020	if ($table === $this->current_table) {
1021	$found = true;
1022	break;
1023	}
1024	next($this->tables_to_process);
1025	}
1026	// Table was dropped between requests — advance to next
1027	if (!$found) {
1028	$this->current_table = null;
1029	$this->state = self::STATE_INIT;
1030	}
1031	}
1032	}
1033
1034	if ($this->current_table) {
1035	$this->current_column_types = $this->get_column_types(
1036	$this->current_table
1037	);
1038	if (empty($this->current_column_types)) {
1039	throw new \RuntimeException(
1040	"Table " . $this->quote_identifier($this->current_table) . " was dropped between export requests " .
1041	"(no columns found in INFORMATION_SCHEMA)"
1042	);
1043	}
1044	}
1045	}
1046	}
1047
1048	/** Returns cached INFORMATION_SCHEMA column metadata for a table. */
1049	private function get_column_types($table_name)
1050	{
1051	if (isset($this->column_type_cache[$table_name])) {
1052	return $this->column_type_cache[$table_name];
1053	}
1054
1055	try {
1056	$database_name = $this->db->query("SELECT DATABASE()")->fetchColumn();
1057
1058	$stmt = $this->db->prepare(
1059	'SELECT COLUMN_NAME, DATA_TYPE, COLUMN_TYPE
1060	FROM INFORMATION_SCHEMA.COLUMNS
1061	WHERE TABLE_SCHEMA = ?
1062	AND TABLE_NAME = ?
1063	ORDER BY ORDINAL_POSITION'
1064	);
1065	$stmt->execute([$database_name, $table_name]);
1066	} catch (\PDOException $e) {
1067	throw new \RuntimeException(
1068	"Failed to get column types for " . $this->quote_identifier($table_name) . ": " . $e->getMessage()
1069	);
1070	}
1071
1072	$columns = [];
1073	while ($row = $stmt->fetch(PDO::FETCH_ASSOC)) {
1074	$columns[$row["COLUMN_NAME"]] = [
1075	"data_type" => $row["DATA_TYPE"],
1076	"column_type" => $row["COLUMN_TYPE"],
1077	];
1078	}
1079
1080	$this->column_type_cache[$table_name] = $columns;
1081
1082	return $columns;
1083	}
1084
1085	/**
1086	* Formats a single column value as a SQL literal.
1087	*
1088	* Numeric types are emitted as bare literals. Everything else — strings,
1089	* binary, dates, enums — goes through FROM_BASE64(). JSON is special:
1090	* MySQL rejects binary-charset input for JSON columns, so we wrap with
1091	* CONVERT(... USING utf8mb4) to decode the base64 into a utf8mb4 string.
1092	* JSON can only be encoded as UTF-8 or UTF-16, and it's typically UTF-8.
1093	* As of this version, we do not support UTF-16-encoded JSON data strings.
1094	*
1095	* @TODO: Support UTF-16-encoded JSON data strings.
1096	*/
1097	private function format_value($value, $data_type)
1098	{
1099	if ($value === null) {
1100	return "NULL";
1101	}
1102
1103	if ($this->is_numeric_type($data_type)) {
1104	return (string) $value;
1105	}
1106
1107	if (strtoupper($data_type) === "JSON") {
1108	if ($value === "") {
1109	return "''";
1110	}
1111	$base64 = base64_encode($value);
1112	return "CONVERT(FROM_BASE64('" . $base64 . "') USING utf8mb4)";
1113	}
1114
1115	// Treat all other data types as strings and encode them as base64. This
1116	// allows us to express all possible text encodings and arbitrary binary values.
1117	if ($value === "") {
1118	return "''";
1119	}
1120	return "FROM_BASE64('" . base64_encode($value) . "')";
1121	}
1122
1123	/**
1124	* Estimates the byte length of format_value()'s output without actually
1125	* encoding. Used by format_row_for_insert() to decide whether a row
1126	* would exceed max_statement_size before doing the expensive encoding.
1127	*/
1128	private function estimate_formatted_size($value, $data_type)
1129	{
1130	if ($value === null) {
1131	return 4; // NULL
1132	}
1133
1134	if ($this->is_numeric_type($data_type)) {
1135	return strlen((string) $value);
1136	}
1137
1138	$len = strlen((string) $value);
1139	if ($len === 0) {
1140	return 2; // ''
1141	}
1142
1143	/** Base64 output is always ceil(n/3)4 bytes. /
1144	$estimated_base64_length = 4 * intdiv($len + 2, 3);
1145	// FROM_BASE64('<data>') => 15 bytes overhead + base64 length
1146	return 15 + $estimated_base64_length;
1147	}
1148
1149	/** Numeric types are emitted as bare literals (no quoting, no base64). */
1150	private function is_numeric_type($data_type)
1151	{
1152	$data_type = strtoupper($data_type);
1153	$numeric_types = [
1154	"TINYINT",
1155	"SMALLINT",
1156	"MEDIUMINT",
1157	"INTEGER",
1158	"INT",
1159	"BIGINT",
1160	"DECIMAL",
1161	"NUMERIC",
1162	"FLOAT",
1163	"DOUBLE",
1164	"REAL",
1165	"BIT",
1166	"YEAR",
1167	];
1168
1169	foreach ($numeric_types as $type) {
1170	if (strpos($data_type, $type) === 0) {
1171	return true;
1172	}
1173	}
1174
1175	return false;
1176	}
1177
1178	/**
1179	* Binary columns are not CAST to BINARY in the SELECT (they already are),
1180	* but they follow the same base64 encoding path as strings.
1181	*/
1182	private function is_binary_type($data_type)
1183	{
1184	$data_type = strtoupper($data_type);
1185	$binary_types = [
1186	"BINARY",
1187	"VARBINARY",
1188	"TINYBLOB",
1189	"BLOB",
1190	"MEDIUMBLOB",
1191	"LONGBLOB",
1192	];
1193
1194	foreach ($binary_types as $type) {
1195	if (strpos($data_type, $type) === 0) {
1196	return true;
1197	}
1198	}
1199
1200	return false;
1201	}
1202
1203	/** Returns the DATA_TYPE string for a column, or throws if unknown. */
1204	private function get_data_type(string $col): string
1205	{
1206	if (!isset($this->current_column_types[$col]["data_type"])) {
1207	throw new \RuntimeException(
1208	"No column type info for '{$col}' in table " .
1209	$this->quote_identifier($this->current_table) .
1210	". This is a bug — INFORMATION_SCHEMA should have returned it."
1211	);
1212	}
1213	return $this->current_column_types[$col]["data_type"];
1214	}
1215
1216	/** Escapes backticks by doubling them: tricky`table → `tricky``table`. */
1217	private function quote_identifier($identifier)
1218	{
1219	return '`' . str_replace('`', '``', $identifier) . '`';
1220	}
1221
1222	/** Auto-detects max_allowed_packet and uses 80% of it. Falls back to 1MB. */
1223	private function detect_max_statement_size()
1224	{
1225	try {
1226	$result = $this->db->query("SELECT @@max_allowed_packet as max_allowed_packet");
1227	$row = $result->fetch(PDO::FETCH_ASSOC);
1228	if ($row && isset($row['max_allowed_packet'])) {
1229	return (int)($row['max_allowed_packet'] * 0.8);
1230	}
1231	} catch (\PDOException $e) {
1232	}
1233
1234	return 1024 * 1024;
1235	}
1236
1237	/**
1238	* Formats a row as a VALUES tuple, splitting oversized columns if needed.
1239	*
1240	* The approach is estimate-first: compute the approximate encoded size of
1241	* each column before doing the actual (expensive) base64 encoding. If the
1242	* row fits within max_statement_size, encode everything. If it doesn't,
1243	* replace the largest non-PK columns with '' and queue their real values
1244	* as UPDATE ... CONCAT() chunks in $this->oversized_queue.
1245	*
1246	* Tables without a primary key can't use the UPDATE fallback (there's no
1247	* stable row identifier for the WHERE clause), so oversized rows in
1248	* PK-less tables are emitted as-is — the import may fail, but that's
1249	* better than silently dropping data.
1250	*/
1251	private function format_row_for_insert($row)
1252	{
1253	$estimated_sizes = [];
1254	$raw_values = [];
1255
1256	foreach ($this->current_column_names as $col) {
1257	$value = $row[$col] ?? null;
1258	$raw_values[$col] = $value;
1259	$data_type = $this->get_data_type($col);
1260	$estimated_sizes[$col] = $this->estimate_formatted_size($value, $data_type);
1261	}
1262
1263	// Estimate the size of "(val1,val2,val3)," — values + commas between them + parens + terminator
1264	$row_size_est = array_sum($estimated_sizes) + count($estimated_sizes) + 3;
1265	$projected_size = $this->current_statement_size + $row_size_est;
1266
1267	if ($projected_size <= $this->max_statement_size) {
1268	$formatted_values = [];
1269	foreach ($this->current_column_names as $col) {
1270	$data_type = $this->get_data_type($col);
1271	$formatted_values[$col] = $this->format_value($raw_values[$col], $data_type);
1272	}
1273	return "(" . implode(",", array_values($formatted_values)) . ")";
1274	}
1275
1276	// The rest of this method deals with rows that are too large to fit into a single INSERT on
1277	// the receiving end.
1278
1279	if (!$this->current_pk_columns \|\| count($this->current_pk_columns) === 0) {
1280	throw new \RuntimeException(
1281	"Row in table " . $this->quote_identifier($this->current_table) .
1282	" exceeds max_statement_size ({$this->max_statement_size} bytes)" .
1283	" but the table has no primary key, so the oversized row" .
1284	" cannot be split into UPDATE ... CONCAT() chunks."
1285	);
1286	}
1287
1288	$this->oversized_pk_values = [];
1289	foreach ($this->current_pk_columns as $pk_col) {
1290	if (!array_key_exists($pk_col, $row)) {
1291	throw new \RuntimeException(
1292	"Primary key column '{$pk_col}' missing from row for table " .
1293	$this->quote_identifier($this->current_table)
1294	);
1295	}
1296	$this->oversized_pk_values[$pk_col] = $row[$pk_col];
1297	}
1298
1299	// Split the largest columns first to bring the row under the limit
1300	$sorted_sizes = $estimated_sizes;
1301	arsort($sorted_sizes);
1302
1303	$this->oversized_queue = [];
1304	$chunked_columns = [];
1305
1306	$excess = $projected_size - $this->max_statement_size;
1307
1308	foreach ($sorted_sizes as $col => $size) {
1309	if (in_array($col, $this->current_pk_columns)) {
1310	continue;
1311	}
1312
1313	if ($size < 1000) {
1314	continue;
1315	}
1316
1317	if ($excess <= 0) {
1318	break;
1319	}
1320
1321	$raw_value = $raw_values[$col];
1322	if ($raw_value === null \|\| $raw_value === '') {
1323	continue;
1324	}
1325
1326	$data_type = $this->get_data_type($col);
1327	$value_length = strlen($raw_value);
1328	$chunk_size = $this->compute_chunk_size($col);
1329
1330	if ($value_length > $chunk_size) {
1331	$chunked_columns[$col] = true;
1332	$excess -= ($size - 2); // Saved bytes (size minus the '' replacement)
1333
1334	$this->oversized_queue[] = [
1335	'column' => $col,
1336	'data_type' => $data_type,
1337	'byte_offset' => 0,
1338	'total_length' => $value_length,
1339	];
1340	}
1341	}
1342
1343	if (empty($chunked_columns)) {
1344	$this->oversized_pk_values = null;
1345	}
1346
1347	$formatted_values = [];
1348	foreach ($this->current_column_names as $col) {
1349	if (isset($chunked_columns[$col])) {
1350	$formatted_values[$col] = "''";
1351	continue;
1352	}
1353	$data_type = $this->get_data_type($col);
1354	$formatted_values[$col] = $this->format_value($raw_values[$col], $data_type);
1355	}
1356
1357	return "(" . implode(",", array_values($formatted_values)) . ")";
1358	}
1359
1360	/**
1361	* Computes the maximum raw byte size of each chunk for the given column,
1362	* such that an UPDATE ... SET col = CONCAT(col, FROM_BASE64('...'))
1363	* statement stays within max_statement_size.
1364	*/
1365	private function compute_chunk_size($column)
1366	{
1367	$quoted_table = $this->quote_identifier($this->current_table);
1368	$quoted_column = $this->quote_identifier($column);
1369	$update_overhead = strlen("UPDATE {$quoted_table} SET {$quoted_column} = CONCAT({$quoted_column}, ) WHERE ;");
1370	$where_clause_size = $this->estimate_pk_where_size();
1371	$total_overhead = $update_overhead + $where_clause_size + 100; // Extra margin
1372
1373	$max_chunk_raw_size = ($this->max_statement_size - $total_overhead);
1374
1375	// Base64 inflates by ~1.33x, plus FROM_BASE64('') wrapper overhead
1376	$max_chunk_raw_size = (int)(($max_chunk_raw_size - 20) / 1.34);
1377	return max($max_chunk_raw_size, 1000);
1378	}
1379
1380	/** Rough strlen() estimate for the WHERE pk1 = v1 AND pk2 = v2 clause. */
1381	private function estimate_pk_where_size()
1382	{
1383	if (!$this->oversized_pk_values) {
1384	/**
1385	* A wild guess. 1KB is probably more than necessary, but we're trying to stay
1386	* on the safe side.
1387	*/
1388	return 1024;
1389	}
1390
1391	$size = 0;
1392	foreach ($this->oversized_pk_values as $col => $value) {
1393	$size += strlen($col) + 10; // `col` =
1394	if ($value === null) {
1395	$size += 10; // IS NULL
1396	} elseif (is_numeric($value)) {
1397	$size += strlen((string)$value);
1398	} else {
1399	$size += strlen((string)$value) * 1.1 + 2; // Quoted
1400	}
1401	$size += 5; // AND
1402	}
1403
1404	return (int)$size;
1405	}
1406
1407	/**
1408	* Emits one UPDATE ... SET col = CONCAT(col, chunk) statement.
1409	*
1410	* Instead of storing the entire column value in memory, this method
1411	* re-reads just the needed chunk from the database using SUBSTRING().
1412	* This keeps the cursor tiny (byte offsets only) while still producing
1413	* the correct UPDATE statements.
1414	*
1415	* Returns false when the queue is drained, which signals the state machine
1416	* to transition back to the state saved in $state_after_oversized.
1417	*/
1418	private function emit_oversized_update()
1419	{
1420	if (empty($this->oversized_queue)) {
1421	if ($this->state_after_oversized === null) {
1422	throw new \RuntimeException(
1423	"State machine bug: state_after_oversized is null when " .
1424	"exiting oversized update loop for table " .
1425	$this->quote_identifier($this->current_table)
1426	);
1427	}
1428	$this->state = $this->state_after_oversized;
1429	$this->state_after_oversized = null;
1430	$this->oversized_pk_values = null;
1431	return false;
1432	}
1433
1434	$current = $this->oversized_queue[0];
1435	$column = $current['column'];
1436	$data_type = $current['data_type'];
1437	$byte_offset = $current['byte_offset'];
1438	$total_length = $current['total_length'];
1439
1440	$chunk_size = $this->compute_chunk_size($column);
1441
1442	// Fetch just the chunk we need from the database using SUBSTRING.
1443	// MySQL's SUBSTRING is 1-indexed, so add 1 to our 0-based offset.
1444	$chunk = $this->fetch_value_substring_from_the_current_oversized_row(
1445	$column,
1446	$byte_offset + 1,
1447	$chunk_size
1448	);
1449
1450	$formatted_chunk = $this->format_value($chunk, $data_type);
1451
1452	$where_parts = [];
1453	foreach ($this->oversized_pk_values as $pk_col => $pk_value) {
1454	$quoted_pk = $this->quote_identifier($pk_col);
1455	if ($pk_value === null) {
1456	$where_parts[] = "{$quoted_pk} IS NULL";
1457	} elseif (is_numeric($pk_value)) {
1458	$where_parts[] = "{$quoted_pk} = {$pk_value}";
1459	} else {
1460	// Use FROM_BASE64() to avoid having to quote() the emitted value.
1461	$where_parts[] = "{$quoted_pk} = FROM_BASE64('" . base64_encode($pk_value) . "')";
1462	}
1463	}
1464	$where_clause = implode(" AND ", $where_parts);
1465
1466	$quoted_table = $this->quote_identifier($this->current_table);
1467	$quoted_column = $this->quote_identifier($column);
1468	$sql = "UPDATE {$quoted_table} SET {$quoted_column} = CONCAT({$quoted_column}, {$formatted_chunk}) WHERE {$where_clause};";
1469
1470	$this->current_sql_fragment = $sql;
1471
1472	$this->oversized_queue[0]['byte_offset'] += strlen($chunk);
1473	if ($this->oversized_queue[0]['byte_offset'] >= $total_length) {
1474	array_shift($this->oversized_queue);
1475	}
1476
1477	return true;
1478	}
1479
1480	/**
1481	* Fetches a substring of a column value from the current table using
1482	* the oversized row's primary key values.
1483	*
1484	* Uses CAST(SUBSTRING(...) AS BINARY) to get raw bytes without charset
1485	* re-encoding — matching the same CAST approach used in the main SELECT.
1486	*/
1487	private function fetch_value_substring_from_the_current_oversized_row(string $column, int $start, int $length): string
1488	{
1489	$quoted_table = $this->quote_identifier($this->current_table);
1490	$quoted_column = $this->quote_identifier($column);
1491
1492	$where_parts = [];
1493	$params = [];
1494	foreach ($this->oversized_pk_values as $pk_col => $pk_value) {
1495	$quoted_pk = $this->quote_identifier($pk_col);
1496	if ($pk_value === null) {
1497	$where_parts[] = "{$quoted_pk} IS NULL";
1498	} else {
1499	$where_parts[] = "{$quoted_pk} = ?";
1500	$params[] = $pk_value;
1501	}
1502	}
1503	$where_clause = implode(" AND ", $where_parts);
1504
1505	$sql = "SELECT CAST(SUBSTRING({$quoted_column}, {$start}, {$length}) AS BINARY)"
1506	. " FROM {$quoted_table} WHERE {$where_clause}";
1507	$stmt = $this->db->prepare($sql);
1508	$stmt->execute($params);
1509	$result = $stmt->fetchColumn();
1510
1511	if ($result === false) {
1512	throw new \RuntimeException(
1513	"Failed to fetch column substring for oversized row: {$column}"
1514	);
1515	}
1516
1517	return $result;
1518	}
1519
1520	/** @return bool */
1521	private function has_pending_oversized_updates()
1522	{
1523	return !empty($this->oversized_queue);
1524	}
1525	}
1526