title: How to merge multiple jdbc inputs in logstash configuration based on condition of single column?
body: <p>I have two tables in sql server i.e. AppDetails and AppBranchDetails.
  I want to read all rows of these two tables and merge based on the condition.</p>
  
  <p>Below are the two queries which I want to run:</p>
  
  <ol>
  <li>select id as colg_id, name, sirname from AppDetails order by id</li>
  <li>select id as branch_id, branch_name, branch_add from AppBranchDetails order by id</li>
  </ol>
  
  <p>From the above two queries, "id" is a primary key which is same for both the tables.</p>
  
  <p>The output will looks like below for <strong>id == 1</strong>:</p>
  
  <pre><code>{
    "name": "ram",
    "sirname": "patil",
    "id": 1,
    "BRANCH": [
      {
        "id": 1,
        "branch_name": "EE",
        "branch_add": "IND"
      },
      {
        "id": 1,
        "branch_name": "ME",
        "branch_add": "IND"
      }
    ]
  }
  </code></pre>
  
  <p>The output will looks like below for <strong>id == 2</strong>:</p>
  
  <pre><code>{
    "name": "sham",
    "sirname": "bhosle",
    "id": 2,
    "BRANCH": [
      {
        "id": 2,
        "branch_name": "SE",
        "branch_add": "US"
      },
      {
        "id": 2,
        "branch_name": "FE",
        "branch_add": "US"
      }
    ]
  }
  </code></pre>
  
  <p>I am trying with below configuration (app.conf):</p>
  
  <pre><code>input {
    jdbc {
      jdbc_connection_string =&gt; "jdbc:sqlserver://x.x.x.x:1433;databaseName=AAA;"
      jdbc_driver_class =&gt; "com.microsoft.sqlserver.jdbc.SQLServerDriver"
      jdbc_user =&gt; "sa"
      jdbc_password =&gt; "sa@111"
      statement =&gt; "select id as colg_id, name, sirname from AppDetails order by id"
      tracking_column =&gt; "colg_id"
      use_column_value =&gt; true
      type =&gt; "college"
    }
    jdbc {
      jdbc_connection_string =&gt; "jdbc:sqlserver://x.x.x.x:1433;databaseName=AAA;"
      jdbc_driver_class =&gt; "com.microsoft.sqlserver.jdbc.SQLServerDriver"
      jdbc_user =&gt; "sa"
      jdbc_password =&gt; "sa@111"
      statement =&gt; "select id as branch_id, branch_name, branch_add from AppBranchDetails order by id"
      tracking_column =&gt; "branch_id"
      use_column_value =&gt; true
      type =&gt; "branch"
    }
  }
  
  filter {
      if [type] == "branch" {
          aggregate {
          task_id =&gt; "%{branch_id}"
          code =&gt; "
              map['BRANCH'] ||= []
              map['BRANCH'] &lt;&lt; event.get('branch_id')
              map['BRANCH'] &lt;&lt; event.get('branch_name')
              map['BRANCH'] &lt;&lt; event.get('branch_add')
              event.cancel()
          "
          push_previous_map_as_event =&gt; true
          timeout =&gt; 5
          }
          mutate {
              remove_field =&gt; [ "@version" , "@timestamp" ]
          }
      }
  }
  
  output {
    stdout { codec =&gt; json_lines }
  }
  </code></pre>
  
  <p>Can anyone please suggest how I can make the result which I have mentioned above.</p>