Putting 4.2.0 on top of 4.0.17
[usit-rt.git] / lib / RT / Handle.pm
index ca6f2e4..a87f72a 100644 (file)
@@ -60,9 +60,9 @@ RT::Handle - RT's database handle
 
 C<RT::Handle> is RT specific wrapper over one of L<DBIx::SearchBuilder::Handle>
 classes. As RT works with different types of DBs we subclass repsective handler
-from L<DBIx::SerachBuilder>. Type of the DB is defined by C<DatabasseType> RT's
-config option. You B<must> load this module only when the configs have been
-loaded.
+from L<DBIx::SearchBuilder>. Type of the DB is defined by L<RT's DatabaseType
+config option|RT_Config/DatabaseType>. You B<must> load this module only when
+the configs have been loaded.
 
 =cut
 
@@ -87,11 +87,20 @@ sub FinalizeDatabaseType {
         use base "DBIx::SearchBuilder::Handle::". RT->Config->Get('DatabaseType');
     };
 
+    my $db_type = RT->Config->Get('DatabaseType');
     if ($@) {
-        die "Unable to load DBIx::SearchBuilder database handle for '". RT->Config->Get('DatabaseType') ."'.\n".
+        die "Unable to load DBIx::SearchBuilder database handle for '$db_type'.\n".
             "Perhaps you've picked an invalid database type or spelled it incorrectly.\n".
             $@;
     }
+
+    # We use COLLATE NOCASE to enforce case insensitivity on the normally
+    # case-sensitive SQLite, LOWER() approach works, but lucks performance
+    # due to absence of functional indexes
+    if ($db_type eq 'SQLite') {
+        no strict 'refs'; no warnings 'redefine';
+        *DBIx::SearchBuilder::Handle::SQLite::CaseSensitive = sub {0};
+    }
 }
 
 =head2 Connect
@@ -123,16 +132,12 @@ sub Connect {
         ($version) = $version =~ /^(\d+\.\d+)/;
         $self->dbh->do("SET NAMES 'utf8'") if $version >= 4.1;
     }
-
-
-    if ( $db_type eq 'Pg' ) {
+    elsif ( $db_type eq 'Pg' ) {
         my $version = $self->DatabaseVersion;
         ($version) = $version =~ /^(\d+\.\d+)/;
         $self->dbh->do("SET bytea_output = 'escape'") if $version >= 9.0;
     }
 
-
-
     $self->dbh->{'LongReadLen'} = RT->Config->Get('MaxAttachmentSize');
 }
 
