logstash postgres

gistfile1.txt

filter {
  if [type] == 'postgres_csv' {
    csv {
      columns => [
        "pg_timestamp",
        "user_name",
        "database_name",
        "process_id",
        "connection_from",
        "session_id",
        "session_line_num",
        "command_tag text,",
        "session_start_time",
        "virtual_transaction_id",
        "transaction_id",
        "error_severity",
        "sql_state_code",
        "log_message",
        "detail",
        "hint",
        "internal_query",
        "internal_query_pos",
        "context",
        "query",
        "query_pos",
        "location",
        "application_name"
      ]
    }

    mutate {
      gsub => [ "pg_timestamp", " \w{3}$", "" ]
    }

    date {
      match => [ "pg_timestamp", "yyyy-MM-dd HH:mm:ss.SSS" ]
    }

    if [log_message] =~ "duration" {
      grok {
        match => [ "log_message", "duration: %{NUMBER:duration_ms} ms  (execute|statement)(\s\S+)?: %{PG_STATEMENT:statement}" ]
      }
      mutate {
        convert => [ "duration_ms", "float" ]
      }
    }
  }
}

hi does't have input , output