CreateAndDropTableDdlTest.cpp

/*
 * SPDX-FileCopyrightText: Copyright (c) 2020-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 * SPDX-License-Identifier: Apache-2.0
 */

/**
 * @file CreateAndDropTableDdlTest.cpp
 * @brief Test suite for CREATE and DROP DDL commands for tables and foreign tables
 *
 */

#include <gtest/gtest.h>
#include <boost/filesystem.hpp>

#include "Calcite/Calcite.h"
#include "Catalog/ForeignTable.h"
#include "Catalog/TableDescriptor.h"
#include "DBHandlerTestHelpers.h"
#include "DataMgr/FileMgr/FileBuffer.h"
#include "Fragmenter/FragmentDefaultValues.h"
#include "Shared/SysDefinitions.h"
#include "Shared/scope.h"
#include "TestHelpers.h"
#include "Utils/DdlUtils.h"

class MockCalcite : public Calcite {
 public:
  struct UpdateMetadataError : public std::runtime_error {
    UpdateMetadataError() : std::runtime_error("UpdateMetadataError") {}
  };
  void updateMetadata(std::string catalog, std::string table) override {
    throw UpdateMetadataError();
  }
};

extern bool g_enable_fsi;
extern bool g_enable_s3_fsi;

using namespace std;
using namespace TestHelpers;

namespace bf = boost::filesystem;

namespace {
struct ColumnAttributes {
  std::string column_name;
  bool not_null{false};
  int size{-1};
  SQLTypes type;
  SQLTypes sub_type{SQLTypes::kNULLT};
  int precision{0};
  int scale{0};
  EncodingType encoding_type{EncodingType::kENCODING_NONE};
  int encoding_size{0};
};
}  // namespace

class CreateAndDropTableDdlTest : public DBHandlerTestFixture {
 protected:
  void SetUp() override {
    g_enable_fsi = true;
    DBHandlerTestFixture::SetUp();
  }

  std::string getCreateTableQuery(const ddl_utils::TableType table_type,
                                  const std::string& table_name,
                                  const std::string& columns,
                                  bool if_not_exists = false,
                                  bool or_replace = false,
                                  bool temporary_table = false) const {
    return getCreateTableQuery(
        table_type, table_name, columns, {}, if_not_exists, or_replace, temporary_table);
  }

  std::string getCreateTableQuery(const ddl_utils::TableType table_type,
                                  const std::string& table_name,
                                  const std::string& columns,
                                  std::map<std::string, std::string> options,
                                  bool if_not_exists = false,
                                  bool or_replace = false,
                                  bool temporary_table = false) const {
    std::string query{"CREATE "};
    if (or_replace) {
      query += "OR REPLACE ";
    }
    if (temporary_table) {
      query += "TEMPORARY ";
    }
    if (table_type == ddl_utils::TableType::FOREIGN_TABLE) {
      query += "FOREIGN TABLE ";
    } else {
      query += "TABLE ";
    }
    if (if_not_exists) {
      query += "IF NOT EXISTS ";
    }
    query += table_name + columns;
    if (table_type == ddl_utils::TableType::FOREIGN_TABLE) {
      query += " SERVER default_local_delimited";
      options["file_path"] = "'" + getTestFilePath() + "'";
    }
    if (!options.empty()) {
      query += " WITH (";
      bool is_first = true;
      for (auto& [key, value] : options) {
        if (is_first) {
          is_first = false;
        } else {
          query += ", ";
        }
        query += key + " = " + value;
      }
      query += ")";
    }
    query += ";";
    return query;
  }

  std::string getDropTableQuery(const ddl_utils::TableType table_type,
                                const std::string& table_name,
                                bool if_exists = false) {
    std::string query{"DROP "};
    if (table_type == ddl_utils::TableType::FOREIGN_TABLE) {
      query += "FOREIGN TABLE ";
    } else {
      query += "TABLE ";
    }

    if (if_exists) {
      query += "IF EXISTS ";
    }
    query += table_name + ";";
    return query;
  }

  std::string getTestFilePath() const {
    return bf::canonical("../../Tests/FsiDataFiles/example_1.csv").string();
  }

  int createTestUser() {
    sql("CREATE USER test_user (password = 'test_pass');");
    sql("GRANT ACCESS ON DATABASE " + shared::kDefaultDbName + " TO test_user;");
    Catalog_Namespace::UserMetadata user_metadata;
    Catalog_Namespace::SysCatalog::instance().getMetadataForUser("test_user",
                                                                 user_metadata);
    return user_metadata.userId;
  }

  void dropTestUser() {
    loginAdmin();
    sql("DROP USER IF EXISTS test_user;");
  }

  std::string getTableDirPath() {
    const auto& catalog = getCatalog();
    auto base_path = catalog.getDataMgr().getGlobalFileMgr()->getBasePath();
    auto td = catalog.getMetadataForTable("test_foreign_table", false);
    CHECK(td);
    auto table_dir_path = base_path + "table_" + std::to_string(catalog.getDatabaseId()) +
                          "_" + std::to_string(td->tableId);
    return table_dir_path;
  }
};

class CreateTableTest : public CreateAndDropTableDdlTest,
                        public testing::WithParamInterface<ddl_utils::TableType> {
 protected:
  void SetUp() override {
    CreateAndDropTableDdlTest::SetUp();
    sql("DROP VIEW IF EXISTS test_view");
    sql(getDropTableQuery(GetParam(), "test_table", true));
    dropTestUser();
  }

  void TearDown() override {
    g_enable_fsi = true;
    sql("DROP VIEW IF EXISTS test_view");
    sql(getDropTableQuery(GetParam(), "test_table", true));
    dropTestUser();
    CreateAndDropTableDdlTest::TearDown();
  }

  /**
   *
   * @param td - table details returned from the catalog
   * @param table_type - table type
   * @param table_name - expected table name
   * @param column_count - expected number of columns in the table
   * @param user_id - id of user who owns the table. Default value is 0 (admin user id)
   * @param max_fragment_size - expected maximum table fragment size
   * used
   */
  void assertTableDetails(const TableDescriptor* td,
                          const ddl_utils::TableType table_type,
                          const std::string& table_name,
                          const int column_count,
                          const int user_id = shared::kRootUserId,
                          const int max_fragment_size = DEFAULT_FRAGMENT_ROWS) {
    EXPECT_EQ(table_name, td->tableName);
    EXPECT_EQ(Fragmenter_Namespace::FragmenterType::INSERT_ORDER, td->fragType);
    EXPECT_EQ(max_fragment_size, td->maxFragRows);
    EXPECT_EQ(size_t(DEFAULT_MAX_CHUNK_SIZE), size_t(td->maxChunkSize));
    EXPECT_EQ(size_t(DEFAULT_PAGE_SIZE), size_t(td->fragPageSize));
    EXPECT_EQ(DEFAULT_MAX_ROWS, td->maxRows);
    EXPECT_EQ(user_id, td->userId);
    EXPECT_EQ(Data_Namespace::MemoryLevel::DISK_LEVEL, td->persistenceLevel);
    EXPECT_FALSE(td->isView);
    EXPECT_EQ(0, td->nShards);
    EXPECT_EQ(0, td->shardedColumnId);
    EXPECT_EQ("[]", td->keyMetainfo);
    EXPECT_EQ("", td->fragments);
    EXPECT_EQ("", td->partitions);

    if (table_type == ddl_utils::TableType::FOREIGN_TABLE) {
      auto foreign_table = dynamic_cast<const foreign_storage::ForeignTable*>(td);
      ASSERT_NE(nullptr, foreign_table);
      EXPECT_EQ(column_count + 1, td->nColumns);  // +1 for rowid column
      EXPECT_FALSE(td->hasDeletedCol);
      EXPECT_EQ(StorageType::FOREIGN_TABLE, foreign_table->storageType);
      ASSERT_TRUE(foreign_table->options.find("FILE_PATH") !=
                  foreign_table->options.end());
      EXPECT_EQ(getTestFilePath(), foreign_table->options.find("FILE_PATH")->second);
      EXPECT_EQ("default_local_delimited", foreign_table->foreign_server->name);
    } else {
      EXPECT_EQ(column_count + 2, td->nColumns);  // +2 for rowid and $deleted$ columns
      EXPECT_TRUE(td->hasDeletedCol);
      EXPECT_TRUE(td->storageType.empty());
    }
  }

  void assertColumnDetails(const ColumnAttributes expected,
                           const ColumnDescriptor* column) {
    EXPECT_EQ(expected.column_name, column->columnName);
    EXPECT_TRUE(column->sourceName.empty());
    EXPECT_TRUE(column->chunks.empty());
    EXPECT_FALSE(column->isSystemCol);
    EXPECT_FALSE(column->isVirtualCol);
    EXPECT_TRUE(column->virtualExpr.empty());
    EXPECT_FALSE(column->isDeletedCol);

    auto& type_info = column->columnType;
    EXPECT_EQ(expected.not_null, type_info.get_notnull());
    EXPECT_EQ(expected.encoding_type, type_info.get_compression());
    EXPECT_EQ(expected.precision, type_info.get_dimension());
    EXPECT_EQ(expected.precision, type_info.get_precision());
    EXPECT_EQ(expected.precision, type_info.get_input_srid());
    EXPECT_EQ(expected.scale, type_info.get_scale());
    EXPECT_EQ(expected.scale, type_info.get_output_srid());
    EXPECT_EQ(expected.size, type_info.get_size());
    EXPECT_EQ(expected.type, type_info.get_type());
    EXPECT_EQ(expected.sub_type, type_info.get_subtype());

    // Comp param contains dictionary id for encoded strings
    if (type_info.get_compression() != kENCODING_DICT) {
      EXPECT_EQ(expected.encoding_size, type_info.get_comp_param());
    }
  }
};

TEST_P(CreateTableTest, BooleanAndNumberTypes) {
  std::string query =
      getCreateTableQuery(GetParam(),
                          "test_table",
                          "(bl BOOLEAN, bint BIGINT, bint8 BIGINT ENCODING FIXED(8), "
                          "bint16 BIGINT ENCODING FIXED(16), "
                          "bint32 BIGINT ENCODING FIXED(32), dc DECIMAL(5, 2), dc1 "
                          "DECIMAL(3), db DOUBLE, fl FLOAT, i INTEGER, "
                          "i8 INTEGER ENCODING FIXED(8), i16 INTEGER ENCODING FIXED(16), "
                          "si SMALLINT, si8 SMALLINT ENCODING FIXED(8), "
                          "ti TINYINT)");
  sql(query);

  auto& catalog = getCatalog();
  auto table = catalog.getMetadataForTable("test_table", false);
  assertTableDetails(table, GetParam(), "test_table", 15);

  auto columns = catalog.getAllColumnMetadataForTable(table->tableId, true, true, true);
  auto it = columns.begin();
  auto column = *it;
  ColumnAttributes expected_attributes{};
  expected_attributes.column_name = "bl";
  expected_attributes.size = 1;
  expected_attributes.type = kBOOLEAN;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "bint";
  expected_attributes.size = 8;
  expected_attributes.type = SQLTypes::kBIGINT;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "bint8";
  expected_attributes.size = 1;
  expected_attributes.type = kBIGINT;
  expected_attributes.encoding_type = kENCODING_FIXED;
  expected_attributes.encoding_size = 8;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "bint16";
  expected_attributes.size = 2;
  expected_attributes.type = kBIGINT;
  expected_attributes.encoding_type = kENCODING_FIXED;
  expected_attributes.encoding_size = 16;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "bint32";
  expected_attributes.size = 4;
  expected_attributes.type = kBIGINT;
  expected_attributes.encoding_type = kENCODING_FIXED;
  expected_attributes.encoding_size = 32;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "dc";
  expected_attributes.size = 4;
  expected_attributes.type = kDECIMAL;
  expected_attributes.encoding_type = kENCODING_FIXED;
  expected_attributes.encoding_size = 32;
  expected_attributes.precision = 5;
  expected_attributes.scale = 2;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "dc1";
  expected_attributes.size = 2;
  expected_attributes.type = kDECIMAL;
  expected_attributes.encoding_type = kENCODING_FIXED;
  expected_attributes.encoding_size = 16;
  expected_attributes.precision = 3;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "db";
  expected_attributes.size = 8;
  expected_attributes.type = kDOUBLE;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "fl";
  expected_attributes.size = 4;
  expected_attributes.type = kFLOAT;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "i";
  expected_attributes.size = 4;
  expected_attributes.type = kINT;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "i8";
  expected_attributes.size = 1;
  expected_attributes.type = kINT;
  expected_attributes.encoding_type = kENCODING_FIXED;
  expected_attributes.encoding_size = 8;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "i16";
  expected_attributes.size = 2;
  expected_attributes.type = kINT;
  expected_attributes.encoding_type = kENCODING_FIXED;
  expected_attributes.encoding_size = 16;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "si";
  expected_attributes.size = 2;
  expected_attributes.type = kSMALLINT;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "si8";
  expected_attributes.size = 1;
  expected_attributes.type = kSMALLINT;
  expected_attributes.encoding_type = kENCODING_FIXED;
  expected_attributes.encoding_size = 8;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "ti";
  expected_attributes.size = 1;
  expected_attributes.type = kTINYINT;
  assertColumnDetails(expected_attributes, column);
}

