· 9 years ago · Apr 11, 2017, 07:12 PM
1######################################################################
2#
3# EPrints::Database::mysql
4#
5######################################################################
6#
7#
8######################################################################
9
10
11=pod
12
13=for Pod2Wiki
14
15=head1 NAME
16
17B<EPrints::Database::mysql> - custom database methods for MySQL DB
18
19=head1 SYNOPSIS
20
21 $c->{dbdriver} = 'mysql';
22 # $c->{dbhost} = 'localhost';
23 # $c->{dbport} = '3316';
24 $c->{dbname} = 'myrepo';
25 $c->{dbuser} = 'bob';
26 $c->{dbpass} = 'asecret';
27 # $c->{dbengine} = 'InnoDB';
28
29=head1 DESCRIPTION
30
31MySQL database wrapper.
32
33Foreign keys will be defined if you use a DB engine that supports them (e.g. InnoDB).
34
35=head2 MySQL-specific Annoyances
36
37MySQL does not support sequences.
38
39MySQL is (by default) lax about truncation.
40
41=head1 METHODS
42
43=over 4
44
45=cut
46
47######################################################################
48#
49# INSTANCE VARIABLES:
50#
51# $self->{session}
52# The EPrints::Session which is associated with this database
53# connection.
54#
55# $self->{debug}
56# If true then SQL is logged.
57#
58# $self->{dbh}
59# The handle on the actual database connection.
60#
61######################################################################
62
63package EPrints::Database::mysql;
64
65use EPrints;
66
67use EPrints::Database qw( :sql_types );
68@ISA = qw( EPrints::Database );
69
70our $I18L = {
71 en => {
72 collate => "utf8_general_ci",
73 },
74 de => {
75 collate => "utf8_unicode_ci",
76 },
77};
78
79use strict;
80
81######################################################################
82=pod
83
84=item $version = $db->get_server_version
85
86Return the database server version.
87
88=cut
89######################################################################
90
91sub get_server_version
92{
93 my( $self ) = @_;
94
95 my $sql = "SELECT VERSION();";
96 my( $version ) = $self->{dbh}->selectrow_array( $sql );
97 return "MySQL $version";
98}
99
100sub mysql_version_from_dbh
101{
102 my( $dbh ) = @_;
103 my $sql = "SELECT VERSION();";
104 my( $version ) = $dbh->selectrow_array( $sql );
105 $version =~ m/^(\d+).(\d+).(\d+)/;
106 return $1*10000+$2*100+$3;
107}
108
109sub create
110{
111 my( $self, $username, $password ) = @_;
112
113 my $repo = $self->{session}->get_repository;
114
115 my $dbh = DBI->connect( EPrints::Database::build_connection_string(
116 dbdriver => "mysql",
117 dbhost => $repo->get_conf("dbhost"),
118 dbsock => $repo->get_conf("dbsock"),
119 dbport => $repo->get_conf("dbport"),
120 dbname => "mysql", ),
121 $username,
122 $password,
123 {
124 AutoCommit => 1,
125 RaiseError => 0,
126 PrintError => 0,
127 } );
128
129 return undef if !defined $dbh;
130
131 $dbh->{RaiseError} = 1;
132
133 my $dbuser = $repo->get_conf( "dbuser" );
134 my $dbpass = $repo->get_conf( "dbpass" );
135 my $dbname = $repo->get_conf( "dbname" );
136
137 my $rc = 1;
138
139 $rc &&= $dbh->do( "CREATE DATABASE IF NOT EXISTS ".$dbh->quote_identifier( $dbname )." DEFAULT CHARACTER SET ".$dbh->quote( $self->get_default_charset ) );
140
141 $rc &&= $dbh->do( "GRANT ALL PRIVILEGES ON ".$dbh->quote_identifier( $dbname ).".* TO ".$dbh->quote_identifier( $dbuser )."\@".$dbh->quote("localhost")." IDENTIFIED BY ".$dbh->quote( $dbpass ) );
142
143 $dbh->disconnect;
144
145 $self->connect();
146
147 return 0 if !defined $self->{dbh};
148
149 return $rc;
150}
151
152######################################################################
153=pod
154
155=item $n = $db->create_counters()
156
157Create and initialise the counters.
158
159=cut
160######################################################################
161
162sub create_counters
163{
164 my( $self ) = @_;
165
166 my $counter_ds = $self->{session}->get_repository->get_dataset( "counter" );
167
168 # The table creation SQL
169 my $table = $counter_ds->get_sql_table_name;
170
171 my $rc = $self->_create_table( $table, ["countername"], [
172 $self->get_column_type( "countername", SQL_VARCHAR, SQL_NOT_NULL , 255 ),
173 $self->get_column_type( "counter", SQL_INTEGER, SQL_NOT_NULL ),
174 ]);
175
176 $rc &&= $self->SUPER::create_counters();
177
178 # Everything OK
179 return $rc;
180}
181
182######################################################################
183=pod
184
185=item $boolean = $db->has_table( $tablename )
186
187Return true if the a table of the given name exists in the database.
188
189=cut
190######################################################################
191
192sub has_table
193{
194 my( $self, $tablename ) = @_;
195
196 my $sth = $self->prepare("SHOW TABLES LIKE ".$self->quote_value($tablename));
197 $sth->execute;
198 my $rc = defined $sth->fetch ? 1 : 0;
199 $sth->finish;
200
201 return $rc;
202}
203
204######################################################################
205=pod
206
207=item $boolean = $db->has_column( $tablename, $columnname )
208
209Return true if the a table of the given name has a column named $columnname in the database.
210
211=cut
212######################################################################
213
214sub has_column
215{
216 my( $self, $table, $column ) = @_;
217
218 my $rc = 0;
219
220 my $sth = $self->{dbh}->column_info( undef, undef, $table, '%' );
221 while(!$rc && (my $row = $sth->fetch))
222 {
223 my $column_name = $row->[$sth->{NAME_lc_hash}{column_name}];
224 $rc = 1 if $column_name eq $column;
225 }
226 $sth->finish;
227
228 return $rc;
229}
230
231sub connect
232{
233 my( $self ) = @_;
234
235 my $rc = $self->SUPER::connect();
236
237 if( $rc )
238 {
239 # always try to reconnect
240 $self->{dbh}->{mysql_auto_reconnect} = 1;
241
242 $self->do("SET NAMES 'utf8'");
243 }
244 elsif( $DBI::err == 1040 )
245 {
246 EPrints->abort( "Error connecting to MySQL server: $DBI::errstr. To fix this increase max_connections in my.cnf:\n\n[mysqld]\nmax_connections=300\n" );
247 }
248
249 return $rc;
250}
251
252######################################################################
253=pod
254
255=item $success = $db->has_counter( $counter )
256
257Returns true if $counter exists.
258
259=cut
260######################################################################
261
262sub has_counter
263{
264 my( $self, $name ) = @_;
265
266 my $sql = "SELECT 1 FROM `counters` WHERE `countername`=".$self->quote_value( $name );
267
268 my $sth = $self->prepare($sql);
269 $self->execute( $sth, $sql );
270
271 return defined $sth->fetch;
272}
273
274sub create_counter
275{
276 my( $self, $name ) = @_;
277
278 return $self->insert( "counters", ["countername", "counter"], [$name, 0] );
279}
280
281sub drop_counter
282{
283 my( $self, $name ) = @_;
284
285 return $self->delete_from( "counters", ["countername"], [$name] );
286}
287
288sub remove_counters
289{
290 my( $self ) = @_;
291
292 my $counter_ds = $self->{session}->get_repository->get_dataset( "counter" );
293 my $table = $counter_ds->get_sql_table_name;
294
295 $self->drop_table( $table );
296}
297
298######################################################################
299=pod
300
301=item $n = $db->counter_next( $counter )
302
303Return the next unused value for the named counter. Returns undef if
304the counter doesn't exist.
305
306=cut
307######################################################################
308
309sub counter_next
310{
311 my( $self, $counter ) = @_;
312
313 my $ds = $self->{session}->get_repository->get_dataset( "counter" );
314
315 # Update the counter
316 my $table = $ds->get_sql_table_name;
317 my $sql = "UPDATE ".$self->quote_identifier($table)." SET counter=".
318 "LAST_INSERT_ID(counter+1) WHERE ".$self->quote_identifier("countername")." = ".$self->quote_value($counter);
319
320 # Send to the database
321 my $rows_affected = $self->do( $sql );
322
323 # Return with an error if unsuccessful
324 return( undef ) unless( $rows_affected==1 );
325
326 # Get the value of the counter
327 $sql = "SELECT LAST_INSERT_ID();";
328 my @row = $self->{dbh}->selectrow_array( $sql );
329
330 return( $row[0] );
331}
332
333######################################################################
334=pod
335
336=item $db->counter_minimum( $counter, $value )
337
338Ensure that the counter is set no lower that $value. This is used when
339importing eprints which may not be in scrict sequence.
340
341=cut
342######################################################################
343
344sub counter_minimum
345{
346 my( $self, $counter, $value ) = @_;
347
348 my $ds = $self->{session}->get_repository->get_dataset( "counter" );
349
350 $value+=0; # ensure numeric!
351
352 # Update the counter to be at least $value
353 my $sql = "UPDATE ".$ds->get_sql_table_name()." SET counter="
354 . "CASE WHEN $value>counter THEN $value ELSE counter END"
355 . " WHERE countername = ".$self->quote_value($counter);
356 $self->do( $sql );
357}
358
359
360######################################################################
361=pod
362
363=item $db->counter_reset( $counter )
364
365Return the counter. Use with cautiuon.
366
367=cut
368######################################################################
369
370sub counter_reset
371{
372 my( $self, $counter ) = @_;
373
374 my $ds = $self->{session}->get_repository->get_dataset( "counter" );
375
376 # Update the counter
377 my $sql = "UPDATE ".$ds->get_sql_table_name()." ";
378 $sql.="SET counter=0 WHERE countername = ".$self->quote_value($counter);
379
380 # Send to the database
381 $self->do( $sql );
382}
383
384sub _cache_from_SELECT
385{
386 my( $self, $cachemap, $dataset, $select_sql ) = @_;
387
388 my $cache_table = $cachemap->get_sql_table_name;
389 my $Q_pos = $self->quote_identifier( "pos" );
390 my $key_field = $dataset->get_key_field();
391 my $Q_keyname = $self->quote_identifier($key_field->get_sql_name);
392
393 $self->do("SET \@i=0");
394
395 my $sql = "";
396 $sql .= "INSERT INTO ".$self->quote_identifier( $cache_table );
397 $sql .= "($Q_pos, $Q_keyname)";
398 $sql .= " SELECT \@i:=\@i+1, $Q_keyname";
399 # MariaDB does not order sub-queries unless limited. Using limit of 2^31-1 in case any system is using a signed 32-bit integer.
400 my $limit = " LIMIT 2147483647";
401 $limit = "" if $select_sql =~ /LIMIT/;
402 $sql .= " FROM ($select_sql$limit) ".$self->quote_identifier( "S" );
403
404 $self->do( $sql );
405}
406
407sub get_default_charset { "utf8" }
408
409sub get_default_collation
410{
411 my( $self, $langid ) = @_;
412
413 return "utf8_bin";
414}
415
416# Not supported by DBD::mysql?
417sub get_primary_key
418{
419 my( $self, $table ) = @_;
420
421 my $sth = $self->prepare( "DESCRIBE ".$self->quote_identifier($table) );
422 $sth->execute;
423
424 my @COLS;
425 while(my $row = $sth->fetch)
426 {
427 push @COLS, $row->[0] if $row->[3] eq 'PRI';
428 }
429
430 return @COLS;
431}
432
433sub get_number_of_keys
434{
435 my( $self, $table ) = @_;
436 my $sth = $self->prepare( "DESCRIBE ".$self->quote_identifier($table) );
437 $sth->execute;
438
439 my $NUM_KEYS = 0;
440 while(my $row = $sth->fetch)
441 {
442 if ( $row->[3] eq 'PRI' or $row->[3] eq 'MUL' or $row->[3] eq 'UNI' )
443 {
444 ++$NUM_KEYS;
445 }
446 }
447 return $NUM_KEYS;
448}
449
450sub get_column_collation
451{
452 my( $self, $table, $column ) = @_;
453
454 my $sth = $self->prepare( "SHOW FULL COLUMNS FROM ".$self->quote_identifier($table)." LIKE ".$self->quote_value($column) );
455 $sth->execute;
456
457 my $collation;
458 while(my $row = $sth->fetch)
459 {
460 $collation = $row->[$sth->{NAME_lc_hash}{"collation"}];
461 }
462
463 return $collation;
464}
465
466# We'll do quote here, because DBD::mysql::quote_identifier is really slow
467sub quote_identifier
468{
469 my( $self, @parts ) = @_;
470
471 # we shouldn't get identifiers with '`' in
472 return join(".", map {
473 $_ =~ m/`/ ?
474 EPrints::abort "Bad character in database identifier: $_" :
475 "`$_`"
476 } @parts);
477}
478
479sub _rename_table_field
480{
481 my( $self, $table, $field, $old_name ) = @_;
482
483 my $rc = 1;
484
485 my @names = $field->get_sql_names;
486 my @types = $field->get_sql_type( $self->{session} );
487
488 # work out what the old columns are called
489 my @old_names;
490 {
491 local $field->{name} = $old_name;
492 @old_names = $field->get_sql_names;
493 }
494
495 my @column_sql;
496 for(my $i = 0; $i < @names; ++$i)
497 {
498 push @column_sql, sprintf("CHANGE %s %s",
499 $self->quote_identifier($old_names[$i]),
500 $types[$i]
501 );
502 }
503
504 $rc &&= $self->do( "ALTER TABLE ".$self->quote_identifier($table)." ".join(",", @column_sql));
505
506 return $rc;
507}
508
509sub _rename_field_ordervalues_lang
510{
511 my( $self, $dataset, $field, $old_name, $langid ) = @_;
512
513 my $order_table = $dataset->get_ordervalues_table_name( $langid );
514
515 my $sql_field = $field->create_ordervalues_field( $self->{session}, $langid );
516
517 my( $col ) = $sql_field->get_sql_type( $self->{session} );
518
519 my $sql = sprintf("ALTER TABLE %s CHANGE %s %s",
520 $self->quote_identifier($order_table),
521 $self->quote_identifier($old_name),
522 $col
523 );
524
525 return $self->do( $sql );
526}
527
528sub prepare_regexp
529{
530 my( $self, $col, $value ) = @_;
531
532 return "$col REGEXP $value";
533}
534
535sub sql_LIKE
536{
537 my( $self ) = @_;
538
539 return " COLLATE utf8_general_ci LIKE ";
540}
541
542# This is a hacky method to support CI username/email lookups. Should be
543# implemented as an option on searching (bigger change of search mechanisms?).
544
545sub ci_lookup
546{
547 my( $self, $field, $value ) = @_;
548
549 return if !defined $value; # Can't do a CI match on 'NULL'
550
551 my $table = $field->dataset->get_sql_table_name;
552
553 my $sql =
554 "SELECT ".$self->quote_identifier( $field->get_sql_name ).
555 " FROM ".$self->quote_identifier( $table ).
556 " WHERE ".$self->quote_identifier( $field->get_sql_name )."=".$self->quote_value( $value )." COLLATE utf8_general_ci";
557
558 my $sth = $self->prepare( $sql );
559 $self->execute( $sth, $sql );
560
561 my( $real_value ) = $sth->fetchrow_array;
562
563 $sth->finish;
564
565 return defined $real_value ? $real_value : $value;
566}
567
568sub duplicate_error { $DBI::err == 1062 }
569sub retry_error { $DBI::err == 2006 }
570
571sub type_info
572{
573 my( $self, $data_type ) = @_;
574
575 if( $data_type eq SQL_CLOB )
576 {
577 return {
578 TYPE_NAME => "longtext",
579 CREATE_PARAMS => "",
580 COLUMN_SIZE => 2 ** 31,
581 };
582 }
583 else
584 {
585 return $self->SUPER::type_info( $data_type );
586 }
587}
588
589# use MySQL 4.0 compatible "SHOW INDEX"
590# This method gets the entire SHOW INDEX response and builds a look-up table of
591# keys with their *ordered* columns. This ensures even if MySQL is weird and
592# returns out of order results we won't break.
593sub index_name
594{
595 my( $self, $table, @cols ) = @_;
596
597 my $hash = sub { join ':', map { $self->quote_identifier( $_ ) } @_ };
598
599 my $needle = &$hash( @cols );
600 my %indexes;
601
602 my $sth = $self->prepare("SHOW INDEX FROM ".$self->quote_identifier( $table ));
603 $sth->execute;
604
605 my( $key_name, $seq, $col_name );
606 $sth->bind_col( $sth->{NAME_uc_hash}->{KEY_NAME} + 1, \$key_name );
607 $sth->bind_col( $sth->{NAME_uc_hash}->{SEQ_IN_INDEX} + 1, \$seq );
608 $sth->bind_col( $sth->{NAME_uc_hash}->{COLUMN_NAME} + 1, \$col_name );
609
610 while($sth->fetch)
611 {
612 $indexes{$key_name} ||= [];
613 $indexes{$key_name}[$seq - 1] = $col_name;
614 }
615 foreach $key_name (keys %indexes)
616 {
617 return $key_name if
618 $needle eq &$hash( @{$indexes{$key_name}} );
619 }
620
621 return undef;
622}
623
624sub _create_table
625{
626 my( $self, $table, $primary_key, $columns ) = @_;
627
628 my $sql = "";
629
630 $sql .= "CREATE TABLE ".$self->quote_identifier($table)." (";
631 $sql .= join(', ', @$columns);
632 if( @$primary_key )
633 {
634 $sql .= ", PRIMARY KEY(".join(', ', map { $self->quote_identifier($_) } @$primary_key).")";
635 }
636 $sql .= ")";
637 $sql .= " DEFAULT CHARSET=".$self->get_default_charset;
638
639 my $engine = $self->{session}->config( "dbengine" );
640 $sql .= " ENGINE=$engine" if $engine;
641
642 return $self->do($sql);
643}
644
6451; # For use/require success
646
647######################################################################
648=pod
649
650=back
651
652=cut
653
654
655=head1 COPYRIGHT
656
657=for COPYRIGHT BEGIN
658
659Copyright 2000-2011 University of Southampton.
660
661=for COPYRIGHT END
662
663=for LICENSE BEGIN
664
665This file is part of EPrints L<http://www.eprints.org/>.
666
667EPrints is free software: you can redistribute it and/or modify it
668under the terms of the GNU Lesser General Public License as published
669by the Free Software Foundation, either version 3 of the License, or
670(at your option) any later version.
671
672EPrints is distributed in the hope that it will be useful, but WITHOUT
673ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
674FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
675License for more details.
676
677You should have received a copy of the GNU Lesser General Public
678License along with EPrints. If not, see L<http://www.gnu.org/licenses/>.
679
680=for LICENSE END