rg/pg_array

.gitignore

@@ -1,3 +1,4 @@
+spec/database.yml
 *.gem
 *.rbc
 .bundle

lib/sequel/adapters/redshift.rb

@@ -1,31 +1,118 @@
 require 'sequel/adapters/postgres'
+require 'sequel/adapters/shared/redshift'
 
 module Sequel
   module Redshift
     include Postgres
 
     class Database < Postgres::Database
+      include Sequel::Redshift::DatabaseMethods
+
       set_adapter_scheme :redshift
 
+      DIST_KEY = ' DISTKEY'.freeze
+      SORT_KEY = ' SORTKEY'.freeze
+
+      # The order of column modifiers to use when defining a column.
+      COLUMN_DEFINITION_ORDER = [:collate, :default, :primary_key, :dist_key, :sort_key, :null, :unique, :auto_increment, :references]
+
+      def dataset_class_default
+        Sequel::Redshift::Dataset
+      end
+
+      # We need to change these default settings because they correspond to
+      # Postgres configuration variables which do not exist in Redshift
+      def adapter_initialize
+        @opts.merge!(
+          force_standard_strings: false,
+          client_min_messages:    false
+        )
+        super
+      end
+
+      def table_exists?(name)
+        sql = <<~SQL
+          SELECT EXISTS (
+            SELECT * FROM pg_catalog.pg_class c
+            JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
+            WHERE  n.nspname = 'public'
+            AND  c.relname = '#{name}'
+            AND  c.relkind = 'r'
+          );
+        SQL
+        fetch(sql).first.fetch(:"?column?")
+      end
+
       def column_definition_primary_key_sql(sql, column)
-        result = super
-        result << ' IDENTITY' if result
-        result
+        if column[:primary_key] && column[:type] == Integer
+          sql << ' IDENTITY'
+        end
+      end
+
+      # Add DISTKEY SQL fragment to column creation SQL.
+      def column_definition_dist_key_sql(sql, column)
+        if column[:dist_key]
+          sql << DIST_KEY
+        end
+      end
+
+      # Add SORTKEY SQL fragment to column creation SQL.
+      def column_definition_sort_key_sql(sql, column)
+        if column[:sort_key]
+          sql << SORT_KEY
+        end
       end
 
       def serial_primary_key_options
         # redshift doesn't support serial type
         super.merge(serial: false)
       end
+
+      # DROP TABLE IF EXISTS is now supported by Redshift
+      def supports_drop_table_if_exists?
+        true
+      end
+
+      # None of the alter table operation are combinable.
+      def combinable_alter_table_op?(op)
+        false
+      end
+
+      def supports_transactional_ddl?
+        false
+      end
+
+      def supports_savepoints?
+        false
+      end
     end
 
     class Dataset < Postgres::Dataset
       Database::DatasetClass = self
 
-      # Redshift doesn't support RETURNING statement
-      def insert_returning_sql(sql)
-        # do nothing here
-        sql
+      Dataset.def_sql_method(self, :select, [['if opts[:values]', %w'values order limit'], ['elsif server_version >= 80400', %w'with select distinct columns from join where group having window compounds order limit lock'], ['else', %w'with select distinct columns from join where group having compounds order limit lock']])
+
+      def initialize(db)
+        @db = db
+        @opts = {disable_insert_returning: true}.freeze
+        @cache = {}
+        freeze
+      end
+
+      def supports_cte?(type = :select)
+        true if type == :select
+      end
+
+      def supports_insert_select?
+        false
+      end
+
+      def supports_returning?(type)
+        false
+      end
+
+      def supports_window_functions?
+        true
       end
     end
   end

lib/sequel/adapters/shared/redshift.rb