TEST_P(CreateTableTest, DateAndTimestampTypes) {
  std::string query = getCreateTableQuery(
      GetParam(),
      "test_table",
      "(dt DATE, dt16 DATE ENCODING FIXED(16), dt16_days DATE ENCODING DAYS(16), t TIME, "
      "t32 TIME ENCODING FIXED(32), tp0 TIMESTAMP(0), tp3 TIMESTAMP(3), tp6 "
      "TIMESTAMP(6), "
      "tp9 TIMESTAMP(9), tp32 TIMESTAMP ENCODING FIXED(32))");
  sql(query);

  auto& catalog = getCatalog();
  auto table = catalog.getMetadataForTable("test_table", false);
  assertTableDetails(table, GetParam(), "test_table", 10);

  auto columns = catalog.getAllColumnMetadataForTable(table->tableId, true, true, true);
  auto it = columns.begin();
  auto column = *it;
  ColumnAttributes expected_attributes{};
  expected_attributes.column_name = "dt";
  expected_attributes.size = 4;
  expected_attributes.type = kDATE;
  expected_attributes.encoding_type = kENCODING_DATE_IN_DAYS;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "dt16";
  expected_attributes.size = 2;
  expected_attributes.type = kDATE;
  expected_attributes.encoding_type = kENCODING_DATE_IN_DAYS;
  expected_attributes.encoding_size = 16;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "dt16_days";
  expected_attributes.size = 2;
  expected_attributes.type = kDATE;
  expected_attributes.encoding_type = kENCODING_DATE_IN_DAYS;
  expected_attributes.encoding_size = 16;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "t";
  expected_attributes.size = 8;
  expected_attributes.type = kTIME;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "t32";
  expected_attributes.size = 4;
  expected_attributes.type = kTIME;
  expected_attributes.encoding_type = kENCODING_FIXED;
  expected_attributes.encoding_size = 32;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "tp0";
  expected_attributes.size = 8;
  expected_attributes.type = kTIMESTAMP;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "tp3";
  expected_attributes.size = 8;
  expected_attributes.type = kTIMESTAMP;
  expected_attributes.precision = 3;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "tp6";
  expected_attributes.size = 8;
  expected_attributes.type = kTIMESTAMP;
  expected_attributes.precision = 6;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "tp9";
  expected_attributes.size = 8;
  expected_attributes.type = kTIMESTAMP;
  expected_attributes.precision = 9;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "tp32";
  expected_attributes.size = 4;
  expected_attributes.type = kTIMESTAMP;
  expected_attributes.encoding_type = kENCODING_FIXED;
  expected_attributes.encoding_size = 32;
  assertColumnDetails(expected_attributes, column);
}

TEST_P(CreateTableTest, TextTypes) {
  std::string query = getCreateTableQuery(
      GetParam(),
      "test_table",
      "(t TEXT ENCODING DICT, t8 TEXT ENCODING DICT(8), t16 TEXT ENCODING DICT(16), "
      "t32 TEXT ENCODING DICT(32), t_non_encoded TEXT ENCODING NONE)");
  sql(query);

  auto& catalog = getCatalog();
  auto table = catalog.getMetadataForTable("test_table", false);
  assertTableDetails(table, GetParam(), "test_table", 5);

  auto columns = catalog.getAllColumnMetadataForTable(table->tableId, true, true, true);
  auto it = columns.begin();
  auto column = *it;
  ColumnAttributes expected_attributes{};
  expected_attributes.column_name = "t";
  expected_attributes.size = 4;
  expected_attributes.type = kTEXT;
  expected_attributes.encoding_type = kENCODING_DICT;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "t8";
  expected_attributes.size = 1;
  expected_attributes.type = kTEXT;
  expected_attributes.encoding_type = kENCODING_DICT;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "t16";
  expected_attributes.size = 2;
  expected_attributes.type = kTEXT;
  expected_attributes.encoding_type = kENCODING_DICT;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "t32";
  expected_attributes.size = 4;
  expected_attributes.type = kTEXT;
  expected_attributes.encoding_type = kENCODING_DICT;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "t_non_encoded";
  expected_attributes.type = kTEXT;
  assertColumnDetails(expected_attributes, column);
}

TEST_P(CreateTableTest, GeoTypes) {
  std::string query = getCreateTableQuery(
      GetParam(),
      "test_table",
      "(ls LINESTRING, mpoly MULTIPOLYGON, p POINT, poly POLYGON, p1 GEOMETRY(POINT), "
      "p2 GEOMETRY(POINT, 4326), p3 GEOMETRY(POINT, 4326) ENCODING NONE, p4 "
      "GEOMETRY(POINT, 900913), "
      "ls1 GEOMETRY(LINESTRING, 4326) ENCODING COMPRESSED(32), ls2 GEOMETRY(LINESTRING, "
      "4326) ENCODING NONE, "
      "poly1 GEOMETRY(POLYGON, 4326) ENCODING COMPRESSED(32), mpoly1 "
      "GEOMETRY(MULTIPOLYGON, 4326))");
  sql(query);

  auto& catalog = getCatalog();
  auto table = catalog.getMetadataForTable("test_table", false);
  /**
   * LINESTRING adds 2 additional columns, MULTIPOLYGON adds 4 additional columns,
   * POINT adds 1 additional column, and POLYGON adds 3 additional columns when
   * expanded.
   */
  assertTableDetails(table, GetParam(), "test_table", 37);

  auto columns = catalog.getAllColumnMetadataForTable(table->tableId, true, true, true);
  auto it = columns.begin();
  auto column = *it;
  ColumnAttributes expected_attributes{};
  expected_attributes.column_name = "ls";
  expected_attributes.type = kLINESTRING;
  expected_attributes.sub_type = kGEOMETRY;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 3);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "mpoly";
  expected_attributes.type = kMULTIPOLYGON;
  expected_attributes.sub_type = kGEOMETRY;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 5);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "p";
  expected_attributes.type = kPOINT;
  expected_attributes.sub_type = kGEOMETRY;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 2);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "poly";
  expected_attributes.type = kPOLYGON;
  expected_attributes.sub_type = kGEOMETRY;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 4);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "p1";
  expected_attributes.type = kPOINT;
  expected_attributes.sub_type = kGEOMETRY;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 2);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "p2";
  expected_attributes.type = kPOINT;
  expected_attributes.sub_type = kGEOMETRY;
  expected_attributes.precision = 4326;
  expected_attributes.scale = 4326;
  expected_attributes.encoding_type = kENCODING_GEOINT;
  expected_attributes.encoding_size = 32;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 2);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "p3";
  expected_attributes.type = kPOINT;
  expected_attributes.sub_type = kGEOMETRY;
  expected_attributes.precision = 4326;
  expected_attributes.scale = 4326;
  expected_attributes.encoding_type = kENCODING_NONE;
  expected_attributes.encoding_size = 0;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 2);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "p4";
  expected_attributes.type = kPOINT;
  expected_attributes.sub_type = kGEOMETRY;
  expected_attributes.precision = 900913;
  expected_attributes.scale = 900913;
  expected_attributes.encoding_type = kENCODING_NONE;
  expected_attributes.encoding_size = 0;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 2);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "ls1";
  expected_attributes.type = kLINESTRING;
  expected_attributes.sub_type = kGEOMETRY;
  expected_attributes.precision = 4326;
  expected_attributes.scale = 4326;
  expected_attributes.encoding_type = kENCODING_GEOINT;
  expected_attributes.encoding_size = 32;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 3);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "ls2";
  expected_attributes.type = kLINESTRING;
  expected_attributes.sub_type = kGEOMETRY;
  expected_attributes.precision = 4326;
  expected_attributes.scale = 4326;
  expected_attributes.encoding_type = kENCODING_NONE;
  expected_attributes.encoding_size = 0;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 3);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "poly1";
  expected_attributes.type = kPOLYGON;
  expected_attributes.sub_type = kGEOMETRY;
  expected_attributes.precision = 4326;
  expected_attributes.scale = 4326;
  expected_attributes.encoding_type = kENCODING_GEOINT;
  expected_attributes.encoding_size = 32;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 4);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "mpoly1";
  expected_attributes.type = kMULTIPOLYGON;
  expected_attributes.sub_type = kGEOMETRY;
  expected_attributes.precision = 4326;
  expected_attributes.scale = 4326;
  expected_attributes.encoding_type = kENCODING_GEOINT;
  expected_attributes.encoding_size = 32;
  assertColumnDetails(expected_attributes, column);
}

TEST_P(CreateTableTest, ArrayTypes) {
  std::string query = getCreateTableQuery(
      GetParam(),
      "test_table",
      R"((t TINYINT[], t2 TINYINT[1], i INTEGER[], i2 INTEGER[1], bint BIGINT[], bint2 BIGINT[1],
      txt TEXT[] ENCODING DICT(32), txt2 TEXT[1] ENCODING DICT(32), f FLOAT[], f2 FLOAT[1], d DOUBLE[], d2 DOUBLE[1], dc DECIMAL(18,6)[], dc2 DECIMAL(18,6)[1], b BOOLEAN[], b2 BOOLEAN[1], dt DATE[], dt2 DATE[1], tm TIME[], tm2 TIME[1], tp TIMESTAMP[], tp2 TIMESTAMP[1], tp3 TIMESTAMP(3)[], tp4 TIMESTAMP(9)[1]))");
  sql(query);

  auto& catalog = getCatalog();
  auto table = catalog.getMetadataForTable("test_table", false);
  assertTableDetails(table, GetParam(), "test_table", 24);

  auto columns = catalog.getAllColumnMetadataForTable(table->tableId, true, true, true);
  auto it = columns.begin();
  auto column = *it;
  ColumnAttributes expected_attributes{};
  expected_attributes.column_name = "t";
  expected_attributes.type = kARRAY;
  expected_attributes.sub_type = kTINYINT;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "t2";
  expected_attributes.size = 1;
  expected_attributes.type = kARRAY;
  expected_attributes.sub_type = kTINYINT;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "i";
  expected_attributes.type = kARRAY;
  expected_attributes.sub_type = kINT;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "i2";
  expected_attributes.size = 4;
  expected_attributes.type = kARRAY;
  expected_attributes.sub_type = kINT;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "bint";
  expected_attributes.type = kARRAY;
  expected_attributes.sub_type = kBIGINT;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "bint2";
  expected_attributes.size = 8;
  expected_attributes.type = kARRAY;
  expected_attributes.sub_type = kBIGINT;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "txt";
  expected_attributes.type = kARRAY;
  expected_attributes.sub_type = kTEXT;
  expected_attributes.encoding_type = kENCODING_DICT;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "txt2";
  expected_attributes.size = 4;
  expected_attributes.type = kARRAY;
  expected_attributes.sub_type = kTEXT;
  expected_attributes.encoding_type = kENCODING_DICT;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "f";
  expected_attributes.type = kARRAY;
  expected_attributes.sub_type = kFLOAT;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "f2";
  expected_attributes.size = 4;
  expected_attributes.type = kARRAY;
  expected_attributes.sub_type = kFLOAT;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "d";
  expected_attributes.type = kARRAY;
  expected_attributes.sub_type = kDOUBLE;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "d2";
  expected_attributes.size = 8;
  expected_attributes.type = kARRAY;
  expected_attributes.sub_type = kDOUBLE;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "dc";
  expected_attributes.type = kARRAY;
  expected_attributes.sub_type = kDECIMAL;
  expected_attributes.precision = 18;
  expected_attributes.scale = 6;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "dc2";
  expected_attributes.size = 8;
  expected_attributes.type = kARRAY;
  expected_attributes.sub_type = kDECIMAL;
  expected_attributes.precision = 18;
  expected_attributes.scale = 6;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "b";
  expected_attributes.type = kARRAY;
  expected_attributes.sub_type = kBOOLEAN;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "b2";
  expected_attributes.size = 1;
  expected_attributes.type = kARRAY;
  expected_attributes.sub_type = kBOOLEAN;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "dt";
  expected_attributes.type = kARRAY;
  expected_attributes.sub_type = kDATE;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "dt2";
  expected_attributes.size = 8;
  expected_attributes.type = kARRAY;
  expected_attributes.sub_type = kDATE;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "tm";
  expected_attributes.type = kARRAY;
  expected_attributes.sub_type = kTIME;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "tm2";
  expected_attributes.size = 8;
  expected_attributes.type = kARRAY;
  expected_attributes.sub_type = kTIME;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "tp";
  expected_attributes.type = kARRAY;
  expected_attributes.sub_type = kTIMESTAMP;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "tp2";
  expected_attributes.size = 8;
  expected_attributes.type = kARRAY;
  expected_attributes.sub_type = kTIMESTAMP;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "tp3";
  expected_attributes.size = -1;
  expected_attributes.type = kARRAY;
  expected_attributes.sub_type = kTIMESTAMP;
  expected_attributes.precision = 3;
  assertColumnDetails(expected_attributes, column);

  std::advance(it, 1);
  column = *it;
  expected_attributes = {};
  expected_attributes.column_name = "tp4";
  expected_attributes.size = 8;
  expected_attributes.type = kARRAY;
  expected_attributes.sub_type = kTIMESTAMP;
  expected_attributes.precision = 9;
  assertColumnDetails(expected_attributes, column);
}

TEST_P(CreateTableTest, FixedEncodingForNonNumberOrTimeType) {
  std::string query =
      getCreateTableQuery(GetParam(), "test_table", "(col1 POINT ENCODING FIXED(8))");
  EXPECT_ANY_THROW(sql(query));
}

TEST_P(CreateTableTest, DictEncodingNonTextType) {
  std::string query =
      getCreateTableQuery(GetParam(), "test_table", "(col1 INTEGER ENCODING DICT)");
  queryAndAssertException(query,
                          "col1: Dictionary encoding is only supported on "
                          "string or string array columns.");
}

TEST_P(CreateTableTest, CompressedEncodingNonWGS84GeoType) {
  std::string query = getCreateTableQuery(
      GetParam(), "test_table", "(col1 GEOMETRY(POINT, 900913) ENCODING COMPRESSED(32))");
  queryAndAssertException(
      query, "col1: COMPRESSED encoding is only supported on WGS84 geo columns.");
}

