database_join_processor.rb in activewarehouse-etl-1.0.0.rc1

- old
+ new

@@ -19,10 +19,13 @@
       def initialize(control, configuration)
         super
         @target = configuration[:target]
         @query = configuration[:query]
         @fields = configuration[:fields]
+        raise ControlError, ":target must be specified" unless @target
+        raise ControlError, ":query must be specified" unless @query
+        raise ControlError, ":fields must be specified" unless @fields
       end
       
       # Get a String identifier for the source
       def to_s
         "#{host}/#{database}"
@@ -38,13 +41,31 @@
         end
 
         ETL::Engine.logger.debug("Executing select: #{q}")
         res = connection.execute(q)
 
-        res.each_hash do |r|
-          @fields.each do |field|
-            row[field.to_sym] = r[field]
-          end
+        # TODO - refactor this and move it (and similar code around) to adapter_extensions
+        case connection.class.name
+          when "ActiveRecord::ConnectionAdapters::PostgreSQLAdapter";
+            res.each do |r|
+              @fields.each do |field|
+                row[field.to_sym] = r[field.to_s]
+              end
+            end
+          when "ActiveRecord::ConnectionAdapters::Mysql2Adapter";
+            res.each(:as => :hash) do |r|
+              @fields.each do |field|
+                row[field.to_sym] = r[field.to_s]
+              end
+            end
+          when "ActiveRecord::ConnectionAdapters::MysqlAdapter";
+            res.each_hash do |r|
+              @fields.each do |field|
+                row[field.to_sym] = r[field.to_s]
+              end
+            end
+            res.free
+          else raise "Unsupported adapter #{connection.class} for this destination"
         end
 
         return row
       end
 