@@ -0,0 +1,78 @@
+module Sequel
+  module Redshift
+    module DatabaseMethods
+
+      REDSHIFT_STRING_MAXIMUM_SIZE = 255
+
+      # Redshift does not support arrays (type of pg_index.indkey is INT2VECTOR),
+      # and because of that we can't determine the primary key - so we set it to false.
+      #
+      # The workaround for now is to use `set_primary_key` inside the sequel model.
+      def schema_parse_table(table_name, opts)
+        m  = output_identifier_meth(opts[:dataset])
+
+        ds = metadata_dataset.select(:pg_attribute__attname___name,
+            SQL::Cast.new(:pg_attribute__atttypid, :integer).as(:oid),
+            SQL::Cast.new(:basetype__oid, :integer).as(:base_oid),
+            SQL::Function.new(:format_type, :basetype__oid, :pg_type__typtypmod).as(:db_base_type),
+            SQL::Function.new(:format_type, :pg_type__oid, :pg_attribute__atttypmod).as(:db_type),
+            SQL::Function.new(:pg_get_expr, :pg_attrdef__adbin, :pg_class__oid).as(:default),
+            SQL::BooleanExpression.new(:NOT, :pg_attribute__attnotnull).as(:allow_null)).
+          from(:pg_class).
+          join(:pg_attribute, :attrelid=>:oid).
+          join(:pg_type, :oid=>:atttypid).
+          left_outer_join(:pg_type___basetype, :oid=>:typbasetype).
+          left_outer_join(:pg_attrdef, :adrelid=>:pg_class__oid, :adnum=>:pg_attribute__attnum).
+          filter(:pg_attribute__attisdropped=>false).
+          filter{|o| o.pg_attribute__attnum > 0}.
+          filter(:pg_class__oid=>regclass_oid(table_name, opts)).
+          order(:pg_attribute__attnum)
+
+        ds.map do |row|
+          row[:default] = nil if blank_object?(row[:default])
+          if row[:base_oid]
+            row[:domain_oid] = row[:oid]
+            row[:oid] = row.delete(:base_oid)
+            row[:db_domain_type] = row[:db_type]
+            row[:db_type] = row.delete(:db_base_type)
+          else
+            row.delete(:base_oid)
+            row.delete(:db_base_type)
+          end
+          row[:type] = schema_column_type(row[:db_type])
+          row[:primary_key] = false
+          [m.call(row.delete(:name)), row]
+        end
+      end
+
+      # Redshift changes text to varchar with maximum size of 256, and it complains if you will give text column
+      def type_literal_generic_string(column)
+        "#{column[:fixed] ? 'CHAR' : 'VARCHAR'}(#{column[:size] || REDSHIFT_STRING_MAXIMUM_SIZE})"
+      end
+
+      # The version of the PostgreSQL server, used for determining capability.
+      def server_version(server=nil)
+        @server_version ||= 80002
+      end
+
+      def create_table_suffix_sql(name, options)
+        sql = create_table_attributes(name, options)
+        "#{super}#{sql}"
+      end
+
+      def create_table_attributes(name, options)
+        sql = String.new
+        sql << " DISTSTYLE #{options[:diststyle].upcase}" if options[:diststyle]
+        sql << " DISTKEY(#{options[:distkey]})" if options[:distkey]
+        sql << " SORTKEY(#{Array(options[:sortkey]).join(', ')})" if options[:sortkey]
+        sql
+      end
+
+      def create_table_as_sql(name, sql, options)
+        result  = create_table_prefix_sql(name, options)
+        result += create_table_attributes(name, options)
+        result += " AS #{sql}"
+      end
+    end
+  end
+end

lib/sequel/extensions/median.rb

@@ -0,0 +1,55 @@
+# frozen-string-literal: true
+#
+module Sequel
+  module SQL
+    module Builders
+      # Return a median expression.
+      def median(*a)
+        Median.new(*a)
+      end
+    end
+
+    # The Median class represents a median function for Redshift and Postgresql databases.
+    class Median < GenericExpression
+      include OrderMethods
+
+      module DatasetMethods
+        # Append the SQL fragment for the Median expression to the SQL query.
+        def median_sql_append(sql, sa)
+          if defined?(super)
+            return super
+          end
+
+          expr = sa.expr
+
+          case db_type = db.adapter_scheme
+          when :postgres
+
+            literal_append(sql, Function.new(:percentile_disc, 0.5))
+            sql << " WITHIN GROUP (ORDER BY "
+            identifier_append(sql, expr)
+            sql << ")"
+
+          when :redshift
+
+            literal_append(sql, Function.new(:median, expr))
+
+          else
+            raise Error, "median is not implemented on #{db.database_type}"
+          end
+        end
+      end
+
+      attr_reader :expr
+
+      # Set the expression and separator
+      def initialize(expr)
+        @expr = expr
+      end
+
+      to_s_method :median_sql
+    end
+  end
+
+  Dataset.register_extension(:median, SQL::Median::DatasetMethods)
+end