TEST_P(CreateTableTest, CompressedEncodingNon32Bit) {
  std::string query = getCreateTableQuery(
      GetParam(), "test_table", "(col1 GEOMETRY(POINT, 4326) ENCODING COMPRESSED(16))");
  queryAndAssertException(query,
                          "col1: only 32-bit COMPRESSED geo encoding is supported");
}

TEST_P(CreateTableTest, DaysEncodingNonDateType) {  // Param for DECIMAL and NUMERIC
  std::string query =
      getCreateTableQuery(GetParam(), "test_table", "(col1 TIME ENCODING DAYS(16))");
  queryAndAssertException(query,
                          "col1: Days encoding is only supported for DATE columns.");
}

TEST_P(CreateTableTest, NonEncodedDictArray) {
  std::string query =
      getCreateTableQuery(GetParam(), "test_table", "(col1 TEXT[] ENCODING NONE)");
  queryAndAssertException(query,
                          "col1: Array of strings must be dictionary encoded. "
                          "Specify ENCODING DICT");
}

TEST_P(CreateTableTest, FixedLengthArrayOfVarLengthType) {
  std::string query =
      getCreateTableQuery(GetParam(), "test_table", "(col1 LINESTRING[5])");
  EXPECT_ANY_THROW(sql(query));
}

TEST_P(CreateTableTest, UnsupportedTimestampPrecision) {
  std::string query =
      getCreateTableQuery(GetParam(), "test_table", "(col1 TIMESTAMP(10))");
  queryAndAssertException(query,
                          "Only TIMESTAMP(n) where n = (0,3,6,9) are supported now.");
}

TEST_P(CreateTableTest, UnsupportedTimePrecision) {
  std::string query = getCreateTableQuery(GetParam(), "test_table", "(col1 TIME(1))");
  queryAndAssertException(query, "Only TIME(0) is supported now.");
}

TEST_P(CreateTableTest, NotNullColumnConstraint) {
  sql(getCreateTableQuery(GetParam(), "test_table", "(col1 INTEGER NOT NULL)"));

  auto& catalog = getCatalog();
  auto table = catalog.getMetadataForTable("test_table", false);
  auto column = catalog.getMetadataForColumn(table->tableId, "col1");

  ASSERT_TRUE(column->columnType.get_notnull());
}

TEST_P(CreateTableTest, DuplicateColumnNames) {
  std::string query =
      getCreateTableQuery(GetParam(), "test_table", "(col1 INTEGER, col1 INTEGER)");
  queryAndAssertException(query, "Column 'col1' defined more than once");
}

TEST_P(CreateTableTest, ExistingTableWithIfNotExists) {
  sql(getCreateTableQuery(GetParam(), "test_table", "(col1 INTEGER)"));
  sql(getCreateTableQuery(GetParam(), "test_table", "(col1 INTEGER)", true));

  auto& catalog = getCatalog();
  auto table = catalog.getMetadataForTable("test_table", false);
  assertTableDetails(table, GetParam(), "test_table", 1);

  auto column = catalog.getMetadataForColumn(table->tableId, "col1");
  ColumnAttributes expected_attributes{};
  expected_attributes.column_name = "col1";
  expected_attributes.size = 4;
  expected_attributes.type = kINT;
  assertColumnDetails(expected_attributes, column);
}

TEST_P(CreateTableTest, ExistingTableWithoutIfNotExists) {
  std::string query = getCreateTableQuery(GetParam(), "test_table", "(col1 INTEGER)");
  sql(query);
  queryAndAssertException(query,
                          "Table or View with name \"test_table\" already exists.");
}

TEST_P(CreateTableTest, UnauthorizedUser) {
  createTestUser();
  login("test_user", "test_pass");
  std::string query = getCreateTableQuery(GetParam(), "test_table", "(col1 INTEGER)");

  if (GetParam() == ddl_utils::TableType::FOREIGN_TABLE) {
    queryAndAssertException(query,
                            "Foreign table \"test_table\" will not be "
                            "created. User has no CREATE TABLE privileges.");
  } else {
    queryAndAssertException(query,
                            "Table test_table will not be created. User has "
                            "no create privileges.");
  }
}

TEST_P(CreateTableTest, AuthorizedUser) {
  auto user_id = createTestUser();
  sql("GRANT CREATE TABLE ON DATABASE " + shared::kDefaultDbName + " TO test_user;");
  login("test_user", "test_pass");

  sql(getCreateTableQuery(GetParam(), "test_table", "(col1 INTEGER)"));

  auto& catalog = getCatalog();
  auto table = catalog.getMetadataForTable("test_table", false);
  assertTableDetails(table, GetParam(), "test_table", 1, user_id);

  auto column = catalog.getMetadataForColumn(table->tableId, "col1");
  ColumnAttributes expected_attributes{};
  expected_attributes.column_name = "col1";
  expected_attributes.size = 4;
  expected_attributes.type = kINT;
  assertColumnDetails(expected_attributes, column);
}

TEST_P(CreateTableTest, WithFragmentSizeOption) {
  sql(getCreateTableQuery(
      GetParam(), "test_table", "(col1 INTEGER)", {{"fragment_size", "10"}}));

  auto& catalog = getCatalog();
  auto table = catalog.getMetadataForTable("test_table", false);
  assertTableDetails(table, GetParam(), "test_table", 1, 0, 10);

  auto column = catalog.getMetadataForColumn(table->tableId, "col1");
  ColumnAttributes expected_attributes{};
  expected_attributes.column_name = "col1";
  expected_attributes.size = 4;
  expected_attributes.type = kINT;
  assertColumnDetails(expected_attributes, column);
}

TEST_P(CreateTableTest, MaxFragmentSizeExceeded) {
  auto query = (getCreateTableQuery(
      GetParam(), "test_table", "(col1 INTEGER)", {{"fragment_size", "3147483646"}}));
  queryAndAssertException(
      query,
      "Invalid value \"3147483646\" provided for FRAGMENT_SIZE "
      "option, expected a positive integer between 1 and 2147483647.");
}

TEST_P(CreateTableTest, OverflowFragmentSize) {
  auto query = (getCreateTableQuery(GetParam(),
                                    "test_table",
                                    "(col1 INTEGER)",
                                    {{"fragment_size", "9223372036854775808"}}));
  if (GetParam() == ddl_utils::TableType::FOREIGN_TABLE) {
    queryAndAssertException(
        query,
        "Invalid value \"9223372036854775808\" provided for FRAGMENT_SIZE "
        "option, expected a positive integer between 1 and 2147483647.");
  } else {
    queryAndAssertException(query, "Unable to handle literal for fragment_size");
  }
}

TEST_P(CreateTableTest, ZeroFragmentSize) {
  auto query = (getCreateTableQuery(
      GetParam(), "test_table", "(col1 INTEGER)", {{"fragment_size", "0"}}));
  queryAndAssertException(query,
                          "Invalid value \"0\" provided for FRAGMENT_SIZE option, "
                          "expected a positive integer between 1 and 2147483647.");
}

TEST_P(CreateTableTest, NegativeFragmentSize) {
  auto query = (getCreateTableQuery(
      GetParam(), "test_table", "(col1 INTEGER)", {{"fragment_size", "-1"}}));
  queryAndAssertException(query,
                          "Invalid value \"-1\" provided for FRAGMENT_SIZE option, "
                          "expected a positive integer between 1 and 2147483647.");
}

TEST_P(CreateTableTest, UnsupportedGeometrySyntax) {
  auto query = getCreateTableQuery(GetParam(), "test_table", "(geom GEOMETRY)");
  queryAndAssertException(query, "Unsupported type \"GEOMETRY\" specified.");
}

TEST_P(CreateTableTest, UnsupportedGeographySyntax) {
  auto query = getCreateTableQuery(GetParam(), "test_table", "(geom GEOGRAPHY)");
  queryAndAssertPartialException(query, "Incorrect syntax near the keyword 'GEOGRAPHY'");
}

INSTANTIATE_TEST_SUITE_P(CreateAndDropTableDdlTest,
                         CreateTableTest,
                         testing::Values(ddl_utils::TableType::TABLE,
                                         ddl_utils::TableType::FOREIGN_TABLE),
                         [](const auto& param_info) {
                           return ddl_utils::table_type_enum_to_string(param_info.param);
                         });

class NegativePrecisionOrDimensionTest
    : public CreateAndDropTableDdlTest,
      public testing::WithParamInterface<std::tuple<ddl_utils::TableType, std::string>> {
};

TEST_P(NegativePrecisionOrDimensionTest, NegativePrecisionOrDimension) {
  const auto& [table_type, data_type] = GetParam();
  try {
    sql(getCreateTableQuery(table_type, "test_table", "(col1 " + data_type + "(-1))"));
    FAIL() << "An exception should have been thrown for this test case.";
  } catch (const TDBException& e) {
    if (table_type == ddl_utils::TableType::FOREIGN_TABLE) {
      ASSERT_TRUE(e.error_msg.find("SQL Error") != std::string::npos);
    }
  }
}

INSTANTIATE_TEST_SUITE_P(
    CreateTableTest,
    NegativePrecisionOrDimensionTest,
    testing::Combine(testing::Values(ddl_utils::TableType::TABLE,
                                     ddl_utils::TableType::FOREIGN_TABLE),
                     testing::Values("CHAR", "VARCHAR", "DECIMAL", "NUMERIC")),
    [](const auto& param_info) {
      return ddl_utils::table_type_enum_to_string(std::get<0>(param_info.param)) + "_" +
             std::get<1>(param_info.param);
    });

class PrecisionAndScaleTest
    : public CreateAndDropTableDdlTest,
      public testing::WithParamInterface<std::tuple<ddl_utils::TableType, std::string>> {
};

TEST_P(PrecisionAndScaleTest, MaxPrecisionExceeded) {
  const auto& [table_type, data_type] = GetParam();
  std::string query =
      getCreateTableQuery(table_type, "test_table", "(col1 " + data_type + "(20))");
  queryAndAssertException(query,
                          "DECIMAL and NUMERIC precision cannot be larger than 18.");
}

TEST_P(PrecisionAndScaleTest, ScaleNotLessThanPrecision) {
  const auto& [table_type, data_type] = GetParam();
  std::string query =
      getCreateTableQuery(table_type, "test_table", "(col1 " + data_type + "(10, 10))");
  queryAndAssertException(query,
                          "DECIMAL and NUMERIC must have precision larger than scale.");
}

INSTANTIATE_TEST_SUITE_P(
    CreateTableTest,
    PrecisionAndScaleTest,
    testing::Combine(testing::Values(ddl_utils::TableType::TABLE,
                                     ddl_utils::TableType::FOREIGN_TABLE),
                     testing::Values("DECIMAL", "NUMERIC")),
    [](const auto& param_info) {
      return ddl_utils::table_type_enum_to_string(std::get<0>(param_info.param)) + "_" +
             std::get<1>(param_info.param);
    });

class CreateForeignTableTest : public CreateAndDropTableDdlTest {
 protected:
  void SetUp() override {
    CreateAndDropTableDdlTest::SetUp();
    sql("DROP FOREIGN TABLE IF EXISTS test_foreign_table;");
    dropTestUser();
  }

  void TearDown() override {
    g_enable_fsi = true;
    g_enable_s3_fsi = false;
    loginAdmin();
    sql("DROP FOREIGN TABLE IF EXISTS test_foreign_table;");
    sql("DROP SERVER IF EXISTS test_server;");
    sql("DROP SERVER IF EXISTS test_server_ce;");
    dropTestUser();
    CreateAndDropTableDdlTest::TearDown();
  }

  void assertOptionEquals(const foreign_storage::ForeignTable* table,
                          const std::string& key,
                          const std::string& value) {
    if (const auto& opt_it = table->options.find(key); opt_it != table->options.end()) {
      ASSERT_EQ(opt_it->second, value);
    } else {
      FAIL() << "Expected value for option " << key;
    }
  }

  const foreign_storage::ForeignTable* getForeignTable(
      const std::string& filename) const {
    auto table = getCatalog().getMetadataForTable(filename, false);
    CHECK(table);
    auto foreign_table = dynamic_cast<const foreign_storage::ForeignTable*>(table);
    return foreign_table;
  }

  std::unique_ptr<const foreign_storage::ForeignTable> getForeignTableFromStorage(
      const std::string& filename) const {
    return getCatalog().getForeignTableFromStorage(getForeignTable(filename)->tableId);
  }

  // Asserts option is as expected for in-memory table then again in catalog storage.
  void assertOptionEquals(const std::string& key, const std::string& value) {
    assertOptionEquals(getForeignTable("test_foreign_table"), key, value);
    assertOptionEquals(
        getForeignTableFromStorage("test_foreign_table").get(), key, value);
  }
};

TEST_F(CreateForeignTableTest, NonExistentServer) {
  std::string query{
      "CREATE FOREIGN TABLE test_foreign_table(col1 INTEGER) SERVER "
      "non_existent_server;"};
  queryAndAssertException(
      query,
      "Foreign Table with name \"test_foreign_table\" can not be created. "
      "Associated foreign server with name \"non_existent_server\" does not exist.");
}

TEST_F(CreateForeignTableTest, DefaultCsvFileServerName) {
  sql("CREATE FOREIGN TABLE test_foreign_table(col1 INTEGER) "
      "SERVER default_local_delimited WITH (file_path = '" +
      getTestFilePath() + "');");
  ASSERT_NE(nullptr, getCatalog().getMetadataForTable("test_foreign_table", false));
}

