/*
 * Licensed to Metamarkets Group Inc. (Metamarkets) under one
 * or more contributor license agreements. See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership. Metamarkets licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License. You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied. See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package io.druid.metadata;

import com.fasterxml.jackson.databind.ObjectMapper;
import io.druid.java.util.common.StringUtils;
import org.joda.time.DateTime;
import org.skife.jdbi.v2.Handle;
import org.skife.jdbi.v2.Query;

import javax.annotation.Nullable;
import java.util.Map;

public class SQLServerMetadataStorageActionHandler<EntryType, StatusType, LogType, LockType>
    extends SQLMetadataStorageActionHandler<EntryType, StatusType, LogType, LockType>
{
  public SQLServerMetadataStorageActionHandler(
      SQLMetadataConnector connector,
      ObjectMapper jsonMapper,
      MetadataStorageActionHandlerTypes<EntryType, StatusType, LogType, LockType> types,
      String entryTypeName,
      String entryTable,
      String logTable,
      String lockTable
  )
  {
    super(connector, jsonMapper, types, entryTypeName, entryTable, logTable, lockTable);
  }

  @Override
  protected Query<Map<String, Object>> createInactiveStatusesSinceQuery(
      Handle handle, DateTime timestamp, @Nullable Integer maxNumStatuses, @Nullable String datasource
  )
  {
    String sql = maxNumStatuses == null ? "SELECT " : "SELECT TOP :n ";

    sql += StringUtils.format(
        "    id, "
        + "  status_payload, "
        + "  created_date, "
        + "  datasource, "
        + "  payload "
        + "FROM "
        + "  %s "
        + "WHERE "
        + getWhereClauseForInactiveStatusesSinceQuery(datasource)
        + "ORDER BY created_date DESC",
        getEntryTable()
    );
    if (maxNumStatuses != null) {
      sql += " LIMIT :n";
    }
    Query<Map<String, Object>> query = handle.createQuery(sql).bind("start", timestamp.toString());

    if (maxNumStatuses != null) {
      query = query.bind("n", maxNumStatuses);
    }
    if (datasource != null) {
      query = query.bind("ds", datasource);
    }
    return query;
  }
  private String getWhereClauseForInactiveStatusesSinceQuery(@Nullable String datasource)
  {
    String sql = StringUtils.format("active = FALSE AND created_date >= :start ");
    if (datasource != null) {
      sql += " AND datasource = :ds ";
    }
    return sql;
  }
}

@@ -1930,18 +1930,20 @@
 ECT TOP 
+(
 :n
+)
  %22;%0A%0A   
@@ -2295,73 +2295,8 @@
  );%0A
-    if (maxNumStatuses != null) %7B%0A      sql += %22 LIMIT :n%22;%0A    %7D
 %0A   

Fix SQL Server select query in createInactiveStatusesSinceQuery() method. (#5901)