content: /*
   * Copyright 2021 4Paradigm
   *
   * Licensed 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._4paradigm.openmldb.jdbc;
  
  import com._4paradigm.openmldb.SQLInsertRow;
  import com._4paradigm.openmldb.SQLInsertRows;
  import com._4paradigm.openmldb.common.Pair;
  import com._4paradigm.openmldb.proto.NS;
  import com._4paradigm.openmldb.sdk.Column;
  import com._4paradigm.openmldb.sdk.DAGNode;
  import com._4paradigm.openmldb.sdk.Schema;
  import com._4paradigm.openmldb.sdk.SdkOption;
  import com._4paradigm.openmldb.sdk.SqlExecutor;
  import com._4paradigm.openmldb.sdk.impl.SqlClusterExecutor;
  import com._4paradigm.openmldb.sdk.utils.AIOSUtil;
  
  import org.testng.Assert;
  import org.testng.annotations.DataProvider;
  import org.testng.annotations.Test;
  import org.testng.collections.Maps;
  
  import com.google.gson.Gson;
  import java.io.IOException;
  import java.nio.file.Files;
  import java.nio.file.Path;
  import java.nio.file.Paths;
  import java.sql.PreparedStatement;
  import java.sql.ResultSetMetaData;
  import java.sql.SQLException;
  import java.sql.Types;
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.NoSuchElementException;
  import java.util.Arrays;
  
  public class SQLRouterSmokeTest {
      public static SqlExecutor clusterExecutor;
      public static SqlExecutor lightClusterExecutor;
      public static SqlExecutor standaloneExecutor;
  
      static {
          try {
              SdkOption option = new SdkOption();
              option.setZkPath(TestConfig.ZK_PATH);
              option.setZkCluster(TestConfig.ZK_CLUSTER);
              option.setSessionTimeout(200000);
              clusterExecutor = new SqlClusterExecutor(option);
              setOnlineMode(clusterExecutor);
              option.setLight(true);
              lightClusterExecutor = new SqlClusterExecutor(option);
              setOnlineMode(lightClusterExecutor);
              // create standalone router
              SdkOption standaloneOption = new SdkOption();
              standaloneOption.setHost(TestConfig.HOST);
              standaloneOption.setPort(TestConfig.PORT);
              standaloneOption.setClusterMode(false);
              standaloneOption.setSessionTimeout(20000);
              standaloneExecutor = new SqlClusterExecutor(standaloneOption);
          } catch (Exception e) {
              e.printStackTrace();
          }
      }
  
      static void setOnlineMode(SqlExecutor executor) {
          java.sql.Statement state = executor.getStatement();
          try {
              state.execute("SET @@execute_mode='online';");
              state.close();
          } catch (Exception e) {
              e.printStackTrace();
          }
      }
  
      @Test
      void testMoreOptions() throws Exception {
          SdkOption option = new SdkOption();
          option.setZkPath(TestConfig.ZK_PATH);
          option.setZkCluster(TestConfig.ZK_CLUSTER);
          option.setSessionTimeout(200000);
          option.setMaxSqlCacheSize(100);
          option.setZkLogLevel(2);
          SqlExecutor tmp = new SqlClusterExecutor(option);
      }
  
      @DataProvider(name = "executor")
      public Object[] executor() {
          return new Object[] { clusterExecutor, lightClusterExecutor, standaloneExecutor };
      }
  
      @Test(dataProvider = "executor")
      public void testSmoke(SqlExecutor router) {
          try {
              String dbname = "SQLRouterSmokeTest" + System.currentTimeMillis();
              String tableName = "tsql1010";
  
              // create db
              boolean ok = router.createDB(dbname);
              Assert.assertTrue(ok);
              String ddl = String.format("create table %s (col1 bigint, col2 string, index(key=col2, ts=col1));",
                      tableName);
              // create table
              ok = router.executeDDL(dbname, ddl);
              Assert.assertTrue(ok);
              NS.TableInfo info = router.getTableInfo(dbname, tableName);
              Assert.assertEquals(info.getName(), tableName);
  
              // insert normal (1000, 'hello')
              String insert = "insert into tsql1010 values(1000, 'hello');";
              ok = router.executeInsert(dbname, insert);
              Assert.assertTrue(ok);
              // insert placeholder (1001, 'world')
              String insertPlaceholder = "insert into tsql1010 values(?, ?);";
              PreparedStatement impl = router.getInsertPreparedStmt(dbname, insertPlaceholder);
              impl.setLong(1, 1001);
              impl.setString(2, "world");
              ok = impl.execute();
              Assert.assertTrue(ok);
              // insert placeholder batch (1002, 'hi'), (1003, 'word')
              SQLInsertRows insertRows = router.getInsertRows(dbname, insertPlaceholder);
              SQLInsertRow row1 = insertRows.NewRow();
              row1.Init(2);
              row1.AppendInt64(1002);
              row1.AppendString("hi");
              SQLInsertRow row2 = insertRows.NewRow();
              row2.Init(4);
              row2.AppendInt64(1003);
              row2.AppendString("word");
              ok = router.executeInsert(dbname, insertPlaceholder, insertRows);
              Assert.assertTrue(ok);
  
              // select
              String select1 = "select * from tsql1010;";
              SQLResultSet rs1 = (SQLResultSet) router.executeSQL(dbname, select1);
  
              Assert.assertEquals(2, rs1.GetInternalSchema().getColumnList().size());
              Assert.assertEquals(Types.BIGINT, rs1.GetInternalSchema().getColumnType(0));
              Assert.assertEquals(Types.VARCHAR, rs1.GetInternalSchema().getColumnType(1));
  
              List<Long> col1Insert = new ArrayList<>();
              List<String> col2Insert = new ArrayList<>();
              while (rs1.next()) {
                  col1Insert.add(rs1.getLong(1));
                  col2Insert.add(rs1.getString(2));
              }
              Collections.sort(col1Insert);
              Collections.sort(col2Insert);
  
              Assert.assertEquals(col1Insert,
                      Arrays.asList(Long.valueOf(1000), Long.valueOf(1001), Long.valueOf(1002), Long.valueOf(1003)));
              Assert.assertEquals(col2Insert, Arrays.asList("hello", "hi", "word", "world"));
              rs1.close();
  
              String select2 = "select col1 from tsql1010;";
              SQLResultSet rs2 = (SQLResultSet) router .executeSQL(dbname, select2);
              Assert.assertEquals(1, rs2.GetInternalSchema().size());
              Assert.assertEquals(Types.BIGINT, rs2.GetInternalSchema().getColumnType(0));
  
              List<Long> col1InsertRes = new ArrayList<>();
              while (rs2.next()) {
                  col1InsertRes.add(rs2.getLong(1));
              }
              Collections.sort(col1InsertRes);
              Assert.assertEquals(col1InsertRes,
                      Arrays.asList(Long.valueOf(1000), Long.valueOf(1001), Long.valueOf(1002), Long.valueOf(1003)));
              rs2.close();
  
              String select3 = "select col2 from tsql1010;";
              SQLResultSet rs3 = (SQLResultSet) router .executeSQL(dbname, select3);
              Assert.assertEquals(1, rs3.GetInternalSchema().size());
              Assert.assertEquals(Types.VARCHAR, rs3.GetInternalSchema().getColumnType(0));
  
              List<String> col2InsertRes = new ArrayList<>();
              while (rs3.next()) {
                  col2InsertRes.add(rs3.getString(1));
              }
              Collections.sort(col2InsertRes);
              Assert.assertEquals(col2InsertRes, Arrays.asList("hello", "hi", "word", "world"));
              rs3.close();
  
              // parameterized query
              String parameterizedQuerySql = "select col1, col2 from tsql1010 where col2 = ? and col1 < ?;";
              PreparedStatement query_statement = router.getPreparedStatement(dbname, parameterizedQuerySql);
              // col2 = "hi" and col1 < 1003
              {
                  query_statement.setString(1, "hi");
                  query_statement.setLong(2, 1003);
                  SQLResultSet rs4 = (SQLResultSet) query_statement .executeQuery();
                  Assert.assertEquals(2, rs4.GetInternalSchema().size());
                  Assert.assertEquals(Types.BIGINT, rs4.GetInternalSchema().getColumnType(0));
                  Assert.assertEquals(Types.VARCHAR, rs4.GetInternalSchema().getColumnType(1));
                  Assert.assertTrue(rs4.next());
                  Assert.assertEquals(1002, rs4.getLong(1));
                  Assert.assertEquals("hi", rs4.getString(2));
                  Assert.assertFalse(rs4.next());
                  rs4.close();
              }
              // col2 = "hi" and col1 < 1002
              {
                  query_statement.setString(1, "hi");
                  query_statement.setLong(2, 1002);
                  SQLResultSet rs4 = (SQLResultSet) query_statement .executeQuery();
                  Assert.assertEquals(2, rs4.GetInternalSchema().size());
                  Assert.assertEquals(Types.BIGINT, rs4.GetInternalSchema().getColumnType(0));
                  Assert.assertEquals(Types.VARCHAR, rs4.GetInternalSchema().getColumnType(1));
                  Assert.assertFalse(rs4.next());
                  rs4.close();
              }
              // col2 = "world" and col1 < 1003
              {
                  query_statement.setString(1, "world");
                  query_statement.setLong(2, 1003);
                  SQLResultSet rs4 = (SQLResultSet) query_statement .executeQuery();
                  Assert.assertEquals(2, rs4.GetInternalSchema().size());
                  Assert.assertEquals(Types.BIGINT, rs4.GetInternalSchema().getColumnType(0));
                  Assert.assertEquals(Types.VARCHAR, rs4.GetInternalSchema().getColumnType(1));
                  Assert.assertTrue(rs4.next());
                  Assert.assertEquals(1001, rs4.getLong(1));
                  Assert.assertEquals("world", rs4.getString(2));
                  Assert.assertFalse(rs4.next());
                  rs4.close();
              }
              // col2 = "hello" and col1 < 1003
              {
                  query_statement.setString(1, "hello");
                  query_statement.setLong(2, 1003);
                  SQLResultSet rs4 = (SQLResultSet) query_statement .executeQuery();
                  Assert.assertEquals(2, rs4.GetInternalSchema().size());
                  Assert.assertEquals(Types.BIGINT, rs4.GetInternalSchema().getColumnType(0));
                  Assert.assertEquals(Types.VARCHAR, rs4.GetInternalSchema().getColumnType(1));
                  Assert.assertTrue(rs4.next());
                  Assert.assertEquals(1000, rs4.getLong(1));
                  Assert.assertEquals("hello", rs4.getString(2));
                  Assert.assertFalse(rs4.next());
                  rs4.close();
              }
              // col2 = "word" and col1 < 1003
              {
                  query_statement.setString(1, "word");
                  query_statement.setLong(2, 1003);
                  SQLResultSet rs4 = (SQLResultSet) query_statement .executeQuery();
                  Assert.assertEquals(2, rs4.GetInternalSchema().size());
                  Assert.assertEquals(Types.BIGINT, rs4.GetInternalSchema().getColumnType(0));
                  Assert.assertEquals(Types.VARCHAR, rs4.GetInternalSchema().getColumnType(1));
                  Assert.assertFalse(rs4.next());
                  rs4.close();
              }
              // drop table
              String drop = "drop table tsql1010;";
              ok = router.executeDDL(dbname, drop);
              Assert.assertTrue(ok);
              // insert into deleted table, can't get insert row
              SQLInsertRow insertRow = router.getInsertRow(dbname, insertPlaceholder);
              Assert.assertNull(insertRow);
              // drop database
              ok = router.dropDB(dbname);
              Assert.assertTrue(ok);
          } catch (Exception e) {
              e.printStackTrace();
              Assert.fail();
          }
      }
  
      @Test(dataProvider = "executor")
      public void testParameterizedQueryFail(SqlExecutor router) {
          try {
              String dbname = "SQLRouterSmokeTest" + System.currentTimeMillis();
              // create db
              router.dropDB(dbname);
              boolean ok = router.createDB(dbname);
              Assert.assertTrue(ok);
              String ddl = "create table tsql1010 ( col1 bigint, col2 string, index(key=col2, ts=col1));";
              // create table
              ok = router.executeDDL(dbname, ddl);
              Assert.assertTrue(ok);
              // parameterized query
              String parameterizedQuerySql = "select col1, col2 from tsql1010 where col2 = ? and col1 < ?;";
              PreparedStatement query_statement = router.getPreparedStatement(dbname, parameterizedQuerySql);
              // missing 2nd parameter
              {
                  query_statement.setString(1, "hi");
                  SQLResultSet rs4 = (SQLResultSet) query_statement .executeQuery();
                  Assert.fail("executeQuery is expected to throw exception");
                  rs4.close();
              }
          } catch (Exception e) {
              e.printStackTrace();
          }
      }
  
      @Test(dataProvider = "executor")
      public void testInsertMeta(SqlExecutor router) {
          String dbname = "SQLRouterSmokeTest" + System.currentTimeMillis();
          // create db
          router.dropDB(dbname);
          boolean ok = router.createDB(dbname);
          Assert.assertTrue(ok);
          String ddl = "create table tsql1010 ( col1 bigint, col2 date, col3 string, col4 string, col5 int, index(key=col3, ts=col1));";
          // create table
          ok = router.executeDDL(dbname, ddl);
          Assert.assertTrue(ok);
          java.sql.Date d1 = new java.sql.Date(2019 - 1900, 1 - 1, 1);
          java.sql.Date d2 = new java.sql.Date(2019 - 1900, 2 - 1, 20);
          java.sql.Date d3 = new java.sql.Date(2019 - 1900, 3 - 1, 3);
          java.sql.Date d4 = new java.sql.Date(2019 - 1900, 4 - 1, 24);
          java.sql.Date d5 = new java.sql.Date(2019 - 1900, 5 - 1, 5);
          String date1 = String.format("%s-%02d-%02d", d1.getYear() + 1900, d1.getMonth() + 1, d1.getDate());
          String fullInsert = String.format("insert into tsql1010 values(1000, '%s', 'guangdong', '广州', 1);", date1);
          try {
  
              PreparedStatement ps = router.getInsertPreparedStmt(dbname, fullInsert);
              Assert.assertEquals(ps.getMetaData().getColumnCount(), 0);
              ps.close();
          } catch (Exception e) {
              Assert.fail();
          }
  
          String insert1 = "insert into tsql1010 values(?, '2019-01-12', 'xx', 'xx', 1);";
          try {
              PreparedStatement ps = router.getInsertPreparedStmt(dbname, insert1);
              Assert.assertEquals(ps.getMetaData().getColumnCount(), 1);
              String col = ps.getMetaData().getColumnName(1);
              Assert.assertEquals(col, "col1");
              int type = ps.getMetaData().getColumnType(1);
              Assert.assertEquals(Types.BIGINT, type);
          } catch (Exception e) {
              Assert.fail();
          }
  
      }
  
      @Test(dataProvider = "executor")
      public void testInsertPreparedState(SqlExecutor router) {
          try {
              String dbname = "SQLRouterSmokeTest" + System.currentTimeMillis();
              // create db
              router.dropDB(dbname);
              boolean ok = router.createDB(dbname);
              Assert.assertTrue(ok);
              String ddl = "create table tsql1010 ( col1 bigint, col2 date, col3 string, col4 string, col5 int," +
                      " index(key=col3, ts=col1));";
              // create table
              ok = router.executeDDL(dbname, ddl);
              Assert.assertTrue(ok);
              // insert normal
              java.sql.Date d1 = new java.sql.Date(2019 - 1900, 1 - 1, 1);
              java.sql.Date d2 = new java.sql.Date(2019 - 1900, 2 - 1, 2);
              java.sql.Date d3 = new java.sql.Date(2019 - 1900, 3 - 1, 3);
              java.sql.Date d4 = new java.sql.Date(2019 - 1900, 4 - 1, 4);
              java.sql.Date d5 = new java.sql.Date(2019 - 1900, 5 - 1, 5);
              String date1 = String.format("%s-%02d-%02d", d1.getYear() + 1900, d1.getMonth() + 1, d1.getDate());
              String fullInsert = String.format("insert into tsql1010 values(1000, '%s', 'guangdong', '广州', 1);", date1);
              ok = router.executeInsert(dbname, fullInsert);
              Assert.assertTrue(ok);
              Object[][] datas = new Object[][] {
                      { 1000L, d1, "guangdong", "广州", 1 },
                      { 1001L, d2, "jiangsu", "nanjing", 2 },
                      { 1002L, d3, "sandong", "jinan", 3 },
                      { 1003L, d4, "zhejiang", "hangzhou", 4 },
                      { 1004L, d5, "henan", "zhenzhou", 5 },
              };
              // insert placeholder
              String date2 = String.format("%s-%s-%s", d2.getYear() + 1900, d2.getMonth() + 1, d2.getDate());
              String insert = String.format("insert into tsql1010 values(?, '%s', 'jiangsu', 'nanjing', 2);", date2);
              PreparedStatement impl = router.getInsertPreparedStmt(dbname, insert);
              impl.setLong(1, 1001);
              try {
                  impl.setInt(2, 1002);
              } catch (Exception e) {
                  Assert.assertEquals("out of data range", e.getMessage());
              }
              ok = impl.execute();
              Assert.assertTrue(ok);
  
              // custom insert order
              insert = "insert into tsql1010 (col1, col3, col2, col4, col5) values (1002, ?, ?, 'jinan', 3);";
              PreparedStatement impl2 = router.getInsertPreparedStmt(dbname, insert);
              ResultSetMetaData metaData = impl2.getMetaData();
              Assert.assertEquals(metaData.getColumnCount(), 2);
              Assert.assertEquals(metaData.getColumnName(1), "col3");
              Assert.assertEquals(metaData.getColumnType(1), Types.VARCHAR);
              Assert.assertEquals(metaData.getColumnTypeName(1), "string");
              try {
                  impl2.setString(2, "c");
              } catch (Exception e) {
                  Assert.assertTrue(e.getMessage().contains("set string failed"));
              }
              impl2.setString(1, "sandong");
              impl2.setDate(2, d3);
              ok = impl2.execute();
              Assert.assertTrue(ok);
  
              insert = "insert into tsql1010 values(?, ?, ?, ?, ?);";
              PreparedStatement impl3 = router.getInsertPreparedStmt(dbname, insert);
              impl3.setLong(1, 1003);
              impl3.setString(3, "zhejiang");
              try {
                  impl3.setString(3, "zhejiangxx");
                  Assert.fail();
              } catch (Exception e) {
                  Assert.assertTrue(true);
              }
              impl3.setString(4, "hangzhou");
              impl3.setDate(2, d4);
              impl3.setInt(5, 3);
              impl3.setInt(5, 4);
              impl3.closeOnCompletion();
              Assert.assertTrue(impl3.isCloseOnCompletion());
              ok = impl3.execute();
              Assert.assertTrue(ok);
              try {
                  impl3.execute();
              } catch (Exception e) {
                  Assert.assertEquals("InsertPreparedStatement closed", e.getMessage());
              }
              insert = "insert into tsql1010 values(?, ?, ?, 'zhenzhou', 5);";
              PreparedStatement impl4 = router.getInsertPreparedStmt(dbname, insert);
              impl4.close();
              Assert.assertTrue(impl4.isClosed());
              PreparedStatement impl5 = router.getInsertPreparedStmt(dbname, insert);
              impl5.setLong(1, 1004);
              impl5.setDate(2, d5);
              impl5.setString(3, "henan");
              ok = impl5.execute();
              Assert.assertTrue(ok);
              // select
              String select1 = "select * from tsql1010;";
              SQLResultSet rs1 = (SQLResultSet) router .executeSQL(dbname, select1);
              Assert.assertEquals(5, rs1.GetInternalSchema().size());
              Assert.assertEquals(Types.BIGINT, rs1.GetInternalSchema().getColumnType(0));
              Assert.assertEquals(Types.DATE, rs1.GetInternalSchema().getColumnType(1));
              Assert.assertEquals(Types.VARCHAR, rs1.GetInternalSchema().getColumnType(2));
              Assert.assertEquals(Types.VARCHAR, rs1.GetInternalSchema().getColumnType(3));
              Assert.assertEquals(Types.INTEGER, rs1.GetInternalSchema().getColumnType(4));
              while (rs1.next()) {
                  int idx = rs1.getInt(5);
                  int suffix = idx - 1;
                  Assert.assertTrue(suffix < datas.length);
                  Assert.assertEquals(datas[suffix][0], rs1.getLong(1));
                  Assert.assertEquals(datas[suffix][4], idx);
                  Assert.assertEquals(datas[suffix][2], rs1.getString(3));
                  Assert.assertEquals(datas[suffix][3], rs1.getString(4));
                  java.sql.Date tmpDate = (java.sql.Date) datas[suffix][1];
                  java.sql.Date getData = rs1.getDate(2);
                  Assert.assertEquals(tmpDate.getYear(), getData.getYear());
                  Assert.assertEquals(tmpDate.getDate(), getData.getDate());
                  Assert.assertEquals(tmpDate.getMonth(), getData.getMonth());
              }
              rs1.close();
  
              String select2 = "select col1 from tsql1010;";
              SQLResultSet rs2 = (SQLResultSet) router .executeSQL(dbname, select2);
              Assert.assertEquals(1, rs2.GetInternalSchema().size());
              Assert.assertEquals(Types.BIGINT, rs2.GetInternalSchema().getColumnType(0));
              rs2.close();
              // drop table
              String drop = "drop table tsql1010;";
              ok = router.executeDDL(dbname, drop);
              Assert.assertTrue(ok);
              // insert into deleted table
              ok = router.executeInsert(dbname, fullInsert);
              Assert.assertFalse(ok);
              // drop database
              ok = router.dropDB(dbname);
              Assert.assertTrue(ok);
          } catch (Exception e) {
              e.printStackTrace();
              Assert.fail();
          }
      }
  
      @Test(dataProvider = "executor")
      public void testInsertPreparedStateBatch(SqlExecutor router) {
          Object[][] batchData = new Object[][] {
                  {
                          "insert into tsql1010 values(?, ?, 'zhao', 1.0, null, 'z');",
                          new Object[][] {
                                  { 1000l, 1l }, { 1001l, 2l }, { 1002l, 3l }, { 1003l, 4l }, }
                  },
                  {
                          "insert into tsql1010 values(?, ?, 'zhao', 1.0, null, 'z');",
                          new Object[][] {
                                  { 1004l, 5l }, { 1005l, 6l }, { 1006l, 7l }, { 1007l, 8l }, }
                  },
                  {
                          "insert into tsql1010 values(?, ?, ?, 2.0, null, ?);",
                          "insert into tsql1010 values(1008, 9, 'zhao', 2.0, null, 'z');",
                          "insert into tsql1010 values(1009, 10, 'zhao', 2.0, null, 'z');",
                          "insert into tsql1010 values(1010, 11, 'zhao', 2.0, null, 'z');",
                  }
          };
          try {
              String dbname = "SQLRouterSmokeTest" + System.currentTimeMillis();
              // create db
              router.dropDB(dbname);
              boolean ok = router.createDB(dbname);
              Assert.assertTrue(ok);
              String ddl = "create table tsql1010 ( col1 bigint, col2 bigint, col3 string, col4 float, col5 date, col6 string, index(key=col2, ts=col1));";
              // create table
              ok = router.executeDDL(dbname, ddl);
              Assert.assertTrue(ok);
              // insert normal
              int i = 0;
              String insertPlaceholder = "insert into tsql1010 values(?, 2, 'taiyuan', 2.0);";
              PreparedStatement impl = router.getInsertPreparedStmt(dbname, (String) batchData[i][0]);
              Object[][] datas1 = (Object[][]) batchData[i][1];
              for (int j = 0; j < datas1.length; j++) {
                  try {
                      impl.setInt(2, 1002);
                  } catch (Exception e) {
                      Assert.assertTrue(e.getMessage().contains("set int failed"));
                  }
                  try {
                      // set failed, so the row is uncompleted, appending row will be failed
                      impl.execute();
                  } catch (Exception e) {
                      if (j > 0) {
                          // j > 0, addBatch has been called
                          Assert.assertEquals(e.getMessage(), "please use executeBatch");
                      } else {
                          Assert.assertTrue(e.getMessage().contains("cannot get index value"));
                      }
                  }
                  impl.setLong(1, (Long) datas1[j][0]);
                  impl.setLong(2, (Long) datas1[j][1]);
                  impl.addBatch();
              }
              try {
                  ok = impl.execute();
              } catch (Exception e) {
                  Assert.assertEquals(e.getMessage(), "please use executeBatch");
              }
              impl.executeBatch();
              Assert.assertTrue(ok);
              String select1 = "select * from tsql1010;";
              SQLResultSet rs1 = (SQLResultSet) router .executeSQL(dbname, select1);
              Assert.assertEquals(6, rs1.GetInternalSchema().size());
              rs1.close();
              i++;
              PreparedStatement impl2 = router.getInsertPreparedStmt(dbname, (String) batchData[i][0]);
              datas1 = (Object[][]) batchData[i][1];
              // value setting error won't break anything
              for (int j = 0; j < datas1.length; j++) {
                  try {
                      impl2.setInt(2, 1002);
                  } catch (Exception e) {
                      Assert.assertTrue(e.getMessage().contains("set int failed"));
                  }
                  try {
                      impl2.execute();
                  } catch (Exception e) {
                      if (j > 0) {
                          Assert.assertEquals(e.getMessage(), "please use executeBatch");
                      } else {
                          Assert.assertTrue(e.getMessage().contains("cannot get index value"));
                      }
                  }
                  impl2.setLong(1, (Long) datas1[j][0]);
                  impl2.setLong(2, (Long) datas1[j][1]);
                  impl2.addBatch();
              }
  
              try {
                  ok = impl2.execute();
              } catch (Exception e) {
                  Assert.assertEquals(e.getMessage(), "please use executeBatch");
              }
              i++;
              // can't use addBatch(String sql)
              Object[] datas2 = batchData[i];
              try {
                  impl2.addBatch((String) datas2[0]);
                  Assert.fail();
              } catch (Exception e) {
                  Assert.assertTrue(true);
              }
  
              int[] result = impl.executeBatch();
              int[] expected = new int[result.length];
              Assert.assertEquals(result, expected);
  
              String select2 = "select * from tsql1010;";
              SQLResultSet rs2 = (SQLResultSet) router .executeSQL(dbname, select1);
              Assert.assertEquals(6, rs2.GetInternalSchema().size());
              int recordCnt = 0;
              while (rs2.next()) {
                  recordCnt++;
              }
              Assert.assertEquals(datas1.length, recordCnt);
              rs2.close();
              // drop table
              String drop = "drop table tsql1010;";
              ok = router.executeDDL(dbname, drop);
              Assert.assertTrue(ok);
              // insert into deleted table
              ok = router.executeInsert(dbname, "insert into tsql1010 values(1009, 10, 'zhao', 2.0, null, 'z')");
              Assert.assertFalse(ok);
              // drop database
              ok = router.dropDB(dbname);
              Assert.assertTrue(ok);
          } catch (Exception e) {
              e.printStackTrace();
              Assert.fail();
          }
      }
  
      @Test
      public void testDDLParseMethods() throws SQLException {
          Map<String, Map<String, Schema>> schemaMaps = new HashMap<>();
          Schema sch = new Schema(Collections.singletonList(new Column("c1", Types.VARCHAR)));
          Map<String, Schema> dbSchema = new HashMap<>();
          dbSchema.put("t1", sch);
          schemaMaps.put("db1", dbSchema);
          List<String> ddls = SqlClusterExecutor.genDDL("select c1 from t1;", schemaMaps);
          Assert.assertEquals(ddls.size(), 1);
          List<Column> schema = SqlClusterExecutor.genOutputSchema("select c1 from t1;", schemaMaps).getColumnList();
          Assert.assertEquals(schema.size(), 1);
          Assert.assertEquals(schema.get(0).getColumnName(), "c1");
          Assert.assertEquals(schema.get(0).getSqlType(), Types.VARCHAR);
  
          // if input schema is null or empty
          try {
              SqlClusterExecutor.genDDL("", null);
              Assert.fail("null input schema will throw an exception");
          } catch (SQLException ignored) {
          }
          try {
              SqlClusterExecutor.genOutputSchema("", null);
              Assert.fail("null input schema will throw an exception");
          } catch (NullPointerException ignored) {
          }
          try {
              SqlClusterExecutor.genDDL("", Maps.<String, Map<String, Schema>>newHashMap());
              Assert.fail("null input schema will throw an exception");
          } catch (SQLException ignored) {
          }
          try {
              SqlClusterExecutor.genOutputSchema("", Maps.<String, Map<String, Schema>>newHashMap());
              Assert.fail("empty input schema will throw an exception");
          } catch (NoSuchElementException ignored) {
          }
  
          // if parse fails, genDDL will create table without index
          List<String> res1 = SqlClusterExecutor.genDDL("select not_exist from t1;", schemaMaps);
          Assert.assertEquals(res1.size(), 1);
          Assert.assertFalse(res1.get(0).contains("index"));
          // if parse fails, the output schema result is empty, can't convert to
          // sdk.Schema
          try {
              SqlClusterExecutor.genOutputSchema("select not_exist from t1;", schemaMaps);
          } catch (SQLException ignored) {
          }
  
          // multi db genOutputSchema
          schemaMaps.put("db2", dbSchema);
          schema = SqlClusterExecutor
                  .genOutputSchema("select t11.c1 from db1.t1 t1 last join db2.t1 t11 on t1.c1==t11.c1;", schemaMaps)
                  .getColumnList();
          Assert.assertEquals(schema.size(), 1);
  
          // get dependence tables
          List<Pair<String, String>> tables = SqlClusterExecutor.getDependentTables(
                  "select t11.c1 from db1.t1 t1 last join db2.t1 t11 on t1.c1==t11.c1;", "",
                  schemaMaps);
          Assert.assertEquals(tables.size(), 2);
      }
  
      @Test
      public void testValidateSQL() throws SQLException {
          Map<String, Map<String, Schema>> schemaMaps = new HashMap<>();
          Schema sch = new Schema(Collections.singletonList(new Column("c1", Types.VARCHAR)));
          Map<String, Schema> dbSchema = new HashMap<>();
          dbSchema.put("t1", sch);
          schemaMaps.put("db1", dbSchema);
          dbSchema = new HashMap<>();
          dbSchema.put("t2", sch);
          // one more db, if no used db, will use the first one db1
          schemaMaps.put("db2", dbSchema);
  
          List<String> ret = SqlClusterExecutor.validateSQLInBatch("select c1 from t1;", schemaMaps);
          Assert.assertEquals(ret.size(), 0);
          // t2 is in db2, db1 is the used db
          ret = SqlClusterExecutor.validateSQLInBatch("select c1 from t2;", schemaMaps);
          Assert.assertEquals(ret.size(), 2);
          // used db won't effect <db>.<table> style
          ret = SqlClusterExecutor.validateSQLInBatch("select c1 from db1.t1;", schemaMaps);
          Assert.assertEquals(ret.size(), 0);
  
          ret = SqlClusterExecutor.validateSQLInBatch("swlect c1 from t1;", schemaMaps);
          Assert.assertEquals(ret.size(), 2);
          Assert.assertTrue(ret.get(0).contains("Syntax error"));
  
          ret = SqlClusterExecutor.validateSQLInBatch("select foo(c1) from t1;", schemaMaps);
          Assert.assertEquals(ret.size(), 2);
          Assert.assertTrue(ret.get(0).contains("Fail to resolve expression"));
  
          // if has the same name tables, the first one will be used
          schemaMaps = new HashMap<>();
          Schema sch2 = new Schema(Collections.singletonList(new Column("c2", Types.VARCHAR)));
          dbSchema = new HashMap<>();
          dbSchema.put("t1", sch);
          schemaMaps.put("db1", dbSchema);
          dbSchema = new HashMap<>();
          dbSchema.put("t1", sch2);
          schemaMaps.put("db2", dbSchema);
  
          ret = SqlClusterExecutor.validateSQLInBatch("select c1 from t1;", schemaMaps);
          Assert.assertEquals(ret.size(), 0);
          // t1 is db1.t1
          ret = SqlClusterExecutor.validateSQLInBatch("select c2 from t1;", schemaMaps);
          Assert.assertEquals(ret.size(), 2);
  
          // if input schema is null or empty
          try {
              SqlClusterExecutor.validateSQLInBatch("", null);
              Assert.fail("null input schema will throw an exception");
          } catch (NullPointerException ignored) {
          }
  
          try {
              SqlClusterExecutor.validateSQLInBatch("", Maps.<String, Map<String, Schema>>newHashMap());
              Assert.fail("null input schema will throw an exception");
          } catch (NoSuchElementException ignored) {
          }
  
          ret = SqlClusterExecutor.validateSQLInRequest("select count(c1) from t1;", schemaMaps);
          Assert.assertEquals(ret.size(), 2);
          Assert.assertTrue(ret.get(0).contains("Aggregate over a table cannot be supported in online serving"));
          dbSchema = new HashMap<>();
          dbSchema.put("t3", new Schema(Arrays.asList(new Column("c1", Types.VARCHAR),
                  new Column("c2", Types.BIGINT))));
          schemaMaps.put("db3", dbSchema);
          // t3 is in db3, <db>.<table> style is ok
          ret = SqlClusterExecutor.validateSQLInRequest("select count(c1) over w1 from db3.t3 window " +
                  "w1 as(partition by c1 order by c2 rows between unbounded preceding and current row);", schemaMaps);
          Assert.assertEquals(ret.size(), 0);
      }
  
      @Test
      public void testMergeSQL() throws SQLException {
          // full table pattern
          List<String> sqls = Arrays.asList(
                  // some features in current row
                  "select c1 from main",
                  // window
                  "select sum(c1) over w1 of2 from main window w1 as (partition by c1 order by c2 rows between unbounded preceding and current row);",
                  // join
                  "select t1.c2 of3 from main last join t1 on main.c1==t1.c1;",
                  // join in order
                  "select t1.c2 of4 from main last join t1 order by t1.c2 on main.c1==t1.c1;",
                  // window union
                  "select sum(c2) over w1 from main window w1 as (union (select \"\" as id, * from t1) partition by c1 order by c2 rows between unbounded preceding and current row)");
  
          // validate merged sql
          Map<String, Map<String, Schema>> schemaMaps = new HashMap<>();
          HashMap<String, Schema> dbSchema = new HashMap<>();
          dbSchema.put("main", new Schema(Arrays.asList(new Column("id", Types.VARCHAR), new Column("c1", Types.BIGINT),
                  new Column("c2", Types.BIGINT))));
          dbSchema.put("t1", new Schema(Arrays.asList(new Column("c1", Types.BIGINT), new Column("c2", Types.BIGINT))));
          schemaMaps.put("foo", dbSchema);
  
          String filtered = SqlClusterExecutor.mergeSQL(sqls, "foo", Arrays.asList("id"), schemaMaps);
          Assert.assertEquals(filtered,
                  "select `c1`, `of2`, `of3`, `of4`, `sum(c2)over w1` from (select foo.main.id as merge_id_0, c1 from main) as out0 "
                          + "last join "
                          + "(select foo.main.id as merge_id_1, sum(c1) over w1 of2 from main window w1 as (partition by c1 order by c2 rows between unbounded preceding and current row)) as out1 on out0.merge_id_0 = out1.merge_id_1 "
                          + "last join "
                          + "(select foo.main.id as merge_id_2, t1.c2 of3 from main last join t1 on main.c1==t1.c1) as out2 "
                          + "on out0.merge_id_0 = out2.merge_id_2" + " last join "
                          + "(select foo.main.id as merge_id_3, t1.c2 of4 from main last join t1 order by t1.c2 on main.c1==t1.c1) as out3 "
                          + "on out0.merge_id_0 = out3.merge_id_3 " + "last join"
                          + " (select foo.main.id as merge_id_4, sum(c2) over w1 from main window w1 as (union (select \"\" as id, * from t1) partition by c1 order by c2 rows between unbounded preceding and current row)) as out4 "
                          + "on out0.merge_id_0 = out4.merge_id_4;");
  
          // add a function col without rename
          List<String> sqls2 = new ArrayList<>(sqls);
          sqls2.add("select int(`c1`) from main");
          String merged1 = SqlClusterExecutor.mergeSQL(sqls2, "foo", Arrays.asList("id"), schemaMaps);
          Assert.assertFalse(merged1.startsWith("select * from "));
          // add a ambiguous col-int(c1), throw exception
          try {
              sqls2.add("select int(`c1`) from main");
              SqlClusterExecutor.mergeSQL(sqls2, "foo", Arrays.asList("id"), schemaMaps);
              Assert.fail("ambiguous col should throw exception");
          } catch (SQLException e) {
              Assert.assertTrue(e.getMessage().contains("ambiguous"));
          }
  
          // join keys
          String mergedKeys = SqlClusterExecutor.mergeSQL(sqls, "foo", Arrays.asList("id", "c1", "c2"), schemaMaps);
          Assert.assertEquals(mergedKeys,
                  "select `c1`, `of2`, `of3`, `of4`, `sum(c2)over w1` from "
                          + "(select foo.main.id as merge_id_0, foo.main.c1 as merge_c1_0, foo.main.c2 as merge_c2_0, c1 from main) as out0 "
                          + "last join "
                          + "(select foo.main.id as merge_id_1, foo.main.c1 as merge_c1_1, foo.main.c2 as merge_c2_1, sum(c1) over w1 of2 from main window w1 as "
                          + "(partition by c1 order by c2 rows between unbounded preceding and current row)) as out1 "
                          + "on out0.merge_id_0 = out1.merge_id_1 and out0.merge_c1_0 = out1.merge_c1_1 and out0.merge_c2_0 = out1.merge_c2_1 "
                          + "last join "
                          + "(select foo.main.id as merge_id_2, foo.main.c1 as merge_c1_2, foo.main.c2 as merge_c2_2, t1.c2 of3 from main last join t1 on main.c1==t1.c1) as out2 "
                          + "on out0.merge_id_0 = out2.merge_id_2 and out0.merge_c1_0 = out2.merge_c1_2 and out0.merge_c2_0 = out2.merge_c2_2 "
                          + "last join "
                          + "(select foo.main.id as merge_id_3, foo.main.c1 as merge_c1_3, foo.main.c2 as merge_c2_3, t1.c2 of4 from main last join t1 order by t1.c2 on main.c1==t1.c1) as out3 "
                          + "on out0.merge_id_0 = out3.merge_id_3 and out0.merge_c1_0 = out3.merge_c1_3 and out0.merge_c2_0 = out3.merge_c2_3 "
                          + "last join "
                          + "(select foo.main.id as merge_id_4, foo.main.c1 as merge_c1_4, foo.main.c2 as merge_c2_4, sum(c2) over w1 from main "
                          + "window w1 as (union (select \"\" as id, * from t1) partition by c1 order by c2 rows between unbounded preceding and current row)) as out4 "
                          + "on out0.merge_id_0 = out4.merge_id_4 and out0.merge_c1_0 = out4.merge_c1_4 and out0.merge_c2_0 = out4.merge_c2_4;");
  
          // main table patterns
          sqls = Arrays.asList(
                  "select c1 from main",
                  "select c2 from foo.main",
                  "select bar.id from main as bar");
          String mainRenamed = SqlClusterExecutor.mergeSQL(sqls, "foo", Arrays.asList("id", "c1", "c2"), schemaMaps);
          Assert.assertEquals(mainRenamed,
                  "select `c1`, `c2`, `id` from (select foo.main.id as merge_id_0, foo.main.c1 as merge_c1_0, foo.main.c2 as merge_c2_0, c1 from main) as out0 "
                          + "last join "
                          + "(select foo.main.id as merge_id_1, foo.main.c1 as merge_c1_1, foo.main.c2 as merge_c2_1, c2 from foo.main) as out1 "
                          + "on out0.merge_id_0 = out1.merge_id_1 and out0.merge_c1_0 = out1.merge_c1_1 and out0.merge_c2_0 = out1.merge_c2_1 "
                          + "last join "
                          + "(select foo.main.id as merge_id_2, foo.main.c1 as merge_c1_2, foo.main.c2 as merge_c2_2, bar.id from main as bar) as out2 "
                          + "on out0.merge_id_0 = out2.merge_id_2 and out0.merge_c1_0 = out2.merge_c1_2 and out0.merge_c2_0 = out2.merge_c2_2;");
          // add one more db aaa, mergeSQL won't use the unrelated db
          schemaMaps.put("aaa", new HashMap<String, Schema>());
          String twoDB = SqlClusterExecutor.mergeSQL(sqls, "foo", Arrays.asList("id", "c1", "c2"), schemaMaps);
  
          // no used db, all tables are <db>.<table>
          sqls = Arrays.asList("select c1 from foo.main",
                  "select t1.c2 from foo.main as main last join foo.t1 as t1 on main.c1==t1.c1",
                  "select id from foo.main");
          String noUsedDB = SqlClusterExecutor.mergeSQL(sqls, "", Arrays.asList("id", "c1", "c2"), schemaMaps);
          Assert.assertEquals(noUsedDB,
                  "select `c1`, `c2`, `id` from "
                          + "(select foo.main.id as merge_id_0, foo.main.c1 as merge_c1_0, foo.main.c2 as merge_c2_0, c1 from foo.main) as out0 "
                          + "last join "
                          + "(select foo.main.id as merge_id_1, foo.main.c1 as merge_c1_1, foo.main.c2 as merge_c2_1, t1.c2 from foo.main as main "
                          + "last join foo.t1 as t1 on main.c1==t1.c1) as out1 "
                          + "on out0.merge_id_0 = out1.merge_id_1 and out0.merge_c1_0 = out1.merge_c1_1 and out0.merge_c2_0 = out1.merge_c2_1 "
                          + "last join "
                          + "(select foo.main.id as merge_id_2, foo.main.c1 as merge_c1_2, foo.main.c2 as merge_c2_2, id from foo.main) as out2 "
                          + "on out0.merge_id_0 = out2.merge_id_2 and out0.merge_c1_0 = out2.merge_c1_2 and out0.merge_c2_0 = out2.merge_c2_2;");
  
  
          // case in java quickstart
          String demoResult = SqlClusterExecutor.mergeSQL(Arrays.asList(
                  // 单表直出特征
                  "select c1 from main;",
                  // 单表聚合特征
                  "select sum(c1) over w1 of2 from main window w1 as (partition by c1 order by c2 rows between unbounded preceding and current row);",
                  // 多表特征
                  "select t1.c2 of4 from main last join t1 order by t1.c2 on main.c1==t1.c1;",
                  // 多表聚合特征
                  "select sum(c2) over w1 from main window w1 as (union (select \"\" as id, * from t1) partition by c1 order by c2 rows between unbounded preceding and current row);"),
                  "db", Arrays.asList("id", "c1"), Collections.singletonMap("db", dbSchema));
          Assert.assertEquals(demoResult, "select `c1`, `of2`, `of4`, `sum(c2)over w1` from "
                  + "(select db.main.id as merge_id_0, db.main.c1 as merge_c1_0, c1 from main) as out0 " + "last join "
                  + "(select db.main.id as merge_id_1, db.main.c1 as merge_c1_1, sum(c1) over w1 of2 from main window w1 as (partition by c1 order by c2 rows between unbounded preceding and current row)) as out1 "
                  + "on out0.merge_id_0 = out1.merge_id_1 and out0.merge_c1_0 = out1.merge_c1_1 " + "last join "
                  + "(select db.main.id as merge_id_2, db.main.c1 as merge_c1_2, t1.c2 of4 from main last join t1 order by t1.c2 on main.c1==t1.c1) as out2 "
                  + "on out0.merge_id_0 = out2.merge_id_2 and out0.merge_c1_0 = out2.merge_c1_2 last join "
                  + "(select db.main.id as merge_id_3, db.main.c1 as merge_c1_3, sum(c2) over w1 from main window w1 as (union (select \"\" as id, * from t1) partition by c1 order by c2 rows between unbounded preceding and current row)) as out3 "
                  + "on out0.merge_id_0 = out3.merge_id_3 and out0.merge_c1_0 = out3.merge_c1_3;");
      }
  
      @Test(dataProvider = "executor")
      public void testSQLToDag(SqlExecutor router) throws SQLException {
          String sql = " WITH q1 as (WITH q3 as (select * from t1 LIMIT 10), q4 as (select * from t2) select * from q3 left join q4 on q3.id = q4.id),"
                  +
                  "q2 as (select * from t3)" +
                  "select * from q1 last join q2 on q1.id = q2.id";
  
          DAGNode dag = router.SQLToDAG(sql);
  
          Assert.assertEquals(dag.name, "");
          Assert.assertEquals(dag.sql, "SELECT\n" +
                  "  *\n" +
                  "FROM\n" +
                  "  q1\n" +
                  "  LAST JOIN\n" +
                  "  q2\n" +
                  "  ON q1.id = q2.id\n");
          Assert.assertEquals(dag.producers.size(), 2);
  
          DAGNode input1 = dag.producers.get(0);
          Assert.assertEquals(input1.name, "q1");
          Assert.assertEquals(input1.sql, "SELECT\n" +
                  "  *\n" +
                  "FROM\n" +
                  "  q3\n" +
                  "  LEFT JOIN\n" +
                  "  q4\n" +
                  "  ON q3.id = q4.id\n");
          Assert.assertEquals(2, input1.producers.size());
  
          DAGNode input2 = dag.producers.get(1);
          Assert.assertEquals(input2.name, "q2");
          Assert.assertEquals(input2.sql, "SELECT\n" +
                  "  *\n" +
                  "FROM\n" +
                  "  t3\n");
          Assert.assertEquals(input2.producers.size(), 0);
  
          DAGNode q1In1 = input1.producers.get(0);
          Assert.assertEquals(q1In1.producers.size(), 0);
          Assert.assertEquals(q1In1.name, "q3");
          Assert.assertEquals(q1In1.sql, "SELECT\n" +
                  "  *\n" +
                  "FROM\n" +
                  "  t1\n" +
                  "LIMIT 10\n");
  
          DAGNode q1In2 = input1.producers.get(1);
          Assert.assertEquals(q1In2.producers.size(), 0);
          Assert.assertEquals(q1In2.name, "q4");
          Assert.assertEquals(q1In2.sql, "SELECT\n" +
                  "  *\n" +
                  "FROM\n" +
                  "  t2\n");
      }
  
      private void testMergeDAGSQLCase(String input, String output, String error) {
          Exception exception = null;    
          try {
              DAGNode dag = AIOSUtil.parseAIOSDAG(input);
              Map<String, Map<String, Schema>> tableSchema = AIOSUtil.parseAIOSTableSchema(input, "usedDB");
              String merged = SqlClusterExecutor.mergeDAGSQL(dag);
              System.out.println(merged);
              Assert.assertEquals(merged, output);
              List<String> errors = SqlClusterExecutor.validateSQLInRequest(merged, "usedDB", tableSchema);
              if (!errors.isEmpty()) {
                  throw new SQLException("merged sql is invalid: " + errors +
                          "\n, merged sql: " + merged + "\n, table schema: " + tableSchema);
              }
          } catch (Exception e) {
              e.printStackTrace();
              exception = e;
          }
          if (error == null) {
              Assert.assertTrue(exception == null);
          } else {
              Assert.assertTrue(exception.toString().contains(error));
          }
      }
  
      @Test
      public void testMergeDAGSQL() throws IOException {
          System.out.println("user.dir: " + System.getProperty("user.dir"));
          ArrayList<Path> inputs = new ArrayList<>();
          ArrayList<Path> outputs = new ArrayList<>();
          inputs.add(Paths.get("src/test/data/aiosdagsql/input1.json"));
          outputs.add(Paths.get("src/test/data/aiosdagsql/output1.sql"));
          for (int i = 0; i < inputs.size(); ++i) {
              String input = new String(Files.readAllBytes(inputs.get(i)));
              String output = new String(Files.readAllBytes(outputs.get(i)));
              testMergeDAGSQLCase(input, output, null);
          }
      }
  
      @Test
      public void testMergeDAGSQLError() throws IOException {
          System.out.println("user.dir: " + System.getProperty("user.dir"));
          ArrayList<Path> inputs = new ArrayList<>();
          ArrayList<Path> outputs = new ArrayList<>();
          inputs.add(Paths.get("src/test/data/aiosdagsql/error1.json"));
          outputs.add(Paths.get("src/test/data/aiosdagsql/error1.sql"));
          for (int i = 0; i < inputs.size(); ++i) {
              String input = new String(Files.readAllBytes(inputs.get(i)));
              String output = new String(Files.readAllBytes(outputs.get(i)));
              testMergeDAGSQLCase(input, output, "Fail to resolve expression");
          }
      }
  
  }
  
  