TEST_F(CreateForeignTableTest, DefaultParquetFileServerName) {
  sql("CREATE FOREIGN TABLE test_foreign_table(col1 INTEGER) "
      "SERVER default_local_parquet WITH (file_path = '" +
      getTestFilePath() + "');");
  ASSERT_NE(nullptr, getCatalog().getMetadataForTable("test_foreign_table", false));
}

TEST_F(CreateForeignTableTest, InvalidTableOption) {
  std::string query{
      "CREATE FOREIGN TABLE test_foreign_table(col1 INTEGER) "
      "SERVER default_local_delimited WITH (invalid_option = 'value');"};
  queryAndAssertException(query, "Invalid foreign table option \"INVALID_OPTION\".");
}

TEST_F(CreateForeignTableTest, WrongTableOptionCharacterSize) {
  std::string query{
      "CREATE FOREIGN TABLE test_foreign_table(col1 INTEGER) "
      "SERVER default_local_delimited WITH (delimiter = ',,', file_path = '" +
      getTestFilePath() + "');"};
  queryAndAssertException(query,
                          "Invalid value specified for option \"DELIMITER\". "
                          "Expected a single character, \"\\n\" or  \"\\t\".");
}

TEST_F(CreateForeignTableTest, InvalidTableOptionBooleanValue) {
  std::string query{
      "CREATE FOREIGN TABLE test_foreign_table(col1 INTEGER) "
      "SERVER default_local_delimited WITH (header = 'value', file_path = '" +
      getTestFilePath() + "');"};
  queryAndAssertException(
      query,
      "Invalid boolean value specified for \"HEADER\" foreign table option. "
      "Value must be either 'true' or 'false'.");
}

TEST_F(CreateForeignTableTest, FsiDisabled) {
  g_enable_fsi = false;
  std::string query = getCreateTableQuery(
      ddl_utils::TableType::FOREIGN_TABLE, "test_foreign_table", "(col1 INTEGER)");
  // Exception differs depending on which parser is enabled
  EXPECT_ANY_THROW(sql(query));
}

TEST_F(CreateForeignTableTest, DefaultServerWrapperPathMissingCsv) {
  queryAndAssertException(
      "CREATE FOREIGN TABLE test_foreign_table(col1 INTEGER) "
      "SERVER default_local_delimited;",
      "No file_path found for Foreign Table \"test_foreign_table\". "
      "Table must have either set a \"FILE_PATH\" "
      "option, or its parent server must have set a \"BASE_PATH\" option.");
}

TEST_F(CreateForeignTableTest, DefaultServerWrapperPathMissingParquet) {
  queryAndAssertException(
      "CREATE FOREIGN TABLE test_foreign_table(col1 INTEGER) "
      "SERVER default_local_parquet;",
      "No file_path found for Foreign Table \"test_foreign_table\". "
      "Table must have either set a \"FILE_PATH\" "
      "option, or its parent server must have set a \"BASE_PATH\" option.");
}

TEST_F(CreateForeignTableTest, ServerPathPresentWrapperPathMissing) {
  sql("CREATE SERVER test_server FOREIGN DATA WRAPPER delimited_file WITH (storage_type "
      "= "
      "'LOCAL_FILE', base_path = '" +
      bf::canonical("../../Tests/FsiDataFiles/example_1_dir").string() + "');");
  sql("CREATE FOREIGN TABLE test_foreign_table(t TEXT, i INTEGER[]) "
      "SERVER test_server;");
  sql("SELECT * FROM test_foreign_table;");
}

TEST_F(CreateForeignTableTest, ServerPathPresentWrapperPathEmpty) {
  sql("CREATE SERVER test_server FOREIGN DATA WRAPPER delimited_file WITH (storage_type "
      "= "
      "'LOCAL_FILE', base_path = '" +
      bf::canonical("../../Tests/FsiDataFiles/example_1_dir").string() + "');");
  sql("CREATE FOREIGN TABLE test_foreign_table(t TEXT, i INTEGER[]) "
      "SERVER test_server WITH (file_path = '');");
  sql("SELECT * FROM test_foreign_table;");
}

TEST_F(CreateForeignTableTest, ServerPathMissingWrapperPathMissing) {
  sql("CREATE SERVER test_server FOREIGN DATA WRAPPER delimited_file WITH (storage_type "
      "= "
      "'LOCAL_FILE');");
  queryAndAssertException(
      "CREATE FOREIGN TABLE test_foreign_table(t TEXT, i INTEGER[]) "
      "SERVER test_server;",
      "No file_path found for Foreign Table \"test_foreign_table\". "
      "Table must have either set a \"FILE_PATH\" "
      "option, or its parent server must have set a \"BASE_PATH\" option.");
}

TEST_F(CreateForeignTableTest, UnsupportedOption) {
  queryAndAssertException(
      "CREATE FOREIGN TABLE test_foreign_table(col1 INTEGER) "
      "SERVER default_local_delimited WITH (file_path = '" +
          getTestFilePath() + "', shard_count = 4);",
      "Invalid foreign table option \"SHARD_COUNT\".");
}

TEST_F(CreateForeignTableTest, ServerPathMissingWrapperPathRelative) {
  sql("CREATE FOREIGN TABLE test_foreign_table(col1 INTEGER) "
      "SERVER default_local_delimited WITH (file_path = "
      "'../../Tests/FsiDataFiles/0.csv');");
  sql("SELECT * FROM test_foreign_table;");
}

TEST_F(CreateForeignTableTest, S3SelectWrongServer) {
  std::string query =
      "CREATE FOREIGN TABLE test_foreign_table (t TEXT) "s +
      "SERVER default_local_delimited WITH (S3_ACCESS_TYPE = 'S3_SELECT', "
      "file_path = '../../Tests/FsiDataFiles/0.csv');";
  queryAndAssertException(
      query,
      "The \"S3_ACCESS_TYPE\" option is only valid for foreign tables using "
      "servers with \"STORAGE_TYPE\" option value of \"AWS_S3\".");
}

TEST_F(CreateForeignTableTest, UnauthorizedServerUsage) {
  sql("CREATE SERVER test_server FOREIGN DATA WRAPPER delimited_file "
      "WITH (storage_type = 'LOCAL_FILE', base_path = '');");
  createTestUser();
  sql("GRANT CREATE TABLE ON DATABASE " + shared::kDefaultDbName + " TO test_user;");
  login("test_user", "test_pass");
  queryAndAssertException(
      "CREATE FOREIGN TABLE test_foreign_table(t TEXT, i INTEGER[]) "
      "SERVER test_server WITH (file_path = '../../Tests/FsiDataFiles/0.csv');",
      "Current user does not have USAGE privilege on foreign server: "
      "test_server");
}

TEST_F(CreateForeignTableTest, AuthorizedServerUsage) {
  sql("CREATE SERVER test_server FOREIGN DATA WRAPPER delimited_file "
      "WITH (storage_type = 'LOCAL_FILE', base_path = '');");
  createTestUser();
  sql("GRANT CREATE TABLE ON DATABASE " + shared::kDefaultDbName + " TO test_user;");
  sql("GRANT USAGE ON SERVER test_server TO test_user;");
  login("test_user", "test_pass");
  sql("CREATE FOREIGN TABLE test_foreign_table(t TEXT, i INTEGER[]) "
      "SERVER test_server WITH (file_path = '" +
      getTestFilePath() + "');");
  sql("SELECT * FROM test_foreign_table;");
}

TEST_F(CreateForeignTableTest, AuthorizedDatabaseServerUser) {
  sql("CREATE SERVER test_server FOREIGN DATA WRAPPER delimited_file "
      "WITH (storage_type = 'LOCAL_FILE', base_path = '');");
  createTestUser();
  sql("GRANT CREATE TABLE ON DATABASE " + shared::kDefaultDbName + " TO test_user;");
  sql("GRANT SERVER USAGE ON DATABASE " + shared::kDefaultDbName + " TO test_user;");
  login("test_user", "test_pass");
  sql("CREATE FOREIGN TABLE test_foreign_table(t TEXT, i INTEGER[]) "
      "SERVER test_server WITH (file_path = '" +
      getTestFilePath() + "');");
  sql("SELECT * FROM test_foreign_table;");
}

TEST_F(CreateForeignTableTest, NonSuserServerOwnerUsage) {
  createTestUser();
  sql("GRANT CREATE TABLE ON DATABASE " + shared::kDefaultDbName + " TO test_user;");
  sql("GRANT CREATE SERVER ON DATABASE " + shared::kDefaultDbName + " TO test_user;");
  login("test_user", "test_pass");
  sql("CREATE SERVER test_server FOREIGN DATA WRAPPER delimited_file "
      "WITH (storage_type = 'LOCAL_FILE', base_path = '');");
  sql("CREATE FOREIGN TABLE test_foreign_table(t TEXT, i INTEGER[]) "
      "SERVER test_server WITH (file_path = '" +
      getTestFilePath() + "');");
  sql("SELECT * FROM test_foreign_table;");
}

TEST_F(CreateForeignTableTest, RevokedServerUsage) {
  sql("CREATE SERVER test_server FOREIGN DATA WRAPPER delimited_file "
      "WITH (storage_type = 'LOCAL_FILE', base_path = '');");
  createTestUser();
  sql("GRANT CREATE TABLE ON DATABASE " + shared::kDefaultDbName + " TO test_user;");
  sql("GRANT USAGE ON SERVER test_server TO test_user;");
  sql("REVOKE USAGE ON SERVER test_server FROM test_user;");
  login("test_user", "test_pass");
  queryAndAssertException(
      "CREATE FOREIGN TABLE test_foreign_table(t TEXT, i INTEGER[]) "
      "SERVER test_server WITH (file_path = '../../Tests/FsiDataFiles/0.csv');",
      "Current user does not have USAGE privilege on foreign server: "
      "test_server");
}

TEST_F(CreateForeignTableTest, RevokedDatabaseServerUsage) {
  sql("CREATE SERVER test_server FOREIGN DATA WRAPPER delimited_file "
      "WITH (storage_type = 'LOCAL_FILE', base_path = '');");
  createTestUser();
  sql("GRANT CREATE TABLE ON DATABASE " + shared::kDefaultDbName + " TO test_user;");
  sql("GRANT SERVER USAGE ON DATABASE " + shared::kDefaultDbName + " TO test_user;");
  sql("REVOKE SERVER USAGE ON DATABASE " + shared::kDefaultDbName + " FROM test_user;");
  login("test_user", "test_pass");
  queryAndAssertException(
      "CREATE FOREIGN TABLE test_foreign_table(t TEXT, i INTEGER[]) "
      "SERVER test_server WITH (file_path = '../../Tests/FsiDataFiles/0.csv');",
      "Current user does not have USAGE privilege on foreign server: "
      "test_server");
}

TEST_F(CreateForeignTableTest, TableDirectoryIsNotCreated) {
  sql(getCreateTableQuery(ddl_utils::TableType::FOREIGN_TABLE,
                          "test_foreign_table",
                          "(t TEXT, i INTEGER[])"));
  sql("SELECT * FROM test_foreign_table;");
  ASSERT_FALSE(boost::filesystem::exists(getTableDirPath()));
}

TEST_F(CreateForeignTableTest, RegexParserWrapperWithMissingLineRegex) {
  std::string query = "CREATE FOREIGN TABLE test_foreign_table (t TEXT) "s +
                      "SERVER default_local_regex_parsed WITH ("
                      "file_path = '../../Tests/FsiDataFiles/0.csv');";
  queryAndAssertException(
      query, "Foreign table options must contain a non-empty value for \"LINE_REGEX\".");
}

TEST_F(CreateForeignTableTest, RegexParserWrapperWithEmptyLineRegex) {
  std::string query = "CREATE FOREIGN TABLE test_foreign_table (t TEXT) "s +
                      "SERVER default_local_regex_parsed WITH (LINE_REGEX = '', "
                      "file_path = '../../Tests/FsiDataFiles/0.csv');";
  queryAndAssertException(
      query, "Foreign table options must contain a non-empty value for \"LINE_REGEX\".");
}

TEST_F(CreateForeignTableTest, RegexParserWrapperWithInvalidLineRegex) {
  std::string query = "CREATE FOREIGN TABLE test_foreign_table (t TEXT) "s +
                      "SERVER default_local_regex_parsed WITH (LINE_REGEX = '[', "
                      "file_path = '../../Tests/FsiDataFiles/0.csv');";
  queryAndAssertException(
      query,
      "Error parsing LINE_REGEX \"[\": Character set declaration starting with [ "
      "terminated prematurely - either no ] was found or the set had no content.  The "
      "error occurred while parsing the regular expression: '[>>>HERE>>>'.");
}

TEST_F(CreateForeignTableTest, RegexParserWrapperWithEmptyLineStartRegex) {
  std::string query = "CREATE FOREIGN TABLE test_foreign_table (t TEXT) "s +
                      "SERVER default_local_regex_parsed WITH (LINE_REGEX = '(.*)', "
                      "LINE_START_REGEX = '', "
                      "file_path = '../../Tests/FsiDataFiles/0.csv');";
  queryAndAssertException(query,
                          "Foreign table option \"LINE_START_REGEX\", when set, must "
                          "contain a non-empty value.");
}

TEST_F(CreateForeignTableTest, RegexParserWrapperWithInvalidLineStartRegex) {
  std::string query = "CREATE FOREIGN TABLE test_foreign_table (t TEXT) "s +
                      "SERVER default_local_regex_parsed WITH (LINE_REGEX = '(.*)', "
                      "LINE_START_REGEX = '[', "
                      "file_path = '../../Tests/FsiDataFiles/0.csv');";
  queryAndAssertException(
      query,
      "Error parsing LINE_START_REGEX \"[\": Character set declaration starting with [ "
      "terminated prematurely - either no ] was found or the set had no content.  The "
      "error occurred while parsing the regular expression: '[>>>HERE>>>'.");
}

