From 76ac043c7db4e9e1d85b1cd83a382ad12676f957 Mon Sep 17 00:00:00 2001 From: harsh mahajan Date: Mon, 12 Oct 2026 19:48:10 +0530 Subject: [PATCH 1/2] fix(csv): reject rows that do not parse instead of changing them A CSV import stored NULL or a cast value for cells that do not parse as their column type, trimmed text, read a quoted "null" as NULL, kept the UTF-8 byte order mark in the first header, aborted on a row with the wrong number of columns, and let one non-UTF-8 value drop its whole batch without an error. Each such row is now rejected with its row number and counted as an error, while the rest of the file is imported. Text keeps its whitespace, only an unquoted null is NULL, and an empty cell is NULL for every column except text. The export quotes the text "null" so it reads back unchanged. --- src/Migration/Destinations/CSV.php | 12 +- src/Migration/Sources/CSV.php | 327 +++++++++++++++++++++-------- 2 files changed, 250 insertions(+), 89 deletions(-) diff --git a/src/Migration/Destinations/CSV.php b/src/Migration/Destinations/CSV.php index d05f78dd..21e08461 100644 --- a/src/Migration/Destinations/CSV.php +++ b/src/Migration/Destinations/CSV.php @@ -240,8 +240,14 @@ protected function writeCSVLine($handle, array $fields): bool $parts = []; foreach ($fields as $field) { + if ($field === null) { + $parts[] = 'null'; + continue; + } + $field = (string)$field; - if (\strpbrk($field, $this->delimiter . "\n\r" . $this->enclosure) !== false) { + // Quoted, the text "null" is not read back as NULL. + if ($field === 'null' || \strpbrk($field, $this->delimiter . "\n\r" . $this->enclosure) !== false) { $parts[] = $this->enclosure . \str_replace($this->enclosure, $this->enclosure . $this->enclosure, $field) . $this->enclosure; } else { $parts[] = $field; @@ -316,10 +322,10 @@ protected function resourceToCSVData(Row $resource): array /** * Convert a single value to CSV-compatible format */ - protected function convertValueToCSV(mixed $value): string + protected function convertValueToCSV(mixed $value): ?string { if (\is_null($value)) { - return 'null'; + return null; } if (\is_bool($value)) { return $value ? 'true' : 'false'; diff --git a/src/Migration/Sources/CSV.php b/src/Migration/Sources/CSV.php index 0b6b725c..862dc3bd 100644 --- a/src/Migration/Sources/CSV.php +++ b/src/Migration/Sources/CSV.php @@ -18,6 +18,19 @@ class CSV extends Source { + private const TEXT_TYPES = [ + Column::TYPE_STRING, + Column::TYPE_VARCHAR, + Column::TYPE_TEXT, + Column::TYPE_MEDIUMTEXT, + Column::TYPE_LONGTEXT, + ]; + + /** + * Rejected rows listed in the errors; every one is still counted. + */ + private const MAX_ROW_ERRORS = 100; + private const ALLOWED_INTERNALS = [ '$id' => true, '$permissions' => true, @@ -40,6 +53,8 @@ class CSV extends Source private bool $downloaded = false; + private int $rejectedRows = 0; + public function __construct( string $resourceId, string $filePath, @@ -221,6 +236,7 @@ private function exportRows(int $batchSize): void $columnTypes = []; $requiredColumns = []; $manyToManyKeys = []; + $textKeys = []; foreach ($columns as $column) { $key = $column['key']; @@ -253,9 +269,14 @@ private function exportRows(int $batchSize): void if ($isArray && $type !== Column::TYPE_RELATIONSHIP) { $arrayKeys[$key] = true; } + + // Formats (email, enum, ip, url) have no empty value, so they are not text here. + if (\in_array($type, self::TEXT_TYPES, true) && empty($column['format'])) { + $textKeys[$key] = true; + } } - $this->withCsvStream(function ($stream, $delimiter) use ($columnTypes, $databaseType, $requiredColumns, $manyToManyKeys, $arrayKeys, $tablePayload, $batchSize) { + $this->withCsvStream(function ($stream, $delimiter) use ($columnTypes, $databaseType, $requiredColumns, $manyToManyKeys, $arrayKeys, $textKeys, $tablePayload, $batchSize) { $headers = \fgetcsv($stream, 0, $delimiter, '"', '"'); if (!\is_array($headers) || \count($headers) === 0) { @@ -265,108 +286,111 @@ private function exportRows(int $batchSize): void $this->validateCSVHeaders($headers, $columnTypes, $requiredColumns); $buffer = []; + $rowNumber = 1; // The header is row 1, as in a spreadsheet. + + while (true) { + $offset = \ftell($stream); + $row = \fgetcsv($stream, 0, $delimiter, '"', '"'); + if ($row === false) { + break; + } + + $rowNumber++; + + if ($row === [null]) { + continue; // Blank line + } - while (($row = \fgetcsv($stream, 0, $delimiter, '"', '"')) !== false) { if (\count($row) !== \count($headers)) { - throw new \Exception('CSV row does not match the number of header columns.', Exception::CODE_VALIDATION); + $this->rejectRow($databaseType, $tablePayload, null, "Row $rowNumber has " . \count($row) . ' columns, but the header has ' . \count($headers) . '.'); + continue; } - $data = \array_combine($headers, $row); - if ($data === false) { + // PDO drops a whole batch without an error when one value is not UTF-8. + if (!\mb_check_encoding($row, 'UTF-8')) { + $this->rejectRow($databaseType, $tablePayload, null, "Row $rowNumber is not valid UTF-8."); continue; } + $data = \array_combine($headers, $row); $parsedData = $data; + $quoted = null; + + /** + * Parsing logic for best compatibility with spec and 3rd party tools. + * - 'null' unquoted literal is converted to null. + * - text keeps its whitespace and an empty cell stays an empty string. + * - an empty cell of any other column is null. + * - other values are parsed as per their type; a value that does not parse rejects the row. + */ + try { + foreach ($headers as $index => $key) { + $type = $columnTypes[$key] ?? null; + + if (!isset($type) && $key !== '$permissions') { + // Skip unknown columns + continue; + } - foreach ($data as $key => $value) { - $parsedValue = \trim($value); - $type = $columnTypes[$key] ?? null; - - if (!isset($type) && $key !== '$permissions') { - // Skip unknown columns - continue; - } - - if (isset($manyToManyKeys[$key])) { - if ($parsedValue === '') { - $parsedData[$key] = []; - } else { - // Split on commas, trim whitespace, drop empty tokens, and reindex - $ids = \explode(',', $parsedValue); - $ids = \array_map(\trim(...), $ids); - $ids = \array_filter($ids, static fn ($id) => $id !== ''); - $parsedData[$key] = \array_values($ids); + $parsedValue = isset($textKeys[$key]) ? $row[$index] : \trim($row[$index]); + + if (isset($manyToManyKeys[$key])) { + if ($parsedValue === '') { + $parsedData[$key] = []; + } else { + // Split on commas, trim whitespace, drop empty tokens, and reindex + $ids = \explode(',', $parsedValue); + $ids = \array_map(\trim(...), $ids); + $ids = \array_filter($ids, static fn ($id) => $id !== ''); + $parsedData[$key] = \array_values($ids); + } + continue; } - continue; - } - if (isset($arrayKeys[$key])) { - if ($parsedValue === '') { - $parsedData[$key] = []; - } else { - // Try to decode as JSON first - $arrayValues = \json_decode($parsedValue, true); - - // If JSON decode fails, fall back to comma-separated parsing - if (\json_last_error() !== JSON_ERROR_NONE) { - // Remove empty strings from comma-separated parsing - $arrayValues = \str_getcsv($parsedValue, ',', '"', '"'); - $arrayValues = \array_map(\trim(...), $arrayValues); - $arrayValues = \array_filter($arrayValues, fn ($item) => $item !== ''); - $arrayValues = \array_values($arrayValues); // Re-index array + if (\trim($parsedValue) === 'null') { + $quoted ??= $this->quotedFields($stream, $offset, $delimiter); + + if (!($quoted[$index] ?? false)) { + $parsedData[$key] = null; + continue; } + } - if (!\is_array($arrayValues)) { - throw new \Exception("Invalid array format for column '$key': $parsedValue", Exception::CODE_VALIDATION); + if (isset($arrayKeys[$key])) { + if ($parsedValue === '') { + $parsedData[$key] = []; + } else { + // Try to decode as JSON first + $arrayValues = \json_decode($parsedValue, true); + + // If JSON decode fails, fall back to comma-separated parsing + if (\json_last_error() !== JSON_ERROR_NONE) { + // Remove empty strings from comma-separated parsing + $arrayValues = \str_getcsv($parsedValue, ',', '"', '"'); + $arrayValues = \array_map(\trim(...), $arrayValues); + $arrayValues = \array_filter($arrayValues, fn ($item) => $item !== ''); + $arrayValues = \array_values($arrayValues); // Re-index array + } + + if (!\is_array($arrayValues)) { + throw new \InvalidArgumentException("'$key' must be an array"); + } + + $parsedData[$key] = \array_map(fn ($item) => $this->parseValue($key, $item, $type), $arrayValues); } + continue; + } - $parsedData[$key] = array_map(function ($item) use ($type) { - return match ($type) { - Column::TYPE_INTEGER,Column::TYPE_BIG_INT => is_numeric($item) ? (int)$item : null, - Column::TYPE_FLOAT => is_numeric($item) ? (float)$item : null, - Column::TYPE_BOOLEAN => filter_var($item, FILTER_VALIDATE_BOOLEAN, FILTER_NULL_ON_FAILURE), - default => $item, - }; - }, $arrayValues); + if ($parsedValue === '') { + $parsedData[$key] = isset($textKeys[$key]) ? '' : null; + continue; } - continue; - } - /** - * Parsing logic for best compatibility with spec and 3rd party tools. - * - 'null' unquoted literal is converted to null. - * - missing strings stay empty strings for best compatibility. - * - missing numbers, booleans, and datetime's are converted to null. - * - other values are parsed as per their type. - */ - - $parsedData[$key] = match ($parsedValue) { - 'null' => null, // 'null' string is converted to null - '' => match ($type) { - Column::TYPE_INTEGER, - Column::TYPE_BIG_INT, - Column::TYPE_FLOAT, - Column::TYPE_BOOLEAN, - Column::TYPE_DATETIME, - Column::TYPE_RELATIONSHIP => null, // primitive types default to null - default => '', // but empty string stays empty string for compatibility - }, - default => match ($type) { - Column::TYPE_INTEGER, Column::TYPE_BIG_INT => \is_numeric($parsedValue) ? (int)$parsedValue : null, - Column::TYPE_FLOAT => \is_numeric($parsedValue) ? (float)$parsedValue : null, - Column::TYPE_BOOLEAN => \filter_var( - $parsedValue, - filter: FILTER_VALIDATE_BOOLEAN, - options: FILTER_NULL_ON_FAILURE - ), - Column::TYPE_POINT, - Column::TYPE_LINE, - Column::TYPE_POLYGON, - Column::TYPE_VECTOR, - Column::TYPE_OBJECT => \is_string($parsedValue) ? json_decode($parsedValue, true) : null, - default => $parsedValue, - }, - }; + $parsedData[$key] = $this->parseValue($key, $parsedValue, $type); + } + } catch (\InvalidArgumentException $e) { + $this->rejectRow($databaseType, $tablePayload, $data['$id'] ?? null, "Row $rowNumber: {$e->getMessage()}."); + continue; } $rowId = $parsedData['$id'] ?? 'unique()'; @@ -395,6 +419,132 @@ private function exportRows(int $batchSize): void }); } + /** + * @throws \InvalidArgumentException when the value is not of the column's type. + */ + private function parseValue(string $key, mixed $value, ?string $type): mixed + { + switch ($type) { + case Column::TYPE_INTEGER: + case Column::TYPE_BIG_INT: + if (\is_numeric($value)) { + $number = $value + 0; + + // Spreadsheets can write whole numbers as 1.0 or 1e3. + if (\is_int($number) || (\floor($number) === $number && \abs($number) < 2 ** 63)) { + return (int) $number; + } + } + + throw new \InvalidArgumentException("'$key' must be an integer"); + + case Column::TYPE_FLOAT: + if (\is_numeric($value)) { + return (float) $value; + } + + throw new \InvalidArgumentException("'$key' must be a number"); + + case Column::TYPE_BOOLEAN: + $boolean = \is_bool($value) ? $value : \filter_var($value, FILTER_VALIDATE_BOOLEAN, FILTER_NULL_ON_FAILURE); + if ($boolean !== null) { + return $boolean; + } + + throw new \InvalidArgumentException("'$key' must be true or false"); + + case Column::TYPE_POINT: + case Column::TYPE_LINE: + case Column::TYPE_POLYGON: + case Column::TYPE_VECTOR: + case Column::TYPE_OBJECT: + $decoded = \json_decode((string) $value, true); + if (\json_last_error() === JSON_ERROR_NONE) { + return $decoded; + } + + throw new \InvalidArgumentException("'$key' must be JSON"); + + default: + return $value; + } + } + + /** + * Report a row that is not imported, and count it as an error. + * + * @param array $table + */ + private function rejectRow(string $databaseType, array $table, ?string $rowId, string $message): void + { + $row = Appwrite::getRecord($databaseType, [ + 'id' => $rowId ?? '', + 'table' => $table, + 'data' => [], + 'permissions' => [], + ]); + $row->setStatus(UtopiaResource::STATUS_ERROR, $message); + $this->cache->update($row); + + // A malformed file would otherwise add an error for each of its rows. + if (++$this->rejectedRows > self::MAX_ROW_ERRORS) { + return; + } + + $this->addError(new Exception( + UtopiaResource::TYPE_ROW, + Transfer::GROUP_DATABASES, + resourceId: $rowId, + message: $message, + code: Exception::CODE_VALIDATION, + )); + } + + /** + * Whether each field of the record at $offset was enclosed in quotes, which fgetcsv() does not tell. + * + * @param resource $stream + * @return array + */ + private function quotedFields($stream, int $offset, string $delimiter): array + { + $end = (int) \ftell($stream); + \fseek($stream, $offset); + $record = (string) \fread($stream, \max(1, $end - $offset)); + \fseek($stream, $end); + + $quoted = []; + $position = 0; + + while (true) { + // Like fgetcsv(), skip whitespace before an opening quote. + $start = $position + \strspn($record, \str_replace($delimiter, '', " \t"), $position); + $isQuoted = ($record[$start] ?? '') === '"'; + $quoted[] = $isQuoted; + + if ($isQuoted) { + // Find the closing quote, stepping over escaped ("") ones. + $position = $start + 1; + while (($position = \strpos($record, '"', $position)) !== false && ($record[$position + 1] ?? '') === '"') { + $position += 2; + } + + if ($position === false) { + break; + } + } + + $position = \strpos($record, $delimiter, $position); + if ($position === false) { + break; + } + + $position++; + } + + return $quoted; + } + /** * @throws \Exception */ @@ -496,6 +646,11 @@ private function withCsvStream(callable $callback): void $delimiter = $this->delimiter($stream); + // Excel's "CSV UTF-8" starts with a byte order mark, which would become part of the first header. + if (\fread($stream, 3) !== "\xEF\xBB\xBF") { + \rewind($stream); + } + try { $callback($stream, $delimiter); } finally { From 0d81ae309b3fae1ea8afb9b8dd721621eecd3dbf Mon Sep 17 00:00:00 2001 From: harsh mahajan Date: Mon, 12 Oct 2026 19:48:10 +0530 Subject: [PATCH 2/2] fix(appwrite): count rows as imported only once their batch is written Rows were marked as succeeded when buffered, so a batch the database rejected still counted every row but one as a success. A rejected last row also skipped the write, dropping the rows buffered before it. The batch is now written even when its last row is rejected, a failed write marks every row in it as an error, and row counters are updated after the write. --- src/Migration/Destinations/Appwrite.php | 111 +++++++++++++++--------- 1 file changed, 68 insertions(+), 43 deletions(-) diff --git a/src/Migration/Destinations/Appwrite.php b/src/Migration/Destinations/Appwrite.php index 46ca9c5e..7b399fac 100644 --- a/src/Migration/Destinations/Appwrite.php +++ b/src/Migration/Destinations/Appwrite.php @@ -157,6 +157,13 @@ class Appwrite extends Destination */ private array $rowBuffer = []; + /** + * The rows whose documents are in $rowBuffer. + * + * @var array + */ + private array $rowBufferResources = []; + /** * Overwrite-mode orphan tracking, keyed by (database, table). Orphans are * destination keys not in `attributeKeys` / `indexKeys`. Entries removed @@ -297,6 +304,7 @@ public function run( private function resetRunState(): void { $this->rowBuffer = []; + $this->rowBufferResources = []; $this->orphansByTable = []; $this->processedTwoWayPairs = []; $this->provisioningDatabases = []; @@ -546,6 +554,7 @@ protected function import(array $resources, callable $callback): void } $total = \count($resources); + $records = []; foreach ($resources as $index => $resource) { $resource->setStatus(Resource::STATUS_PROCESSING); @@ -587,9 +596,17 @@ protected function import(array $resources, callable $callback): void $this->dbForPlatform->setPreserveDates(false); } + // Rows are counted once their batch is written, since writing it can still fail them. + if ($responseResource instanceof Row) { + $records[] = $responseResource; + continue; + } + $this->cache->update($responseResource); } + $this->cache->updateAll($records); + $callback($resources); } @@ -1551,13 +1568,7 @@ protected function createIndex(Index $resource): bool return true; } - /** - * @throws AuthorizationException - * @throws DatabaseException - * @throws StructureException - * @throws Exception - */ - protected function createRecord(Row $resource, bool $isLast): bool + private function bufferRecord(Row $resource): bool { if ($resource->getId() == 'unique()') { $resource->setId(ID::unique()); @@ -1621,8 +1632,25 @@ protected function createRecord(Row $resource, bool $isLast): bool '$id' => $resource->getId(), '$permissions' => $resource->getPermissions(), ], $data)); + $this->rowBufferResources[] = $resource; + + return true; + } + + /** + * Rows are written in one batch, so a failed write fails every row in it. + * + * @throws AuthorizationException + * @throws DatabaseException + * @throws StructureException + * @throws Exception + */ + protected function createRecord(Row $resource, bool $isLast): bool + { + $buffered = $this->bufferRecord($resource); - if ($isLast) { + // Write the batch even when its last row was rejected, or the rows before it are lost. + if ($isLast && !empty($this->rowBuffer)) { try { $database = $this->dbForProject->getDocument( self::META_DATABASES, @@ -1668,48 +1696,45 @@ protected function createRecord(Row $resource, bool $isLast): bool } $collectionId = $this->tableCollectionId($database, $table); - try { - match ($this->onDuplicate) { - OnDuplicate::Overwrite => $dbForDatabases->skipRelationshipsExistCheck( - fn () => $dbForDatabases->upsertDocuments($collectionId, $this->rowBuffer) - ), - OnDuplicate::Skip => $dbForDatabases->skipDuplicates( - fn () => $dbForDatabases->skipRelationshipsExistCheck( - fn () => $dbForDatabases->createDocuments($collectionId, $this->rowBuffer) - ) - ), - OnDuplicate::Fail => $dbForDatabases->skipRelationshipsExistCheck( + match ($this->onDuplicate) { + OnDuplicate::Overwrite => $dbForDatabases->skipRelationshipsExistCheck( + fn () => $dbForDatabases->upsertDocuments($collectionId, $this->rowBuffer) + ), + OnDuplicate::Skip => $dbForDatabases->skipDuplicates( + fn () => $dbForDatabases->skipRelationshipsExistCheck( fn () => $dbForDatabases->createDocuments($collectionId, $this->rowBuffer) - ), - }; - } catch (DuplicateException $e) { - $resource->setStatus(Resource::STATUS_ERROR, 'Document already exists'); - $this->addError(new Exception( - resourceName: $resource->getName(), - resourceGroup: $resource->getGroup(), - resourceId: $resource->getId(), - message: 'Document already exists', - previous: $e, - )); - return false; - } catch (StructureException $e) { - $resource->setStatus(Resource::STATUS_ERROR, $e->getMessage()); - $this->addError(new Exception( - resourceName: $resource->getName(), - resourceGroup: $resource->getGroup(), - resourceId: $resource->getId(), - message: $e->getMessage(), - previous: $e, - )); - return false; + ) + ), + OnDuplicate::Fail => $dbForDatabases->skipRelationshipsExistCheck( + fn () => $dbForDatabases->createDocuments($collectionId, $this->rowBuffer) + ), + }; + } catch (\Throwable $e) { + $message = $e instanceof DuplicateException ? 'Document already exists' : $e->getMessage(); + + foreach ($this->rowBufferResources as $row) { + $row->setStatus(Resource::STATUS_ERROR, $message); } + + if (!$e instanceof DuplicateException && !$e instanceof StructureException) { + throw $e; + } + + $this->addError(new Exception( + resourceName: $resource->getName(), + resourceGroup: $resource->getGroup(), + resourceId: $resource->getId(), + message: $message, + previous: $e, + )); + return false; } finally { $this->rowBuffer = []; + $this->rowBufferResources = []; } } - - return true; + return $buffered; } /** Relationships route through deleteRelationship since deleteAttribute throws for VAR_RELATIONSHIP. */