root/branches/release-38/lib/MT/ObjectDriver/DDL/Pg.pm @ 2387

Revision 2387, 9.4 kB (checked in by takayama, 19 months ago)

Fixed BugId:79636
* Added drop_index_sql for Pg
  • Property svn:keywords set to Id Revision
Line 
1# Movable Type (r) Open Source (C) 2001-2008 Six Apart, Ltd.
2# This program is distributed under the terms of the
3# GNU General Public License, version 2.
4#
5# $Id$
6
7package MT::ObjectDriver::DDL::Pg;
8
9use strict;
10use warnings;
11use base qw( MT::ObjectDriver::DDL );
12
13sub can_add_column { 1 }
14sub can_drop_column { 1 }
15sub can_alter_column { 0 }
16
17sub index_defs {
18    my $ddl = shift;
19    my ($class) = @_;
20    my $driver = $class->driver;
21    my $dbh = $driver->r_handle;
22    my $field_prefix = $class->datasource;
23    my $table_name = $class->table_name;
24    my $sth = $dbh->prepare(<<SQL)
25SELECT cidx.relname as index_name, idx.indisunique, idx.indisprimary, idx.indnatts, idx.indkey, am.amname
26FROM pg_index idx
27INNER JOIN pg_class cidx ON idx.indexrelid = cidx.oid
28INNER JOIN pg_class ctbl ON idx.indrelid = ctbl.oid
29INNER JOIN pg_am am ON cidx.relam = am.oid
30WHERE ctbl.relname = '$table_name'
31SQL
32        or return undef;
33    $sth->execute or return undef;
34
35    my $defs = {};
36    while (my $row = $sth->fetchrow_hashref) {
37        next if 1 == $row->{'indisprimary'};
38
39        my $key = $row->{'index_name'};
40        next unless $key =~ m/^(mt_)?\Q$field_prefix\E_/;
41        $key = 'mt_' . $key unless $key =~ m/^mt_/;
42
43        my $type = $row->{'amname'};
44        # ignore fulltext or other unrecognized indexes for now
45        next unless $type eq 'btree';
46
47        my $is_unique = $row->{'indisunique'};
48        $key =~ s/^mt_\Q$field_prefix\E_//;
49
50        my $indkeys = $row->{'indkey'};
51        $indkeys =~ s/\s+/,/g;
52
53        my $sth_att = $dbh->prepare(<<ATTSQL)
54SELECT attnum, attname
55FROM pg_attribute att
56INNER JOIN pg_class ctbl ON att.attrelid = ctbl.oid
57WHERE att.attnum IN ($indkeys)
58AND ctbl.relname = '$table_name'
59ATTSQL
60            or next;
61        $sth_att->execute or next;
62        my $row_att = $sth_att->fetchall_hashref('attnum');
63        $sth_att->finish;
64
65        my $cols;
66        if ( 1 == $row->{'indnatts'} ) {
67            # $indkeys have column's attnum
68            my $col = $row_att->{$indkeys}->{'attname'};
69            $col =~ s/^\Q$field_prefix\E_//;
70            $cols = [ $col ];
71        }
72        else {
73            my @cols;
74            for my $indkey ( split ',', $indkeys ) {
75                my $col = $row_att->{$indkey}->{'attname'};
76                $col =~ s/^\Q$field_prefix\E_//;
77                push @cols, $col;
78            }
79            $cols = \@cols;
80        }
81        if ( $is_unique ) {
82            $defs->{$key} = { 'unique' => 1, 'columns' => $cols };
83        }
84        else {
85            if ((@$cols == 1) && ($key eq $cols->[0])) {
86                $defs->{$key} = 1;
87            } else {
88                $defs->{$key} = { 'columns' => $cols };
89            }
90        }
91    }
92    $sth->finish;
93    return $defs;
94}
95
96sub column_defs {
97    my $ddl = shift;
98    my ($class) = @_;
99
100    my $table_name = $class->table_name;
101    my $field_prefix = $class->datasource;
102    my $dbh = $class->driver->r_handle;
103    return undef unless $dbh;
104
105    local $dbh->{RaiseError} = 0;
106    my $attr = $dbh->func($table_name, 'table_attributes') or return undef;
107    return undef unless @$attr;
108
109    my $defs = {};
110    foreach my $col (@$attr) {
111        my $coltype = $ddl->db2type($col->{TYPE});
112        my $colname = lc $col->{NAME};
113        $colname =~ s/^\Q$field_prefix\E_//i;
114        $defs->{$colname}{type} = $coltype;
115        if ( $coltype eq 'string') {
116            if (defined $col->{SIZE}) {
117                $defs->{$colname}{size} = $col->{SIZE};
118            } else {
119                $defs->{$colname}{type} = 'text';
120            }
121        }
122        if ( $col->{NOTNULL} ) {
123            $defs->{$colname}{not_null} = 1;
124        }
125        if ( $col->{PRIMARY_KEY} ) {
126            $defs->{$colname}{key} = 1;
127        }
128    }
129    $defs;
130}
131
132sub drop_sequence {
133    my $ddl = shift;
134    my ($class) = @_;
135    my $driver = $class->driver;
136    my $dbh = $driver->rw_handle;
137
138    # do this, but ignore error since it usually means the
139    # sequence didn't exist to begin with
140    if (my $col = $class->properties->{primary_key}) {
141        ## If it's a complex primary key, use the second half.
142        if(ref $col) {
143            $col = $col->[1];
144        }
145        my $def = $class->column_def($col);
146        if (exists($def->{auto}) && $def->{auto}) {
147        #if ($def->{type} eq 'integer') {
148            my $seq = $driver->dbd->sequence_name($class);
149            local $dbh->{RaiseError} = 0;
150            $dbh->do('DROP SEQUENCE ' . $seq);
151        }
152    }
153    1;
154}
155
156sub create_sequence {
157    my $ddl = shift;
158    my ($class) = @_;
159
160    my $driver = $class->driver;
161    my $dbh = $driver->rw_handle;
162
163    if ( my $col = $class->properties->{primary_key} ) {
164        ## If it's a complex primary key, use the second half.
165        if(ref $col) {
166            $col = $col->[1];
167        }
168        my $def = $class->column_def($col);
169        if (exists($def->{auto}) && $def->{auto}) {
170        #if ($def->{type} eq 'integer') {
171            my $seq = $driver->dbd->sequence_name($class);
172            my $table_name = $class->table_name;
173            my $field_prefix = $class->datasource;
174            my $max_sql = 'SELECT MAX(' . $field_prefix . '_' . $col . ') FROM ' . $table_name;
175            my ($start) = $dbh->selectrow_array($max_sql);
176
177            $dbh->do('CREATE SEQUENCE ' . $seq . 
178                ($start ? (' START ' . ($start + 1)) : ''));
179        }
180    }
181    1;
182}
183
184sub type2db {
185    my $ddl = shift;
186    my ($def) = @_;
187    my $type = $def->{type};
188    if ($type eq 'string') {
189        return 'varchar(' . $def->{size} . ')';
190    } elsif ($type eq 'smallint' ) {
191        return 'smallint';
192    } elsif ($type eq 'bigint' ) {
193        return 'bigint';
194    } elsif ($type eq 'boolean') { 
195        return 'smallint';
196    } elsif ($type eq 'datetime') {
197        return 'timestamp';
198    } elsif ($type eq 'timestamp') {
199        return 'timestamp';
200    } elsif ($type eq 'integer') {
201        return 'integer';
202    } elsif ($type eq 'blob') {
203        return 'bytea';
204    } elsif ($type eq 'text') {
205        return 'text';
206    } elsif ($type eq 'float') {
207        return 'float';
208    }
209    Carp::croak("undefined type: " . $type);
210}
211
212sub column_sql {
213    my $ddl = shift;
214    my ($class, $name) = @_;
215
216    # ugly but we need to return the sql to express
217    # a column differently based on whether we are declaring
218    # the column for creating a table or for altering a column.
219    # postgres 7.x does not support the 'not null' and 'default'
220    # keywords when altering the column.
221    if ((caller(1))[3] =~ m/::create_table_sql$/) {
222        my $def = $class->column_def($name);
223        return $ddl->SUPER::column_sql($class, $name);
224    }
225
226    my $field_prefix = $class->datasource;
227    my $def = $class->column_def($name);
228    my $type = $ddl->type2db($def);
229    return $field_prefix . '_' . $name . ' ' . $type;
230}
231
232sub add_column_sql { 
233    my $ddl = shift;
234    my ($class, $name) = @_;
235    my $sql = $ddl->column_sql($class, $name);
236    my $driver = $class->driver;
237    my $table_name = $class->table_name;
238    my $field_prefix = $class->datasource;
239    my $dbh = $driver->r_handle;
240    my @stmt = ("ALTER TABLE $table_name ADD $sql");
241
242    my $def = $class->column_def($name);
243    my $default_value;
244    if (exists $def->{default}) {
245        $default_value = $def->{default};
246        if (($def->{type} =~ m/time/) || $driver->dbd->is_date_col($name)) {
247            $default_value = $dbh->quote($driver->dbd->ts2db($default_value));
248        } elsif ($def->{type} !~ m/int|float|boolean/) {
249            $default_value = $dbh->quote($default_value);
250        }
251        push @stmt, "ALTER TABLE $table_name ALTER COLUMN ${field_prefix}_${name} SET DEFAULT " . $default_value;
252    }
253    if ($def->{key}) {
254        push @stmt, "ALTER TABLE $table_name ADD PRIMARY KEY (${field_prefix}_${name})";
255    } elsif (($def->{not_null}) 
256          && (70300 < $dbh->{pg_server_version})
257          && (exists $def->{default})) {
258          #postgresql under 7.3.0 does not support not null in alter table
259          #plus,we can't set not null unless there are no rows contains null
260        push @stmt, "UPDATE $table_name SET ${field_prefix}_${name} = $default_value WHERE ${field_prefix}_${name} IS NULL";
261        push @stmt, "ALTER TABLE $table_name ALTER COLUMN ${field_prefix}_${name} SET NOT NULL";
262    }
263    return @stmt;
264}
265
266sub alter_column_sql {
267    my $ddl = shift;
268    my ($class, $name) = @_;
269    my $sql = $ddl->SUPER::alter_column_sql(@_);
270    $sql =~ s/\bMODIFY\b/ALTER COLUMN/;
271    return $sql;
272}
273
274sub cast_column_sql {
275    my $ddl = shift;
276    my ($class, $name, $from_def) = @_;
277    my $field_prefix = $class->datasource;
278    my $def = $class->column_def($name);
279    if (($from_def->{type} eq 'text') && ($def->{type} eq 'blob')) {
280        return "cast(decode(${field_prefix}_$name, 'escape') as " . $ddl->type2db($def) . ')';
281    } elsif (($from_def->{type} eq 'blob') && ($def->{type} eq 'text')) {
282        return "cast(encode(${field_prefix}_$name, 'escape') as " . $ddl->type2db($def) . ')';
283    } else {
284        return "cast(${field_prefix}_$name as " . $ddl->type2db($def) . ')';
285    }
286}
287
288sub drop_index_sql {
289    my $ddl = shift;
290    my ($class, $key) = @_;
291    my $table_name = $class->table_name;
292
293    my $props = $class->properties;
294    my $indexes = $props->{indexes};
295    return q() unless exists($indexes->{$key});
296
297    if (ref $indexes->{$key} eq 'HASH') {
298        my $idx_info = $indexes->{$key};
299        if ($idx_info->{unique} && $ddl->can_add_constraint) {
300            return "ALTER TABLE $table_name DROP CONSTRAINT ${table_name}_$key";
301        }
302    }
303
304    return "DROP INDEX ${table_name}_$key";
305}
306
3071;
Note: See TracBrowser for help on using the browser.