package DBIx::Custom;

our $VERSION = '0.1694';

use 5.008001;

use Object::Simple -base;

use Carp 'croak';

use DBI;

use DBIx::Custom::Result;

use DBIx::Custom::Query;

use DBIx::Custom::QueryBuilder;

use DBIx::Custom::Where;

use DBIx::Custom::Model;

use DBIx::Custom::Tag;

use DBIx::Custom::Util qw/_array_to_hash _subname/;

use Encode qw/encode encode_utf8 decode_utf8/;

use constant DEBUG => $ENV{DBIX_CUSTOM_DEBUG} || 0;

use constant DEBUG_ENCODING => $ENV{DBIX_CUSTOM_DEBUG_ENCODING} || 'UTF-8';

our @COMMON_ARGS = qw/bind_type table query filter id primary_key

                      type_rule_off type/;

has [qw/connector dsn password quote user/],

    cache => 0,

    cache_method => sub {

        sub {

            my $self = shift;

            $self->{_cached} ||= {};

            if (@_ > 1) {

                $self->{_cached}{$_[0]} = $_[1];

            }

            else {

                return $self->{_cached}{$_[0]};

            }

        }

    },

    dbi_option => sub { {} },

    default_dbi_option => sub {

        {

            RaiseError => 1,

            PrintError => 0,

            AutoCommit => 1

        }

    },

    filters => sub {

        {

            encode_utf8 => sub { encode_utf8($_[0]) },

            decode_utf8 => sub { decode_utf8($_[0]) }

        }

    },

    models => sub { {} },

    query_builder => sub { DBIx::Custom::QueryBuilder->new },

    result_class  => 'DBIx::Custom::Result',

    safety_character => '\w',

    stash => sub { {} };

our $AUTOLOAD;

