Add support for date columns to jdbc static and streaming filters (#171)

Added support for SQL DATE columns to the jdbc static and streaming filters.

Before this changes, querying data from tables with DATE columns would produce a missing converter error caused by the RubyDate fields.

Author: edmocosta

.ci/setup.sql

@@ -6,7 +6,10 @@ create database jdbc_streaming_db;
 create table reference_table (
     ip VARCHAR(50) NOT NULL,
     name VARCHAR(50) NOT NULL,
-    location VARCHAR(50) NOT NULL
+    location VARCHAR(50) NOT NULL,
+    entry_date DATE NOT NULL,
+    entry_time TIME NOT NULL,
+    timestamp TIMESTAMP NOT NULL
 );
 
 
@@ -23,7 +26,10 @@ DO $$
                 INSERT INTO reference_table
                 VALUES ((SELECT FORMAT(ipTemplate, counter)),
                         (SELECT FORMAT(nameTemplate, counter)),
-                        (SELECT FORMAT(locationTemplate, counter)));
+                        (SELECT FORMAT(locationTemplate, counter)),
+                        '2003-02-01',
+                        '10:05:00',
+                        '2003-02-01 01:02:03');
                 counter = counter + 1;
             END LOOP;
 END $$;
@@ -37,13 +43,16 @@ create database jdbc_static_db;
 create table reference_table (
     ip VARCHAR(50) NOT NULL,
     name VARCHAR(50) NOT NULL,
-    location VARCHAR(50) NOT NULL
+    location VARCHAR(50) NOT NULL,
+    entry_date DATE NOT NULL,
+    entry_time TIME NOT NULL,
+    timestamp TIMESTAMP NOT NULL
 );
 
 
-INSERT INTO reference_table VALUES ('10.1.1.1', 'ldn-server-1', 'LDN-2-3-4');
-INSERT INTO reference_table VALUES ('10.2.1.1', 'nyc-server-1', 'NYC-5-2-8');
-INSERT INTO reference_table VALUES ('10.3.1.1', 'mv-server-1', 'MV-9-6-4');
+INSERT INTO reference_table VALUES ('10.1.1.1', 'ldn-server-1', 'LDN-2-3-4','2003-02-01', '10:05:00', '2003-02-01 01:02:03');
+INSERT INTO reference_table VALUES ('10.2.1.1', 'nyc-server-1', 'NYC-5-2-8','2003-02-01', '10:05:00', '2003-02-01 01:02:03');
+INSERT INTO reference_table VALUES ('10.3.1.1', 'mv-server-1', 'MV-9-6-4', '2003-02-01', '10:05:00', '2003-02-01 01:02:03');
 
 create DATABASE jdbc_input_db;
 
@@ -52,8 +61,11 @@ create DATABASE jdbc_input_db;
 CREATE TABLE employee (
     emp_no integer NOT NULL,
     first_name VARCHAR (50) NOT NULL,
-    last_name VARCHAR (50) NOT NULL
+    last_name VARCHAR (50) NOT NULL,
+    entry_date DATE NOT NULL,
+    entry_time TIME NOT NULL,
+    timestamp TIMESTAMP NOT NULL
 );
 
