/**
 * Licensed to 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 org.apache.drill.store.kudu;

import java.util.ArrayList;
import java.util.List;

import org.junit.Test;
import org.kududb.ColumnSchema;
import org.kududb.Schema;
import org.kududb.Type;
import org.kududb.client.Insert;
import org.kududb.client.KuduClient;
import org.kududb.client.KuduScanner;
import org.kududb.client.KuduSession;
import org.kududb.client.KuduTable;
import org.kududb.client.ListTablesResponse;
import org.kududb.client.PartialRow;
import org.kududb.client.RowResult;
import org.kududb.client.RowResultIterator;


public class TestKuduConnect {
  static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestKuduConnect.class);

  public static final String KUDU_MASTER = "172.31.1.99";
  public static final String KUDU_TABLE = "demo";

  @Test
  public void abc() throws Exception {

    try (KuduClient client = new KuduClient.KuduClientBuilder(KUDU_MASTER).build()) {

      ListTablesResponse tables = client.getTablesList(KUDU_TABLE);
      if (!tables.getTablesList().isEmpty()) {
        client.deleteTable(KUDU_TABLE);
      }
      ;

      List<ColumnSchema> columns = new ArrayList<>(5);
      columns.add(new ColumnSchema.ColumnSchemaBuilder("key", Type.INT32).key(true).build());
      columns.add(new ColumnSchema.ColumnSchemaBuilder("binary", Type.BINARY).nullable(false).build());
      columns.add(new ColumnSchema.ColumnSchemaBuilder("boolean", Type.BOOL).nullable(true).build());
      columns.add(new ColumnSchema.ColumnSchemaBuilder("float", Type.FLOAT).nullable(false).build());
      columns.add(new ColumnSchema.ColumnSchemaBuilder("string", Type.STRING).nullable(true).build());

      Schema schema = new Schema(columns);
      client.createTable(KUDU_TABLE, schema);

      KuduTable table = client.openTable(KUDU_TABLE);

      KuduSession session = client.newSession();
      for (int i = 0; i < 3; i++) {
        Insert insert = table.newInsert();
        PartialRow row = insert.getRow();
        row.addInt(0, i);
        row.addBinary(1, ("Row " + i).getBytes());
        row.addBoolean(2, i % 2 == 0);
        row.addFloat(3, i + 0.01f);
        row.addString(4, ("Row " + i));
        session.apply(insert);
      }

      List<String> projectColumns = new ArrayList<>(1);
      projectColumns.add("float");
      KuduScanner scanner = client.newScannerBuilder(table)
          .setProjectedColumnNames(projectColumns)
          .build();
      while (scanner.hasMoreRows()) {
        RowResultIterator results = scanner.nextRows();
        while (results.hasNext()) {
          RowResult result = results.next();
          System.out.println(result.toStringLongFormat());
        }
      }
    }
  }
}

@@ -991,16 +991,61 @@
 b.Type;%0A
+import org.kududb.client.CreateTableBuilder;%0A
 import o
@@ -1380,16 +1380,102 @@
 erator;%0A
+import org.kududb.client.SessionConfiguration;%0A%0Aimport static org.kududb.Type.STRING;%0A
 %0A%0Apublic
@@ -1657,16 +1657,17 @@
 .1.99%22;%0A
+%0A
   public
@@ -1678,69 +1678,82 @@
 tic 
-final String KUDU_TABLE = %22demo%22;%0A%0A  @Test%0A  public void abc(
+void createKuduTable(String tableName, int tablets, int replicas, int rows
 ) th
@@ -1912,26 +1912,25 @@
 lesList(
-KUDU_TABLE
+tableName
 );%0A     
@@ -1998,26 +1998,25 @@
 teTable(
-KUDU_TABLE
+tableName
 );%0A     
@@ -2021,16 +2021,8 @@
    %7D
-%0A      ;
 %0A%0A  
@@ -2627,51 +2627,345 @@
 s);%0A
-      client.createTable(KUDU_TABLE, schema
+%0A      CreateTableBuilder builder = new CreateTableBuilder();%0A      builder.setNumReplicas(replicas);%0A      for (int i = 1; i %3C tablets; i++) %7B%0A        PartialRow splitRow = schema.newPartialRow();%0A        splitRow.addInt(%22key%22, i*1000);%0A        builder.addSplitRow(splitRow);%0A      %7D%0A%0A      client.createTable(tableName, schema, builder
 );%0A%0A
@@ -3009,18 +3009,17 @@
 ble(
-KUDU_TABLE
+tableName
 );%0A%0A
@@ -3063,24 +3063,100 @@
 wSession();%0A
+      session.setFlushMode(SessionConfiguration.FlushMode.AUTO_FLUSH_SYNC);%0A
       for (i
@@ -3169,17 +3169,20 @@
  0; i %3C 
-3
+rows
 ; i++) %7B
@@ -3982,11 +3982,222 @@
    %7D%0A  %7D
+%0A%0A  @Test%0A  public void abc() throws Exception %7B%0A    createKuduTable(%22demo%22, 1, 1, 3);%0A  %7D%0A%0A  @Test%0A  public void def() throws Exception %7B%0A    createKuduTable(%22demo-large-splits%22, 6, /* replicas */ 1, 6000);%0A  %7D
 %0A%7D%0A

Create table with multiple tablets in kudu test.