TEST_F(CreateForeignTableTest, NonAppendModeWithFileRollOff) {
  std::string query = "CREATE FOREIGN TABLE test_foreign_table (t TEXT) "s +
                      "SERVER default_local_regex_parsed WITH ("
                      "ALLOW_FILE_ROLL_OFF = 'true', "
                      "file_path = '../../Tests/FsiDataFiles/0.csv');";
  queryAndAssertException(query,
                          "The \"ALLOW_FILE_ROLL_OFF\" option can only be set to 'true' "
                          "for foreign tables with append refresh updates.");
}

TEST_F(CreateForeignTableTest, LargeRefreshDateTime) {
  sql("CREATE FOREIGN TABLE test_foreign_table(col1 INTEGER) "
      "SERVER default_local_delimited WITH (file_path = '" +
      getTestFilePath() +
      "', refresh_interval='1H', refresh_start_date_time='3000-01-01T10:10:10Z', "
      "refresh_timing_type='SCHEDULED', refresh_update_type='APPEND');");
  resetCatalog();
  // Login again to force a catalog reload.
  loginAdmin();

  auto td = getCatalog().getMetadataForTable("test_foreign_table", false);
  ASSERT_NE(td, nullptr);

  auto foreign_table = dynamic_cast<const foreign_storage::ForeignTable*>(td);
  ASSERT_NE(foreign_table, nullptr);

  constexpr int64_t next_refresh_time_epoch{32503716610};
  ASSERT_EQ(foreign_table->next_refresh_time, next_refresh_time_epoch);

  auto stored_foreign_table =
      getCatalog().getForeignTableFromStorage(foreign_table->tableId);
  ASSERT_EQ(foreign_table->options, stored_foreign_table->options);
  ASSERT_EQ(foreign_table->next_refresh_time, stored_foreign_table->next_refresh_time);
}

#ifdef EE_FSI_ODBC
TEST_F(CreateForeignTableTest, ODBCAppend) {
  sql("CREATE SERVER test_server FOREIGN DATA WRAPPER odbc "
      "WITH (DATA_SOURCE_NAME = 'dsn_name');");
  sql("CREATE FOREIGN TABLE test_foreign_table (i INTEGER) SERVER test_server WITH "
      "(sql_select = 'SELECT * FROM bar', sql_order_by = 'some_column;', "
      "REFRESH_UPDATE_TYPE = 'APPEND');");
  assertOptionEquals("REFRESH_UPDATE_TYPE", "APPEND");
  assertOptionEquals("SQL_SELECT", "SELECT * FROM bar");
  assertOptionEquals("SQL_ORDER_BY", "some_column;");
}

TEST_F(CreateForeignTableTest, ODBCNoSql) {
  sql("CREATE SERVER test_server FOREIGN DATA WRAPPER odbc "
      "WITH (DATA_SOURCE_NAME = 'dsn_name');");
  queryAndAssertException(
      "CREATE FOREIGN TABLE test_foreign_table (i INTEGER) SERVER test_server;",
      "Foreign table options must contain a value for \"SQL_SELECT\".");
}

TEST_F(CreateForeignTableTest, ODBCWithBufferSizeOption) {
  sql("CREATE SERVER test_server FOREIGN DATA WRAPPER odbc "
      "WITH (DATA_SOURCE_NAME = 'dsn_name');");
  sql("CREATE FOREIGN TABLE test_foreign_table (i INTEGER) SERVER test_server WITH "
      "(sql_select = 'SELECT * from bar', sql_order_by = 'some_column', "
      "buffer_size = '4');");
  assertOptionEquals("BUFFER_SIZE", "4");
}

TEST_F(CreateForeignTableTest, ODBCWithInvalidBufferSizeOption) {
  sql("CREATE SERVER test_server FOREIGN DATA WRAPPER odbc "
      "WITH (DATA_SOURCE_NAME = 'dsn_name');");
  queryAndAssertException(
      "CREATE FOREIGN TABLE test_foreign_table (i INTEGER) SERVER test_server WITH "
      "(sql_select = 'SELECT * from bar', sql_order_by = 'some_column', "
      "buffer_size = '-1');",
      "Can not parse string '-1' into a positive integer while validating "
      "option 'BUFFER_SIZE'");
}

#endif
#if defined(HAVE_AWS_S3)

class CreateS3ForeignTableTest : public CreateForeignTableTest {
  void SetUp() override {
    CreateForeignTableTest::SetUp();
    g_enable_s3_fsi = true;
    sql("CREATE SERVER test_server FOREIGN DATA WRAPPER delimited_file WITH "
        "(storage_type = "
        "'AWS_S3', s3_bucket = 'test_bucket', aws_region = 'test_region');");
    sql("CREATE SERVER test_server_ce FOREIGN DATA WRAPPER delimited_file WITH "
        "(storage_type = "
        "'AWS_S3', s3_bucket = 'test_bucket', aws_region = 'test_region', "
        "s3_endpoint='test_endpoint');");
  }

  void TearDown() override { CreateForeignTableTest::TearDown(); }
};

TEST_F(CreateS3ForeignTableTest, InvalidS3AccessType) {
  std::string query = "CREATE FOREIGN TABLE test_foreign_table (t TEXT) "s +
                      "SERVER test_server WITH (S3_ACCESS_TYPE = 'Invalid_Type', "
                      "file_path = '../../Tests/FsiDataFiles/0.csv');";
  queryAndAssertException(
      query,
      "Invalid value provided for the \"S3_ACCESS_TYPE\" option. Value must "
      "be one of the following: S3_DIRECT, S3_SELECT.");
}

TEST_F(CreateS3ForeignTableTest, S3AccessTypeDirect) {
  std::string query = "CREATE FOREIGN TABLE test_foreign_table (t TEXT) "s +
                      "SERVER test_server WITH (S3_ACCESS_TYPE = 'S3_DIRECT', "
                      "file_path = '../../Tests/FsiDataFiles/0.csv');";
  sql(query);
}

TEST_F(CreateS3ForeignTableTest, S3AccessTypeSelect) {
  std::string query = "CREATE FOREIGN TABLE test_foreign_table (t TEXT) "s +
                      "SERVER test_server WITH (S3_ACCESS_TYPE = 'S3_SELECT', "
                      "file_path = '../../Tests/FsiDataFiles/0.csv');";
  sql(query);
}

TEST_F(CreateS3ForeignTableTest, S3SelectAndFileRollOff) {
  std::string query = "CREATE FOREIGN TABLE test_foreign_table (t TEXT) "s +
                      "SERVER test_server WITH (S3_ACCESS_TYPE = 'S3_SELECT', "
                      "REFRESH_UPDATE_TYPE = 'append', ALLOW_FILE_ROLL_OFF = 'true', "
                      "file_path = '../../Tests/FsiDataFiles/0.csv');";
  queryAndAssertException(
      query, "File roll off is not currently supported with S3 select access type.");
}

TEST_F(CreateS3ForeignTableTest, S3SelectAndCustomEndpoint) {
  std::string query = "CREATE FOREIGN TABLE test_foreign_table (t TEXT) "s +
                      "SERVER test_server_ce WITH (S3_ACCESS_TYPE = 'S3_SELECT', "
                      "file_path = '../../Tests/FsiDataFiles/0.csv');";
  queryAndAssertException(query,
                          "S3_SELECT is not supported when using custom s3 endpoints.");
}

class S3SelectDisallowedTypeTest : public CreateS3ForeignTableTest,
                                   public testing::WithParamInterface<std::string> {};

TEST_P(S3SelectDisallowedTypeTest, S3Select) {
  std::string query = "CREATE FOREIGN TABLE test_foreign_table (g " + GetParam() + ") "s +
                      "SERVER test_server WITH (S3_ACCESS_TYPE = 'S3_SELECT', "
                      "file_path = '../../Tests/FsiDataFiles/0.csv');";
  queryAndAssertException(
      query,
      "Geo Types and Array Types not currently supported with S3_ACCESS_TYPE "
      ": S3_SELECT. Please use S3_ACCESS_TYPE : S3_DIRECT");
}

INSTANTIATE_TEST_SUITE_P(DisallowedTypes,
                         S3SelectDisallowedTypeTest,
                         ::testing::Values("POINT",
                                           "LINESTRING",
                                           "POLYGON",
                                           "MULTIPOLYGON",
                                           "INT []",
                                           "INT [4]"));

#endif  // defined(HAVE_AWS_S3)

class CreateTableInThrift : public DBHandlerTestFixture {
 protected:
  void SetUp() override {
    DBHandlerTestFixture::SetUp();
    sql("DROP TABLE IF EXISTS test_table");
  }

  void TearDown() override {
    sql("DROP TABLE IF EXISTS test_table");
    DBHandlerTestFixture::TearDown();
  }
};

TEST_F(CreateTableInThrift, ThriftCreateTableWithDefaults) {
  auto [handler, session] = getDbHandlerAndSessionId();
  TColumnType c1, c2, c3;
  c1.col_id = 1;
  c1.col_name = "idx";
  c1.col_type.type = TDatumType::INT;
  c2.col_id = 2;
  c2.col_name = "s";
  c2.col_type.type = TDatumType::STR;
  c2.__set_default_value("'default str'");
  c3.col_id = 3;
  c3.col_name = "sa";
  c3.col_type.is_array = true;
  c3.col_type.type = TDatumType::STR;
  c3.col_type.encoding = TEncodingType::DICT;
  c3.__set_default_value("ARRAY['a', 'b', 'c']");
  TRowDescriptor row_desc = {c1, c2, c3};
  handler->create_table(session, "test_table", row_desc);
  TTableDetails details;
  handler->get_table_details(details, session, "test_table");
  auto created_row_desc = details.row_desc;
  EXPECT_EQ(created_row_desc.size(), row_desc.size());
  for (size_t i = 0; i < row_desc.size(); ++i) {
    EXPECT_EQ(created_row_desc[i].col_name, row_desc[i].col_name);
    EXPECT_EQ(created_row_desc[i].default_value, row_desc[i].default_value);
    EXPECT_EQ(created_row_desc[i].col_type.is_array, row_desc[i].col_type.is_array);
    EXPECT_EQ(created_row_desc[i].col_type.type, row_desc[i].col_type.type);
  }
}

TEST_F(CreateTableInThrift, ThriftCreateTableDifferentEncodings) {
  struct ColumnDef {
    TDatumType::type type;
    TEncodingType::type encoding;
    int comp_param;
  };
  const vector<ColumnDef> columns = {{TDatumType::TINYINT, TEncodingType::NONE, 0},
                                     {TDatumType::SMALLINT, TEncodingType::NONE, 0},
                                     {TDatumType::SMALLINT, TEncodingType::FIXED, 8},
                                     {TDatumType::INT, TEncodingType::NONE, 0},
                                     {TDatumType::INT, TEncodingType::FIXED, 8},
                                     {TDatumType::INT, TEncodingType::FIXED, 16},
                                     {TDatumType::BIGINT, TEncodingType::NONE, 0},
                                     {TDatumType::BIGINT, TEncodingType::FIXED, 8},
                                     {TDatumType::BIGINT, TEncodingType::FIXED, 16},
                                     {TDatumType::BIGINT, TEncodingType::FIXED, 32},
                                     {TDatumType::FLOAT, TEncodingType::NONE, 0},
                                     {TDatumType::DOUBLE, TEncodingType::NONE, 0},
                                     {TDatumType::DECIMAL, TEncodingType::NONE, 0},
                                     {TDatumType::STR, TEncodingType::NONE, 0},
                                     {TDatumType::STR, TEncodingType::DICT, 8},
                                     {TDatumType::STR, TEncodingType::DICT, 16},
                                     {TDatumType::STR, TEncodingType::DICT, 32},
                                     {TDatumType::TIME, TEncodingType::NONE, 0},
                                     {TDatumType::TIME, TEncodingType::FIXED, 32},
                                     {TDatumType::TIMESTAMP, TEncodingType::NONE, 0},
                                     {TDatumType::TIMESTAMP, TEncodingType::FIXED, 32},
                                     {TDatumType::DATE, TEncodingType::DATE_IN_DAYS, 16},
                                     {TDatumType::DATE, TEncodingType::DATE_IN_DAYS, 32},
                                     {TDatumType::BOOL, TEncodingType::NONE, 0},
                                     {TDatumType::POINT, TEncodingType::NONE, 0},
                                     {TDatumType::LINESTRING, TEncodingType::NONE, 0},
                                     {TDatumType::POLYGON, TEncodingType::NONE, 0},
                                     {TDatumType::MULTIPOLYGON, TEncodingType::NONE, 0}};
  TRowDescriptor row_desc;
  for (size_t i = 0; i < columns.size(); ++i) {
    TColumnType c;
    c.col_name = "c" + to_string(i);
    c.col_type.type = columns[i].type;
    c.col_type.encoding = columns[i].encoding;
    c.col_type.comp_param = columns[i].comp_param;
    if (c.col_type.type == TDatumType::POINT ||
        c.col_type.type == TDatumType::LINESTRING ||
        c.col_type.type == TDatumType::POLYGON ||
        c.col_type.type == TDatumType::MULTIPOLYGON) {
      c.col_type.precision = static_cast<int>(SQLTypes::kGEOGRAPHY);
    }
    row_desc.push_back(move(c));
  }
  auto [handler, session] = getDbHandlerAndSessionId();
  handler->create_table(session, "test_table", row_desc);
  TTableDetails details;
  handler->get_table_details(details, session, "test_table");
  auto created_row_desc = details.row_desc;
  EXPECT_EQ(created_row_desc.size(), row_desc.size());
  for (size_t i = 0; i < row_desc.size(); ++i) {
    EXPECT_EQ(created_row_desc[i].col_name, row_desc[i].col_name);
    EXPECT_EQ(created_row_desc[i].col_type.type, row_desc[i].col_type.type);
    EXPECT_EQ(created_row_desc[i].col_type.encoding, row_desc[i].col_type.encoding);
    EXPECT_EQ(created_row_desc[i].col_type.comp_param, row_desc[i].col_type.comp_param);
  }
}