@@ -224,7 +229,6 @@ sub SystemDSN {
 
 sub CheckIntegrity {
     my $self = shift;
-    $self = new $self unless ref $self;
 
     unless ($RT::Handle and $RT::Handle->dbh) {
         local $@;
@@ -237,13 +241,13 @@ sub CheckIntegrity {
     my $test_user = RT::CurrentUser->new;
     $test_user->Load('RT_System');
     unless ( $test_user->id ) {
-        return (0, 'no system user', "Couldn't find RT_System user in the DB '". $self->DSN ."'");
+        return (0, 'no system user', "Couldn't find RT_System user in the DB '". $RT::Handle->DSN ."'");
     }
 
     $test_user = RT::CurrentUser->new;
     $test_user->Load('Nobody');
     unless ( $test_user->id ) {
-        return (0, 'no nobody user', "Couldn't find Nobody user in the DB '". $self->DSN ."'");
+        return (0, 'no nobody user', "Couldn't find Nobody user in the DB '". $RT::Handle->DSN ."'");
     }
 
     return $RT::Handle->dbh;
@@ -527,6 +531,33 @@ sub InsertSchema {
     return (1);
 }
 
+sub InsertIndexes {
+    my $self      = shift;
+    my $dbh       = shift;
+    my $base_path = shift || $RT::EtcPath;
+
+    my $db_type = RT->Config->Get('DatabaseType');
+
+    $dbh = $self->dbh if !$dbh && ref $self;
+    return (0, "No DBI handle provided") unless $dbh;
+
+    return (0, "'$base_path' doesn't exist") unless -e $base_path;
+
+    my $path;
+    if ( -d $base_path ) {
+        $path = File::Spec->catfile( $base_path, "indexes");
+        return (0, "Couldn't find indexes file")
+            unless -e $path;
+    } else {
+        $path = $base_path;
+    }
+
+    local $@;
+    eval { require $path; 1 }
+        or return (0, "Couldn't execute '$path': " . $@);
+    return (1);
+}
+
 =head1 GetVersionFile
 
 Takes base name of the file as argument, scans for <base name>-<version> named
@@ -682,10 +713,9 @@ sub InsertInitialData {
 
         $group = RT::Group->new( RT->SystemUser );
         my ( $val, $msg ) = $group->_Create(
-            Type        => $name,
             Domain      => 'SystemInternal',
             Description => 'Pseudogroup for internal use',  # loc
-            Name        => '',
+            Name        => $name,
             Instance    => '',
         );
         return ($val, $msg) unless $val;
@@ -725,20 +755,18 @@ sub InsertInitialData {
 
     # system role groups
     foreach my $name (qw(Owner Requestor Cc AdminCc)) {
-        my $group = RT::Group->new( RT->SystemUser );
-        $group->LoadSystemRoleGroup( $name );
+        my $group = RT->System->RoleGroup( $name );
         if ( $group->id ) {
             push @warns, "System role '$name' already exists.";
             next;
         }
 
         $group = RT::Group->new( RT->SystemUser );
-        my ( $val, $msg ) = $group->_Create(
-            Type        => $name,
-            Domain      => 'RT::System-Role',
-            Description => 'SystemRolegroup for internal use',  # loc
-            Name        => '',
-            Instance    => '',
+        my ( $val, $msg ) = $group->CreateRoleGroup(
+            Name                => $name,
+            Object              => RT->System,
+            Description         => 'SystemRolegroup for internal use',  # loc
+            InsideTransaction   => 0,
         );
         return ($val, $msg) unless $val;
     }
@@ -787,6 +815,7 @@ sub InsertData {
         $RT::Logger->debug("Creating groups...");
         foreach my $item (@Groups) {
             my $new_entry = RT::Group->new( RT->SystemUser );
+            $item->{'Domain'} ||= 'UserDefined';
             my $member_of = delete $item->{'MemberOf'};
             my ( $return, $msg ) = $new_entry->_Create(%$item);
             unless ( $return ) {
@@ -832,6 +861,7 @@ sub InsertData {
     if ( @Users ) {
         $RT::Logger->debug("Creating users...");
         foreach my $item (@Users) {
+            my $member_of = delete $item->{'MemberOf'};
             if ( $item->{'Name'} eq 'root' && $root_password ) {
                 $item->{'Password'} = $root_password;
             }
@@ -842,6 +872,37 @@ sub InsertData {
             } else {
                 $RT::Logger->debug( $return ."." );
             }
+            if ( $member_of ) {
+                $member_of = [ $member_of ] unless ref $member_of eq 'ARRAY';
+                foreach( @$member_of ) {
+                    my $parent = RT::Group->new($RT::SystemUser);
+                    if ( ref $_ eq 'HASH' ) {
+                        $parent->LoadByCols( %$_ );
+                    }
+                    elsif ( !ref $_ ) {
+                        $parent->LoadUserDefinedGroup( $_ );
+                    }
+                    else {
+                        $RT::Logger->error(
+                            "(Error: wrong format of MemberOf field."
+                            ." Should be name of user defined group or"
+                            ." hash reference with 'column => value' pairs."
+                            ." Use array reference to add to multiple groups)"
+                        );
+                        next;
+                    }
+                    unless ( $parent->Id ) {
+                        $RT::Logger->error("(Error: couldn't load group to add member)");
+                        next;
+                    }
+                    my ( $return, $msg ) = $parent->AddMember( $new_entry->Id );
+                    unless ( $return ) {
+                        $RT::Logger->error( $msg );
+                    } else {
+                        $RT::Logger->debug( $return ."." );
+                    }
+                }
+            }
         }
         $RT::Logger->debug("done.");
     }
@@ -864,13 +925,16 @@ sub InsertData {
             my $new_entry = RT::CustomField->new( RT->SystemUser );
             my $values    = delete $item->{'Values'};
 
-            my @queues;
-            # if ref then it's list of queues, so we do things ourself
-            if ( exists $item->{'Queue'} && ref $item->{'Queue'} ) {
+            # Back-compat for the old "Queue" argument
+            if ( exists $item->{'Queue'} ) {
                 $item->{'LookupType'} ||= 'RT::Queue-RT::Ticket';
-                @queues = @{ delete $item->{'Queue'} };
+                $RT::Logger->warn("Queue provided for non-ticket custom field")
+                    unless $item->{'LookupType'} =~ /^RT::Queue-/;
+                $item->{'ApplyTo'} = delete $item->{'Queue'};
             }
 
+            my $apply_to = delete $item->{'ApplyTo'};
+
             if ( $item->{'BasedOn'} ) {
                 if ( $item->{'LookupType'} ) {
                     my $basedon = RT::CustomField->new($RT::SystemUser);
@@ -896,29 +960,38 @@ sub InsertData {
             }
 
             foreach my $value ( @{$values} ) {
-                my ( $return, $msg ) = $new_entry->AddValue(%$value);
+                ( $return, $msg ) = $new_entry->AddValue(%$value);
                 $RT::Logger->error( $msg ) unless $return;
             }
 
-            # apply by default
-            if ( !@queues && !exists $item->{'Queue'} && $item->{LookupType} ) {
-                my $ocf = RT::ObjectCustomField->new(RT->SystemUser);
-                $ocf->Create( CustomField => $new_entry->Id );
-            }
-
-            for my $q (@queues) {
-                my $q_obj = RT::Queue->new(RT->SystemUser);
-                $q_obj->Load($q);
-                unless ( $q_obj->Id ) {
-                    $RT::Logger->error("Could not find queue ". $q );
-                    next;
+            my $class = $new_entry->RecordClassFromLookupType;
+            if ($class) {
+                if ($new_entry->IsOnlyGlobal and $apply_to) {
+                    $RT::Logger->warn("ApplyTo provided for global custom field ".$new_entry->Name );
+                    undef $apply_to;
+                }
+                if ( !$apply_to ) {
+                    # Apply to all by default
+                    my $ocf = RT::ObjectCustomField->new(RT->SystemUser);
+                    ( $return, $msg) = $ocf->Create( CustomField => $new_entry->Id );
+                    $RT::Logger->error( $msg ) unless $return and $ocf->Id;
+                } else {
+                    $apply_to = [ $apply_to ] unless ref $apply_to;
+                    for my $name ( @{ $apply_to } ) {
+                        my $obj = $class->new(RT->SystemUser);
+                        $obj->Load($name);
+                        if ( $obj->Id ) {
+                            my $ocf = RT::ObjectCustomField->new(RT->SystemUser);
+                            ( $return, $msg ) = $ocf->Create(
+                                CustomField => $new_entry->Id,
+                                ObjectId    => $obj->Id,
+                            );
+                            $RT::Logger->error( $msg ) unless $return and $ocf->Id;
+                        } else {
+                            $RT::Logger->error("Could not find $class $name to apply ".$new_entry->Name." to" );
+                        }
+                    }
                 }
-                my $OCF = RT::ObjectCustomField->new(RT->SystemUser);
-                ( $return, $msg ) = $OCF->Create(
-                    CustomField => $new_entry->Id,
-                    ObjectId    => $q_obj->Id,
-                );
-                $RT::Logger->error( $msg ) unless $return and $OCF->Id;
             }
         }
 
@@ -934,17 +1007,31 @@ sub InsertData {
             if ( $item->{'CF'} ) {
                 $object = RT::CustomField->new( RT->SystemUser );
                 my @columns = ( Name => $item->{'CF'} );
+                push @columns, LookupType => $item->{'LookupType'} if $item->{'LookupType'};
                 push @columns, Queue => $item->{'Queue'} if $item->{'Queue'} and not ref $item->{'Queue'};
-                $object->LoadByName( @columns );
+                my ($ok, $msg) = $object->LoadByName( @columns );
+                unless ( $ok ) {
+                    RT->Logger->error("Unable to load CF ".$item->{CF}.": $msg");
+                    next;
+                }
             } elsif ( $item->{'Queue'} ) {
                 $object = RT::Queue->new(RT->SystemUser);
-                $object->Load( $item->{'Queue'} );
+                my ($ok, $msg) = $object->Load( $item->{'Queue'} );
+                unless ( $ok ) {
+                    RT->Logger->error("Unable to load queue ".$item->{Queue}.": $msg");
+                    next;
+                }
+            } elsif ( $item->{ObjectType} and $item->{ObjectId}) {
+                $object = $item->{ObjectType}->new(RT->SystemUser);
+                my ($ok, $msg) = $object->Load( $item->{ObjectId} );
+                unless ( $ok ) {
+                    RT->Logger->error("Unable to load ".$item->{ObjectType}." ".$item->{ObjectId}.": $msg");
+                    next;
+                }
             } else {
                 $object = $RT::System;
             }
 
-            $RT::Logger->error("Couldn't load object") and next unless $object and $object->Id;
-
             # Group rights or user rights?
             if ( $item->{'GroupDomain'} ) {
                 $princ = RT::Group->new(RT->SystemUser);
@@ -953,12 +1040,11 @@ sub InsertData {
                 } elsif ( $item->{'GroupDomain'} eq 'SystemInternal' ) {
                   $princ->LoadSystemInternalGroup( $item->{'GroupType'} );
                 } elsif ( $item->{'GroupDomain'} eq 'RT::System-Role' ) {
-                  $princ->LoadSystemRoleGroup( $item->{'GroupType'} );
+                  $princ->LoadRoleGroup( Object => RT->System, Name => $item->{'GroupType'} );
                 } elsif ( $item->{'GroupDomain'} eq 'RT::Queue-Role' &&
                           $item->{'Queue'} )
                 {
-                  $princ->LoadQueueRoleGroup( Type => $item->{'GroupType'},
-                                              Queue => $object->id);
+                  $princ->LoadRoleGroup( Object => $object, Name => $item->{'GroupType'} );
                 } else {
                   $princ->Load( $item->{'GroupId'} );
                 }
@@ -1048,14 +1134,21 @@ sub InsertData {
             my @queues = ref $item->{'Queue'} eq 'ARRAY'? @{ $item->{'Queue'} }: $item->{'Queue'} || 0;
             push @queues, 0 unless @queues; # add global queue at least
 
+            my ( $return, $msg ) = $new_entry->Create( %$item, Queue => shift @queues );
+            unless ( $return ) {
+                $RT::Logger->error( $msg );
+                next;
+            }
+            else {
+                $RT::Logger->debug( $return ."." );
+            }
             foreach my $q ( @queues ) {
-                my ( $return, $msg ) = $new_entry->Create( %$item, Queue => $q );
-                unless ( $return ) {
-                    $RT::Logger->error( $msg );
-                }
-                else {
-                    $RT::Logger->debug( $return ."." );
-                }
+                my ($return, $msg) = $new_entry->AddToObject(
+                    ObjectId => $q,
+                    Stage    => $item->{'Stage'},
+                );
+                $RT::Logger->error( "Couldn't apply scrip to $q: $msg" )
+                    unless $return;
             }
         }
         $RT::Logger->debug("done.");
@@ -1191,6 +1284,356 @@ sub FillIn {
     return $sql;
 }
 
+sub Indexes {
+    my $self = shift;
+
+    my %res;
+
+    my $db_type = RT->Config->Get('DatabaseType');
+    my $dbh = $self->dbh;
+
+    my $list;
+    if ( $db_type eq 'mysql' ) {
+        $list = $dbh->selectall_arrayref(
+            'select distinct table_name, index_name from information_schema.statistics where table_schema = ?',
+            undef, scalar RT->Config->Get('DatabaseName')
+        );
+    }
+    elsif ( $db_type eq 'Pg' ) {
+        $list = $dbh->selectall_arrayref(
+            'select tablename, indexname from pg_indexes where schemaname = ?',
+            undef, 'public'
+        );
+    }
+    elsif ( $db_type eq 'SQLite' ) {
+        $list = $dbh->selectall_arrayref(
+            'select tbl_name, name from sqlite_master where type = ?',
+            undef, 'index'
+        );
+    }
+    elsif ( $db_type eq 'Oracle' ) {
+        $list = $dbh->selectall_arrayref(
+            'select table_name, index_name from dba_indexes where index_name NOT LIKE ? AND lower(Owner) = ?',
+            undef, 'SYS_%$$', lc RT->Config->Get('DatabaseUser'),
+        );
+    }
+    else {
+        die "Not implemented";
+    }
+    push @{ $res{ lc $_->[0] } ||= [] }, lc $_->[1] foreach @$list;
+    return %res;
+}
+
+sub IndexesThatBeginWith {
+    my $self = shift;
+    my %args = (Table => undef, Columns => [], @_);
+
+    my %indexes = $self->Indexes;
+
+    my @check = @{ $args{'Columns'} };
+
+    my @list;
+    foreach my $index ( @{ $indexes{ lc $args{'Table'} } || [] } ) {
+        my %info = $self->IndexInfo( Table => $args{'Table'}, Name => $index );
+        next if @{ $info{'Columns'} } < @check;
+        my $check = join ',', @check;
+        next if join( ',', @{ $info{'Columns'} } ) !~ /^\Q$check\E(?:,|$)/i;
+
+        push @list, \%info;
+    }
+    return sort { @{ $a->{'Columns'} } <=> @{ $b->{'Columns'} } } @list;
+}
+
+sub IndexInfo {
+    my $self = shift;
+    my %args = (Table => undef, Name => undef, @_);
+
+    my $db_type = RT->Config->Get('DatabaseType');
+    my $dbh = $self->dbh;
+
+    my %res = (
+        Table => lc $args{'Table'},
+        Name => lc $args{'Name'},
+    );
+    if ( $db_type eq 'mysql' ) {
+        my $list = $dbh->selectall_arrayref(
+            'select NON_UNIQUE, COLUMN_NAME, SUB_PART
+            from information_schema.statistics
+            where table_schema = ? AND LOWER(table_name) = ? AND index_name = ?
+            ORDER BY SEQ_IN_INDEX',
+            undef, scalar RT->Config->Get('DatabaseName'), lc $args{'Table'}, $args{'Name'},
+        );
+        return () unless $list && @$list;
+        $res{'Unique'} = $list->[0][0]? 0 : 1;
+        $res{'Functional'} = 0;
+        $res{'Columns'} = [ map $_->[1], @$list ];
+    }
+    elsif ( $db_type eq 'Pg' ) {
+        my $index = $dbh->selectrow_hashref(
+            'select ix.*, pg_get_expr(ix.indexprs, ix.indrelid) as functions
+            from
+                pg_class t, pg_class i, pg_index ix
+            where
+                t.relname ilike ?
+                and t.relkind = ?
+                and i.relname ilike ?
+                and ix.indrelid = t.oid
+                and ix.indexrelid = i.oid
+            ',
+            undef, $args{'Table'}, 'r', $args{'Name'},
+        );
+        return () unless $index && keys %$index;
+        $res{'Unique'} = $index->{'indisunique'};
+        $res{'Functional'} = (grep $_ == 0, split ' ', $index->{'indkey'})? 1 : 0;
+        $res{'Columns'} = [ map int($_), split ' ', $index->{'indkey'} ];
+        my $columns = $dbh->selectall_hashref(
+            'select a.attnum, a.attname
+            from pg_attribute a where a.attrelid = ?',
+            'attnum', undef, $index->{'indrelid'}
+        );
+        if ($index->{'functions'}) {
+            # XXX: this is good enough for us
+            $index->{'functions'} = [ split /,\s+/, $index->{'functions'} ];
+        }
+        foreach my $e ( @{ $res{'Columns'} } ) {
+            if (exists $columns->{$e} ) {
+                $e = $columns->{$e}{'attname'};
+            }
+            elsif ( !$e ) {
+                $e = shift @{ $index->{'functions'} };
+            }
+        }
+
+        foreach my $column ( @{$res{'Columns'}} ) {
+            next unless $column =~ s/^lower\( \s* \(? (\w+) \)? (?:::text)? \s* \)$/$1/ix;
+            $res{'CaseInsensitive'}{ lc $1 } = 1;
+        }
+    }
+    elsif ( $db_type eq 'SQLite' ) {
+        my $list = $dbh->selectall_arrayref("pragma index_info('$args{'Name'}')");
+        return () unless $list && @$list;
+
+        $res{'Functional'} = 0;
+        $res{'Columns'} = [ map $_->[2], @$list ];
+
+        $list = $dbh->selectall_arrayref("pragma index_list('$args{'Table'}')");
+        $res{'Unique'} = (grep lc $_->[1] eq lc $args{'Name'}, @$list)[0][2]? 1 : 0;
+    }
+    elsif ( $db_type eq 'Oracle' ) {
+        my $index = $dbh->selectrow_hashref(
+            'select uniqueness, funcidx_status from dba_indexes
+            where lower(table_name) = ? AND lower(index_name) = ? AND LOWER(Owner) = ?',
+            undef, lc $args{'Table'}, lc $args{'Name'}, lc RT->Config->Get('DatabaseUser'),
+        );
+        return () unless $index && keys %$index;
+        $res{'Unique'} = $index->{'uniqueness'} eq 'UNIQUE'? 1 : 0;
+        $res{'Functional'} = $index->{'funcidx_status'}? 1 : 0;
+
+        my %columns = map @$_, @{ $dbh->selectall_arrayref(
+            'select column_position, column_name from dba_ind_columns
+            where lower(table_name) = ? AND lower(index_name) = ? AND LOWER(index_owner) = ?',
+            undef, lc $args{'Table'}, lc $args{'Name'}, lc RT->Config->Get('DatabaseUser'),
+        ) };
+        $columns{ $_->[0] } = $_->[1] foreach @{ $dbh->selectall_arrayref(
+            'select column_position, column_expression from dba_ind_expressions
+            where lower(table_name) = ? AND lower(index_name) = ? AND LOWER(index_owner) = ?',
+            undef, lc $args{'Table'}, lc $args{'Name'}, lc RT->Config->Get('DatabaseUser'),
+        ) };
+        $res{'Columns'} = [ map $columns{$_}, sort { $a <=> $b } keys %columns ];
+
+        foreach my $column ( @{$res{'Columns'}} ) {
+            next unless $column =~ s/^lower\( \s* " (\w+) " \s* \)$/$1/ix;
+            $res{'CaseInsensitive'}{ lc $1 } = 1;
+        }
+    }
+    else {
+        die "Not implemented";
+    }
+    $_ = lc $_ foreach @{ $res{'Columns'} };
+    return %res;
+}
+
+sub DropIndex {
+    my $self = shift;
+    my %args = (Table => undef, Name => undef, @_);
+
+    my $db_type = RT->Config->Get('DatabaseType');
+    my $dbh = $self->dbh;
+    local $dbh->{'PrintError'} = 0;
+    local $dbh->{'RaiseError'} = 0;
+
+    my $res;
+    if ( $db_type eq 'mysql' ) {
+        $args{'Table'} = $self->_CanonicTableNameMysql( $args{'Table'} );
+        $res = $dbh->do(
+            'drop index '. $dbh->quote_identifier($args{'Name'}) ." on $args{'Table'}",
+        );
+    }
+    elsif ( $db_type eq 'Pg' ) {
+        $res = $dbh->do("drop index $args{'Name'} CASCADE");
+    }
+    elsif ( $db_type eq 'SQLite' ) {
+        $res = $dbh->do("drop index $args{'Name'}");
+    }
+    elsif ( $db_type eq 'Oracle' ) {
+        my $user = RT->Config->Get('DatabaseUser');
+        $res = $dbh->do("drop index $user.$args{'Name'}");
+    }
+    else {
+        die "Not implemented";
+    }
+    my $desc = $self->IndexDescription( %args );
+    return ($res, $res? "Dropped $desc" : "Couldn't drop $desc: ". $dbh->errstr);
+}
+
+sub _CanonicTableNameMysql {
+    my $self = shift;
+    my $table = shift;
+    return $table unless $table;
+    # table name can be case sensitivity in DDL
+    # use LOWER to workaround mysql "bug"
+    return ($self->dbh->selectrow_array(
+        'SELECT table_name
+        FROM information_schema.tables
+        WHERE table_schema = ? AND LOWER(table_name) = ?',
+        undef, scalar RT->Config->Get('DatabaseName'), lc $table
+    ))[0] || $table;
+}
+
+sub DropIndexIfExists {
+    my $self = shift;
+    my %args = (Table => undef, Name => undef, @_);
+
+    my %indexes = $self->Indexes;
+    return (1, ucfirst($self->IndexDescription( %args )) ." doesn't exists")
+        unless grep $_ eq lc $args{'Name'},
+        @{ $indexes{ lc $args{'Table'} } || []};
+    return $self->DropIndex(%args);
+}
+
+sub CreateIndex {
+    my $self = shift;
+    my %args = ( Table => undef, Name => undef, Columns => [], CaseInsensitive => {}, @_ );
+
+    $args{'Table'} = $self->_CanonicTableNameMysql( $args{'Table'} )
+        if RT->Config->Get('DatabaseType') eq 'mysql';
+
+    my $name = $args{'Name'};
+    unless ( $name ) {
+        my %indexes = $self->Indexes;
+        %indexes = map { $_ => 1 } @{ $indexes{ lc $args{'Table'} } || [] };
+        my $i = 1;
+        $i++ while $indexes{ lc($args{'Table'}).$i };
+        $name = lc($args{'Table'}).$i;
+    }
+
+    my @columns = @{ $args{'Columns'} };
+    if ( $self->CaseSensitive ) {
+        foreach my $column ( @columns ) {
+            next unless $args{'CaseInsensitive'}{ lc $column };
+            $column = "LOWER($column)";
+        }
+    }
+
+    my $sql = "CREATE"
+        . ($args{'Unique'}? ' UNIQUE' : '')
+        ." INDEX $name ON $args{'Table'}"
+        ."(". join( ', ', @columns ) .")"
+    ;
+
+    my $res = $self->dbh->do( $sql );
+    unless ( $res ) {
+        return (
+            undef, "Failed to create ". $self->IndexDescription( %args )
+                ." (sql: $sql): ". $self->dbh->errstr
+        );
+    }
+    return ($name, "Created ". $self->IndexDescription( %args ) );
+}
+
+sub IndexDescription {
+    my $self = shift;
+    my %args = (@_);
+
+    my $desc =
+        ($args{'Unique'}? 'unique ' : '')
+        .'index'
+        . ($args{'Name'}? " $args{'Name'}" : '')
+        . ( @{$args{'Columns'}||[]}?
+            " ("
+            . join(', ', @{$args{'Columns'}})
+            . (@{$args{'Optional'}||[]}? '['. join(', ', '', @{$args{'Optional'}}).']' : '' )
+            .")"
+            : ''
+        )
+        . ($args{'Table'}? " on $args{'Table'}" : '')
+    ;
+    return $desc;
+}
+
+sub MakeSureIndexExists {
+    my $self = shift;
+    my %args = ( Table => undef, Columns => [], Optional => [], @_ );
+
+    my @list = $self->IndexesThatBeginWith(
+        Table => $args{'Table'}, Columns => [@{$args{'Columns'}}, @{$args{'Optional'}}],
+    );
+    if (@list) {
+        RT->Logger->debug( ucfirst $self->IndexDescription(
+            Table => $args{'Table'}, Columns => [@{$args{'Columns'}}, @{$args{'Optional'}}],
+        ). ' exists.' );
+        return;
+    }
+
+    @list = $self->IndexesThatBeginWith(
+        Table => $args{'Table'}, Columns => $args{'Columns'},
+    );
+    if ( !@list ) {
+        my ($status, $msg) = $self->CreateIndex(
+            Table => $args{'Table'}, Columns => [@{$args{'Columns'}}, @{$args{'Optional'}}],
+        );
+        my $method = $status ? 'debug' : 'warning';
+        RT->Logger->$method($msg);
+    }
+    else {
+        RT->Logger->info(
+            ucfirst $self->IndexDescription(
+                %{$list[0]}
+            )
+            .' exists, you may consider replacing it with '
+            . $self->IndexDescription(
+                Table => $args{'Table'}, Columns => [@{$args{'Columns'}}, @{$args{'Optional'}}],
+            )
+        );
+    }
+}
+
+sub DropIndexesThatArePrefix {
+    my $self = shift;
+    my %args = ( Table => undef, Columns => [], @_ );
+
+    my @list = $self->IndexesThatBeginWith(
+        Table => $args{'Table'}, Columns => [$args{'Columns'}[0]],
+    );
+
+    my $checking = join ',', map lc $_, @{ $args{'Columns'} }, '';
+    foreach my $i ( splice @list ) {
+        my $columns = join ',', @{ $i->{'Columns'} }, '';
+        next unless $checking =~ /^\Q$columns/i;
+
+        push @list, $i;
+    }
+    pop @list;
+
+    foreach my $i ( @list ) {
+        my ($status, $msg) = $self->DropIndex(
+            Table => $i->{'Table'}, Name => $i->{'Name'},
+        );
+        my $method = $status ? 'debug' : 'warning';
+        RT->Logger->$method($msg);
+    }
+}
+
 # log a mason stack trace instead of a Carp::longmess because it's less painful
 # and uses mason component paths properly
 sub _LogSQLStatement {
@@ -1203,6 +1646,13 @@ sub _LogSQLStatement {
     push @{$self->{'StatementLog'}} , ([Time::HiRes::time(), $statement, [@bind], $duration, HTML::Mason::Exception->new->as_string]);
 }
 
+# helper in a few cases where we do SQL by hand
+sub __MakeClauseCaseInsensitive {
+    my $self = shift;
+    return join ' ', @_ unless $self->CaseSensitive;
+    my ($field, $op, $value) = $self->_MakeClauseCaseInsensitive(@_);
+    return "$field $op $value";
+}
 
 sub _TableNames {
     my $self = shift;