sub AUTOLOAD {

    my $self = shift;

    # Method name

    my ($package, $mname) = $AUTOLOAD =~ /^([\w\:]+)\:\:(\w+)$/;

    # Call method

    $self->{_methods} ||= {};

    if (my $method = $self->{_methods}->{$mname}) {

        return $self->$method(@_)

    }

    elsif ($self->{dbh} && (my $dbh_method = $self->dbh->can($mname))) {

        $self->dbh->$dbh_method(@_);

    }

    else {

        croak qq{Can't locate object method "$mname" via "$package" }

            . _subname;

    }

}

sub assign_param {

    my ($self, $param) = @_;

    # Create set tag

    my @params;

    my $safety = $self->safety_character;

    my $q = $self->_quote;

    foreach my $column (keys %$param) {

        croak qq{"$column" is not safety column name } . _subname

          unless $column =~ /^[$safety\.]+$/;

        my $column_quote = "$q$column$q";

        $column_quote =~ s/\./$q.$q/;

        push @params, "$column_quote = :$column";

    }

    my $tag = join(', ', @params);

    return $tag;

}

sub column {

    my $self = shift;

    my $option = pop if ref $_[-1] eq 'HASH';

    my $real_table = shift;

    my $columns = shift;

    my $table = $option->{alias} || $real_table;

    # Columns

    unless ($columns) {

        $columns ||= $self->model($real_table)->columns;

    }

    # Reserved word quote

    my $q = $self->_quote;

    # Separator

    my $separator = $self->separator;

    # Column clause

    my @column;

    $columns ||= [];

    push @column, "$q$table$q.$q$_$q as $q${table}${separator}$_$q"

      for @$columns;

    return join (', ', @column);

}

sub connect {

    my $self = ref $_[0] ? shift : shift->new(@_);;

    # Connect

    $self->dbh;

    return $self;

}

sub create_query {

    my ($self, $source) = @_;

    # Cache

    my $cache = $self->cache;

    # Query

    my $query;

    # Get cached query

    if ($cache) {

        # Get query

        my $q = $self->cache_method->($self, $source);

        # Create query

        if ($q) {

            $query = DBIx::Custom::Query->new($q);

            $query->filters($self->filters);

        }

    }

    # Create query

    unless ($query) {

        # Create query

        my $builder = $self->query_builder;

        $query = $builder->build_query($source);

        # Remove reserved word quote

        if (my $q = $self->_quote) {

            $_ =~ s/$q//g for @{$query->columns}

        }

        # Save query to cache

        $self->cache_method->(

            $self, $source,

            {

                sql     => $query->sql, 

                columns => $query->columns,

                tables  => $query->tables

            }

        ) if $cache;

    }

    # Prepare statement handle

    my $sth;

    eval { $sth = $self->dbh->prepare($query->{sql})};

    if ($@) {

        $self->_croak($@, qq{. Following SQL is executed.\n}

                        . qq{$query->{sql}\n} . _subname);

    }

    # Set statement handle

    $query->sth($sth);

    # Set filters

    $query->filters($self->filters);

    return $query;

}

sub dbh {

    my $self = shift;

    # Set

    if (@_) {

        $self->{dbh} = $_[0];

        return $self;

    }

    # Get

    else {

        # From Connction manager

        if (my $connector = $self->connector) {

            croak "connector must have dbh() method " . _subname

              unless ref $connector && $connector->can('dbh');

            $self->{dbh} = $connector->dbh;

        }

        # Connect

        $self->{dbh} ||= $self->_connect;

        # Quote

        if (!defined $self->reserved_word_quote && !defined $self->quote) {

            my $driver = $self->{dbh}->{Driver}->{Name};

            my $quote = $driver eq 'mysql' ? '`' : '"';

            $self->quote($quote);

        }

        return $self->{dbh};

    }

}

our %DELETE_ARGS

  = map { $_ => 1 } @COMMON_ARGS, qw/where append allow_delete_all where_param/;

sub delete {

    my ($self, %args) = @_;

    # Check arguments

    foreach my $name (keys %args) {

        croak qq{"$name" is wrong option } . _subname

          unless $DELETE_ARGS{$name};

    }

    # Arguments

    my $table = $args{table} || '';

    croak qq{"table" option must be specified. } . _subname

      unless $table;

    my $where            = delete $args{where} || {};

    my $append           = delete $args{append};

    my $allow_delete_all = delete $args{allow_delete_all};

    my $where_param      = delete $args{where_param} || {};

    my $id = delete $args{id};

    my $primary_key = delete $args{primary_key};

    croak "update method primary_key option " .

          "must be specified when id is specified " . _subname

      if defined $id && !defined $primary_key;

    $primary_key = [$primary_key] unless ref $primary_key eq 'ARRAY';

    # Where

    $where = $self->_create_param_from_id($id, $primary_key) if $id;

    my $where_clause = '';

    if (ref $where) {

        $where = $self->_where_to_obj($where);

        $where_param = keys %$where_param

                     ? $self->merge_param($where_param, $where->param)

                     : $where->param;

        # String where

        $where_clause = $where->to_string;

    }

    elsif ($where) { $where_clause = "where $where" }

    croak qq{"where" must be specified } . _subname

      if $where_clause eq '' && !$allow_delete_all;

    # Delete statement

    my @sql;

    my $q = $self->_quote;

    push @sql, "delete from $q$table$q $where_clause";

    push @sql, $append if $append;

    my $sql = join(' ', @sql);

    # Execute query

    return $self->execute(

        $sql,

        param => $where_param,

        table => $table,

        %args

    );

}

sub delete_all { shift->delete(allow_delete_all => 1, @_) }

sub DESTROY { }

sub create_model {

    my $self = shift;

    # Arguments

    my $args = ref $_[0] eq 'HASH' ? $_[0] : {@_};

    $args->{dbi} = $self;

    my $model_class = delete $args->{model_class} || 'DBIx::Custom::Model';

    my $model_name  = delete $args->{name};

    my $model_table = delete $args->{table};

    $model_name ||= $model_table;

    # Create model

    my $model = $model_class->new($args);

    $model->name($model_name) unless $model->name;

    $model->table($model_table) unless $model->table;

    # Apply filter

    my $filter = ref $model->filter eq 'HASH'

               ? [%{$model->filter}]

               : $model->filter;

    warn "DBIx::Custom::Model filter method is DEPRECATED!"

      if @$filter;

    $self->_apply_filter($model->table, @$filter);

    # Set model

    $self->model($model->name, $model);

    return $self->model($model->name);

}

sub each_column {

    my ($self, $cb) = @_;

    # Iterate all tables

    my $sth_tables = $self->dbh->table_info;

    while (my $table_info = $sth_tables->fetchrow_hashref) {

        # Table

        my $table = $table_info->{TABLE_NAME};

        # Iterate all columns

        my $sth_columns = $self->dbh->column_info(undef, undef, $table, '%');

        while (my $column_info = $sth_columns->fetchrow_hashref) {

            my $column = $column_info->{COLUMN_NAME};

            $self->$cb($table, $column, $column_info);

        }

    }

}

our %EXECUTE_ARGS = map { $_ => 1 } @COMMON_ARGS, 'param';

sub execute {

    my $self = shift;

    my $query = shift;

    my $param;

    $param = shift if @_ % 2;

    my %args = @_;

    # Arguments

    my $p = delete $args{param} || {};

    $param ||= $p;

    my $tables = delete $args{table} || [];

    $tables = [$tables] unless ref $tables eq 'ARRAY';

    my $filter = delete $args{filter};

    $filter = _array_to_hash($filter);

    my $bind_type = delete $args{bind_type} || delete $args{type};

    $bind_type = _array_to_hash($bind_type);

    my $type_rule_off = delete $args{type_rule_off};

    my $query_return = delete $args{query};

    # Check argument names

    foreach my $name (keys %args) {

        croak qq{"$name" is wrong option } . _subname

          unless $EXECUTE_ARGS{$name};

    }

    # Create query

    $query = $self->create_query($query) unless ref $query;

    return $query if $query_return;

    $filter ||= $query->filter;

    # Tables

    unshift @$tables, @{$query->tables};

    my $main_table = pop @$tables;

    $tables = $self->_remove_duplicate_table($tables, $main_table);

    if (my $q = $self->_quote) {

        $_ =~ s/$q//g for @$tables;

    }

    # Type rule

    my $type_filter = {};

    unless ($type_rule_off) {

        foreach my $name (keys %$param) {

            my $table;

            my $column;

            if ($name =~ /(?:(.+)\.)?(.+)/) {

                $table = $1;

                $column = $2;

            }

            $table ||= $main_table;

            my $into = $self->{_into} || {};

            if (defined $table && $into->{$table} &&

                (my $rule = $into->{$table}->{$column}))

            {

                $type_filter->{$column} = $rule;

                $type_filter->{"$table.$column"} = $rule;

            }

        }

    }

    # Applied filter

    my $applied_filter = {};

    foreach my $table (@$tables) {

        $applied_filter = {

            %$applied_filter,

            %{$self->{filter}{out}->{$table} || {}}

        }

    }

    $filter = {%$applied_filter, %$filter};

    # Replace filter name to code

    foreach my $column (keys %$filter) {

        my $name = $filter->{$column};

        if (!defined $name) {

            $filter->{$column} = undef;

        }

        elsif (ref $name ne 'CODE') {

          croak qq{Filter "$name" is not registered" } . _subname

            unless exists $self->filters->{$name};

          $filter->{$column} = $self->filters->{$name};

        }

    }

    # Create bind values

    my $bind = $self->_create_bind_values(

        $param,

        $query->columns,

        $filter,

        $type_filter,

        $bind_type

    );

    # Execute

    my $sth = $query->sth;

    my $affected;

    eval {

        for (my $i = 0; $i < @$bind; $i++) {

            my $bind_type = $bind->[$i]->{bind_type};

            $sth->bind_param(

                $i + 1,

                $bind->[$i]->{value},

                $bind_type ? $bind_type : ()

            );

        }

        $affected = $sth->execute;

    };

    if ($@) {

        $self->_croak($@, qq{. Following SQL is executed.\n}

                        . qq{$query->{sql}\n} . _subname);

    }

    # DEBUG message

    if (DEBUG) {

        print STDERR "SQL:\n" . $query->sql . "\n";

        my @output;

        foreach my $b (@$bind) {

            my $value = $b->{value};

            $value = 'undef' unless defined $value;

            $value = encode(DEBUG_ENCODING(), $value)

              if utf8::is_utf8($value);

            push @output, $value;

        }

        print STDERR "Bind values: " . join(', ', @output) . "\n\n";

    }

    # Select statement

    if ($sth->{NUM_OF_FIELDS}) {

        # Filter

        my $filter = {};

        $filter->{in}  = {};

        $filter->{end} = {};

        push @$tables, $main_table if $main_table;

        foreach my $table (@$tables) {

            foreach my $way (qw/in end/) {

                $filter->{$way} = {

                    %{$filter->{$way}},

                    %{$self->{filter}{$way}{$table} || {}}

                };

            }

        }

        # Result

        my $result = $self->result_class->new(

            sth => $sth,

            filters => $self->filters,

            default_filter => $self->{default_in_filter},

            filter => $filter->{in} || {},

            end_filter => $filter->{end} || {},

            type_rule => \%{$self->type_rule->{from}},

        );

        return $result;

    }

    # Not select statement

    else { return $affected }

}

our %INSERT_ARGS = map { $_ => 1 } @COMMON_ARGS, qw/param/;

sub insert {

    my $self = shift;

    # Arguments

    my $param;

    $param = shift if @_ % 2;

    my %args = @_;

    my $table  = delete $args{table};

    croak qq{"table" option must be specified } . _subname

      unless $table;

    my $p = delete $args{param} || {};

    $param  ||= $p;

    my $append = delete $args{append} || '';

    my $id = delete $args{id};

    my $primary_key = delete $args{primary_key};

    croak "insert method primary_key option " .

          "must be specified when id is specified " . _subname

      if defined $id && !defined $primary_key;

    $primary_key = [$primary_key] unless ref $primary_key eq 'ARRAY';

    # Check arguments

    foreach my $name (keys %args) {

        croak qq{"$name" is wrong option } . _subname

          unless $INSERT_ARGS{$name};

    }

    # Merge parameter

    if ($id) {

        my $id_param = $self->_create_param_from_id($id, $primary_key);

        $param = $self->merge_param($id_param, $param);

    }

    # Reserved word quote

    my $q = $self->_quote;

    # Insert statement

    my @sql;

    push @sql, "insert into $q$table$q " . $self->insert_param($param);

    push @sql, $append if $append;

    my $sql = join (' ', @sql);

    # Execute query

    return $self->execute(

        $sql,

        param => $param,

        table => $table,

        %args

    );

}

sub insert_param {

    my ($self, $param) = @_;

    # Create insert parameter tag

    my $safety = $self->safety_character;

    my $q = $self->_quote;

    my @columns;

    my @placeholders;

    foreach my $column (keys %$param) {

        croak qq{"$column" is not safety column name } . _subname

          unless $column =~ /^[$safety\.]+$/;

        my $column_quote = "$q$column$q";

        $column_quote =~ s/\./$q.$q/;

        push @columns, $column_quote;

        push @placeholders, ":$column";

    }

    return '(' . join(', ', @columns) . ') ' . 'values ' .

           '(' . join(', ', @placeholders) . ')'

}

sub include_model {

    my ($self, $name_space, $model_infos) = @_;

    # Name space

    $name_space ||= '';

    # Get Model infomations

    unless ($model_infos) {

        # Load name space module

        croak qq{"$name_space" is invalid class name } . _subname

          if $name_space =~ /[^\w:]/;

        eval "use $name_space";

        croak qq{Name space module "$name_space.pm" is needed. $@ }

            . _subname

          if $@;

        # Search model modules

        my $path = $INC{"$name_space.pm"};

        $path =~ s/\.pm$//;

        opendir my $dh, $path

          or croak qq{Can't open directory "$path": $! } . _subname

        $model_infos = [];

        while (my $module = readdir $dh) {

            push @$model_infos, $module

              if $module =~ s/\.pm$//;

        }

        close $dh;

    }

    # Include models

    foreach my $model_info (@$model_infos) {

        # Load model

        my $model_class;

        my $model_name;

        my $model_table;

        if (ref $model_info eq 'HASH') {

            $model_class = $model_info->{class};

            $model_name  = $model_info->{name};

            $model_table = $model_info->{table};

            $model_name  ||= $model_class;

            $model_table ||= $model_name;

        }

        else { $model_class = $model_name = $model_table = $model_info }

        my $mclass = "${name_space}::$model_class";

        croak qq{"$mclass" is invalid class name } . _subname

          if $mclass =~ /[^\w:]/;

        unless ($mclass->can('isa')) {

            eval "use $mclass";

            croak "$@ " . _subname if $@;

        }

        # Create model

        my $args = {};

        $args->{model_class} = $mclass if $mclass;

        $args->{name}        = $model_name if $model_name;

        $args->{table}       = $model_table if $model_table;

        $self->create_model($args);

    }

    return $self;

}

sub merge_param {

    my ($self, @params) = @_;

    # Merge parameters

    my $merge = {};

    foreach my $param (@params) {

        foreach my $column (keys %$param) {

            my $param_is_array = ref $param->{$column} eq 'ARRAY' ? 1 : 0;

            if (exists $merge->{$column}) {

                $merge->{$column} = [$merge->{$column}]

                  unless ref $merge->{$column} eq 'ARRAY';

                push @{$merge->{$column}},

                  ref $param->{$column} ? @{$param->{$column}} : $param->{$column};

            }

            else {

                $merge->{$column} = $param->{$column};

            }

        }

    }

    return $merge;

}

sub method {

    my $self = shift;

    # Register method

    my $methods = ref $_[0] eq 'HASH' ? $_[0] : {@_};

    $self->{_methods} = {%{$self->{_methods} || {}}, %$methods};

    return $self;

}

sub model {

    my ($self, $name, $model) = @_;

    # Set model

    if ($model) {

        $self->models->{$name} = $model;

        return $self;

    }

    # Check model existance

    croak qq{Model "$name" is not included } . _subname

      unless $self->models->{$name};

    # Get model

    return $self->models->{$name};

}

sub mycolumn {

    my ($self, $table, $columns) = @_;

    # Create column clause

    my @column;

    my $q = $self->_quote;

    $columns ||= [];

    push @column, "$q$table$q.$q$_$q as $q$_$q" for @$columns;

    return join (', ', @column);

}

sub new {

    my $self = shift->SUPER::new(@_);

    # Check attributes

    my @attrs = keys %$self;

    foreach my $attr (@attrs) {

        croak qq{"$attr" is wrong name } . _subname

          unless $self->can($attr);

    }

    # DEPRECATED!

    $self->query_builder->{tags} = {

        '?'     => \&DBIx::Custom::Tag::placeholder,

        '='     => \&DBIx::Custom::Tag::equal,

        '<>'    => \&DBIx::Custom::Tag::not_equal,

        '>'     => \&DBIx::Custom::Tag::greater_than,

        '<'     => \&DBIx::Custom::Tag::lower_than,

        '>='    => \&DBIx::Custom::Tag::greater_than_equal,

        '<='    => \&DBIx::Custom::Tag::lower_than_equal,

        'like'  => \&DBIx::Custom::Tag::like,

        'in'    => \&DBIx::Custom::Tag::in,

        'insert_param' => \&DBIx::Custom::Tag::insert_param,

        'update_param' => \&DBIx::Custom::Tag::update_param

    };

    return $self;

}

sub not_exists { bless {}, 'DBIx::Custom::NotExists' }

sub register_filter {

    my $self = shift;

    # Register filter

    my $filters = ref $_[0] eq 'HASH' ? $_[0] : {@_};

    $self->filters({%{$self->filters}, %$filters});

    return $self;

}

our %SELECT_ARGS

  = map { $_ => 1 } @COMMON_ARGS,

                    qw/column where relation join param where_param wrap

                       prefix/;

sub select {

    my ($self, %args) = @_;

    # Arguments

    my $table = delete $args{table};

    my $tables = ref $table eq 'ARRAY' ? $table

               : defined $table ? [$table]

               : [];

    my $columns   = delete $args{column};

    my $where     = delete $args{where} || {};

    my $append    = delete $args{append};

    my $join      = delete $args{join} || [];

    croak qq{"join" must be array reference } . _subname

      unless ref $join eq 'ARRAY';

    my $relation = delete $args{relation};

    warn "select() relation option is DEPRECATED! use join option instead"

      if $relation;

    my $param = delete $args{param} || {}; # DEPRECATED!

    warn "select() param option is DEPRECATED! use where_param option instead"

      if keys %$param;

    my $where_param = delete $args{where_param} || $param || {};

    my $wrap = delete $args{wrap};

    my $id = delete $args{id};

    my $primary_key = delete $args{primary_key};

    croak "update method primary_key option " .

          "must be specified when id is specified " . _subname

      if defined $id && !defined $primary_key;

    $primary_key = [$primary_key] unless ref $primary_key eq 'ARRAY';

    my $prefix = delete $args{prefix};

    # Check arguments

    foreach my $name (keys %args) {

        croak qq{"$name" is wrong option } . _subname

          unless $SELECT_ARGS{$name};

    }

    # Add relation tables(DEPRECATED!);

    $self->_add_relation_table($tables, $relation);

    # Select statement

    my @sql;

    push @sql, 'select';

    # Reserved word quote

    my $q = $self->_quote;

    # Prefix

    push @sql, $prefix if defined $prefix;

    # Column clause

    if ($columns) {

        $columns = [$columns] unless ref $columns eq 'ARRAY';

        foreach my $column (@$columns) {

            if (ref $column eq 'HASH') {

                $column = $self->column(%$column) if ref $column eq 'HASH';

            }

            elsif (ref $column eq 'ARRAY') {

                croak "Format must be [COLUMN, as => ALIAS] " . _subname

                  unless @$column == 3 && $column->[1] eq 'as';

                $column = join(' ', $column->[0], 'as', $q . $column->[2] . $q);

            }

            unshift @$tables, @{$self->_search_tables($column)};

            push @sql, ($column, ',');

        }

        pop @sql if $sql[-1] eq ',';

    }

    else { push @sql, '*' }

    # Table

    push @sql, 'from';

    if ($relation) {

        my $found = {};

        foreach my $table (@$tables) {

            push @sql, ("$q$table$q", ',') unless $found->{$table};

            $found->{$table} = 1;

        }

    }

    else {

        my $main_table = $tables->[-1] || '';

        push @sql, "$q$main_table$q";

    }

    pop @sql if ($sql[-1] || '') eq ',';

    croak "Not found table name " . _subname

      unless $tables->[-1];

    # Add tables in parameter

    unshift @$tables,

            @{$self->_search_tables(join(' ', keys %$where_param) || '')};

    # Where

    my $where_clause = '';

    $where = $self->_create_param_from_id($id, $primary_key) if $id;

    if (ref $where) {

        $where = $self->_where_to_obj($where);

        $where_param = keys %$where_param

                     ? $self->merge_param($where_param, $where->param)

                     : $where->param;

        # String where

        $where_clause = $where->to_string;

    }

    elsif ($where) { $where_clause = "where $where" }

    # Add table names in where clause

    unshift @$tables, @{$self->_search_tables($where_clause)};

    # Push join

    $self->_push_join(\@sql, $join, $tables);

    # Add where clause

    push @sql, $where_clause;

    # Relation(DEPRECATED!);

    $self->_push_relation(\@sql, $tables, $relation, $where_clause eq '' ? 1 : 0);

    # Append

    push @sql, $append if $append;

    # Wrap

    if ($wrap) {

        croak "wrap option must be array refrence " . _subname

          unless ref $wrap eq 'ARRAY';

        unshift @sql, $wrap->[0];

        push @sql, $wrap->[1];

    }

    # SQL

    my $sql = join (' ', @sql);

    # Execute query

    my $result = $self->execute(

        $sql,

        param => $where_param, 

        table => $tables,

        %args

    );

    return $result;

}

sub separator {

    my $self = shift;

    if (@_) {

        my $separator = $_[0] || '';

        croak qq{Separator must be "." or "__" or "-" } . _subname

          unless $separator eq '.' || $separator eq '__'

              || $separator eq '-';

        $self->{separator} = $separator;

        return $self;

    }

    return $self->{separator} ||= '.';

}

sub setup_model {

    my $self = shift;

    # Setup model

    $self->each_column(

        sub {

            my ($self, $table, $column, $column_info) = @_;

            if (my $model = $self->models->{$table}) {

                push @{$model->columns}, $column;

            }

        }

    );

    return $self;

}

sub available_data_type {

    my $self = shift;

    my $data_types = '';

    foreach my $i (-1000 .. 1000) {

         my $type_info = $self->dbh->type_info($i);

         my $data_type = $type_info->{DATA_TYPE};

         my $type_name = $type_info->{TYPE_NAME};

         $data_types .= "$data_type ($type_name)\n"

           if defined $data_type;

    }

    return "Data Type maybe equal to Type Name" unless $data_types;

    $data_types = "Data Type (Type name)\n" . $data_types;

    return $data_types;

}

sub available_type_name {

    my $self = shift;

    # Type Names

    my $type_names = {};

    $self->each_column(sub {

        my ($self, $table, $column, $column_info) = @_;

        $type_names->{$column_info->{TYPE_NAME}} = 1

          if $column_info->{TYPE_NAME};

    });

    my @output = sort keys %$type_names;

    unshift @output, "Type Name";

    return join "\n", @output;

}

sub type_rule {

    my $self = shift;

    if (@_) {

        my $type_rule = ref $_[0] eq 'HASH' ? $_[0] : {@_};

        # Into

        $type_rule->{into} = _array_to_hash($type_rule->{into});

        $self->{type_rule} = $type_rule;

        $self->{_into} = {};

        foreach my $type_name (keys %{$type_rule->{into} || {}}) {

            croak qq{type name of into section must be lower case}

              if $type_name =~ /[A-Z]/;

        }

        $self->each_column(sub {

            my ($dbi, $table, $column, $column_info) = @_;

            my $type_name = lc $column_info->{TYPE_NAME};

            if ($type_rule->{into} &&

                (my $filter = $type_rule->{into}->{$type_name}))

            {

                return unless exists $type_rule->{into}->{$type_name};

                if  (defined $filter && ref $filter ne 'CODE') 

                {

                    my $fname = $filter;

                    croak qq{Filter "$fname" is not registered" } . _subname

                      unless exists $self->filters->{$fname};

                    $filter = $self->filters->{$fname};

                }

                $self->{_into}{$table}{$column} = $filter;

            }

        });

        # From

        $type_rule->{from} = _array_to_hash($type_rule->{from});

        foreach my $data_type (keys %{$type_rule->{from} || {}}) {

            croak qq{data type of into section must be lower case or number}

              if $data_type =~ /[A-Z]/;