class DropTableTest : public CreateAndDropTableDdlTest,
                      public testing::WithParamInterface<ddl_utils::TableType> {
 protected:
  void SetUp() override {
    CreateAndDropTableDdlTest::SetUp();
    sql(getCreateTableQuery(GetParam(), "test_table", "(col1 INTEGER)", true));
    dropTestUser();
  }

  void TearDown() override {
    dropTestUser();
    CreateAndDropTableDdlTest::TearDown();
  }
};

TEST_P(DropTableTest, ExistingTable) {
  sql(getDropTableQuery(GetParam(), "test_table"));
  ASSERT_EQ(nullptr, getCatalog().getMetadataForTable("test_table", false));
}

TEST_P(DropTableTest, NonExistingTableWithIfExists) {
  sql(getDropTableQuery(GetParam(), "test_table"));
  sql(getDropTableQuery(GetParam(), "test_table", true));

  ASSERT_EQ(nullptr, getCatalog().getMetadataForTable("test_table", false));
}

TEST_P(DropTableTest, NonExistentTableWithoutIfExists) {
  std::string query = getDropTableQuery(GetParam(), "test_table_2");
  queryAndAssertException(query,
                          "Table/View test_table_2 for catalog " +
                              shared::kDefaultDbName + " does not exist.");
}

TEST_P(DropTableTest, UnauthorizedUser) {
  createTestUser();
  login("test_user", "test_pass");
  std::string query = getDropTableQuery(GetParam(), "test_table");

  if (GetParam() == ddl_utils::TableType::FOREIGN_TABLE) {
    queryAndAssertException(query,
                            "Foreign table \"test_table\" will not be "
                            "dropped. User has no DROP TABLE privileges.");
  } else {
    queryAndAssertException(query,
                            "Table test_table will not be dropped. User has "
                            "no proper privileges.");
  }

  loginAdmin();
  sql(getDropTableQuery(GetParam(), "test_table"));
}

TEST_P(DropTableTest, AuthorizedUser) {
  createTestUser();
  sql("GRANT DROP ON TABLE test_table TO test_user;");
  login("test_user", "test_pass");

  sql(getDropTableQuery(GetParam(), "test_table"));

  ASSERT_EQ(nullptr, getCatalog().getMetadataForTable("test_table", false));
}

TEST_P(CreateTableTest, InvalidSyntax) {
  std::string query = getCreateTableQuery(
      GetParam(), "test_table", "(str TEXT ENCODING DICT(8), f FLOAT i INTEGER)");
  try {
    sql(query);
    FAIL() << "An exception should have been thrown for this test case.";
  } catch (const TDBException& e) {
    if (GetParam() == ddl_utils::TableType::FOREIGN_TABLE) {
      ASSERT_TRUE(e.error_msg.find("SQL Error") != std::string::npos);
    }
  }
}

TEST_P(CreateTableTest, InvalidColumnDefinition) {
  std::string query =
      getCreateTableQuery(GetParam(), "test_table", "(str TEXT ENCODING DICT(8), test)");
  try {
    sql(query);
    FAIL() << "An exception should have been thrown for this test case.";
  } catch (const TDBException& e) {
    if (GetParam() == ddl_utils::TableType::TABLE) {
      ASSERT_TRUE(e.error_msg.find("Column definition for table test_table") !=
                  std::string::npos);
    }
  }
}

TEST_P(CreateTableTest, RealAlias) {
  // TODO(vancouver: support for FSI)
  if (GetParam() == ddl_utils::TableType::FOREIGN_TABLE) {
    LOG(ERROR) << "REAL alias not yet supported for FSI.";
    return;
  }

  std::string query = getCreateTableQuery(GetParam(), "test_table", "(f REAL)");
  sql(query);
  auto td = getCatalog().getMetadataForTable("test_table", false);
  ASSERT_TRUE(td);
  auto cd = getCatalog().getMetadataForColumn(td->tableId, "f");
  ASSERT_TRUE(cd);
  ASSERT_EQ(cd->columnType.get_type(), kFLOAT);
}

TEST_P(CreateTableTest, CreateOrReplaceTable) {
  auto query =
      getCreateTableQuery(GetParam(), "test_table", "(idx INTEGER)", false, true);
  // using a partial exception for the sake of brevity
  if (GetParam() == ddl_utils::TableType::FOREIGN_TABLE) {
    queryAndAssertPartialException(
        query,
        R"(SQL Error: Encountered "FOREIGN" at line 1, column 19.
Was expecting:
    "MODEL" ...)");
  } else {
    queryAndAssertPartialException(query,
                                   R"(SQL Error: Encountered "TABLE" at line 1, column 19.
Was expecting:
    "MODEL" ...)");
  }
}

TEST_P(CreateTableTest, CreateOrReplaceTemporaryTable) {
  if (GetParam() == ddl_utils::TableType::FOREIGN_TABLE) {
    GTEST_SKIP() << "Foreign tables can't be temporary.";
  }
  auto query =
      getCreateTableQuery(GetParam(), "test_table", "(idx INTEGER)", false, true, true);
  // using a partial exception for the sake of brevity
  queryAndAssertPartialException(
      query,
      R"(SQL Error: Encountered "TEMPORARY" at line 1, column 19.
Was expecting:
    "MODEL" ...)");
}

TEST_P(CreateTableTest, CreateOrReplaceView) {
  sql(getCreateTableQuery(GetParam(), "test_table", "(idx INTEGER)"));
  auto query =
      std::string("CREATE OR REPLACE VIEW test_view AS SELECT * FROM test_table");
  // using a partial exception for the sake of brevity
  queryAndAssertPartialException(query,
                                 R"(SQL Error: Encountered "VIEW" at line 1, column 19.
Was expecting:
    "MODEL" ...)");
}

INSTANTIATE_TEST_SUITE_P(CreateAndDropTableDdlTest,
                         DropTableTest,
                         testing::Values(ddl_utils::TableType::TABLE,
                                         ddl_utils::TableType::FOREIGN_TABLE),
                         [](const auto& param_info) {
                           if (param_info.param == ddl_utils::TableType::TABLE) {
                             return "Table";
                           }
                           if (param_info.param == ddl_utils::TableType::FOREIGN_TABLE) {
                             return "ForeignTable";
                           }
                           throw std::runtime_error{"Unexpected parameter type"};
                         });

class DropTableTypeMismatchTest : public CreateAndDropTableDdlTest {};

TEST_F(DropTableTypeMismatchTest, Table_DropCommandForOtherTableTypes) {
  sql(getCreateTableQuery(ddl_utils::TableType::TABLE, "test_table", "(col1 INTEGER)"));

  queryAndAssertException("DROP VIEW test_table;",
                          "test_table is a table. Use DROP TABLE.");
  queryAndAssertException("DROP FOREIGN TABLE test_table;",
                          "test_table is a table. Use DROP TABLE.");

  sql("DROP table test_table;");
  ASSERT_EQ(nullptr, getCatalog().getMetadataForTable("test_table", false));
}

TEST_F(DropTableTypeMismatchTest, View_DropCommandForOtherTableTypes) {
  sql(getCreateTableQuery(ddl_utils::TableType::TABLE, "test_table", "(col1 INTEGER)"));
  sql("CREATE VIEW test_view AS SELECT * FROM test_table;");

  queryAndAssertException("DROP table test_view;", "test_view is a view. Use DROP VIEW.");
  queryAndAssertException("DROP FOREIGN TABLE test_view;",
                          "test_view is a view. Use DROP VIEW.");

  sql("DROP VIEW test_view;");
  ASSERT_EQ(nullptr, getCatalog().getMetadataForTable("test_view", false));
  sql("DROP TABLE test_table");
}

TEST_F(DropTableTypeMismatchTest, ForeignTable_DropCommandForOtherTableTypes) {
  sql(getCreateTableQuery(
      ddl_utils::TableType::FOREIGN_TABLE, "test_foreign_table", "(col1 INTEGER)"));

  queryAndAssertException(
      "DROP table test_foreign_table;",
      "test_foreign_table is a foreign table. Use DROP FOREIGN TABLE.");
  queryAndAssertException(
      "DROP VIEW test_foreign_table;",
      "test_foreign_table is a foreign table. Use DROP FOREIGN TABLE.");

  sql("DROP FOREIGN TABLE test_foreign_table;");
  ASSERT_EQ(nullptr, getCatalog().getMetadataForTable("test_foreign_table", false));
}

class DropForeignTableTest : public CreateAndDropTableDdlTest {
 protected:
  void SetUp() override {
    CreateAndDropTableDdlTest::SetUp();
    sql("DROP FOREIGN TABLE IF EXISTS test_foreign_table;");
  }

  void TearDown() override {
    g_enable_fsi = true;
    sql("DROP FOREIGN TABLE IF EXISTS test_foreign_table;");
    CreateAndDropTableDdlTest::TearDown();
  }
};

TEST_F(DropForeignTableTest, FsiDisabled) {
  sql(getCreateTableQuery(
      ddl_utils::TableType::FOREIGN_TABLE, "test_foreign_table", "(col1 INTEGER)"));

  g_enable_fsi = false;
  queryAndAssertException(
      "DROP table test_foreign_table;",
      "test_foreign_table is a foreign table. Use DROP FOREIGN TABLE.");
}

class CreateViewUnsupportedTest : public CreateAndDropTableDdlTest {
 protected:
  void SetUp() override {
    CreateAndDropTableDdlTest::SetUp();
    sql(getDropTableQuery(ddl_utils::TableType::TABLE, "test_table", true));
    dropTestUser();
  }

  void TearDown() override {
    g_enable_fsi = true;
    sql(getDropTableQuery(ddl_utils::TableType::TABLE, "test_table", true));
    dropTestUser();
    CreateAndDropTableDdlTest::TearDown();
  }
};

TEST_F(CreateViewUnsupportedTest, Basics) {
  sql(getCreateTableQuery(
      ddl_utils::TableType::TABLE, "test_table", "(col1 INTEGER, col2 FLOAT)"));
  queryAndAssertException(
      "CREATE VIEW showcreateviewtest (c1, c2) AS SELECT * FROM showcreatetabletest;",
      "SQL Error: Column list aliases in views are not yet supported.");
}

class CreateNonReservedKeywordsTest : public CreateAndDropTableDdlTest {
 protected:
  void SetUp() override {
    CreateAndDropTableDdlTest::SetUp();
    sql(getDropTableQuery(ddl_utils::TableType::TABLE, "test_table", true));
    dropTestUser();
  }

  void TearDown() override {
    g_enable_fsi = true;
    sql(getDropTableQuery(ddl_utils::TableType::TABLE, "test_table", true));
    dropTestUser();
    CreateAndDropTableDdlTest::TearDown();
  }
};

TEST_F(CreateNonReservedKeywordsTest, NonReservedKeywords) {
  sql(getCreateTableQuery(
      ddl_utils::TableType::TABLE,
      "test_table",
      R"((QUERY text, QUERIES text, SESSIONS int[], TABLES text[], databases DECIMAL(6, 2), servers int, mapping int, owner float, rename double, disk point, cache polygon, stored date))"));
  sql("SELECT query FROM test_table LIMIT 1");
}

TEST_F(CreateNonReservedKeywordsTest, NonReservedAndReserved) {
  EXPECT_ANY_THROW(sql(getCreateTableQuery(
      ddl_utils::TableType::TABLE,
      "test_table",
      R"((POLYGON INT, QUERY text, QUERIES text, SESSIONS int[], TABLES text[], databases DECIMAL(6, 2), servers int, mapping int, owner float, rename double, disk point, cache polygon))")));
}

TEST_F(CreateNonReservedKeywordsTest, NonReservedRename) {
  sql(getCreateTableQuery(ddl_utils::TableType::TABLE, "test_table", R"((QUERY text))"));
  sql("ALTER TABLE test_table RENAME COLUMN QUERY to virtual;");
}

class RenameTableTest : public CreateAndDropTableDdlTest {
 protected:
  void SetUp() override {
    CreateAndDropTableDdlTest::SetUp();

    sql("DROP VIEW IF EXISTS AVIEW");
    sql("DROP VIEW IF EXISTS BVIEW");
    sql(getDropTableQuery(ddl_utils::TableType::TABLE, "A", true));
    sql(getDropTableQuery(ddl_utils::TableType::TABLE, "B", true));
    sql(getDropTableQuery(ddl_utils::TableType::TABLE, "C", true));
    sql(getDropTableQuery(ddl_utils::TableType::TABLE, "D", true));
    sql(getDropTableQuery(ddl_utils::TableType::TABLE, "E", true));
    sql(getDropTableQuery(ddl_utils::TableType::TABLE, "F", true));
    sql(getDropTableQuery(ddl_utils::TableType::TABLE, "G", true));
    sql(getDropTableQuery(ddl_utils::TableType::TABLE, "H", true));
    dropTestUser();

    sql("CREATE TABLE A (Name text);");
    sql("INSERT INTO  A (Name) values ('A');");

    sql("CREATE TABLE B (Name text);");
    sql("INSERT INTO  B (Name) values ('B');");

    sql("CREATE TABLE C (Name text);");
    sql("INSERT INTO  C (Name) values ('C');");

    sql("CREATE TABLE D (Name text);");
    sql("INSERT INTO  D (Name) values ('D');");

    sql("CREATE TABLE E (Name text);");
    sql("INSERT INTO  E (Name) values ('E');");

    // test rename VIEW which uses same machinery
    sql("CREATE VIEW AVIEW as SELECT * FROM A;");
    sql("CREATE VIEW BVIEW as SELECT * FROM B;");
  }

