Skip to content

Fix import precision and transactional finalization #12

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
TonisOrmisson merged 2 commits into from
Sep 8, 2026
Merged

Fix import precision and transactional finalization #12

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2cf4ab7
test: reproduce import precision and transaction integrity failures
TonisOrmisson Sep 8, 2026
e5d30bb
fix: preserve import precision and finalize within native transactions
TonisOrmisson Sep 8, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 4 additions & 5 deletions src/Core/Binary64.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,15 +5,12 @@
namespace OpenStatSpec\Core;

use InvalidArgumentException;
use JsonException;

final class Binary64
{
/**
* Encode a numeric SPSS dictionary value without PDO's default
* Encode a numeric SPSS value without PDO's default
* precision-losing float-to-string conversion.
*
* @throws JsonException
*/
public static function encode(int|float $value): string
{
Expand All @@ -22,6 +19,8 @@ public static function encode(int|float $value): string
throw new InvalidArgumentException('SPSS binary64 dictionary values must be finite.');
}

return json_encode($float, JSON_PRESERVE_ZERO_FRACTION | JSON_THROW_ON_ERROR);
$encoded = sprintf('%.17H', $float);

return strpbrk($encoded, '.eE') === false ? $encoded . '.0' : $encoded;
}
}
119 changes: 68 additions & 51 deletions src/Spss/SpssAdapter.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -134,63 +134,80 @@ public function import(
string $datasetName,
?string $verifiedSourceSha256 = null,
): SpssImportResult {
$this->assertLogicalSourcePath($sourcePath);
$verifiedSourceSha256 = NormativeCatalog::validateSourceSha256($verifiedSourceSha256);
$this->connection->assertClaimedSupported();
$this->ensureCatalogReady();
$sourceFormat = $this->spssFormat($sourcePath);
$journal = new OperationJournal($this->connection->pdo);
$operationId = $journal->start('import', null, $sourcePath, engineDetails: $this->engine->identity(), sourceFormat: $sourceFormat);
$mySqlDefinition = null;
if ($this->connection->pdo->inTransaction()) {
throw new UnsupportedOperation(DiagnosticCode::UnsupportedOperation, 'Import cannot run inside a caller-owned transaction.');
}
$errorMode = $this->connection->pdo->getAttribute(PDO::ATTR_ERRMODE);
try {
if (!in_array($sourceFormat, ['sav', 'zsav'], true)) {
throw new UnsupportedOperation(
DiagnosticCode::UnsupportedSourceFormat,
'This adapter profile supports SAV and ZSAV files only.',
);
}
$source = SpssSourceNormalizer::normalize($this->engine->read($sourcePath));
if ($this->connection->profile->driverName() === 'pgsql') {
(new PostgreSqlWideTableImporter($this->connection->pdo))->import(
$source,
$datasetName,
$sourcePath,
$verifiedSourceSha256,
);
} elseif ($this->connection->profile->driverName() === 'mysql') {
$mySqlDefinition = (new MySqlWideTableImporter(
$this->connection->pdo,
$this->mySqlProfile(),
))->import($source, $datasetName, $sourcePath, $verifiedSourceSha256);
} else {
(new SqliteWideTableImporter($this->connection->pdo))->import(
$source,
$datasetName,
$sourcePath,
$verifiedSourceSha256,
);
}
if ($this->beforeImportFinalization !== null) {
($this->beforeImportFinalization)();
if (!$this->connection->pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION)) {
throw new \RuntimeException('Could not enable PDO exception mode.');
}
$diagnostics = [];
$journal->succeed($operationId, $datasetName, $diagnostics);

return new SpssImportResult($operationId, $datasetName, count($source['data']), $diagnostics);
} catch (Throwable $exception) {
$failure = $exception;
if ($mySqlDefinition instanceof MySqlWideTableDefinition) {
try {
(new MySqlWideTableImporter(
$this->assertLogicalSourcePath($sourcePath);
$verifiedSourceSha256 = NormativeCatalog::validateSourceSha256($verifiedSourceSha256);
$this->connection->assertClaimedSupported();
$this->ensureCatalogReady();
$sourceFormat = $this->spssFormat($sourcePath);
$journal = new OperationJournal($this->connection->pdo);
$operationId = $journal->start('import', null, $sourcePath, engineDetails: $this->engine->identity(), sourceFormat: $sourceFormat);
$finalize = function () use ($journal, $operationId, $datasetName): void {
if ($this->beforeImportFinalization !== null) {
($this->beforeImportFinalization)();
}
$journal->succeed($operationId, $datasetName, []);
};
$mySqlDefinition = null;
try {
if (!in_array($sourceFormat, ['sav', 'zsav'], true)) {
throw new UnsupportedOperation(
DiagnosticCode::UnsupportedSourceFormat,
'This adapter profile supports SAV and ZSAV files only.',
);
}
$source = SpssSourceNormalizer::normalize($this->engine->read($sourcePath));
if ($this->connection->profile->driverName() === 'pgsql') {
(new PostgreSqlWideTableImporter($this->connection->pdo))->import(
$source,
$datasetName,
$sourcePath,
$verifiedSourceSha256,
$finalize,
);
} elseif ($this->connection->profile->driverName() === 'mysql') {
$mySqlDefinition = (new MySqlWideTableImporter(
$this->connection->pdo,
$this->mySqlProfile(),
))->compensateFailure($datasetName, $mySqlDefinition);
} catch (Throwable $cleanupFailure) {
$failure = $cleanupFailure;
))->import($source, $datasetName, $sourcePath, $verifiedSourceSha256);
$finalize();
} else {
(new SqliteWideTableImporter($this->connection->pdo))->import(
$source,
$datasetName,
$sourcePath,
$verifiedSourceSha256,
$finalize,
);
}

return new SpssImportResult($operationId, $datasetName, count($source['data']), []);
} catch (Throwable $exception) {
$failure = $exception;
if ($mySqlDefinition instanceof MySqlWideTableDefinition) {
try {
(new MySqlWideTableImporter(
$this->connection->pdo,
$this->mySqlProfile(),
))->compensateFailure($datasetName, $mySqlDefinition);
} catch (Throwable $cleanupFailure) {
$failure = $cleanupFailure;
}
}
$journal->fail($operationId, null, $failure, sourceItem: $sourcePath);
throw $failure;
}
} finally {
if (!$this->connection->pdo->setAttribute(PDO::ATTR_ERRMODE, $errorMode)) {
throw new \RuntimeException('Could not restore PDO error mode.');
}
$journal->fail($operationId, null, $failure, sourceItem: $sourcePath);
throw $failure;
}
}

Expand Down
129 changes: 73 additions & 56 deletions src/Sql/MySqlWideTableImporter.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,71 +34,88 @@ public function import(
string $sourcePath = "",
?string $verifiedSourceSha256 = null,
): MySqlWideTableDefinition {
$verifiedSourceSha256 = NormativeCatalog::validateSourceSha256($verifiedSourceSha256);
$variables = $source['variables'] ?? null;
$rows = $source['data'] ?? null;
if (!is_array($variables) || !array_is_list($variables) || $variables === []) {
throw new UnsupportedOperation(
DiagnosticCode::InvalidSourceDataset,
'The source dataset must contain an ordered variable list.',
);
}
if (!is_array($rows) || !array_is_list($rows)) {
throw new UnsupportedOperation(
DiagnosticCode::InvalidSourceDataset,
'The source dataset must contain an ordered case list.',
);
if ($this->pdo->inTransaction()) {
throw new UnsupportedOperation(DiagnosticCode::UnsupportedOperation, 'Import cannot run inside a caller-owned transaction.');
}

$this->profile->assertDataset($variables, $rows, $this->pdo);

$schema = new MySqlSchema($this->pdo, $this->profile);
// Complete source, physical-name, and width preflight happens before any DDL.
$definition = $schema->wideTableDefinition($datasetName, $variables);
$v3Metadata = $this->assertSourceMetadata($source, $variables, $definition);

$schema->createCatalog();
$this->pdo->exec($definition->createSql);

$ownedDefinition = $definition;
$errorMode = $this->pdo->getAttribute(PDO::ATTR_ERRMODE);
try {
$this->pdo->beginTransaction();
$this->storeDatasetMetadata($datasetName, $source);
$this->storeTechnicalMetadata($datasetName, $source);
$this->storeCatalogue($datasetName, $variables, $definition);
$this->storeWeightVariable($datasetName, $source['weightVariableName'] ?? null, $definition);
$this->storeDisplayMetadata($datasetName, $source['displayParameters'] ?? []);
$this->storeDictionaryMetadata($datasetName, $variables, $source['valueLabels'] ?? []);
if ($v3Metadata !== null) {
(new SqliteV3MetadataImporter($this->pdo))->storeValidated($datasetName, $v3Metadata);
if (!$this->pdo->setAttribute(PDO::ATTR_ERRMODE, PDO::ERRMODE_EXCEPTION)) {
throw new \RuntimeException('Could not enable PDO exception mode.');
}
$this->insertCases($definition, $rows);
if ($sourcePath !== "" || $verifiedSourceSha256 !== null) {
$datasetId = (new NormativeCatalog($this->pdo))->storeImportedDataset(
$datasetName,
$sourcePath,
$source,
$verifiedSourceSha256,
$verifiedSourceSha256 = NormativeCatalog::validateSourceSha256($verifiedSourceSha256);
$variables = $source['variables'] ?? null;
$rows = $source['data'] ?? null;
if (!is_array($variables) || !array_is_list($variables) || $variables === []) {
throw new UnsupportedOperation(
DiagnosticCode::InvalidSourceDataset,
'The source dataset must contain an ordered variable list.',
);
$ownedDefinition = new MySqlWideTableDefinition(
$definition->tableName,
$definition->createSql,
$definition->columns,
$datasetId,
}
if (!is_array($rows) || !array_is_list($rows)) {
throw new UnsupportedOperation(
DiagnosticCode::InvalidSourceDataset,
'The source dataset must contain an ordered case list.',
);
}
$this->pdo->commit();
} catch (Throwable $exception) {
if ($this->pdo->inTransaction()) {
$this->pdo->rollBack();

$this->profile->assertDataset($variables, $rows, $this->pdo);

$schema = new MySqlSchema($this->pdo, $this->profile);
// Complete source, physical-name, and width preflight happens before any DDL.
$definition = $schema->wideTableDefinition($datasetName, $variables);
$v3Metadata = $this->assertSourceMetadata($source, $variables, $definition);

$schema->createCatalog();
$this->pdo->exec($definition->createSql);

$ownedDefinition = $definition;
try {
if (!$this->pdo->beginTransaction()) {
throw new \RuntimeException('Could not begin import transaction.');
}
$this->storeDatasetMetadata($datasetName, $source);
$this->storeTechnicalMetadata($datasetName, $source);
$this->storeCatalogue($datasetName, $variables, $definition);
$this->storeWeightVariable($datasetName, $source['weightVariableName'] ?? null, $definition);
$this->storeDisplayMetadata($datasetName, $source['displayParameters'] ?? []);
$this->storeDictionaryMetadata($datasetName, $variables, $source['valueLabels'] ?? []);
if ($v3Metadata !== null) {
(new SqliteV3MetadataImporter($this->pdo))->storeValidated($datasetName, $v3Metadata);
}
$this->insertCases($definition, $rows);
if ($sourcePath !== "" || $verifiedSourceSha256 !== null) {
$datasetId = (new NormativeCatalog($this->pdo))->storeImportedDataset(
$datasetName,
$sourcePath,
$source,
$verifiedSourceSha256,
);
$ownedDefinition = new MySqlWideTableDefinition(
$definition->tableName,
$definition->createSql,
$definition->columns,
$datasetId,
);
}
if (!$this->pdo->commit()) {
throw new \RuntimeException('Could not commit import transaction.');
}
} catch (Throwable $exception) {
if ($this->pdo->inTransaction() && !$this->pdo->rollBack()) {
throw new \RuntimeException('Could not roll back import transaction.', previous: $exception);
}

// Catalogue DML was rolled back. Only the physical DDL survives.
$this->dropPhysicalTable($definition);
Comment thread
TonisOrmisson marked this conversation as resolved.
throw $exception;
}

// Catalogue DML was rolled back. Only the physical DDL survives.
$this->dropPhysicalTable($definition);
throw $exception;
return $ownedDefinition;
} finally {
if (!$this->pdo->setAttribute(PDO::ATTR_ERRMODE, $errorMode)) {
throw new \RuntimeException('Could not restore PDO error mode.');
}
Comment thread
TonisOrmisson marked this conversation as resolved.
}

return $ownedDefinition;
}

/**
Expand Down
2 changes: 1 addition & 1 deletion src/Sql/NormativeCatalog.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -395,7 +395,7 @@ private function storeMultipleResponseSets(string $datasetId, array $variableIds
$counted = $set['countedValue'] ?? null;
$countedKind = is_string($counted) ? 'string' : (is_int($counted) || is_float($counted) ? 'numeric' : null);
$this->statement('INSERT INTO multiple_response_set (multiple_response_set_id, dataset_id, source_ordinal, set_name, set_label, set_kind, counted_value_kind, counted_numeric_value, counted_string_value, category_label_behavior, label_source) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)')->execute([
$setId, $datasetId, $setOrdinal + 1, $set['name'], is_string($set['label'] ?? null) ? $set['label'] : null, $set['type'] === 'dichotomy' ? 'MD' : 'MC', $countedKind, is_int($counted) || is_float($counted) ? $counted : null, is_string($counted) ? $counted : null, is_string($set['categoryLabels'] ?? null) ? $set['categoryLabels'] : null, is_string($set['labelSource'] ?? null) ? $set['labelSource'] : null,
$setId, $datasetId, $setOrdinal + 1, $set['name'], is_string($set['label'] ?? null) ? $set['label'] : null, $set['type'] === 'dichotomy' ? 'MD' : 'MC', $countedKind, is_float($counted) ? Binary64::encode($counted) : (is_int($counted) ? $counted : null), is_string($counted) ? $counted : null, is_string($set['categoryLabels'] ?? null) ? $set['categoryLabels'] : null, is_string($set['labelSource'] ?? null) ? $set['labelSource'] : null,
]);
foreach ($this->list($set['variableNames'] ?? null, 'Multiple-response members') as $ordinal => $name) {
$variableOrdinal = $this->variableOrdinal($variables, $name);
Expand Down
Loading