title: sql query to remove redundant records based on changes in multiple columns
body: <p>I am trying to write a efficient select query which will remove redundant records based on changes in column client_id, product_id, interest_rate  with min(actual_from_dt) and max(actual_to_dt) with <code>pk</code> as agrmnt_id.</p>
  <pre><code>select agrmnt_id,min(actual_from_dt),max(actual_to_dt),client_id,product_id,interest_rate from your table group by  agrmnt_id,client_id,product_id,interest_rate
  </code></pre>
  <p><em>Source Table</em></p>
  <div class="s-table-container">
  <table class="s-table">
  <thead>
  <tr>
  <th>sk</th>
  <th>agrmnt_id</th>
  <th>actual_from_dt</th>
  <th>actual_to_dt</th>
  <th>client_id</th>
  <th>product_id</th>
  <th>interest_rate</th>
  </tr>
  </thead>
  <tbody>
  <tr>
  <td>1</td>
  <td>101</td>
  <td>21.02.2015</td>
  <td>17.05.2015</td>
  <td>20</td>
  <td>345</td>
  <td>4%</td>
  </tr>
  <tr>
  <td>2</td>
  <td>101</td>
  <td>18.05.2015</td>
  <td>05.07.2015</td>
  <td>20</td>
  <td>345</td>
  <td>4%</td>
  </tr>
  <tr>
  <td>3</td>
  <td>101</td>
  <td>23.08.2015</td>
  <td>31.12.9999</td>
  <td>20</td>
  <td>345</td>
  <td>4%</td>
  </tr>
  <tr>
  <td>4</td>
  <td>102</td>
  <td>01.01.2016</td>
  <td>30.06.2016</td>
  <td>25</td>
  <td>333</td>
  <td>3.7%</td>
  </tr>
  <tr>
  <td>5</td>
  <td>102</td>
  <td>01.07.2016</td>
  <td>25.07.2016</td>
  <td>25</td>
  <td>333</td>
  <td>3.7%</td>
  </tr>
  <tr>
  <td>6</td>
  <td>102</td>
  <td>26.07.2016</td>
  <td>15.09.2016</td>
  <td>25</td>
  <td>333</td>
  <td>3.7%</td>
  </tr>
  </tbody>
  </table>
  </div>
  <p>Output</p>
  <div class="s-table-container">
  <table class="s-table">
  <thead>
  <tr>
  <th>sk</th>
  <th>agrmnt_id</th>
  <th>actual_from_dt</th>
  <th>actual_to_dt</th>
  <th>client_id</th>
  <th>product_id</th>
  <th>interest_rate</th>
  </tr>
  </thead>
  <tbody>
  <tr>
  <td>1</td>
  <td>101</td>
  <td>21.02.2015</td>
  <td>31.12.9999</td>
  <td>20</td>
  <td>345</td>
  <td>4%</td>
  </tr>
  <tr>
  <td>2</td>
  <td>102</td>
  <td>01.01.2016</td>
  <td>15.09.2016</td>
  <td>25</td>
  <td>333</td>
  <td>3.7%</td>
  </tr>
  </tbody>
  </table>
  </div>
  <pre><code>|== Physical Plan ==
  AdaptiveSparkPlan isFinalPlan=false
  +- == Current Plan ==
     SortAggregate(key=[agrmnt_id#4633944, client_id#4633947, product_id#4633948, interest_rate#4633949], functions=[finalmerge_min(merge min#4633975) AS min(actual_from_dt#4633945)#4633970, finalmerge_max(merge max#4633977) AS max(actual_to_dt#4633946)#4633971])
     +- Sort [agrmnt_id#4633944 ASC NULLS FIRST, client_id#4633947 ASC NULLS FIRST, product_id#4633948 ASC NULLS FIRST, interest_rate#4633949 ASC NULLS FIRST], false, 0
        +- Exchange hashpartitioning(agrmnt_id#4633944, client_id#4633947, product_id#4633948, interest_rate#4633949, 200), true, [id=#7017871]
           +- SortAggregate(key=[agrmnt_id#4633944, client_id#4633947, product_id#4633948, interest_rate#4633949], functions=[partial_min(actual_from_dt#4633945) AS min#4633975, partial_max(actual_to_dt#4633946) AS max#4633977])
              +- Sort [agrmnt_id#4633944 ASC NULLS FIRST, client_id#4633947 ASC NULLS FIRST, product_id#4633948 ASC NULLS FIRST, interest_rate#4633949 ASC NULLS FIRST], false, 0
                 +- FileScan csv [agrmnt_id#4633944,actual_from_dt#4633945,actual_to_dt#4633946,client_id#4633947,product_id#4633948,interest_rate#4633949] Batched: false, DataFilters: [], Format: CSV, PartitionFilters: [], PushedFilters: [], ReadSchema: struct&lt;agrmnt_id:string,actual_from_dt:string,actual_to_dt:string,client_id:string,product_id:str...
  +- == Initial Plan ==
     SortAggregate(key=[agrmnt_id#4633944, client_id#4633947, product_id#4633948, interest_rate#4633949], functions=[finalmerge_min(merge min#4633975) AS min(actual_from_dt#4633945)#4633970, finalmerge_max(merge max#4633977) AS max(actual_to_dt#4633946)#4633971])
     +- Sort [agrmnt_id#4633944 ASC NULLS FIRST, client_id#4633947 ASC NULLS FIRST, product_id#4633948 ASC NULLS FIRST, interest_rate#4633949 ASC NULLS FIRST], false, 0
        +- Exchange hashpartitioning(agrmnt_id#4633944, client_id#4633947, product_id#4633948, interest_rate#4633949, 200), true, [id=#7017871]
           +- SortAggregate(key=[agrmnt_id#4633944, client_id#4633947, product_id#4633948, interest_rate#4633949], functions=[partial_min(actual_from_dt#4633945) AS min#4633975, partial_max(actual_to_dt#4633946) AS max#4633977])
              +- Sort [agrmnt_id#4633944 ASC NULLS FIRST, client_id#4633947 ASC NULLS FIRST, product_id#4633948 ASC NULLS FIRST, interest_rate#4633949 ASC NULLS FIRST], false, 0
                 +- FileScan csv [agrmnt_id#4633944,actual_from_dt#4633945,actual_to_dt#4633946,client_id#4633947,product_id#4633948,interest_rate#4633949] Batched: false, DataFilters: [], Format: CSV, PartitionFilters: [], PushedFilters: [], ReadSchema: struct&lt;agrmnt_id:string,actual_from_dt:string,actual_to_dt:string,client_id:string,product_id:str...
  </code></pre>