  void TearDown() override {
    g_enable_fsi = true;
    sql("DROP VIEW IF EXISTS AVIEW");
    sql("DROP VIEW IF EXISTS BVIEW");
    sql(getDropTableQuery(ddl_utils::TableType::TABLE, "A", true));
    sql(getDropTableQuery(ddl_utils::TableType::TABLE, "B", true));
    sql(getDropTableQuery(ddl_utils::TableType::TABLE, "C", true));
    sql(getDropTableQuery(ddl_utils::TableType::TABLE, "D", true));
    sql(getDropTableQuery(ddl_utils::TableType::TABLE, "E", true));

    dropTestUser();

    CreateAndDropTableDdlTest::TearDown();
  }
};

TEST_F(RenameTableTest, SimpleRename) {
  // rename
  sql("RENAME TABLE A to F;");
  sql("RENAME TABLE B to G, C to H;");

  // verify
  sqlAndCompareResult("SELECT count(*) FROM F WHERE Name = 'A';", {{i(1)}});
  sqlAndCompareResult("SELECT count(*) FROM G WHERE Name = 'B';", {{i(1)}});
  sqlAndCompareResult("SELECT count(*) FROM H WHERE Name = 'C';", {{i(1)}});

  // reset
  sql("RENAME TABLE F TO A, G TO B, H TO C;");
}

TEST_F(RenameTableTest, RenameCalciteError) {
  // Inject a calcite conenction issue so that updating calcite metadata will throw.
  // TODO(Misiu): Replace this with proper Catalog unit tests.
  auto& cat = getCatalog();
  auto old_calcite = cat.getCalciteMgr();
  cat.setCalciteMgr(std::make_shared<MockCalcite>());
  ScopeGuard scope = [&] { cat.setCalciteMgr(old_calcite); };

  queryAndAssertException("RENAME TABLE A to F;", "UpdateMetadataError");
  sqlAndCompareResult("SELECT count(*) FROM A WHERE Name = 'A';", {{i(1)}});
  queryAndAssertException(
      "SELECT count(*) from F;",
      "SQL Error: From line 2, column 6 to line 2, column 8: Object 'F' not found");
}

TEST_F(RenameTableTest, SwapRename) {
  // swap
  sql("RENAME TABLE D to E, E to D");

  // verify
  sqlAndCompareResult("SELECT count(*) FROM D WHERE Name = 'E';", {{i(1)}});
  sqlAndCompareResult("SELECT count(*) FROM E WHERE Name = 'D';", {{i(1)}});

  // rotate
  sql("RENAME TABLE A TO B, B TO C, C TO A");

  // verify
  sqlAndCompareResult("SELECT count(*) FROM A WHERE Name = 'C';", {{i(1)}});
  sqlAndCompareResult("SELECT count(*) FROM B WHERE Name = 'A';", {{i(1)}});
  sqlAndCompareResult("SELECT count(*) FROM C WHERE Name = 'B';", {{i(1)}});

  // reset
  sql("RENAME TABLE D TO E, E TO D;");
  sql("RENAME TABLE C TO B, B TO A, A TO C;");
}

TEST_F(RenameTableTest, SwapRenameRequiresPrivilegesOnEveryTable) {
  createTestUser();
  sql("GRANT ALTER ON TABLE A TO test_user;");

  login("test_user", "test_pass");
  queryAndAssertException("RENAME TABLE A TO B, B TO A;",
                          "Current user does not have the privilege to alter table: B");

  loginAdmin();
}

TEST_F(RenameTableTest, ErrorChecks) {
  // ERROR, expect no change, as would trigger an overwrite of B

  executeLambdaAndAssertException(
      [] { sql("RENAME TABLE A to B"); },
      "Error: Attempted to overwrite and lose data in table: 'B'");

  // verify no change
  sqlAndCompareResult("SELECT count(*) FROM A WHERE Name = 'A';", {{i(1)}});
  sqlAndCompareResult("SELECT count(*) FROM B WHERE Name = 'B';", {{i(1)}});

  // ERROR, expect no change, as would trigger an overwrite
  executeLambdaAndAssertException(
      [] { sql("RENAME TABLE A to C, B TO C"); },
      "Error: Attempted to overwrite and lose data in table: 'C'");

  // verify no name change
  sqlAndCompareResult("SELECT count(*) FROM A WHERE Name = 'A';", {{i(1)}});
  sqlAndCompareResult("SELECT count(*) FROM B WHERE Name = 'B';", {{i(1)}});

  // ERROR, expect no change, table Z is non-existant
  executeLambdaAndAssertException([] { sql("RENAME TABLE Z TO A"); },
                                  "Source table 'Z' does not exist.");

  // verify
  sqlAndCompareResult("SELECT count(*) FROM A WHERE Name = 'A';", {{i(1)}});
}

TEST_F(RenameTableTest, RenameViews) {
  sql("RENAME TABLE AVIEW to XVIEW;");
  sql("ALTER TABLE BVIEW RENAME TO ZVIEW;");

  sqlAndCompareResult("SELECT count(*) FROM XVIEW WHERE Name = 'A';", {{i(1)}});
  sqlAndCompareResult("SELECT count(*) FROM ZVIEW WHERE Name = 'B';", {{i(1)}});

  // reset
  sql("RENAME TABLE XVIEW to AVIEW;");
  sql("ALTER TABLE ZVIEW RENAME TO BVIEW;");
}

class CreateShardedTableTest : public CreateAndDropTableDdlTest {
  void SetUp() override {
    CreateAndDropTableDdlTest::SetUp();
    sql("DROP TABLE IF EXISTS test_table;");
  }

  void TearDown() override {
    sql("DROP TABLE IF EXISTS test_table;");
    CreateAndDropTableDdlTest::TearDown();
  }
};

TEST_F(CreateShardedTableTest, ShardedTableName) {
  sql("CREATE TABLE test_table (i INT, SHARD KEY(i)) WITH "
      "(shard_count = 2);");

  auto& catalog = getCatalog();
  auto logical_table = catalog.getMetadataForTable("test_table", false);
  ASSERT_NE(logical_table, nullptr);
  ASSERT_EQ(2, logical_table->nShards);
  ASSERT_EQ(-1, logical_table->shard);
  ASSERT_EQ("test_table", logical_table->tableName);

  auto physical_tables = catalog.getPhysicalTablesDescriptors(logical_table);
  ASSERT_EQ(static_cast<size_t>(2), physical_tables.size());

  ASSERT_NE(physical_tables[0], nullptr);
  ASSERT_EQ(2, physical_tables[0]->nShards);
  ASSERT_EQ(0, physical_tables[0]->shard);
  ASSERT_EQ("test_table_shard_#1", physical_tables[0]->tableName);

  ASSERT_NE(physical_tables[1], nullptr);
  ASSERT_EQ(2, physical_tables[1]->nShards);
  ASSERT_EQ(1, physical_tables[1]->shard);
  ASSERT_EQ("test_table_shard_#2", physical_tables[1]->tableName);
}

class MaxRollbackEpochsTest : public CreateAndDropTableDdlTest {
 protected:
  void SetUp() override {
    CreateAndDropTableDdlTest::SetUp();
    sql("DROP TABLE IF EXISTS test_table;");
  }

  void TearDown() override {
    sql("DROP TABLE IF EXISTS test_table;");
    CreateAndDropTableDdlTest::TearDown();
  }

  void assertEpochCeilingAndFloor(int64_t epoch_ceiling, int64_t epoch_floor) {
    TQueryResult result;
    sql(result, "show table details test_table;");

    ASSERT_EQ("max_epoch", result.row_set.row_desc[9].col_name);
    ASSERT_EQ(epoch_ceiling, result.row_set.columns[9].data.int_col[0]);

    ASSERT_EQ("min_epoch_floor", result.row_set.row_desc[10].col_name);
    ASSERT_EQ(epoch_floor, result.row_set.columns[10].data.int_col[0]);
  }
};

TEST_F(MaxRollbackEpochsTest, CreateTableDefaultValue) {
  sql("CREATE TABLE test_table (col1 INTEGER);");

  auto& catalog = getCatalog();
  auto table = catalog.getMetadataForTable("test_table", false);
  ASSERT_EQ(table->maxRollbackEpochs, DEFAULT_MAX_ROLLBACK_EPOCHS);

  // Sanity test to ensure that default max_rollback_epoch is in effect
  for (int i = 0; i < DEFAULT_MAX_ROLLBACK_EPOCHS; i++) {
    sql("INSERT INTO test_table VALUES (10);");
  }
  assertEpochCeilingAndFloor(DEFAULT_MAX_ROLLBACK_EPOCHS, 0);

  sql("INSERT INTO test_table VALUES (10);");
  assertEpochCeilingAndFloor(DEFAULT_MAX_ROLLBACK_EPOCHS + 1, 1);
}

class DefaultValuesTest : public DBHandlerTestFixture {
 protected:
  void SetUp() override {
    DBHandlerTestFixture::SetUp();
    sql("DROP TABLE IF EXISTS defval_tbl");
  }
  void TearDown() override {
    sql("DROP TABLE IF EXISTS defval_tbl");
    DBHandlerTestFixture::TearDown();
  }

  void insertDefaultValues() {
    EXPECT_NO_THROW(sql("INSERT INTO defval_tbl(idx) values(1)"));
  }

  void insertNotDefaultValues() {
    EXPECT_NO_THROW(
        sql("INSERT INTO defval_tbl(idx, i, big_i, null_i, int_a, text_a,"
            "t, dt, ls, p, d) values(2, 15, 314958735, NULL, "
            "ARRAY[4, 5, 6], ARRAY['b', 'c'],"
            "'World', '!!!', 'LINESTRING (2 2,3 3, 4 4)',"
            "'POINT (2 3)', '2012-11-24')"));
  }

  void verifyData() {
    sqlAndCompareResult("SELECT * FROM defval_tbl WHERE idx = 1", default_values);
    sqlAndCompareResult("SELECT * FROM defval_tbl WHERE idx = 2", not_default_values);
  }

  const std::vector<std::vector<NullableTargetValue>> default_values = {
      {i(1),
       i(14),
       i(314958734),
       Null,
       array({i(1), i(2), i(3)}),
       array({"a", "b"}),
       "Hello",
       "World",
       "LINESTRING (1 1,2 2,3 3)",
       "POINT (1 2)",
       "2011-10-23"}};

  const std::vector<std::vector<NullableTargetValue>> not_default_values = {
      {i(2),
       i(15),
       i(314958735),
       Null,
       array({i(4), i(5), i(6)}),
       array({"b", "c"}),
       "World",
       "!!!",
       "LINESTRING (2 2,3 3,4 4)",
       "POINT (2 3)",
       "2012-11-24"}};

  std::string col_defs =
      "i integer default 14,"
      "big_i bigint default 314958734,"
      "null_i integer,"
      "int_a integer[] default ARRAY[1, 2, 3],"
      "text_a text[] default ARRAY['a', 'b'],"
      "t text default 'Hello' encoding none ,"
      "dt text default 'World' encoding dict,"
      "ls LINESTRING default 'LINESTRING (1 1,2 2,3 3)',"
      "p POINT default 'POINT (1 2)',"
      "d date default '2011-10-23'";
};

TEST_F(DefaultValuesTest, CreateTableTest) {
  std::string create_table_query =
      "CREATE TABLE defval_tbl (idx integer, " + col_defs + ")";
  sql(create_table_query);
  insertNotDefaultValues();
  insertDefaultValues();
  verifyData();
}

TEST_F(DefaultValuesTest, AlterAfterInsert) {
  std::string create_table_query = "CREATE TABLE defval_tbl (idx integer)";
  sql(create_table_query);
  insertDefaultValues();
  std::string alter_table_query = "ALTER TABLE defval_tbl ADD " + col_defs;
  sql(alter_table_query);
  insertNotDefaultValues();
  verifyData();
}

TEST_F(DefaultValuesTest, InsertAfterAlter) {
  std::string create_table_query = "CREATE TABLE defval_tbl (idx integer)";
  sql(create_table_query);
  std::string alter_table_query = "ALTER TABLE defval_tbl ADD " + col_defs;
  sql(alter_table_query);
  insertDefaultValues();
  insertNotDefaultValues();
  verifyData();
}

TEST_F(DefaultValuesTest, AllowDefaultOnShardedKey) {
  sql("CREATE TABLE defval_tbl (i INTEGER default -1, key text "
      "default 'default', shard key(i)) with (shard_count = 2)");
  sql("INSERT INTO defval_tbl(i, key) VALUES (1, 'Foo')");
  sql("INSERT INTO defval_tbl(key) VALUES ('Bar')");
  sqlAndCompareResult("SELECT i FROM defval_tbl WHERE key = 'Foo'", {{i(1)}});
  sqlAndCompareResult("SELECT i FROM defval_tbl WHERE key = 'Bar'", {{i(-1)}});
}

TEST_F(DefaultValuesTest, DefaultAllowsNulls) {
  std::vector<std::string> create_sqls = {
      "CREATE TABLE defval_tbl (idx INTEGER, a INTEGER DEFAULT 12)",
      "CREATE TABLE defval_tbl (idx INTEGER, a INTEGER NULL DEFAULT 12)"};
  for (auto& create_sql : create_sqls) {
    sql("DROP TABLE IF EXISTS defval_tbl");
    sql(create_sql);
    sql("INSERT INTO defval_tbl(idx, a) VALUES (1, NULL)");
    sqlAndCompareResult("SELECT idx, a FROM defval_tbl", {{i(1), Null}});
  }
}