-INSERT INTO employee VALUES (1, 'David', 'Blenkinsop');
-INSERT INTO employee VALUES (2, 'Mark', 'Guckenheimer');
+INSERT INTO employee VALUES (1, 'David', 'Blenkinsop', '2003-02-01', '10:05:00', '2003-02-01 01:02:03');
+INSERT INTO employee VALUES (2, 'Mark', 'Guckenheimer', '2003-02-01', '10:05:00','2003-02-01 01:02:03');

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 5.5.0 
+  - Feat: add support for SQL `DATE` columns to jdbc static and streaming filters [#171](https://github.com/logstash-plugins/logstash-integration-jdbc/pull/171)
+
 ## 5.4.11
   - Fixes an issue in which any one instance of a JDBC input plugin using `jdbc_default_timezone` changes the behaviour of plugin instances that do _not_ use `jdbc_default_timezone`, ensuring that timezone offsets remain consistent for each instance of the plugin _as configured_ [#151](https://github.com/logstash-plugins/logstash-integration-jdbc/pull/151)
   - Fixes an exception that could occur while reloading `jdbc_static` databases when the underlying connection to the remote has been broken [#165](https://github.com/logstash-plugins/logstash-integration-jdbc/pull/165)

lib/logstash/filters/jdbc/lookup.rb

@@ -1,10 +1,12 @@
 # encoding: utf-8
 require_relative "lookup_result"
 require "logstash/util/loggable"
+require "logstash/plugin_mixins/jdbc/value_handler"
 
 module LogStash module Filters module Jdbc
   class Lookup
     include LogStash::Util::Loggable
+    include LogStash::PluginMixins::Jdbc::ValueHandler
 
     class Sprintfier
       def initialize(param)
@@ -134,15 +136,13 @@ def tag_default(event)
 
     def load_data_from_local(local, query, params, result)
       local.fetch(query, params).each do |row|
-        stringified = row.inject({}){|hash,(k,v)| hash[k.to_s] = v; hash} #Stringify row keys
-        result.push(stringified)
+        result.push(extract_values_from(row))
       end
     end
 
     def load_data_from_prepared(_local, _query, params, result)
       @prepared_statement.call(params).each do |row|
-        stringified = row.inject({}){|hash,(k,v)| hash[k.to_s] = v; hash} #Stringify row keys
-        result.push(stringified)
+        result.push(extract_values_from(row))
       end
     end
 

lib/logstash/plugin_mixins/jdbc/jdbc.rb

@@ -6,13 +6,15 @@
 require_relative "value_tracking"
 require_relative "timezone_proxy"
 require_relative "statement_handler"
+require_relative "value_handler"
 
 java_import java.util.concurrent.locks.ReentrantLock
 
 # Tentative of abstracting JDBC logic to a mixin
 # for potential reuse in other plugins (input/output)
 module LogStash  module PluginMixins module Jdbc
   module Jdbc
+    include LogStash::PluginMixins::Jdbc::ValueHandler
     # This method is called when someone includes this module
     def self.included(base)
       # Add these methods to the 'base' given.
@@ -252,25 +254,6 @@ def get_column_value(row)
         row[@tracking_column.to_sym]
       end
     end
-
-    private
-    #Stringify row keys and decorate values when necessary
-    def extract_values_from(row)
-      Hash[row.map { |k, v| [k.to_s, decorate_value(v)] }]
-    end
-
-    private
-    def decorate_value(value)
-      case value
-      when Time
-        # transform it to LogStash::Timestamp as required by LS
-        LogStash::Timestamp.new(value)
-      when Date, DateTime
-        LogStash::Timestamp.new(value.to_time)
-      else
-        value
-      end
-    end
   end
 end end end
 

lib/logstash/plugin_mixins/jdbc/value_handler.rb

@@ -0,0 +1,24 @@
+# encoding: utf-8
+require "time"
+require "date"
+
+module LogStash module PluginMixins module Jdbc
+  # Provides functions to extract the row's values, ensuring column types
+  # are properly decorated to become coercible to a LogStash::Event.
+  module ValueHandler
+    # Stringify the row keys and decorate values when necessary
+    def extract_values_from(row)
+      Hash[row.map { |k, v| [k.to_s, decorate_value(v)] }]
+    end
+
+    # Decorate the value so it can be used as a LogStash::Event field
+    def decorate_value(value)
+      case value
+      when Date, DateTime
+        value.to_time
+      else
+        value
+      end
+    end
+  end
+end end end

lib/logstash/plugin_mixins/jdbc_streaming/statement_handler.rb

@@ -1,12 +1,15 @@
 # encoding: utf-8
 require "logstash/util/loggable"
+require "logstash/plugin_mixins/jdbc/value_handler"
 
 module LogStash module PluginMixins module JdbcStreaming
   # so as to not clash with the class of the same name and function in the jdbc input
   # this is in the `module JdbcStreaming` namespace
   # this duplication can be removed in a universal plugin
 
   class StatementHandler
+    include LogStash::PluginMixins::Jdbc::ValueHandler
+
     def self.build_statement_handler(plugin)
       klass = plugin.use_prepared_statements ? PreparedStatementHandler : NormalStatementHandler
       klass.new(plugin)
@@ -86,7 +89,7 @@ def cache_lookup(db, event)
     def execute_extract_records(db, params, result)
       dataset = db[statement, params] # returns a Sequel dataset
       dataset.all do |row|
-        result.push row.inject({}){|hash,(k,v)| hash[k.to_s] = v; hash} # Stringify row keys
+        result.push extract_values_from(row)
       end
     end
 
@@ -113,7 +116,7 @@ def cache_lookup(db, event)
     def execute_extract_records(db, params, result)
       records = db.call(name, params) # returns an array of hashes
       records.each do |row|
-        result.push row.inject({}){|hash,(k,v)| hash[k.to_s] = v; hash} #Stringify row keys
+        result.push extract_values_from(row)
       end
     end
 

logstash-integration-jdbc.gemspec

@@ -1,6 +1,6 @@
 Gem::Specification.new do |s|
   s.name = 'logstash-integration-jdbc'
-  s.version         = '5.4.11'
+  s.version         = '5.5.0'
   s.licenses = ['Apache License (2.0)']
   s.summary         = "Integration with JDBC - input and filter plugins"
   s.description     = "This gem is a Logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/logstash-plugin install gemname. This gem is not a stand-alone program"

spec/filters/integration/jdbc_static_spec.rb

@@ -116,6 +116,45 @@ module LogStash module Filters
           plugin.filter(event)
           expect(event.get("server")).to eq([{"ip"=>"10.3.1.1", "name"=>"mv-server-1", "location"=>"MV-9-6-4"}])
         end
+
+        context 'and record with temporal columns' do
+          let(:loader_statement) { "SELECT ip, name, location, entry_date, entry_time, timestamp FROM reference_table" }
+          let(:local_db_objects) do
+            [
+              {
+                "name" => "servers",
+                "columns" => [
+                  %w[ip varchar(64)],
+                  %w[name varchar(64)],
+                  %w[location varchar(64)],
+                  %w[entry_date date],
+                  %w[entry_time time],
+                  %w[timestamp timestamp]
+                ]
+              },
+            ]
+          end
+
+          before(:each) { plugin.register }
+
+          subject { event.get("server").first }
+
+          it "maps the DATE to a Logstash Timestamp" do
+            plugin.filter(event)
+            expect(subject['entry_date']).to eq(LogStash::Timestamp.new(Time.new(2003, 2, 1)))
+          end
+
+          it "maps the TIME field to a Logstash Timestamp" do
+            plugin.filter(event)
+            now = DateTime.now
+            expect(subject['entry_time']).to eq(LogStash::Timestamp.new(Time.new(now.year, now.month, now.day, 10, 5, 0)))
+          end
+
+          it "maps the TIMESTAMP to a Logstash Timestamp" do
+            plugin.filter(event)
+            expect(subject['timestamp']).to eq(LogStash::Timestamp.new(Time.new(2003, 2, 1, 1, 2, 3)))
+          end
+        end
       end
 
       context "under normal conditions when index_columns is not specified" do

spec/filters/integration/jdbcstreaming_spec.rb

@@ -63,7 +63,32 @@ class TestJdbcStreaming < JdbcStreaming
       end
     end
 
-    describe "In Prepared Statement mode, found record - uses row" do
+    describe 'found record with temporal columns' do
+      let(:idx) { 200 }
+      let(:statement) { "SELECT entry_date, entry_time, timestamp FROM reference_table WHERE ip = :ip" }
+
+      before(:each) { plugin.register }
+
+      subject { event.get("server").first }
+
+      it "maps the DATE to a Logstash Timestamp" do
+        plugin.filter(event)
+        expect(subject['entry_date']).to eq(LogStash::Timestamp.new(Time.new(2003, 2, 1)))
+      end
+
+      it "maps the TIME field to a Logstash Timestamp" do
+        plugin.filter(event)
+        now = DateTime.now
+        expect(subject['entry_time']).to eq(LogStash::Timestamp.new(Time.new(now.year, now.month, now.day, 10, 5, 0)))
+      end
+
+      it "maps the TIMESTAMP to a Logstash Timestamp" do
+        plugin.filter(event)
+        expect(subject['timestamp']).to eq(LogStash::Timestamp.new(Time.new(2003, 2, 1, 1, 2, 3)))
+      end
+    end
+
+    context 'prepared statement mode' do
       let(:idx) { 200 }
       let(:statement) { "SELECT name, location FROM reference_table WHERE ip = ?" }
       let(:settings) do
@@ -82,10 +107,37 @@ class TestJdbcStreaming < JdbcStreaming
           "sequel_opts" => {"pool_timeout" => 600}
         }
       end
-      it "fills in the target" do
-        plugin.filter(event)
-        expect(event.get("server")).to eq([{"name" => "ldn-server-#{idx}", "location" => "LDN-#{idx}-2-3"}])
-        expect((event.get("tags") || []) & ["lookup_failed", "default_used_instead"]).to be_empty
+
+      describe "found record - uses row" do
+        it "fills in the target" do
+          plugin.filter(event)
+          expect(event.get("server")).to eq([{"name" => "ldn-server-#{idx}", "location" => "LDN-#{idx}-2-3"}])
+          expect((event.get("tags") || []) & ["lookup_failed", "default_used_instead"]).to be_empty
+        end
+      end
+
+      describe 'found record with temporal columns' do
+        let(:statement) { "SELECT entry_date, entry_time, timestamp FROM reference_table WHERE ip = ?" }
+
+        before(:each) { plugin.register }
+
+        subject { event.get("server").first }
+
+        it "maps the DATE to a Logstash Timestamp" do
+          plugin.filter(event)
+          expect(subject['entry_date']).to eq(LogStash::Timestamp.new(Time.new(2003, 2, 1)))
+        end
+
+        it "maps the TIME field to a Logstash Timestamp" do
+          plugin.filter(event)
+          now = DateTime.now
+          expect(subject['entry_time']).to eq(LogStash::Timestamp.new(Time.new(now.year, now.month, now.day, 10, 5, 0)))
+        end
+
+        it "maps the TIMESTAMP to a Logstash Timestamp" do
+          plugin.filter(event)
+          expect(subject['timestamp']).to eq(LogStash::Timestamp.new(Time.new(2003, 2, 1, 1, 2, 3)))
+        end
       end
     end
 

spec/inputs/integration/integ_spec.rb

@@ -66,6 +66,29 @@
         expect(event.get('first_name')).to eq('David')
       end
     end
+
+    context 'with temporal columns' do
+      let(:settings) do
+        super().merge("statement" => 'SELECT ENTRY_DATE, ENTRY_TIME, TIMESTAMP FROM "employee" WHERE EMP_NO = 2')
+      end
+
+      before(:each) { plugin.run(queue) }
+
+      subject(:event) { queue.pop }
+
+      it "maps the DATE to a Logstash Timestamp" do
+        expect(event.get('entry_date')).to eq(LogStash::Timestamp.new(Time.new(2003, 2, 1)))
+      end
+
+      it "maps the TIME field to a Logstash Timestamp" do
+        now = DateTime.now
+        expect(event.get('entry_time')).to eq(LogStash::Timestamp.new(Time.new(now.year, now.month, now.day, 10, 5, 0)))
+      end
+
+      it "maps the TIMESTAMP to a Logstash Timestamp" do
+        expect(event.get('timestamp')).to eq(LogStash::Timestamp.new(Time.new(2003, 2, 1, 1, 2, 3)))
+      end
+    end
   end
 
   context "when supplying a non-existent library" do