input {
  jdbc {
        jdbc_driver_class => "com.teradata.jdbc.TeraDriver"
        jdbc_connection_string => "jdbc:teradata://<server-name>/DATABASE=<database-name>,DBS_PORT=<port>"
        jdbc_user => "<db-user>"
        jdbc_password => "<db-password>"
		schedule => "*/1 * * * *"
        clean_run => false
        statement => "select SessionID, CAST((((CAST(CAST(CAST(STARTTIME At 'GMT' AS FORMAT 'YYYY-MM-DD') AS VARCHAR(14))  AS DATE FORMAT 'YYYY-MM-DD') - DATE '1970-01-01')*86400 * 1000) + ((EXTRACT(HOUR FROM StartTime at 'gmt')*3600*1000) + (EXTRACT(MINUTE FROM StartTime at 'gmt')*60*1000) + (EXTRACT(SECOND FROM db.STARTTIME AT TIME ZONE 'GMT')*1000)))AS BIGINT) AS t1, StartTime at 'gmt' as t2, trim(qca.ClientIPAddrByClient) as ClientAddr, trim(qca.ServerIPAddrByServer) as ServerAddr,db.UserName,db.QueryID,SqlTextInfo,trim(ErrorText) as ErrorText,trim(AppID) as LogonSource,trim(ClientID) as ClientID, NumResultRows as rowsReturned from dbc.qrylogv db INNER JOIN DBC.QryLogSQLV dbs ON db.QueryID = dbs.QueryID INNER JOIN DBC.QryLogClientAttrV qca ON db.QueryID = qca.QueryID where t1 > :sql_last_value and ClientID <> 'GUC' and sqltextinfo NOT IN ('select * from DBC.Database_Default_JournalsV','COMMIT WORK','SELECT 1','SELECT USER','--isValid') and sqltextinfo not like '%LogonSource%' and sqltextinfo not like'%ApplicationName%' and sqltextinfo not like'%LOCK%' and sqltextinfo not like'%TYPE%' and sqltextinfo not like'%WORK%' order by t1;"
        use_column_value => true
        tracking_column => "t1"
        tracking_column_type => "numeric"
        last_run_metadata_path => "/usr/share/logstash/third_party/.teradata_logstash_jdbc_last_run"
		type => "teradata"
		#Insert the server_name
        add_field => {"server_name" => "<server-name>"}
	}
}	

filter {

	if [type] == "teradata" {
  
		if [errortext] {
		mutate { 
				add_field => { "[Error_Text]" => "%{errortext}"}
				}
		}
	
		mutate { 
			add_field => { 
							"[Session_ID]" => "%{sessionid}" 
							"[Time_Field]" => "%{t2}" 
							"[Client_IP]" => "%{clientaddr}"
							"[Server_IP]" => "%{serveraddr}"
							"[User_Name]" => "%{username}"
							"[Sql_Text_Info]" => "%{sqltextinfo}"
							"[Logon_Source]" => "%{logonsource}"
							"[Query_ID]" => "%{queryid}"
							"[Os_User]" => "%{clientid}"
							"[Server_Hostname]" => "%{server_name}"
							"[Rows_Returned]" => "%{rowsreturned}"
						}
				}
	
		teradatadb_guardium_plugin_filter{}	
	
		mutate { remove_field => ["@version","type","@timestamp","Error_Text","errortext","Session_ID","sessionid","Time_Field","starttime","Client_IP","clientaddr","Server_IP","serveraddr","User_Name","username","Sql_Text_Info","sqltextinfo","Logon_Source","logonsource","Query_ID","queryid","Server_Hostname","t1","t2"] }			
	}	
}