diff options
Diffstat (limited to 'lib/private/DB/Connection.php')
| -rw-r--r-- | lib/private/DB/Connection.php | 329 |
1 files changed, 282 insertions, 47 deletions
diff --git a/lib/private/DB/Connection.php b/lib/private/DB/Connection.php index c886cb20235..f86cbc341a4 100644 --- a/lib/private/DB/Connection.php +++ b/lib/private/DB/Connection.php @@ -13,18 +13,32 @@ use Doctrine\DBAL\Cache\QueryCacheProfile; use Doctrine\DBAL\Configuration; use Doctrine\DBAL\Connections\PrimaryReadReplicaConnection; use Doctrine\DBAL\Driver; +use Doctrine\DBAL\Driver\ServerInfoAwareConnection; use Doctrine\DBAL\Exception; use Doctrine\DBAL\Exception\ConnectionLost; +use Doctrine\DBAL\Platforms\MariaDBPlatform; use Doctrine\DBAL\Platforms\MySQLPlatform; use Doctrine\DBAL\Platforms\OraclePlatform; +use Doctrine\DBAL\Platforms\PostgreSQLPlatform; use Doctrine\DBAL\Platforms\SqlitePlatform; use Doctrine\DBAL\Result; use Doctrine\DBAL\Schema\Schema; use Doctrine\DBAL\Statement; +use OC\DB\QueryBuilder\Partitioned\PartitionedQueryBuilder; +use OC\DB\QueryBuilder\Partitioned\PartitionSplit; use OC\DB\QueryBuilder\QueryBuilder; +use OC\DB\QueryBuilder\Sharded\AutoIncrementHandler; +use OC\DB\QueryBuilder\Sharded\CrossShardMoveHelper; +use OC\DB\QueryBuilder\Sharded\RoundRobinShardMapper; +use OC\DB\QueryBuilder\Sharded\ShardConnectionManager; +use OC\DB\QueryBuilder\Sharded\ShardDefinition; use OC\SystemConfig; use OCP\DB\QueryBuilder\IQueryBuilder; +use OCP\DB\QueryBuilder\Sharded\IShardMapper; use OCP\Diagnostics\IEventLogger; +use OCP\ICacheFactory; +use OCP\IDBConnection; +use OCP\ILogger; use OCP\IRequestId; use OCP\PreConditionNotMetException; use OCP\Profiler\IProfiler; @@ -66,9 +80,35 @@ class Connection extends PrimaryReadReplicaConnection { /** @var array<string, int> */ protected $tableDirtyWrites = []; + protected bool $logDbException = false; + private ?array $transactionBacktrace = null; + protected bool $logRequestId; protected string $requestId; + /** @var array<string, list<string>> */ + protected array $partitions; + /** @var ShardDefinition[] */ + protected array $shards = []; + protected ShardConnectionManager $shardConnectionManager; + protected AutoIncrementHandler $autoIncrementHandler; + protected bool $isShardingEnabled; + + public const SHARD_PRESETS = [ + 'filecache' => [ + 'companion_keys' => [ + 'file_id', + ], + 'companion_tables' => [ + 'filecache_extended', + 'files_metadata', + ], + 'primary_key' => 'fileid', + 'shard_key' => 'storage', + 'table' => 'filecache', + ], + ]; + /** * Initializes a new instance of the Connection class. * @@ -78,7 +118,7 @@ class Connection extends PrimaryReadReplicaConnection { private array $params, Driver $driver, ?Configuration $config = null, - ?EventManager $eventManager = null + ?EventManager $eventManager = null, ) { if (!isset($params['adapter'])) { throw new \Exception('adapter not set'); @@ -92,12 +132,23 @@ class Connection extends PrimaryReadReplicaConnection { parent::__construct($params, $driver, $config, $eventManager); $this->adapter = new $params['adapter']($this); $this->tablePrefix = $params['tablePrefix']; - + $this->isShardingEnabled = isset($this->params['sharding']) && !empty($this->params['sharding']); + + if ($this->isShardingEnabled) { + /** @psalm-suppress InvalidArrayOffset */ + $this->shardConnectionManager = $this->params['shard_connection_manager'] ?? Server::get(ShardConnectionManager::class); + /** @psalm-suppress InvalidArrayOffset */ + $this->autoIncrementHandler = $this->params['auto_increment_handler'] ?? new AutoIncrementHandler( + Server::get(ICacheFactory::class), + $this->shardConnectionManager, + ); + } $this->systemConfig = \OC::$server->getSystemConfig(); $this->clock = Server::get(ClockInterface::class); $this->logger = Server::get(LoggerInterface::class); $this->logRequestId = $this->systemConfig->getValue('db.log_request_id', false); + $this->logDbException = $this->systemConfig->getValue('db.log_exceptions', false); $this->requestId = Server::get(IRequestId::class)->getId(); /** @var \OCP\Profiler\IProfiler */ @@ -110,10 +161,58 @@ class Connection extends PrimaryReadReplicaConnection { $this->_config->setSQLLogger($debugStack); } + /** @var array<string, array{shards: array[], mapper: ?string, from_primary_key: ?int, from_shard_key: ?int}> $shardConfig */ + $shardConfig = $this->params['sharding'] ?? []; + $shardNames = array_keys($shardConfig); + $this->shards = array_map(function (array $config, string $name) { + if (!isset(self::SHARD_PRESETS[$name])) { + throw new \Exception("Shard preset $name not found"); + } + + $shardMapperClass = $config['mapper'] ?? RoundRobinShardMapper::class; + $shardMapper = Server::get($shardMapperClass); + if (!$shardMapper instanceof IShardMapper) { + throw new \Exception("Invalid shard mapper: $shardMapperClass"); + } + return new ShardDefinition( + self::SHARD_PRESETS[$name]['table'], + self::SHARD_PRESETS[$name]['primary_key'], + self::SHARD_PRESETS[$name]['companion_keys'], + self::SHARD_PRESETS[$name]['shard_key'], + $shardMapper, + self::SHARD_PRESETS[$name]['companion_tables'], + $config['shards'], + $config['from_primary_key'] ?? 0, + $config['from_shard_key'] ?? 0, + ); + }, $shardConfig, $shardNames); + $this->shards = array_combine($shardNames, $this->shards); + $this->partitions = array_map(function (ShardDefinition $shard) { + return array_merge([$shard->table], $shard->companionTables); + }, $this->shards); + $this->setNestTransactionsWithSavepoints(true); } /** + * @return IDBConnection[] + */ + public function getShardConnections(): array { + $connections = []; + if ($this->isShardingEnabled) { + foreach ($this->shards as $shardDefinition) { + foreach ($shardDefinition->getAllShards() as $shard) { + if ($shard !== ShardDefinition::MIGRATION_SHARD) { + /** @var ConnectionAdapter $connection */ + $connections[] = $this->shardConnectionManager->getConnection($shardDefinition, $shard); + } + } + } + } + return $connections; + } + + /** * @throws Exception */ public function connect($connectionName = null) { @@ -124,8 +223,6 @@ class Connection extends PrimaryReadReplicaConnection { return parent::connect(); } - $this->lastConnectionCheck[$this->getConnectionName()] = time(); - // Only trigger the event logger for the initial connect call $eventLogger = Server::get(IEventLogger::class); $eventLogger->start('connect:db', 'db connection opened'); @@ -133,6 +230,8 @@ class Connection extends PrimaryReadReplicaConnection { $status = parent::connect(); $eventLogger->end('connect:db'); + $this->lastConnectionCheck[$this->getConnectionName()] = time(); + return $status; } catch (Exception $e) { // throw a new exception to prevent leaking info from the stacktrace @@ -161,18 +260,34 @@ class Connection extends PrimaryReadReplicaConnection { */ public function getQueryBuilder(): IQueryBuilder { $this->queriesBuilt++; - return new QueryBuilder( + + $builder = new QueryBuilder( new ConnectionAdapter($this), $this->systemConfig, $this->logger ); + if ($this->isShardingEnabled && count($this->partitions) > 0) { + $builder = new PartitionedQueryBuilder( + $builder, + $this->shards, + $this->shardConnectionManager, + $this->autoIncrementHandler, + ); + foreach ($this->partitions as $name => $tables) { + $partition = new PartitionSplit($name, $tables); + $builder->addPartition($partition); + } + return $builder; + } else { + return $builder; + } } /** * Gets the QueryBuilder for the connection. * * @return \Doctrine\DBAL\Query\QueryBuilder - * @deprecated please use $this->getQueryBuilder() instead + * @deprecated 8.0.0 please use $this->getQueryBuilder() instead */ public function createQueryBuilder() { $backtrace = $this->getCallerBacktrace(); @@ -185,7 +300,7 @@ class Connection extends PrimaryReadReplicaConnection { * Gets the ExpressionBuilder for the connection. * * @return \Doctrine\DBAL\Query\Expression\ExpressionBuilder - * @deprecated please use $this->getQueryBuilder()->expr() instead + * @deprecated 8.0.0 please use $this->getQueryBuilder()->expr() instead */ public function getExpressionBuilder() { $backtrace = $this->getCallerBacktrace(); @@ -232,10 +347,10 @@ class Connection extends PrimaryReadReplicaConnection { if ($limit === -1 || $limit === null) { $limit = null; } else { - $limit = (int) $limit; + $limit = (int)$limit; } if ($offset !== null) { - $offset = (int) $offset; + $offset = (int)$offset; } if (!is_null($limit)) { $platform = $this->getDatabasePlatform(); @@ -252,10 +367,10 @@ class Connection extends PrimaryReadReplicaConnection { * If the query is parametrized, a prepared statement is used. * If an SQLLogger is configured, the execution is logged. * - * @param string $sql The SQL query to execute. - * @param array $params The parameters to bind to the query, if any. - * @param array $types The types the previous parameters are in. - * @param \Doctrine\DBAL\Cache\QueryCacheProfile|null $qcp The query cache profile, optional. + * @param string $sql The SQL query to execute. + * @param array $params The parameters to bind to the query, if any. + * @param array $types The types the previous parameters are in. + * @param \Doctrine\DBAL\Cache\QueryCacheProfile|null $qcp The query cache profile, optional. * * @return Result The executed statement. * @@ -284,7 +399,7 @@ class Connection extends PrimaryReadReplicaConnection { // Read to a table that has been written to previously // While this might not necessarily mean that we did a read after write it is an indication for a code path to check $this->logger->log( - (int) ($this->systemConfig->getValue('loglevel_dirty_database_queries', null) ?? 0), + (int)($this->systemConfig->getValue('loglevel_dirty_database_queries', null) ?? 0), 'dirty table reads: ' . $sql, [ 'tables' => array_keys($this->tableDirtyWrites), @@ -300,8 +415,13 @@ class Connection extends PrimaryReadReplicaConnection { $sql = $this->finishQuery($sql); $this->queriesExecuted++; - $this->logQueryToFile($sql); - return parent::executeQuery($sql, $params, $types, $qcp); + $this->logQueryToFile($sql, $params); + try { + return parent::executeQuery($sql, $params, $types, $qcp); + } catch (\Exception $e) { + $this->logDatabaseException($e); + throw $e; + } } /** @@ -318,10 +438,7 @@ class Connection extends PrimaryReadReplicaConnection { * @throws Exception */ public function executeUpdate(string $sql, array $params = [], array $types = []): int { - $sql = $this->finishQuery($sql); - $this->queriesExecuted++; - $this->logQueryToFile($sql); - return parent::executeUpdate($sql, $params, $types); + return $this->executeStatement($sql, $params, $types); } /** @@ -330,9 +447,9 @@ class Connection extends PrimaryReadReplicaConnection { * * This method supports PDO binding types as well as DBAL mapping types. * - * @param string $sql The SQL query. - * @param array $params The query parameters. - * @param array $types The parameter types. + * @param string $sql The SQL query. + * @param array $params The query parameters. + * @param array $types The parameter types. * * @return int The number of affected rows. * @@ -345,18 +462,28 @@ class Connection extends PrimaryReadReplicaConnection { } $sql = $this->finishQuery($sql); $this->queriesExecuted++; - $this->logQueryToFile($sql); - return (int)parent::executeStatement($sql, $params, $types); + $this->logQueryToFile($sql, $params); + try { + return (int)parent::executeStatement($sql, $params, $types); + } catch (\Exception $e) { + $this->logDatabaseException($e); + throw $e; + } } - protected function logQueryToFile(string $sql): void { + protected function logQueryToFile(string $sql, array $params): void { $logFile = $this->systemConfig->getValue('query_log_file'); if ($logFile !== '' && is_writable(dirname($logFile)) && (!file_exists($logFile) || is_writable($logFile))) { $prefix = ''; if ($this->systemConfig->getValue('query_log_file_requestid') === 'yes') { $prefix .= Server::get(IRequestId::class)->getId() . "\t"; } + $postfix = ''; + if ($this->systemConfig->getValue('query_log_file_parameters') === 'yes') { + $postfix .= '; ' . json_encode($params); + } + if ($this->systemConfig->getValue('query_log_file_backtrace') === 'yes') { $trace = debug_backtrace(DEBUG_BACKTRACE_IGNORE_ARGS); array_pop($trace); @@ -412,18 +539,28 @@ class Connection extends PrimaryReadReplicaConnection { * @param string $table The table name (will replace *PREFIX* with the actual prefix) * @param array $input data that should be inserted into the table (column name => value) * @param array|null $compare List of values that should be checked for "if not exists" - * If this is null or an empty array, all keys of $input will be compared - * Please note: text fields (clob) must not be used in the compare array + * If this is null or an empty array, all keys of $input will be compared + * Please note: text fields (clob) must not be used in the compare array * @return int number of inserted rows * @throws \Doctrine\DBAL\Exception * @deprecated 15.0.0 - use unique index and "try { $db->insert() } catch (UniqueConstraintViolationException $e) {}" instead, because it is more reliable and does not have the risk for deadlocks - see https://github.com/nextcloud/server/pull/12371 */ public function insertIfNotExist($table, $input, ?array $compare = null) { - return $this->adapter->insertIfNotExist($table, $input, $compare); + try { + return $this->adapter->insertIfNotExist($table, $input, $compare); + } catch (\Exception $e) { + $this->logDatabaseException($e); + throw $e; + } } public function insertIgnoreConflict(string $table, array $values) : int { - return $this->adapter->insertIgnoreConflict($table, $values); + try { + return $this->adapter->insertIgnoreConflict($table, $values); + } catch (\Exception $e) { + $this->logDatabaseException($e); + throw $e; + } } private function getType($value) { @@ -472,22 +609,22 @@ class Connection extends PrimaryReadReplicaConnection { foreach ($values as $name => $value) { $updateQb->set($name, $updateQb->createNamedParameter($value, $this->getType($value))); } - $where = $updateQb->expr()->andX(); + $where = []; $whereValues = array_merge($keys, $updatePreconditionValues); foreach ($whereValues as $name => $value) { if ($value === '') { - $where->add($updateQb->expr()->emptyString( + $where[] = $updateQb->expr()->emptyString( $name - )); + ); } else { - $where->add($updateQb->expr()->eq( + $where[] = $updateQb->expr()->eq( $name, $updateQb->createNamedParameter($value, $this->getType($value)), $this->getType($value) - )); + ); } } - $updateQb->where($where); + $updateQb->where($updateQb->expr()->andX(...$where)); $affected = $updateQb->executeStatement(); if ($affected === 0 && !empty($updatePreconditionValues)) { @@ -537,9 +674,9 @@ class Connection extends PrimaryReadReplicaConnection { $msg = $this->errorCode() . ': '; $errorInfo = $this->errorInfo(); if (!empty($errorInfo)) { - $msg .= 'SQLSTATE = '.$errorInfo[0] . ', '; - $msg .= 'Driver Code = '.$errorInfo[1] . ', '; - $msg .= 'Driver Message = '.$errorInfo[2]; + $msg .= 'SQLSTATE = ' . $errorInfo[0] . ', '; + $msg .= 'Driver Code = ' . $errorInfo[1] . ', '; + $msg .= 'Driver Message = ' . $errorInfo[2]; } return $msg; } @@ -561,13 +698,26 @@ class Connection extends PrimaryReadReplicaConnection { */ public function dropTable($table) { $table = $this->tablePrefix . trim($table); - $schema = $this->getSchemaManager(); + $schema = $this->createSchemaManager(); if ($schema->tablesExist([$table])) { $schema->dropTable($table); } } /** + * Truncate a table data if it exists + * + * @param string $table table name without the prefix + * @param bool $cascade whether to truncate cascading + * + * @throws Exception + */ + public function truncateTable(string $table, bool $cascade) { + $this->executeStatement($this->getDatabasePlatform() + ->getTruncateTableSQL($this->tablePrefix . trim($table), $cascade)); + } + + /** * Check if a table exists * * @param string $table table name without the prefix @@ -577,7 +727,7 @@ class Connection extends PrimaryReadReplicaConnection { */ public function tableExists($table) { $table = $this->tablePrefix . trim($table); - $schema = $this->getSchemaManager(); + $schema = $this->createSchemaManager(); return $schema->tablesExist([$table]); } @@ -585,7 +735,7 @@ class Connection extends PrimaryReadReplicaConnection { $statement = $this->replaceTablePrefix($statement); $statement = $this->adapter->fixupStatement($statement); if ($this->logRequestId) { - return $statement . " /* reqid: " . $this->requestId . " */"; + return $statement . ' /* reqid: ' . $this->requestId . ' */'; } else { return $statement; } @@ -662,6 +812,9 @@ class Connection extends PrimaryReadReplicaConnection { return $migrator->generateChangeScript($toSchema); } else { $migrator->migrate($toSchema); + foreach ($this->getShardConnections() as $shardConnection) { + $shardConnection->migrateToSchema($toSchema); + } } } @@ -682,6 +835,7 @@ class Connection extends PrimaryReadReplicaConnection { public function beginTransaction() { if (!$this->inTransaction()) { + $this->transactionBacktrace = debug_backtrace(DEBUG_BACKTRACE_IGNORE_ARGS); $this->transactionActiveSince = microtime(true); } return parent::beginTransaction(); @@ -691,9 +845,23 @@ class Connection extends PrimaryReadReplicaConnection { $result = parent::commit(); if ($this->getTransactionNestingLevel() === 0) { $timeTook = microtime(true) - $this->transactionActiveSince; + $this->transactionBacktrace = null; $this->transactionActiveSince = null; if ($timeTook > 1) { - $this->logger->debug('Transaction took ' . $timeTook . 's', ['exception' => new \Exception('Transaction took ' . $timeTook . 's')]); + $logLevel = match (true) { + $timeTook > 20 * 60 => ILogger::ERROR, + $timeTook > 5 * 60 => ILogger::WARN, + $timeTook > 10 => ILogger::INFO, + default => ILogger::DEBUG, + }; + $this->logger->log( + $logLevel, + 'Transaction took ' . $timeTook . 's', + [ + 'exception' => new \Exception('Transaction took ' . $timeTook . 's'), + 'timeSpent' => $timeTook, + ] + ); } } return $result; @@ -703,9 +871,23 @@ class Connection extends PrimaryReadReplicaConnection { $result = parent::rollBack(); if ($this->getTransactionNestingLevel() === 0) { $timeTook = microtime(true) - $this->transactionActiveSince; + $this->transactionBacktrace = null; $this->transactionActiveSince = null; if ($timeTook > 1) { - $this->logger->debug('Transaction rollback took longer than 1s: ' . $timeTook, ['exception' => new \Exception('Long running transaction rollback')]); + $logLevel = match (true) { + $timeTook > 20 * 60 => ILogger::ERROR, + $timeTook > 5 * 60 => ILogger::WARN, + $timeTook > 10 => ILogger::INFO, + default => ILogger::DEBUG, + }; + $this->logger->log( + $logLevel, + 'Transaction rollback took longer than 1s: ' . $timeTook, + [ + 'exception' => new \Exception('Long running transaction rollback'), + 'timeSpent' => $timeTook, + ] + ); } } return $result; @@ -713,9 +895,9 @@ class Connection extends PrimaryReadReplicaConnection { private function reconnectIfNeeded(): void { if ( - !isset($this->lastConnectionCheck[$this->getConnectionName()]) || - time() <= $this->lastConnectionCheck[$this->getConnectionName()] + 30 || - $this->isTransactionActive() + !isset($this->lastConnectionCheck[$this->getConnectionName()]) + || time() <= $this->lastConnectionCheck[$this->getConnectionName()] + 30 + || $this->isTransactionActive() ) { return; } @@ -732,4 +914,57 @@ class Connection extends PrimaryReadReplicaConnection { private function getConnectionName(): string { return $this->isConnectedToPrimary() ? 'primary' : 'replica'; } + + /** + * @return IDBConnection::PLATFORM_MYSQL|IDBConnection::PLATFORM_ORACLE|IDBConnection::PLATFORM_POSTGRES|IDBConnection::PLATFORM_SQLITE|IDBConnection::PLATFORM_MARIADB + */ + public function getDatabaseProvider(bool $strict = false): string { + $platform = $this->getDatabasePlatform(); + if ($strict && $platform instanceof MariaDBPlatform) { + return IDBConnection::PLATFORM_MARIADB; + } elseif ($platform instanceof MySQLPlatform) { + return IDBConnection::PLATFORM_MYSQL; + } elseif ($platform instanceof OraclePlatform) { + return IDBConnection::PLATFORM_ORACLE; + } elseif ($platform instanceof PostgreSQLPlatform) { + return IDBConnection::PLATFORM_POSTGRES; + } elseif ($platform instanceof SqlitePlatform) { + return IDBConnection::PLATFORM_SQLITE; + } else { + throw new \Exception('Database ' . $platform::class . ' not supported'); + } + } + + /** + * @internal Should only be used inside the QueryBuilder, ExpressionBuilder and FunctionBuilder + * All apps and API code should not need this and instead use provided functionality from the above. + */ + public function getServerVersion(): string { + /** @var ServerInfoAwareConnection $this->_conn */ + return $this->_conn->getServerVersion(); + } + + /** + * Log a database exception if enabled + * + * @param \Exception $exception + * @return void + */ + public function logDatabaseException(\Exception $exception): void { + if ($this->logDbException) { + if ($exception instanceof Exception\UniqueConstraintViolationException) { + $this->logger->info($exception->getMessage(), ['exception' => $exception, 'transaction' => $this->transactionBacktrace]); + } else { + $this->logger->error($exception->getMessage(), ['exception' => $exception, 'transaction' => $this->transactionBacktrace]); + } + } + } + + public function getShardDefinition(string $name): ?ShardDefinition { + return $this->shards[$name] ?? null; + } + + public function getCrossShardMoveHelper(): CrossShardMoveHelper { + return new CrossShardMoveHelper($this->shardConnectionManager); + } } |