TEST_F(DefaultValuesTest, DefaultNotNull) {
  std::string create_table_query =
      "CREATE TABLE defval_tbl (idx INTEGER, a INTEGER NOT NULL DEFAULT 12)";
  sql(create_table_query);
  queryAndAssertException("INSERT INTO defval_tbl(idx, a) VALUES (1, NULL)",
                          "Cannot insert NULL into column a");
  sql("INSERT INTO defval_tbl(idx) VALUES (1)");
  sqlAndCompareResult("SELECT idx, a FROM defval_tbl", {{i(1), i(12)}});
}

TEST_F(DefaultValuesTest, NullDefault) {
  std::string create_table_query =
      "CREATE TABLE defval_tbl (idx INTEGER, a INTEGER DEFAULT NULL)";
  sql(create_table_query);
  sql("INSERT INTO defval_tbl(idx) VALUES (1)");
  sqlAndCompareResult("SELECT * FROM defval_tbl", {{i(1), Null}});
}

TEST_F(DefaultValuesTest, NotNullWithNullDefault) {
  queryAndAssertException(
      "CREATE TABLE defval_tbl (idx INTEGER, "
      "a INTEGER NOT NULL DEFAULT NULL)",
      "a: cannot set default value to NULL for NOT NULL column");
}

struct DefaultValueParams {
  DefaultValueParams(const SQLTypeInfo& t,
                     const std::string& v,
                     const std::string& e,
                     bool n,
                     const std::string& d)
      : type(t), value(v), error_message(e), is_null(n), description(d) {}

  SQLTypeInfo type;
  std::string value;
  std::string error_message;
  bool is_null;
  std::string description;

  friend std::ostream& operator<<(std::ostream& os, const DefaultValueParams& p) {
    return os << "Type: " << p.type.to_string() << ", Value: " << p.value
              << ", is_null: " << p.is_null;
  }
};

SQLTypeInfo decimal = SQLTypeInfo(SQLTypes::kDECIMAL, 4, 2, false);
SQLTypeInfo integer = SQLTypeInfo(SQLTypes::kINT);
SQLTypeInfo integer_notnull = SQLTypeInfo(SQLTypes::kINT, 0, 0, true);
SQLTypeInfo smallint = SQLTypeInfo(SQLTypes::kSMALLINT);
SQLTypeInfo timestamp = SQLTypeInfo(SQLTypes::kTIMESTAMP);
SQLTypeInfo date_in_days_16 =
    SQLTypeInfo(SQLTypes::kDATE, kENCODING_DATE_IN_DAYS, 16, SQLTypes::kNULLT);
SQLTypeInfo dict_encoded_text = SQLTypeInfo(SQLTypes::kTEXT, false, kENCODING_DICT);
SQLTypeInfo linestring_notnull = SQLTypeInfo(SQLTypes::kLINESTRING, 0, 0, true);
SQLTypeInfo text_array =
    SQLTypeInfo(SQLTypes::kARRAY, kENCODING_DICT, 32, SQLTypes::kTEXT);
SQLTypeInfo integer_array =
    SQLTypeInfo(SQLTypes::kARRAY, kENCODING_NONE, 0, SQLTypes::kINT);
auto fixed_integer_array = integer_array;
auto fixed_text_array = text_array;

class ValidateDefaultValuesTest : public testing::Test,
                                  public testing::WithParamInterface<DefaultValueParams> {
};

TEST_P(ValidateDefaultValuesTest, InvalidLiteralTest) {
  ColumnDescriptor cd(0, 1, "c", GetParam().type, 0);
  try {
    const std::string* value = GetParam().is_null ? nullptr : &GetParam().value;
    ddl_utils::validate_and_set_default_value(cd, value, GetParam().type.get_notnull());
    ASSERT_TRUE(false) << "Expected an error: " << GetParam().error_message;
  } catch (const std::exception& e) {
    ASSERT_EQ(e.what(), GetParam().error_message);
  }
}

INSTANTIATE_TEST_SUITE_P(
    ValidateDefaultValuesTest,
    ValidateDefaultValuesTest,
    testing::Values(
        DefaultValueParams(decimal,
                           "{1, 2, 3}",
                           "Invalid conversion from string to NUMERIC(0,0)",
                           false,
                           "ArrayAsDecimalValue"),
        DefaultValueParams(decimal,
                           "Hello",
                           "Invalid conversion from string to NUMERIC(0,0)",
                           false,
                           "TextAsDecimalValue"),
        DefaultValueParams(
            decimal,
            "123.4",
            "Decimal overflow: value is greater than 10^2 max 10000 value 12340",
            false,
            "DecimalOverflow"),
        DefaultValueParams(
            decimal,
            "123",
            "Decimal overflow: value is greater than 10^2 max 10000 value 12300",
            false,
            "DecimalOverflowNoDot"),
        DefaultValueParams(
            decimal,
            "-123.4",
            "Decimal overflow: value is less than -10^2 min -10000 value -12340",
            false,
            "DecimalUnderflow"),
        DefaultValueParams(integer,
                           "Hello",
                           "Unable to parse Hello to INTEGER",
                           false,
                           "TextAsIntegerValue"),
        DefaultValueParams(integer_notnull,
                           "Null",
                           "c: cannot set default value to NULL for NOT NULL column",
                           false,
                           "NullValueForNotNullInteger"),
        DefaultValueParams(smallint,
                           "999999999",
                           "Integer 999999999 is out of range for SMALLINT",
                           false,
                           "SmallintOverflow"),
        DefaultValueParams(smallint,
                           "-999999999",
                           "Integer -999999999 is out of range for SMALLINT",
                           false,
                           "SmallintUnderflow"),
        DefaultValueParams(dict_encoded_text,
                           std::string(StringDictionary::MAX_STRLEN + 1, 'a'),
                           "String too long for column c was 32768 max is 32767",
                           false,
                           "LongDictEncodedString"),
        DefaultValueParams(linestring_notnull,
                           "",
                           "c: cannot set default value to NULL for NOT NULL column",
                           false,
                           "EmptyNotNullGeo"),
        DefaultValueParams(linestring_notnull,
                           "Hello world",
                           "Unexpected geo literal 'Hello world' for column c: "
                           "GeoGeoFactory Error: unsupported geometry type",
                           false,
                           "HelloGeoString"),
        DefaultValueParams(linestring_notnull,
                           "LINESTRING(1)",
                           "Unexpected geo literal 'LINESTRING(1)' for column c: "
                           "GeoGeoFactory Error: corrupt input data",
                           false,
                           "MalformedGeoString"),
        DefaultValueParams(
            linestring_notnull,
            "POINT(1 -1)",
            "Geo literal 'POINT(1 -1)' doesn't match the type of column column c",
            false,
            "GeoTypeMismatch"),
        DefaultValueParams(integer_array,
                           "{1, 2, 3",
                           "c: arrays should start and end with curly braces",
                           false,
                           "ArrayWithoutClosingBrace"),
        DefaultValueParams(integer_array,
                           "{1, 2, 9999999999}",
                           "Integer 9999999999 is out of range for INTEGER",
                           false,
                           "IntegerArrayOverflow"),
        DefaultValueParams(integer_array,
                           "1, 2, 3",
                           "c: arrays should start and end with curly braces",
                           false,
                           "ArrayWithoutBraces"),
        DefaultValueParams(integer_array,
                           "ARRAY[1, 2]",
                           "c: arrays should start and end with curly braces",
                           false,
                           "UnsupportedArrayFormat"),
        DefaultValueParams(integer_array,
                           "12309834",
                           "c: arrays should start and end with curly braces",
                           false,
                           "LiteralInsteadOfIntArray"),
        DefaultValueParams(integer_array,
                           "{1, 2, Three}",
                           "Unable to parse Three to INTEGER",
                           false,
                           "TextInIntegerArray"),
        DefaultValueParams((fixed_integer_array.set_size(
                                fixed_integer_array.get_elem_type().get_size() * 4),
                            fixed_integer_array),
                           "{1, 2, 3}",
                           "Fixed length array column c expects 4 values, received 3",
                           false,
                           "NotEnoughElementsInFixedIntArray"),
        DefaultValueParams(timestamp,
                           "Hello",
                           "Invalid TIMESTAMP string (Hello)",
                           false,
                           "TextInsteadOfTimestamp"),
        DefaultValueParams(text_array,
                           "Hello",
                           "c: arrays should start and end with curly braces",
                           false,
                           "LiteralInsteadOfTextArray"),
        DefaultValueParams(
            (fixed_text_array.set_size(fixed_text_array.get_elem_type().get_size() * 4),
             fixed_text_array),
            "{a, b, c, d, e}",
            "Fixed length array column c expects 4 values, received 5",
            false,
            "TooManyElementsInFixedTextArray")),
    [](const auto& param_info) { return param_info.param.description; });

class CommentsBeforeCommandTest : public DBHandlerTestFixture {
 protected:
  inline static const std::string kCreateTableCommand{
      "CREATE TABLE test_table (i INTEGER);"};
};

TEST_F(CommentsBeforeCommandTest, SingleLineHyphenCommentBeforeCommand) {
  queryAndAssertException(
      " -- test comment\n" + kCreateTableCommand,
      "SQL statements starting with comments are currently not allowed.");
}

TEST_F(CommentsBeforeCommandTest, SingleLineSlashCommentBeforeCommand) {
  queryAndAssertException(
      "//test comment\n" + kCreateTableCommand,
      "SQL statements starting with comments are currently not allowed.");
}

TEST_F(CommentsBeforeCommandTest, MultiLineCommentBeforeCommand) {
  queryAndAssertException(
      "/* test comment 1\n test comment2*/" + kCreateTableCommand,
      "SQL statements starting with comments are currently not allowed.");
}

class TableOptionsValidationTest : public CreateAndDropTableDdlTest {
 protected:
  void TearDown() override {
    sql(getDropTableQuery(ddl_utils::TableType::TABLE, "test_table", true));
    CreateAndDropTableDdlTest::TearDown();
  }

  std::string getCreateTableQuery(
      const std::map<std::string, std::string>& options) const {
    return CreateAndDropTableDdlTest::getCreateTableQuery(
        ddl_utils::TableType::TABLE, "test_table", "(i INTEGER)", options);
  }
};

TEST_F(TableOptionsValidationTest, ReplicatedPartitionsNotSupported) {
  queryAndAssertException(getCreateTableQuery({{"partitions", "'REPLICATED'"}}),
                          "PARTITIONS='REPLICATED' is not supported");
}

class PageSizeValidationTest : public TableOptionsValidationTest {
 protected:
  std::string getCreateTableQuery(size_t page_size) const {
    return TableOptionsValidationTest::getCreateTableQuery(
        {{"page_size", std::to_string(page_size)}});
  }
};

TEST_F(PageSizeValidationTest, BelowMinPageSize) {
  const auto min_page_size = File_Namespace::FileBuffer::getMinPageSize();
  queryAndAssertException(
      PageSizeValidationTest::getCreateTableQuery(min_page_size - 1),
      "page_size cannot be less than " + std::to_string(min_page_size));
}

TEST_F(PageSizeValidationTest, MinPageSize) {
  sql(PageSizeValidationTest::getCreateTableQuery(
      File_Namespace::FileBuffer::getMinPageSize()));
  sql("INSERT INTO test_table VALUES (1);");
  sqlAndCompareResult("SELECT * FROM test_table;", {{i(1)}});
}

class StringTypeErrorValidationTest : public TableOptionsValidationTest,
                                      public testing::WithParamInterface<std::string> {};

TEST_P(StringTypeErrorValidationTest, StringOptions) {
  const auto& option_name = GetParam();
  queryAndAssertException(getCreateTableQuery({{option_name, "0"}}),
                          "The \"" + option_name + "\" option must be a string.");
}

INSTANTIATE_TEST_SUITE_P(
    OptionTypeValidation,
    StringTypeErrorValidationTest,
    ::testing::Values("partitions", "sort_column", "storage_type", "vacuum"));

class IntTypeErrorValidationTest : public TableOptionsValidationTest,
                                   public testing::WithParamInterface<std::string> {
 protected:
  std::string getCreateTableQuery(const std::map<std::string, std::string>& options,
                                  bool include_shard_key) const {
    return CreateAndDropTableDdlTest::getCreateTableQuery(
        ddl_utils::TableType::TABLE,
        "test_table",
        "(i INTEGER"s + (include_shard_key ? ", SHARD KEY (i)" : "") + ")",
        options);
  }
};

TEST_P(IntTypeErrorValidationTest, IntegerOptions) {
  const auto& option_name = GetParam();
  queryAndAssertException(
      getCreateTableQuery({{option_name, "'test'"}}, option_name == "shard_count"),
      to_upper(option_name) + " must be an integer literal.");
}

INSTANTIATE_TEST_SUITE_P(OptionTypeValidation,
                         IntTypeErrorValidationTest,
                         ::testing::Values("fragment_size",
                                           "max_chunk_size",
                                           "page_size",
                                           "max_rows",
                                           "shard_count",
                                           "max_rollback_epochs"));

int main(int argc, char** argv) {
  g_enable_fsi = true;
  TestHelpers::init_logger_stderr_only(argc, argv);
  testing::InitGoogleTest(&argc, argv);

  int err{0};
  try {
    testing::AddGlobalTestEnvironment(new DBHandlerTestEnvironment);
    err = RUN_ALL_TESTS();
  } catch (const std::exception& e) {
    LOG(ERROR) << e.what();
  }

  g_enable_fsi = false;

  return err;
}