* @var integer Number of write queries for the current transaction
*/
private $mTrxWriteQueryCount = 0;
+ /**
+ * @var integer Number of rows affected by write queries for the current transaction
+ */
+ private $mTrxWriteAffectedRows = 0;
/**
* @var float Like mTrxWriteQueryCount but excludes lock-bound, easy to replicate, queries
*/
return $this->mTrxLevel ? $this->mTrxWriteCallers : [];
}
+ public function pendingWriteRowsAffected() {
+ return $this->mTrxWriteAffectedRows;
+ }
+
/**
* Get the list of method names that have pending write queries or callbacks
* for this transaction
if ( $ret !== false ) {
$this->lastPing = $startTime;
if ( $isWrite && $this->mTrxLevel ) {
- $this->updateTrxWriteQueryTime( $sql, $queryRuntime );
+ $this->updateTrxWriteQueryTime( $sql, $queryRuntime, $this->affectedRows() );
$this->mTrxWriteCallers[] = $fname;
}
}
*
* @param string $sql A SQL write query
* @param float $runtime Total runtime, including RTT
+ * @param integer $affected Affected row count
*/
- private function updateTrxWriteQueryTime( $sql, $runtime ) {
+ private function updateTrxWriteQueryTime( $sql, $runtime, $affected ) {
// Whether this is indicative of replica DB runtime (except for RBR or ws_repl)
$indicativeOfReplicaRuntime = true;
if ( $runtime > self::SLOW_WRITE_SEC ) {
$this->mTrxWriteDuration += $runtime;
$this->mTrxWriteQueryCount += 1;
+ $this->mTrxWriteAffectedRows += $affected;
if ( $indicativeOfReplicaRuntime ) {
$this->mTrxWriteAdjDuration += $runtime;
$this->mTrxWriteAdjQueryCount += 1;
}
public function selectField(
- $table, $var, $cond = '', $fname = __METHOD__, $options = []
+ $table, $var, $cond = '', $fname = __METHOD__, $options = [], $join_conds = []
) {
if ( $var === '*' ) { // sanity
throw new DBUnexpectedError( $this, "Cannot use a * field: got '$var'" );
$options['LIMIT'] = 1;
- $res = $this->select( $table, $var, $cond, $fname, $options );
+ $res = $this->select( $table, $var, $cond, $fname, $options, $join_conds );
if ( $res === false || !$this->numRows( $res ) ) {
return false;
}
}
# Split database and table into proper variables.
- # We reverse the explode so that database.table and table both output
- # the correct table.
+ list( $database, $schema, $prefix, $table ) = $this->qualifiedTableComponents( $name );
+
+ # Quote $table and apply the prefix if not quoted.
+ # $tableName might be empty if this is called from Database::replaceVars()
+ $tableName = "{$prefix}{$table}";
+ if ( $format === 'quoted'
+ && !$this->isQuotedIdentifier( $tableName )
+ && $tableName !== ''
+ ) {
+ $tableName = $this->addIdentifierQuotes( $tableName );
+ }
+
+ # Quote $schema and $database and merge them with the table name if needed
+ $tableName = $this->prependDatabaseOrSchema( $schema, $tableName, $format );
+ $tableName = $this->prependDatabaseOrSchema( $database, $tableName, $format );
+
+ return $tableName;
+ }
+
+ /**
+ * Get the table components needed for a query given the currently selected database
+ *
+ * @param string $name Table name in the form of db.schema.table, db.table, or table
+ * @return array (DB name or "" for default, schema name, table prefix, table name)
+ */
+ protected function qualifiedTableComponents( $name ) {
+ # We reverse the explode so that database.table and table both output the correct table.
$dbDetails = explode( '.', $name, 3 );
if ( count( $dbDetails ) == 3 ) {
list( $database, $schema, $table ) = $dbDetails;
}
}
- # Quote $table and apply the prefix if not quoted.
- # $tableName might be empty if this is called from Database::replaceVars()
- $tableName = "{$prefix}{$table}";
- if ( $format === 'quoted'
- && !$this->isQuotedIdentifier( $tableName )
- && $tableName !== ''
- ) {
- $tableName = $this->addIdentifierQuotes( $tableName );
- }
-
- # Quote $schema and $database and merge them with the table name if needed
- $tableName = $this->prependDatabaseOrSchema( $schema, $tableName, $format );
- $tableName = $this->prependDatabaseOrSchema( $database, $tableName, $format );
-
- return $tableName;
+ return [ $database, $schema, $prefix, $table ];
}
/**
public function insertSelect(
$destTable, $srcTable, $varMap, $conds,
- $fname = __METHOD__, $insertOptions = [], $selectOptions = []
+ $fname = __METHOD__, $insertOptions = [], $selectOptions = [], $selectJoinConds = []
) {
if ( $this->cliMode ) {
// For massive migrations with downtime, we don't want to select everything
$conds,
$fname,
$insertOptions,
- $selectOptions
+ $selectOptions,
+ $selectJoinConds
);
}
$fields[] = $this->fieldNameWithAlias( $sourceColumnOrSql, $dstColumn );
}
$selectOptions[] = 'FOR UPDATE';
- $res = $this->select( $srcTable, implode( ',', $fields ), $conds, $fname, $selectOptions );
+ $res = $this->select(
+ $srcTable, implode( ',', $fields ), $conds, $fname, $selectOptions, $selectJoinConds
+ );
if ( !$res ) {
return false;
}
*/
protected function nativeInsertSelect( $destTable, $srcTable, $varMap, $conds,
$fname = __METHOD__,
- $insertOptions = [], $selectOptions = []
+ $insertOptions = [], $selectOptions = [], $selectJoinConds = []
) {
$destTable = $this->tableName( $destTable );
$insertOptions = $this->makeInsertOptions( $insertOptions );
- if ( !is_array( $selectOptions ) ) {
- $selectOptions = [ $selectOptions ];
- }
-
- list( $startOpts, $useIndex, $tailOpts, $ignoreIndex ) = $this->makeSelectOptions(
- $selectOptions );
-
- if ( is_array( $srcTable ) ) {
- $srcTable = implode( ',', array_map( [ $this, 'tableName' ], $srcTable ) );
- } else {
- $srcTable = $this->tableName( $srcTable );
- }
+ $selectSql = $this->selectSQLText(
+ $srcTable,
+ array_values( $varMap ),
+ $conds,
+ $fname,
+ $selectOptions,
+ $selectJoinConds
+ );
$sql = "INSERT $insertOptions" .
- " INTO $destTable (" . implode( ',', array_keys( $varMap ) ) . ')' .
- " SELECT $startOpts " . implode( ',', $varMap ) .
- " FROM $srcTable $useIndex $ignoreIndex ";
-
- if ( $conds != '*' ) {
- if ( is_array( $conds ) ) {
- $conds = $this->makeList( $conds, self::LIST_AND );
- }
- $sql .= " WHERE $conds";
- }
-
- $sql .= " $tailOpts";
+ " INTO $destTable (" . implode( ',', array_keys( $varMap ) ) . ') ' .
+ $selectSql;
return $this->query( $sql, $fname );
}
return '(' . implode( $glue, $sqls ) . ')';
}
+ public function unionConditionPermutations(
+ $table, $vars, array $permute_conds, $extra_conds = '', $fname = __METHOD__,
+ $options = [], $join_conds = []
+ ) {
+ // First, build the Cartesian product of $permute_conds
+ $conds = [ [] ];
+ foreach ( $permute_conds as $field => $values ) {
+ if ( !$values ) {
+ // Skip empty $values
+ continue;
+ }
+ $values = array_unique( $values ); // For sanity
+ $newConds = [];
+ foreach ( $conds as $cond ) {
+ foreach ( $values as $value ) {
+ $cond[$field] = $value;
+ $newConds[] = $cond; // Arrays are by-value, not by-reference, so this works
+ }
+ }
+ $conds = $newConds;
+ }
+
+ $extra_conds = $extra_conds === '' ? [] : (array)$extra_conds;
+
+ // If there's just one condition and no subordering, hand off to
+ // selectSQLText directly.
+ if ( count( $conds ) === 1 &&
+ ( !isset( $options['INNER ORDER BY'] ) || !$this->unionSupportsOrderAndLimit() )
+ ) {
+ return $this->selectSQLText(
+ $table, $vars, $conds[0] + $extra_conds, $fname, $options, $join_conds
+ );
+ }
+
+ // Otherwise, we need to pull out the order and limit to apply after
+ // the union. Then build the SQL queries for each set of conditions in
+ // $conds. Then union them together (using UNION ALL, because the
+ // product *should* already be distinct).
+ $orderBy = $this->makeOrderBy( $options );
+ $limit = isset( $options['LIMIT'] ) ? $options['LIMIT'] : null;
+ $offset = isset( $options['OFFSET'] ) ? $options['OFFSET'] : false;
+ $all = empty( $options['NOTALL'] ) && !in_array( 'NOTALL', $options );
+ if ( !$this->unionSupportsOrderAndLimit() ) {
+ unset( $options['ORDER BY'], $options['LIMIT'], $options['OFFSET'] );
+ } else {
+ if ( array_key_exists( 'INNER ORDER BY', $options ) ) {
+ $options['ORDER BY'] = $options['INNER ORDER BY'];
+ }
+ if ( $limit !== null && is_numeric( $offset ) && $offset != 0 ) {
+ // We need to increase the limit by the offset rather than
+ // using the offset directly, otherwise it'll skip incorrectly
+ // in the subqueries.
+ $options['LIMIT'] = $limit + $offset;
+ unset( $options['OFFSET'] );
+ }
+ }
+
+ $sqls = [];
+ foreach ( $conds as $cond ) {
+ $sqls[] = $this->selectSQLText(
+ $table, $vars, $cond + $extra_conds, $fname, $options, $join_conds
+ );
+ }
+ $sql = $this->unionQueries( $sqls, $all ) . $orderBy;
+ if ( $limit !== null ) {
+ $sql = $this->limitResult( $sql, $limit, $offset );
+ }
+
+ return $sql;
+ }
+
public function conditional( $cond, $trueVal, $falseVal ) {
if ( is_array( $cond ) ) {
$cond = $this->makeList( $cond, self::LIST_AND );
}
final public function onTransactionPreCommitOrIdle( callable $callback, $fname = __METHOD__ ) {
- if ( $this->mTrxLevel ) {
+ if ( $this->mTrxLevel || $this->getFlag( self::DBO_TRX ) ) {
+ // As long as DBO_TRX is set, writes will accumulate until the load balancer issues
+ // an implicit commit of all peer databases. This is true even if a transaction has
+ // not yet been triggered by writes; make sure $callback runs *after* any such writes.
$this->mTrxPreCommitCallbacks[] = [ $callback, $fname ];
} else {
- // If no transaction is active, then make one for this callback
+ // No transaction is active nor will start implicitly, so make one for this callback
$this->startAtomic( __METHOD__ );
try {
call_user_func( $callback );
$this->mTrxShortId = sprintf( '%06x', mt_rand( 0, 0xffffff ) );
$this->mTrxWriteDuration = 0.0;
$this->mTrxWriteQueryCount = 0;
+ $this->mTrxWriteAffectedRows = 0;
$this->mTrxWriteAdjDuration = 0.0;
$this->mTrxWriteAdjQueryCount = 0;
$this->mTrxWriteCallers = [];
if ( $this->mTrxDoneWrites ) {
$this->mLastWriteTime = microtime( true );
$this->trxProfiler->transactionWritingOut(
- $this->mServer, $this->mDBname, $this->mTrxShortId, $writeTime );
+ $this->mServer,
+ $this->mDBname,
+ $this->mTrxShortId,
+ $writeTime,
+ $this->mTrxWriteAffectedRows
+ );
}
$this->runOnTransactionIdleCallbacks( self::TRIGGER_COMMIT );
$this->mTrxAtomicLevels = [];
if ( $this->mTrxDoneWrites ) {
$this->trxProfiler->transactionWritingOut(
- $this->mServer, $this->mDBname, $this->mTrxShortId );
+ $this->mServer,
+ $this->mDBname,
+ $this->mTrxShortId
+ );
}
$this->mTrxIdleCallbacks = []; // clear
dépôts / lhc / web / wiklou.git / blobdiff