module Sequel
  class Dataset
    # ---------------------
    # :section: 3 - User Methods relating to SQL Creation
    # These are methods you can call to see what SQL will be generated by the dataset.
    # ---------------------
    
    # Returns a DELETE SQL query string.  See +delete+.
    # 
    #   dataset.filter{|o| o.price >= 100}.delete_sql
    #   # => "DELETE FROM items WHERE (price >= 100)"
    def delete_sql
      return static_sql(opts[:sql]) if opts[:sql]
      check_modification_allowed!
      clause_sql(:delete)
    end
    
    # Returns an EXISTS clause for the dataset as a +LiteralString+.
    #
    #   DB.select(1).where(DB[:items].exists)
    #   # SELECT 1 WHERE (EXISTS (SELECT * FROM items))
    def exists
      SQL::PlaceholderLiteralString.new(EXISTS, [self], true)
    end
    
    # Returns an INSERT SQL query string.  See +insert+.
    #
    #   DB[:items].insert_sql(:a=>1)
    #   # => "INSERT INTO items (a) VALUES (1)"
    def insert_sql(*values)
      return static_sql(@opts[:sql]) if @opts[:sql]

      check_modification_allowed!

      columns = []

      case values.size
      when 0
        return insert_sql({})
      when 1
        case vals = values.at(0)
        when Hash
          vals = @opts[:defaults].merge(vals) if @opts[:defaults]
          vals = vals.merge(@opts[:overrides]) if @opts[:overrides]
          values = []
          vals.each do |k,v| 
            columns << k
            values << v
          end
        when Dataset, Array, LiteralString
          values = vals
        end
      when 2
        if (v0 = values.at(0)).is_a?(Array) && ((v1 = values.at(1)).is_a?(Array) || v1.is_a?(Dataset) || v1.is_a?(LiteralString))
          columns, values = v0, v1
          raise(Error, "Different number of values and columns given to insert_sql") if values.is_a?(Array) and columns.length != values.length
        end
      end

      if values.is_a?(Array) && values.empty? && !insert_supports_empty_values? 
        columns = [columns().last]
        values = [DEFAULT]
      end
      clone(:columns=>columns, :values=>values)._insert_sql
    end
    
    # Returns a literal representation of a value to be used as part
    # of an SQL expression. 
    # 
    #   DB[:items].literal("abc'def\\") #=> "'abc''def\\\\'"
    #   DB[:items].literal(:items__id) #=> "items.id"
    #   DB[:items].literal([1, 2, 3]) => "(1, 2, 3)"
    #   DB[:items].literal(DB[:items]) => "(SELECT * FROM items)"
    #   DB[:items].literal(:x + 1 > :y) => "((x + 1) > y)"
    #
    # If an unsupported object is given, an +Error+ is raised.
    def literal_append(sql, v)
      case v
      when Symbol
        literal_symbol_append(sql, v)
      when String
        case v
        when LiteralString
          sql << v
        when SQL::Blob
          literal_blob_append(sql, v)
        else
          literal_string_append(sql, v)
        end
      when Integer
        sql << literal_integer(v)
      when Hash
        literal_hash_append(sql, v)
      when SQL::Expression
        literal_expression_append(sql, v)
      when Float
        sql << literal_float(v)
      when BigDecimal
        sql << literal_big_decimal(v)
      when NilClass
        sql << literal_nil
      when TrueClass
        sql << literal_true
      when FalseClass
        sql << literal_false
      when Array
        literal_array_append(sql, v)
      when Time
        sql << (v.is_a?(SQLTime) ? literal_sqltime(v) : literal_time(v))
      when DateTime
        sql << literal_datetime(v)
      when Date
        sql << literal_date(v)
      when Dataset
        literal_dataset_append(sql, v)
      else
        literal_other_append(sql, v)
      end
    end
    
    # Returns an array of insert statements for inserting multiple records.
    # This method is used by +multi_insert+ to format insert statements and
    # expects a keys array and and an array of value arrays.
    #
    # This method should be overridden by descendants if the support
    # inserting multiple records in a single SQL statement.
    def multi_insert_sql(columns, values)
      values.map{|r| insert_sql(columns, r)}
    end
    
    # Returns a SELECT SQL query string.
    #
    #   dataset.select_sql # => "SELECT * FROM items"
    def select_sql
      return static_sql(@opts[:sql]) if @opts[:sql]
      clause_sql(:select)
    end

    # Same as +select_sql+, not aliased directly to make subclassing simpler.
    def sql
      select_sql
    end
    
    # Returns a TRUNCATE SQL query string.  See +truncate+
    #
    #   DB[:items].truncate_sql # => 'TRUNCATE items'
    def truncate_sql
      if opts[:sql]
        static_sql(opts[:sql])
      else
        check_truncation_allowed!
        raise(InvalidOperation, "Can't truncate filtered datasets") if opts[:where] || opts[:having]
        t = ''
        source_list_append(t, opts[:from])
        _truncate_sql(t)
      end
    end

    # Formats an UPDATE statement using the given values.  See +update+.
    #
    #   DB[:items].update_sql(:price => 100, :category => 'software')
    #   # => "UPDATE items SET price = 100, category = 'software'
    #
    # Raises an +Error+ if the dataset is grouped or includes more
    # than one table.
    def update_sql(values = {})
      return static_sql(opts[:sql]) if opts[:sql]
      check_modification_allowed!
      clone(:values=>values)._update_sql
    end
    
    # ---------------------
    # :section: 9 - Internal Methods relating to SQL Creation
    # These methods, while public, are not designed to be used directly by the end user.
    # ---------------------
    
    # Given a type (e.g. select) and an array of clauses,
    # return an array of methods to call to build the SQL string.
    def self.clause_methods(type, clauses)
      clauses.map{|clause| :"#{type}_#{clause}_sql"}.freeze
    end

    # Map of emulated function names to native function names.
    EMULATED_FUNCTION_MAP = {}

    WILDCARD = LiteralString.new('*').freeze
    ALL = ' ALL'.freeze
    AND_SEPARATOR = " AND ".freeze
    APOS = "'".freeze
    APOS_RE = /'/.freeze
    ARRAY_EMPTY = '(NULL)'.freeze
    AS = ' AS '.freeze
    ASC = ' ASC'.freeze
    BACKSLASH = "\\".freeze
    BOOL_FALSE = "'f'".freeze
    BOOL_TRUE = "'t'".freeze
    BRACKET_CLOSE =  ']'.freeze
    BRACKET_OPEN = '['.freeze
    CASE_ELSE = " ELSE ".freeze
    CASE_END = " END)".freeze
    CASE_OPEN = '(CASE'.freeze
    CASE_THEN = " THEN ".freeze
    CASE_WHEN = " WHEN ".freeze
    CAST_OPEN = 'CAST('.freeze
    COLUMN_REF_RE1 = Sequel::COLUMN_REF_RE1
    COLUMN_REF_RE2 = Sequel::COLUMN_REF_RE2
    COLUMN_REF_RE3 = Sequel::COLUMN_REF_RE3
    COMMA = ', '.freeze
    COMMA_SEPARATOR = COMMA
    CONDITION_FALSE = '(1 = 0)'.freeze
    CONDITION_TRUE = '(1 = 1)'.freeze
    COUNT_FROM_SELF_OPTS = [:distinct, :group, :sql, :limit, :offset, :compounds]
    COUNT_OF_ALL_AS_COUNT = SQL::Function.new(:count, WILDCARD).as(:count)
    DATASET_ALIAS_BASE_NAME = 't'.freeze
    DEFAULT = LiteralString.new('DEFAULT').freeze
    DEFAULT_VALUES = " DEFAULT VALUES".freeze
    DELETE = 'DELETE'.freeze
    DELETE_CLAUSE_METHODS = clause_methods(:delete, %w'delete from where')
    DESC = ' DESC'.freeze
    DISTINCT = " DISTINCT".freeze
    DOT = '.'.freeze
    DOUBLE_APOS = "''".freeze
    DOUBLE_QUOTE = '""'.freeze
    EQUAL = ' = '.freeze
    ESCAPE = " ESCAPE ".freeze
    EXTRACT = 'extract('.freeze
    EXISTS = ['EXISTS '.freeze].freeze
    FOR_UPDATE = ' FOR UPDATE'.freeze
    FORMAT_DATE = "'%Y-%m-%d'".freeze
    FORMAT_DATE_STANDARD = "DATE '%Y-%m-%d'".freeze
    FORMAT_OFFSET = "%+03i%02i".freeze
    FORMAT_TIMESTAMP_RE = /%[Nz]/.freeze
    FORMAT_TIMESTAMP_USEC = ".%06d".freeze
    FORMAT_USEC = '%N'.freeze
    FRAME_ALL = "ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING".freeze
    FRAME_ROWS = "ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW".freeze
    FROM = ' FROM '.freeze
    FUNCTION_EMPTY = '()'.freeze
    GROUP_BY = " GROUP BY ".freeze
    HAVING = " HAVING ".freeze
    INSERT = "INSERT".freeze
    INSERT_CLAUSE_METHODS = clause_methods(:insert, %w'insert into columns values')
    INTO = " INTO ".freeze
    IS_LITERALS = {nil=>'NULL'.freeze, true=>'TRUE'.freeze, false=>'FALSE'.freeze}.freeze
    IS_OPERATORS = ::Sequel::SQL::ComplexExpression::IS_OPERATORS
    LIKE_OPERATORS = ::Sequel::SQL::ComplexExpression::LIKE_OPERATORS
    LIMIT = " LIMIT ".freeze
    N_ARITY_OPERATORS = ::Sequel::SQL::ComplexExpression::N_ARITY_OPERATORS
    NOT_SPACE = 'NOT '.freeze
    NULL = "NULL".freeze
    NULLS_FIRST = " NULLS FIRST".freeze
    NULLS_LAST = " NULLS LAST".freeze
    OFFSET = " OFFSET ".freeze
    ON = ' ON '.freeze
    ON_PAREN = " ON (".freeze
    ORDER_BY = " ORDER BY ".freeze
    ORDER_BY_NS = "ORDER BY ".freeze
    OVER = ' OVER '.freeze
    PAREN_CLOSE = ')'.freeze
    PAREN_OPEN = '('.freeze
    PAREN_SPACE_OPEN = ' ('.freeze
    PARTITION_BY = "PARTITION BY ".freeze
    QUALIFY_KEYS = [:select, :where, :having, :order, :group]
    QUESTION_MARK = '?'.freeze
    QUESTION_MARK_RE = /\?/.freeze
    QUOTE = '"'.freeze
    QUOTE_RE = /"/.freeze
    RETURNING = " RETURNING ".freeze
    SELECT = 'SELECT'.freeze
    SELECT_CLAUSE_METHODS = clause_methods(:select, %w'with select distinct columns from join where group having compounds order limit lock')
    SET = ' SET '.freeze
    SPACE = ' '.freeze
    SQL_WITH = "WITH ".freeze
    SPACE_WITH = " WITH ".freeze
    TILDE = '~'.freeze
    TIMESTAMP_FORMAT = "'%Y-%m-%d %H:%M:%S%N%z'".freeze
    STANDARD_TIMESTAMP_FORMAT = "TIMESTAMP #{TIMESTAMP_FORMAT}".freeze
    TWO_ARITY_OPERATORS = ::Sequel::SQL::ComplexExpression::TWO_ARITY_OPERATORS
    REGEXP_OPERATORS = ::Sequel::SQL::ComplexExpression::REGEXP_OPERATORS
    UNDERSCORE = '_'.freeze
    UPDATE = 'UPDATE'.freeze
    UPDATE_CLAUSE_METHODS = clause_methods(:update, %w'update table set where')
    USING = ' USING ('.freeze
    VALUES = " VALUES ".freeze
    V190 = '1.9.0'.freeze
    WHERE = " WHERE ".freeze

    PUBLIC_APPEND_METHODS = (<<-END).split.map{|x| x.to_sym}
      literal
      aliased_expression_sql
      array_sql
      boolean_constant_sql
      case_expression_sql
      cast_sql
      column_all_sql
      complex_expression_sql
      constant_sql
      delayed_evaluation_sql
      function_sql
      join_clause_sql
      join_on_clause_sql
      join_using_clause_sql
      negative_boolean_constant_sql
      ordered_expression_sql
      placeholder_literal_string_sql
      qualified_identifier_sql
      quote_identifier
      quote_schema_table
      quoted_identifier
      subscript_sql
      window_sql
      window_function_sql
    END
    PRIVATE_APPEND_METHODS = (<<-END).split.map{|x| x.to_sym}
      as_sql
      column_list
      compound_dataset_sql
      expression_list
      literal_array
      literal_blob
      literal_dataset
      literal_expression
      literal_hash
      literal_other
      literal_string
      literal_symbol
      source_list
      subselect_sql
      table_ref
    END

    # For each of the methods in the given array, define a method with
    # that name that returns a string with the SQL fragment that the
    # related *_append method would add.
    #
    # Do not call this method with untrusted input, as that can result in
    # arbitrary code execution.
    def self.def_append_methods(meths)
      meths.each do |meth|
        class_eval(<<-END, __FILE__, __LINE__ + 1)
          def #{meth}(*args, &block)
            s = ''
            #{meth}_append(s, *args, &block)
            s
          end
        END
      end
    end
    def_append_methods(PUBLIC_APPEND_METHODS + PRIVATE_APPEND_METHODS)
    private(*PRIVATE_APPEND_METHODS)

    # SQL fragment for AliasedExpression
    def aliased_expression_sql_append(sql, ae)
      literal_append(sql, ae.expression)
      as_sql_append(sql, ae.aliaz)
    end

    # SQL fragment for Array
    def array_sql_append(sql, a)
      if a.empty?
        sql << ARRAY_EMPTY
      else
        sql << PAREN_OPEN
        expression_list_append(sql, a)
        sql << PAREN_CLOSE
      end
    end

    # SQL fragment for BooleanConstants
    def boolean_constant_sql_append(sql, constant)
      if (constant == true || constant == false) && !supports_where_true?
        sql << (constant == true ? CONDITION_TRUE : CONDITION_FALSE)
      else
        literal_append(sql, constant)
      end
    end

    # SQL fragment for CaseExpression
    def case_expression_sql_append(sql, ce)
      sql << CASE_OPEN
      if ce.expression?
        sql << SPACE
        literal_append(sql, ce.expression)
      end
      w = CASE_WHEN
      t = CASE_THEN
      ce.conditions.each do |c,r|
        sql << w
        literal_append(sql, c)
        sql << t
        literal_append(sql, r)
      end
      sql << CASE_ELSE
      literal_append(sql, ce.default)
      sql << CASE_END
    end

    # SQL fragment for the SQL CAST expression
    def cast_sql_append(sql, expr, type)
      sql << CAST_OPEN
      literal_append(sql, expr)
      sql << AS << db.cast_type_literal(type).to_s
      sql << PAREN_CLOSE
    end

    # SQL fragment for specifying all columns in a given table
    def column_all_sql_append(sql, ca)
      qualified_identifier_sql_append(sql, ca.table, WILDCARD)
    end

    # SQL fragment for the complex expression.
    def complex_expression_sql_append(sql, op, args)
      case op
      when *IS_OPERATORS
        r = args.at(1)
        if r.nil? || supports_is_true?
          raise(InvalidOperation, 'Invalid argument used for IS operator') unless val = IS_LITERALS[r]
          sql << PAREN_OPEN
          literal_append(sql, args.at(0))
          sql << SPACE << op.to_s << SPACE
          sql << val << PAREN_CLOSE
        elsif op == :IS
          complex_expression_sql_append(sql, :"=", args)
        else
          complex_expression_sql_append(sql, :OR, [SQL::BooleanExpression.new(:"!=", *args), SQL::BooleanExpression.new(:IS, args.at(0), nil)])
        end
      when :IN, :"NOT IN"
        cols = args.at(0)
        vals = args.at(1)
        col_array = true if cols.is_a?(Array)
        if vals.is_a?(Array)
          val_array = true
          empty_val_array = vals == []
        end
        if empty_val_array
          literal_append(sql, empty_array_value(op, cols))
        elsif col_array
          if !supports_multiple_column_in?
            if val_array
              expr = SQL::BooleanExpression.new(:OR, *vals.to_a.map{|vs| SQL::BooleanExpression.from_value_pairs(cols.to_a.zip(vs).map{|c, v| [c, v]})})
              literal_append(sql, op == :IN ? expr : ~expr)
            else
              old_vals = vals
              vals = vals.naked if vals.is_a?(Sequel::Dataset)
              vals = vals.to_a
              val_cols = old_vals.columns
              complex_expression_sql_append(sql, op, [cols, vals.map!{|x| x.values_at(*val_cols)}])
            end
          else
            # If the columns and values are both arrays, use array_sql instead of
            # literal so that if values is an array of two element arrays, it
            # will be treated as a value list instead of a condition specifier.
            sql << PAREN_OPEN
            literal_append(sql, cols)
            sql << SPACE << op.to_s << SPACE
            if val_array
              array_sql_append(sql, vals)
            else
              literal_append(sql, vals)
            end
            sql << PAREN_CLOSE
          end
        else
          sql << PAREN_OPEN
          literal_append(sql, cols)
          sql << SPACE << op.to_s << SPACE
          literal_append(sql, vals)
          sql << PAREN_CLOSE
        end
      when :LIKE, :'NOT LIKE'
        sql << PAREN_OPEN
        literal_append(sql, args.at(0))
        sql << SPACE << op.to_s << SPACE
        literal_append(sql, args.at(1))
        sql << ESCAPE
        literal_append(sql, BACKSLASH)
        sql << PAREN_CLOSE
      when :ILIKE, :'NOT ILIKE'
        complex_expression_sql_append(sql, (op == :ILIKE ? :LIKE : :"NOT LIKE"), args.map{|v| Sequel.function(:UPPER, v)})
      when *TWO_ARITY_OPERATORS
        if REGEXP_OPERATORS.include?(op) && !supports_regexp?
          raise InvalidOperation, "Pattern matching via regular expressions is not supported on #{db.database_type}"
        end
        sql << PAREN_OPEN
        literal_append(sql, args.at(0))
        sql << SPACE << op.to_s << SPACE
        literal_append(sql, args.at(1))
        sql << PAREN_CLOSE
      when *N_ARITY_OPERATORS
        sql << PAREN_OPEN
        c = false
        op_str = " #{op} "
        args.each do |a|
          sql << op_str if c
          literal_append(sql, a)
          c ||= true
        end
        sql << PAREN_CLOSE
      when :NOT
        sql << NOT_SPACE
        literal_append(sql, args.at(0))
      when :NOOP
        literal_append(sql, args.at(0))
      when :'B~'
        sql << TILDE
        literal_append(sql, args.at(0))
      when :extract
        sql << EXTRACT << args.at(0).to_s << FROM
        literal_append(sql, args.at(1))
        sql << PAREN_CLOSE
      else
        raise(InvalidOperation, "invalid operator #{op}")
      end
    end
    
    # SQL fragment for constants
    def constant_sql_append(sql, constant)
      sql << constant.to_s
    end

    # SQL fragment for delayed evaluations, evaluating the
    # object and literalizing the returned value.
    def delayed_evaluation_sql_append(sql, callable)
      literal_append(sql, callable.call)
    end

    # SQL fragment specifying an emulated SQL function call.
    # By default, assumes just the function name may need to
    # be emulated, adapters should set an EMULATED_FUNCTION_MAP
    # hash mapping emulated functions to native functions in
    # their dataset class to setup the emulation.
    def emulated_function_sql_append(sql, f)
      _function_sql_append(sql, native_function_name(f.f), f.args)
    end

    # SQL fragment specifying an SQL function call without emulation.
    def function_sql_append(sql, f)
      _function_sql_append(sql, f.f, f.args)
    end

    # SQL fragment specifying a JOIN clause without ON or USING.
    def join_clause_sql_append(sql, jc)
      table = jc.table
      table_alias = jc.table_alias
      table_alias = nil if table == table_alias
      sql << SPACE << join_type_sql(jc.join_type) << SPACE
      identifier_append(sql, table)
      as_sql_append(sql, table_alias) if table_alias
    end

    # SQL fragment specifying a JOIN clause with ON.
    def join_on_clause_sql_append(sql, jc)
      join_clause_sql_append(sql, jc)
      sql << ON
      literal_append(sql, filter_expr(jc.on))
    end

    # SQL fragment specifying a JOIN clause with USING.
    def join_using_clause_sql_append(sql, jc)
      join_clause_sql_append(sql, jc)
      sql << USING
      column_list_append(sql, jc.using)
      sql << PAREN_CLOSE
    end
    
    # SQL fragment for NegativeBooleanConstants
    def negative_boolean_constant_sql_append(sql, constant)
      sql << NOT_SPACE
      boolean_constant_sql_append(sql, constant)
    end

    # SQL fragment for the ordered expression, used in the ORDER BY
    # clause.
    def ordered_expression_sql_append(sql, oe)
      literal_append(sql, oe.expression)
      sql << (oe.descending ? DESC : ASC)
      case oe.nulls
      when :first
        sql << NULLS_FIRST
      when :last
        sql << NULLS_LAST
      end
    end

    # SQL fragment for a literal string with placeholders
    def placeholder_literal_string_sql_append(sql, pls)
      args = pls.args
      str = pls.str
      sql << PAREN_OPEN if pls.parens
      if args.is_a?(Hash)
        re = /:(#{args.keys.map{|k| Regexp.escape(k.to_s)}.join('|')})\b/
        loop do
          previous, q, str = str.partition(re)
          sql << previous
          literal_append(sql, args[($1||q[1..-1].to_s).to_sym]) unless q.empty?
          break if str.empty?
        end
      elsif str.is_a?(Array)
        len = args.length
        str.each_with_index do |s, i|
          sql << s
          literal_append(sql, args[i]) unless i == len
        end
      else
        i = -1
        loop do
          previous, q, str = str.partition(QUESTION_MARK)
          sql << previous
           literal_append(sql, args.at(i+=1)) unless q.empty?
          break if str.empty?
        end
      end
      sql << PAREN_CLOSE if pls.parens
    end

    # SQL fragment for the qualifed identifier, specifying
    # a table and a column (or schema and table).
    # If 3 arguments are given, the 2nd should be the table/qualifier and the third should be
    # column/qualified.  If 2 arguments are given, the 2nd should be an SQL::QualifiedIdentifier.
    def qualified_identifier_sql_append(sql, table, column=(c = table.column; table = table.table; c))
      identifier_append(sql, table)
      sql << DOT
      identifier_append(sql, column)
    end

    # Adds quoting to identifiers (columns and tables). If identifiers are not
    # being quoted, returns name as a string.  If identifiers are being quoted
    # quote the name with quoted_identifier.
    def quote_identifier_append(sql, name)
      if name.is_a?(LiteralString)
        sql << name
      else
        name = name.value if name.is_a?(SQL::Identifier)
        name = input_identifier(name)
        if quote_identifiers?
          quoted_identifier_append(sql, name)
        else
          sql << name
        end
      end
    end

    # Separates the schema from the table and returns a string with them
    # quoted (if quoting identifiers)
    def quote_schema_table_append(sql, table)
      schema, table = schema_and_table(table)
      if schema
        quote_identifier_append(sql, schema)
        sql << DOT
      end
      quote_identifier_append(sql, table)
    end

    # This method quotes the given name with the SQL standard double quote. 
    # should be overridden by subclasses to provide quoting not matching the
    # SQL standard, such as backtick (used by MySQL and SQLite).
    def quoted_identifier_append(sql, name)
      sql << QUOTE << name.to_s.gsub(QUOTE_RE, DOUBLE_QUOTE) << QUOTE
    end

    # Split the schema information from the table, returning two strings,
    # one for the schema and one for the table.  The returned schema may
    # be nil, but the table will always have a string value.
    #
    # Note that this function does not handle tables with more than one
    # level of qualification (e.g. database.schema.table on Microsoft
    # SQL Server).
    def schema_and_table(table_name, sch=(db.default_schema if db))
      sch = sch.to_s if sch
      case table_name
      when Symbol
        s, t, _ = split_symbol(table_name)
        [s||sch, t]
      when SQL::QualifiedIdentifier
        [table_name.table.to_s, table_name.column.to_s]
      when SQL::Identifier
        [sch, table_name.value.to_s]
      when String
        [sch, table_name]
      else
        raise Error, 'table_name should be a Symbol, SQL::QualifiedIdentifier, SQL::Identifier, or String'
      end
    end

    # Splits table_name into an array of strings.
    #
    #   ds.split_qualifiers(:s) # ['s']
    #   ds.split_qualifiers(:t__s) # ['t', 's']
    #   ds.split_qualifiers(Sequel.qualify(:d, :t__s)) # ['d', 't', 's']
    #   ds.split_qualifiers(Sequel.qualify(:h__d, :t__s)) # ['h', 'd', 't', 's']
    def split_qualifiers(table_name, *args)
      case table_name
      when SQL::QualifiedIdentifier
        split_qualifiers(table_name.table, nil) + split_qualifiers(table_name.column, nil)
      else
        sch, table = schema_and_table(table_name, *args)
        sch ? [sch, table] : [table]
      end
    end

    # SQL fragment for specifying subscripts (SQL array accesses)
    def subscript_sql_append(sql, s)
      literal_append(sql, s.f)
      sql << BRACKET_OPEN
      expression_list_append(sql, s.sub)
      sql << BRACKET_CLOSE
    end

    # The SQL fragment for the given window's options.
    def window_sql_append(sql, opts)
      raise(Error, 'This dataset does not support window functions') unless supports_window_functions?
      sql << PAREN_OPEN
      window, part, order, frame = opts.values_at(:window, :partition, :order, :frame)
      space = false
      space_s = SPACE
      if window
        literal_append(sql, window)
        space = true
      end
      if part
        sql << space_s if space
        sql << PARTITION_BY
        expression_list_append(sql, Array(part))
        space = true
      end
      if order
        sql << space_s if space
        sql << ORDER_BY_NS
        expression_list_append(sql, Array(order))
        space = true
      end
      case frame
        when nil
          # nothing
        when :all
          sql << space_s if space
          sql << FRAME_ALL
        when :rows
          sql << space_s if space
          sql << FRAME_ROWS
        when String
          sql << space_s if space
          sql << frame
        else
          raise Error, "invalid window frame clause, should be :all, :rows, a string, or nil"
      end
      sql << PAREN_CLOSE
    end

    # The SQL fragment for the given window function's function and window.
    def window_function_sql_append(sql, function, window)
      literal_append(sql, function)
      sql << OVER
      literal_append(sql, window)
    end

    protected

    # Formats in INSERT statement using the stored columns and values.
    def _insert_sql
      clause_sql(:insert)
    end

    # Formats an UPDATE statement using the stored values.
    def _update_sql
      clause_sql(:update)
    end

    # Return a from_self dataset if an order or limit is specified, so it works as expected
    # with UNION, EXCEPT, and INTERSECT clauses.
    def compound_from_self
      (@opts[:limit] || @opts[:order]) ? from_self : self
    end
    
    private

    # Backbone of function_sql_append and emulated_function_sql_append.
    def _function_sql_append(sql, name, args)
      sql << name.to_s
      if args.empty?
        sql << FUNCTION_EMPTY
      else
        literal_append(sql, args)
      end
    end

    # Formats the truncate statement.  Assumes the table given has already been
    # literalized.
    def _truncate_sql(table)
      "TRUNCATE TABLE #{table}"
    end

    # Returns an appropriate symbol for the alias represented by s.
    def alias_alias_symbol(s)
      case s
      when Symbol
        s
      when String
        s.to_sym
      when SQL::Identifier
        s.value.to_s.to_sym
      else
        raise Error, "Invalid alias for alias_alias_symbol: #{s.inspect}"
      end
    end

    # Returns an appropriate alias symbol for the given object, which can be
    # a Symbol, String, SQL::Identifier, SQL::QualifiedIdentifier, or
    # SQL::AliasedExpression.
    def alias_symbol(sym)
      case sym
      when Symbol
        s, t, a = split_symbol(sym)
        a || s ? (a || t).to_sym : sym
      when String
        sym.to_sym
      when SQL::Identifier
        sym.value.to_s.to_sym
      when SQL::QualifiedIdentifier
        alias_symbol(sym.column)
      when SQL::AliasedExpression
        alias_alias_symbol(sym.aliaz)
      else
        raise Error, "Invalid alias for alias_symbol: #{sym.inspect}"
      end
    end

    # Clone of this dataset usable in aggregate operations.  Does
    # a from_self if dataset contains any parameters that would
    # affect normal aggregation, or just removes an existing
    # order if not.
    def aggregate_dataset
      options_overlap(COUNT_FROM_SELF_OPTS) ? from_self : unordered
    end

    # SQL fragment for specifying an alias.  expression should already be literalized.
    def as_sql_append(sql, aliaz)
      sql << AS
      quote_identifier_append(sql, aliaz)
    end
    
    # Raise an InvalidOperation exception if deletion is not allowed
    # for this dataset
    def check_modification_allowed!
      raise(InvalidOperation, "Grouped datasets cannot be modified") if opts[:group]
      raise(InvalidOperation, "Joined datasets cannot be modified") if !supports_modifying_joins? && joined_dataset?
    end

    # Alias of check_modification_allowed!
    def check_truncation_allowed!
      check_modification_allowed!
    end

    # Prepare an SQL statement by calling all clause methods for the given statement type.
    def clause_sql(type)
      sql = @opts[:append_sql] || sql_string_origin
      send("#{type}_clause_methods").each{|x| send(x, sql)}
      sql
    end

    # Converts an array of column names into a comma seperated string of 
    # column names. If the array is empty, a wildcard (*) is returned.
    def column_list_append(sql, columns)
      if (columns.nil? || columns.empty?)
        sql << WILDCARD
      else
        expression_list_append(sql, columns)
      end
    end

    # Yield each two pair of arguments to the block, which should
    # return a string representing the SQL code for those
    # two arguments.  If more than 2 arguments are provided, all
    # calls to the block # after the first will have a LiteralString
    # as the first argument, representing the application of the block to
    # the previous arguments.
    def complex_expression_arg_pairs(args)
      case args.length
      when 1
        literal(args.at(0))
      when 2
        yield args.at(0), args.at(1)
      else
        args.inject{|m, a| LiteralString.new(yield(m, a))}
      end
    end

    # The SQL to use for the dataset used in a UNION/INTERSECT/EXCEPT clause. 
    def compound_dataset_sql_append(sql, ds)
      subselect_sql_append(sql, ds)
    end

    # The alias to use for datasets, takes a number to make sure the name is unique.
    def dataset_alias(number)
      :"#{DATASET_ALIAS_BASE_NAME}#{number}"
    end
    
    # The strftime format to use when literalizing the time.
    def default_timestamp_format
      requires_sql_standard_datetimes? ? STANDARD_TIMESTAMP_FORMAT : TIMESTAMP_FORMAT
    end

    # The order of methods to call to build the DELETE SQL statement
    def delete_clause_methods
      DELETE_CLAUSE_METHODS
    end

    def delete_delete_sql(sql)
      sql << DELETE
    end

    # Converts an array of expressions into a comma separated string of
    # expressions.
    def expression_list_append(sql, columns)
      c = false
      co = COMMA
      columns.each do |col|
        sql << co if c
        literal_append(sql, col)
        c ||= true
      end
    end

    def empty_array_value(op, cols)
      if Sequel.empty_array_handle_nulls
        c = Array(cols)
        SQL::BooleanExpression.from_value_pairs(c.zip(c), :AND, op == :IN)
      else
        {1 => ((op == :IN) ? 0 : 1)}
      end
    end
    
    # Format the timestamp based on the default_timestamp_format, with a couple
    # of modifiers.  First, allow %N to be used for fractions seconds (if the
    # database supports them), and override %z to always use a numeric offset
    # of hours and minutes.
    def format_timestamp(v)
      v2 = db.from_application_timestamp(v)
      fmt = default_timestamp_format.gsub(FORMAT_TIMESTAMP_RE) do |m|
        if m == FORMAT_USEC
          format_timestamp_usec(v.is_a?(DateTime) ? v.sec_fraction*(RUBY_VERSION < V190 ? 86400000000 : 1000000) : v.usec) if supports_timestamp_usecs?
        else
          if supports_timestamp_timezones?
            # Would like to just use %z format, but it doesn't appear to work on Windows
            # Instead, the offset fragment is constructed manually
            minutes = (v2.is_a?(DateTime) ? v2.offset * 1440 : v2.utc_offset/60).to_i
            format_timestamp_offset(*minutes.divmod(60))
          end
        end
      end
      v2.strftime(fmt)
    end
    
    # Return the SQL timestamp fragment to use for the timezone offset.
    def format_timestamp_offset(hour, minute)
      sprintf(FORMAT_OFFSET, hour, minute)
    end

    # Return the SQL timestamp fragment to use for the fractional time part.
    # Should start with the decimal point.  Uses 6 decimal places by default.
    def format_timestamp_usec(usec)
      sprintf(FORMAT_TIMESTAMP_USEC, usec)
    end

    # Append the value, but special case regular (non-literal, non-blob) strings
    # so that they are considered as identifiers and not SQL strings.
    def identifier_append(sql, v)
      if v.is_a?(String)
        case v
        when LiteralString
          sql << v
        when SQL::Blob
          literal_append(sql, v)
        else
          quote_identifier_append(sql, v)
        end
      else
        literal_append(sql, v)
      end
    end
    alias table_ref_append identifier_append 
    
    # Append all identifiers in args interspersed by commas.
    def identifier_list_append(sql, args)
      c = false
      comma = COMMA
      args.each do |a|
        sql << comma if c
        identifier_append(sql, a)
        c ||= true
      end
    end

    # Modify the identifier returned from the database based on the
    # identifier_output_method.
    def input_identifier(v)
      (i = identifier_input_method) ? v.to_s.send(i) : v.to_s
    end

    # SQL fragment specifying the table to insert INTO
    def insert_into_sql(sql)
      sql << INTO
      source_list_append(sql, @opts[:from])
    end

    # The order of methods to call to build the INSERT SQL statement
    def insert_clause_methods
      INSERT_CLAUSE_METHODS
    end

    # SQL fragment specifying the columns to insert into
    def insert_columns_sql(sql)
      columns = opts[:columns]
      if columns && !columns.empty?
        sql << PAREN_SPACE_OPEN
        identifier_list_append(sql, columns)
        sql << PAREN_CLOSE
      end 
    end

    def insert_insert_sql(sql)
      sql << INSERT
    end

    # SQL fragment specifying the values to insert.
    def insert_values_sql(sql)
      case values = opts[:values]
      when Array
        if values.empty?
          sql << DEFAULT_VALUES
        else
          sql << VALUES
          literal_append(sql, values)
        end
      when Dataset
        sql << SPACE
        subselect_sql_append(sql, values)
      when LiteralString
        sql << SPACE << values
      else
        raise Error, "Unsupported INSERT values type, should be an Array or Dataset: #{values.inspect}"
      end
    end

    # SQL fragment specifying the values to return.
    def insert_returning_sql(sql)
      if opts.has_key?(:returning)
        sql << RETURNING
        column_list_append(sql, Array(opts[:returning]))
      end
    end
    alias delete_returning_sql insert_returning_sql
    alias update_returning_sql insert_returning_sql

    # SQL fragment specifying a JOIN type, converts underscores to
    # spaces and upcases.
    def join_type_sql(join_type)
      "#{join_type.to_s.gsub(UNDERSCORE, SPACE).upcase} JOIN"
    end

    # Whether this dataset is a joined dataset
    def joined_dataset?
     (opts[:from].is_a?(Array) && opts[:from].size > 1) || opts[:join]
    end

    # SQL fragment for Array.  Treats as an expression if an array of all two pairs, or as a SQL array otherwise.
    def literal_array_append(sql, v)
      if Sequel.condition_specifier?(v)
        literal_expression_append(sql, SQL::BooleanExpression.from_value_pairs(v))
      else
        array_sql_append(sql, v)
      end
    end

    # SQL fragment for BigDecimal
    def literal_big_decimal(v)
      d = v.to_s("F")
      v.nan? || v.infinite? ?  "'#{d}'" : d
    end

    # SQL fragment for SQL::Blob
    def literal_blob_append(sql, v)
      literal_string_append(sql, v)
    end

    # SQL fragment for Dataset.  Does a subselect inside parantheses.
    def literal_dataset_append(sql, v)
      sql << PAREN_OPEN
      subselect_sql_append(sql, v)
      sql << PAREN_CLOSE
    end

    # SQL fragment for Date, using the ISO8601 format.
    def literal_date(v)
      if requires_sql_standard_datetimes?
        v.strftime(FORMAT_DATE_STANDARD)
      else
        v.strftime(FORMAT_DATE)
      end
    end

    # SQL fragment for DateTime
    def literal_datetime(v)
      format_timestamp(v)
    end

    # SQL fragment for SQL::Expression, result depends on the specific type of expression.
    def literal_expression_append(sql, v)
      v.to_s_append(self, sql)
    end

    # SQL fragment for false
    def literal_false
      BOOL_FALSE
    end

    # SQL fragment for Float
    def literal_float(v)
      v.to_s
    end

    # SQL fragment for Hash, treated as an expression
    def literal_hash_append(sql, v)
      literal_expression_append(sql, SQL::BooleanExpression.from_value_pairs(v))
    end

    # SQL fragment for Integer
    def literal_integer(v)
      v.to_s
    end

    # SQL fragment for nil
    def literal_nil
      NULL
    end

    # SQL fragment for a type of object not handled by Dataset#literal.
    # Calls +sql_literal+ if object responds to it, otherwise raises an error.
    # Classes implementing +sql_literal+ should call a class-specific method on the dataset
    # provided and should add that method to Sequel::Dataset, allowing for adapters
    # to provide customized literalizations.
    # If a database specific type is allowed, this should be overriden in a subclass.
    def literal_other_append(sql, v)
      if v.respond_to?(:sql_literal_append)
        v.sql_literal_append(self, sql)
      elsif v.respond_to?(:sql_literal)
        sql << v.sql_literal(self)
      else
        raise Error, "can't express #{v.inspect} as a SQL literal"
      end
    end

    # SQL fragment for Sequel::SQLTime, containing just the time part
    def literal_sqltime(v)
      v.strftime("'%H:%M:%S#{format_timestamp_usec(v.usec) if supports_timestamp_usecs?}'")
    end

    # SQL fragment for String.  Doubles \ and ' by default.
    def literal_string_append(sql, v)
      sql << APOS << v.gsub(APOS_RE, DOUBLE_APOS) << APOS
    end

    # Converts a symbol into a column name. This method supports underscore
    # notation in order to express qualified (two underscores) and aliased
    # (three underscores) columns:
    #
    #   dataset.literal(:abc) #=> "abc"
    #   dataset.literal(:abc___a) #=> "abc AS a"
    #   dataset.literal(:items__abc) #=> "items.abc"
    #   dataset.literal(:items__abc___a) #=> "items.abc AS a"
    def literal_symbol_append(sql, v)
      c_table, column, c_alias = split_symbol(v)
      if c_table
        quote_identifier_append(sql, c_table)
        sql << DOT
      end
      quote_identifier_append(sql, column)
      as_sql_append(sql, c_alias) if c_alias
    end

    # SQL fragment for Time
    def literal_time(v)
      format_timestamp(v)
    end

    # SQL fragment for true
    def literal_true
      BOOL_TRUE
    end

    # Get the native function name given the emulated function name.
    def native_function_name(emulated_function)
      self.class.const_get(:EMULATED_FUNCTION_MAP).fetch(emulated_function, emulated_function)
    end

    # Returns a qualified column name (including a table name) if the column
    # name isn't already qualified.
    def qualified_column_name(column, table)
      if Symbol === column 
        c_table, column, _ = split_symbol(column)
        unless c_table
          case table
          when Symbol
            schema, table, t_alias = split_symbol(table)
            t_alias ||= Sequel::SQL::QualifiedIdentifier.new(schema, table) if schema
          when Sequel::SQL::AliasedExpression
            t_alias = table.aliaz
          end
          c_table = t_alias || table
        end
        ::Sequel::SQL::QualifiedIdentifier.new(c_table, column)
      else
        column
      end
    end
    
    # Qualify the given expression e to the given table.
    def qualified_expression(e, table)
      Qualifier.new(self, table).transform(e)
    end

    # The order of methods to call to build the SELECT SQL statement
    def select_clause_methods
      SELECT_CLAUSE_METHODS
    end

    # Modify the sql to add the columns selected
    def select_columns_sql(sql)
      sql << SPACE
      column_list_append(sql, @opts[:select])
    end

    # Modify the sql to add the DISTINCT modifier
    def select_distinct_sql(sql)
      if distinct = @opts[:distinct]
        sql << DISTINCT
        unless distinct.empty?
          sql << ON_PAREN
          expression_list_append(sql, distinct)
          sql << PAREN_CLOSE
        end
      end
    end

    # Modify the sql to add a dataset to the via an EXCEPT, INTERSECT, or UNION clause.
    # This uses a subselect for the compound datasets used, because using parantheses doesn't
    # work on all databases.  I consider this an ugly hack, but can't I think of a better default.
    def select_compounds_sql(sql)
      return unless c = @opts[:compounds]
      c.each do |type, dataset, all|
        sql << SPACE << type.to_s.upcase
        sql << ALL if all
        sql << SPACE
        compound_dataset_sql_append(sql, dataset)
      end
    end

    # Modify the sql to add the list of tables to select FROM
    def select_from_sql(sql)
      if f = @opts[:from]
        sql << FROM
        source_list_append(sql, f)
      end
    end
    alias delete_from_sql select_from_sql

    # Modify the sql to add the expressions to GROUP BY
    def select_group_sql(sql)
      if group = @opts[:group]
        sql << GROUP_BY
        if go = @opts[:group_options]
          if uses_with_rollup?
            expression_list_append(sql, group)
            sql << SPACE_WITH << go.to_s.upcase
          else
            sql << go.to_s.upcase << PAREN_OPEN
            expression_list_append(sql, group)
            sql << PAREN_CLOSE
          end
        else
          expression_list_append(sql, group)
        end
      end
    end

    # Modify the sql to add the filter criteria in the HAVING clause
    def select_having_sql(sql)
      if having = @opts[:having]
        sql << HAVING
        literal_append(sql, having)
      end
    end

    # Modify the sql to add the list of tables to JOIN to
    def select_join_sql(sql)
      if js = @opts[:join]
        js.each{|j| literal_append(sql, j)}
      end
    end

    # Modify the sql to limit the number of rows returned and offset
    def select_limit_sql(sql)
      if l = @opts[:limit]
        sql << LIMIT
        literal_append(sql, l)
      end
      if o = @opts[:offset]
        sql << OFFSET
        literal_append(sql, o)
      end
    end
  
    # Modify the sql to support the different types of locking modes.
    def select_lock_sql(sql)
      case l = @opts[:lock]
      when :update
        sql << FOR_UPDATE
      when String
        sql << SPACE << l
      end
    end

    # Modify the sql to add the expressions to ORDER BY
    def select_order_sql(sql)
      if o = @opts[:order]
        sql << ORDER_BY
        expression_list_append(sql, o)
      end
    end
    alias delete_order_sql select_order_sql
    alias update_order_sql select_order_sql

    def select_select_sql(sql)
      sql << SELECT
    end

    # Modify the sql to add the filter criteria in the WHERE clause
    def select_where_sql(sql)
      if w = @opts[:where]
        sql << WHERE
        literal_append(sql, w)
      end
    end
    alias delete_where_sql select_where_sql
    alias update_where_sql select_where_sql
    
    # SQL Fragment specifying the WITH clause
    def select_with_sql(sql)
      ws = opts[:with]
      return if !ws || ws.empty?
      sql << select_with_sql_base
      c = false
      comma = COMMA
      ws.each do |w|
        sql << comma if c
        quote_identifier_append(sql, w[:name])
        if args = w[:args]
         sql << PAREN_OPEN
         identifier_list_append(sql, args)
         sql << PAREN_CLOSE
        end
        sql << AS
        literal_dataset_append(sql, w[:dataset])
        c ||= true
      end
      sql << SPACE
    end
    alias delete_with_sql select_with_sql
    alias insert_with_sql select_with_sql
    alias update_with_sql select_with_sql
    
    # The base keyword to use for the SQL WITH clause
    def select_with_sql_base
      SQL_WITH
    end

    # Converts an array of source names into into a comma separated list.
    def source_list_append(sql, sources)
      raise(Error, 'No source specified for query') if sources.nil? || sources == []
      identifier_list_append(sql, sources)
    end
    
    # Delegate to Sequel.split_symbol.
    def split_symbol(sym)
      Sequel.split_symbol(sym)
    end

    # The string that is appended to to create the SQL query, the empty
    # string by default
    def sql_string_origin
      ''
    end
    
    # SQL to use if this dataset uses static SQL.  Since static SQL
    # can be a PlaceholderLiteralString in addition to a String,
    # we literalize nonstrings.
    def static_sql(sql)
      if append_sql = @opts[:append_sql]
        if sql.is_a?(String)
          append_sql << sql
        else
          literal_append(append_sql, sql)
        end
      else
        if sql.is_a?(String)
          sql
        else
          literal(sql)
        end
      end
    end

    # SQL fragment for a subselect using the given database's SQL.
    def subselect_sql_append(sql, ds)
      ds.clone(:append_sql=>sql).sql
    end

    # The order of methods to call to build the UPDATE SQL statement
    def update_clause_methods
      UPDATE_CLAUSE_METHODS
    end

    # SQL fragment specifying the tables from with to delete.
    # Includes join table if modifying joins is allowed.
    def update_table_sql(sql)
      sql << SPACE
      source_list_append(sql, @opts[:from])
      select_join_sql(sql) if supports_modifying_joins?
    end

    # The SQL fragment specifying the columns and values to SET.
    def update_set_sql(sql)
      values = opts[:values]
      sql << SET
      if values.is_a?(Hash)
        values = opts[:defaults].merge(values) if opts[:defaults]
        values = values.merge(opts[:overrides]) if opts[:overrides]
        c = false
        eq = EQUAL
        values.each do |k, v|
          sql << COMMA if c
          if k.is_a?(String) && !k.is_a?(LiteralString)
            quote_identifier_append(sql, k)
          else
            literal_append(sql, k)
          end
          sql << eq
          literal_append(sql, v)
          c ||= true
        end
      else
        sql << values
      end
    end

    def update_update_sql(sql)
      sql << UPDATE
    end
  end
end