destination.rb in darrell-activewarehouse-etl-0.9.1.6

- old
+ new

@@ -207,11 +207,16 @@
         unless has_natural_key?(row)
           buffer << row
           return
         end
         
-        @timestamp = Time.now
+        @timestamp = case configuration[:scd][:timestamp]
+                     when Time, Date then configuration[:scd][:timestamp]
+                     when Symbol then row[configuration[:scd][:timestamp]]
+                     when nil then Time.now
+                     else raise "Unknown timestamp: #{configuration[:scd][:timestamp].inspect}. Use Time or Date for a specific time, a symbol for a value from each row, or nil for the current time"
+                     end
 
         # See if the scd_fields of the current record have changed
         # from the last time this record was loaded into the data
         # warehouse. If they match then throw away this row (no need
         # to process). If they do not match then the record is an
@@ -316,10 +321,16 @@
           delete_outdated_record
           
           ETL::Engine.logger.debug "expiring original record"
           @existing_row[scd_end_date_field] = @timestamp
           @existing_row[scd_latest_version_field] = false
+
+          if configuration[:scd][:merge_nils]
+            scd_fields(row).each do |f|
+              row[f] ||= @existing_row[f]
+            end
+          end
           
           buffer << @existing_row
 
         elsif scd_type == 1
           # SCD Type 1: only the new row should be added
@@ -379,20 +390,19 @@
       end
       
       # Check whether non-scd fields have changed since the last
       # load of this record.
       def has_scd_field_changes?(row)
-        scd_fields(row).any? { |csd_field| 
-          ETL::Engine.logger.debug "Row: #{row.inspect}"
-          ETL::Engine.logger.debug "Existing Row: #{@existing_row.inspect}"
-          ETL::Engine.logger.debug "comparing: #{row[csd_field].to_s} != #{@existing_row[csd_field].to_s}"
-          if row[csd_field].to_s != @existing_row[csd_field].to_s
-            x=true
-          else
-            x=false
-          end
-          ETL::Engine.logger.debug "Fields differ?: #{x}"
-          x
+        fields = scd_fields(row)
+        ETL::Engine.logger.debug "         Row: %s" % row.slice(*fields).inspect
+        ETL::Engine.logger.debug "Existing Row: %s" % @existing_row.slice(*fields).inspect
+
+        fields.any? { |csd_field| 
+          mismatch = configuration[:scd][:merge_nils] ? !row[csd_field].nil? : true
+          mismatch = mismatch && (row[csd_field].to_s != @existing_row[csd_field].to_s)
+
+          ETL::Engine.logger.debug "#{csd_field}: " + (mismatch ? row[csd_field].to_s + " != " + @existing_row[csd_field].to_s : @existing_row[csd_field].to_s)
+          mismatch
         }
       end
       
       # Check whether non-scd fields have changed since the last
       # load of this record.