/**
 * Copyright 2016 StreamSets Inc.
 *
 * Licensed under the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF 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 com.streamsets.pipeline.stage.it;

import com.google.common.collect.ImmutableList;
import com.streamsets.pipeline.api.Field;
import com.streamsets.pipeline.api.Record;
import com.streamsets.pipeline.api.impl.Utils;
import com.streamsets.pipeline.sdk.RecordCreator;
import com.streamsets.pipeline.stage.HiveMetastoreTargetBuilder;
import com.streamsets.pipeline.stage.destination.hive.HiveMetastoreTarget;
import com.streamsets.pipeline.stage.processor.hive.HiveMetadataProcessor;
import com.streamsets.pipeline.stage.HiveMetadataProcessorBuilder;
import org.apache.commons.lang3.tuple.ImmutablePair;
import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.Types;
import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.Map;
import com.streamsets.pipeline.stage.ParametrizedUtils;

/**
 * Run all combination of old/new way of creating tables with internal/external for case when the table in HMS
 * doesn't exists before starting the pipeline.
 */
@RunWith(Parameterized.class)
public class ColdStartIT extends BaseHiveMetadataPropagationIT {

  private static Logger LOG = LoggerFactory.getLogger(ColdStartIT.class);

  @Parameterized.Parameters(name = "db({0}),storedAsAvro({1}),external({2})")
  public static Collection<Object[]> data() {
    return ParametrizedUtils.crossProduct(
        new String[] {"default", "custom"},
        new Boolean[] {true, false},
        new Boolean[] {true, false}
    );
  }

  private String database;
  private boolean storedAsAvro;
  private boolean external;
  public ColdStartIT(String database, boolean storedAsAvro, boolean external) {
    this.database = database;
    this.storedAsAvro = storedAsAvro;
    this.external = external;
  }


  @Test
  public void testColdStart() throws  Exception {
    LOG.info(Utils.format("Starting cold start with database({}), storedAsAvro({}) and external({})", database, storedAsAvro, external));
    executeUpdate(Utils.format("create database if not exists {}", database));

    HiveMetadataProcessor processor = new HiveMetadataProcessorBuilder()
      .database(database)
      .tablePathTemplate(getDefaultWareHouseDir())
      .partitionPathTemplate("dt=super-secret")
      .external(external)
      .build();

    HiveMetastoreTarget hiveTarget = new HiveMetastoreTargetBuilder()
        .storedAsAvro(storedAsAvro)
        .build();

    Map<String, Field> map = new LinkedHashMap<>();
    map.put("name", Field.create(Field.Type.STRING, "StreamSets"));
    Record record = RecordCreator.create("s", "s:1");
    record.set(Field.create(map));

    processRecords(processor, hiveTarget, ImmutableList.of(record));

    assertTableExists(Utils.format("{}.tbl", database));
    assertQueryResult(Utils.format("select * from {}.tbl", database), new QueryValidator() {
      @Override
      public void validateResultSet(ResultSet rs) throws Exception {
        assertResultSetStructure(rs,
            new ImmutablePair("tbl.name", Types.VARCHAR),
            new ImmutablePair("tbl.dt", Types.VARCHAR)
        );

        Assert.assertTrue("Table tbl doesn't contain any rows", rs.next());
        Assert.assertEquals("StreamSets", rs.getString(1));
        Assert.assertFalse("Table tbl contains more then one row", rs.next());
      }
    });
  }
}

@@ -2375,16 +2375,20 @@
 ring%5B%5D %7B
+%22%22, 
 %22default
@@ -2401,16 +2401,16 @@
 stom%22%7D,%0A
-
         
@@ -2956,87 +2956,8 @@
 l));
-%0A    executeUpdate(Utils.format(%22create database if not exists %7B%7D%22, database));
 %0A%0A  
@@ -3494,16 +3494,16 @@
 %22s:1%22);%0A
-
     reco
@@ -3526,24 +3526,355 @@
 ate(map));%0A%0A
+    // Empty string as a database was already configured, so we replace it with%0A    // string %22default%22, so that we don't have to do that for every single method%0A    // call from now on.%0A    if(database.isEmpty()) %7B%0A      database = %22default%22;%0A    %7D%0A%0A    executeUpdate(Utils.format(%22create database if not exists %7B%7D%22, database));%0A%0A
     processR

Add empty database value to ColdStartIT