Refactor insert plan for better error handling and more efficiency (#1063)

* Better include path checks

* Fix validator script

* Fix local includes in existing files

* functools.reduce() instead of reduce()

* Refactor insert for error handling

* Add tests for exception handling

* Move exception handling from planner to query transformer

* Fix linting

* Resolve merge conflict (remove indicators)

* Extract schema from catalog instead of physical table

* Fix linting

* Add braces

* Re-fix linting

* Refactor bad insert tests

* Add database deletion in test

* Lint insert tests file

* Add not null test

* Remove obsolete commented code and add important todo

* Remove exception type getter

* Remove changed rows equality checks

* Reduce nesting

* Fix insertion segfault

* Remove rows_changed variable completely

* Code cleanup

* Add more not null logic

* Add GetColumnNames() implementation

* Make specify a map

* Format code

* Minor test changes

* Format tests

* Cleanups

* Add semicolon, lint

* Use commit instead of abort

* Make specified a set

* Add missing txn commit

Author: schedutron

src/catalog/table_catalog.cpp

@@ -257,6 +257,22 @@ TableCatalogObject::GetColumnObjects(bool cached_only) {
   return column_objects;
 }
 
+/* @brief   get all column objects of this table into cache
+ * @return  map from column name to cached column object
+ */
+std::unordered_map<std::string, std::shared_ptr<ColumnCatalogObject>>
+TableCatalogObject::GetColumnNames(bool cached_only) {
+  if (!valid_column_objects && !cached_only) {
+    auto column_objects = GetColumnObjects();
+    std::unordered_map<std::string, std::shared_ptr<ColumnCatalogObject> > column_names;
+    for (auto& pair : column_objects) {
+      auto column = pair.second;
+      column_names[column->GetColumnName()] = column;
+    }
+  }
+  return column_names;
+}
+
 /* @brief   get column object with column id from cache
  * @param   column_id
  * @param   cached_only   if cached only, return nullptr on a cache miss

src/optimizer/query_to_operator_transformer.cpp

@@ -11,21 +11,18 @@
 //===----------------------------------------------------------------------===//
 
 #include <cmath>
-#include "settings/settings_manager.h"
 
+#include "catalog/column_catalog.h"
 #include "catalog/database_catalog.h"
+#include "catalog/table_catalog.h"
 #include "expression/expression_util.h"
 #include "expression/subquery_expression.h"
 #include "optimizer/operator_expression.h"
 #include "optimizer/operators.h"
 #include "optimizer/query_node_visitor.h"
 #include "optimizer/query_to_operator_transformer.h"
-
-#include "planner/seq_scan_plan.h"
-
 #include "parser/statements.h"
-
-#include "catalog/manager.h"
+#include "settings/settings_manager.h"
 
 using std::vector;
 using std::shared_ptr;
@@ -245,11 +242,71 @@ void QueryToOperatorTransformer::Visit(parser::InsertStatement *op) {
     op->select->Accept(this);
     insert_expr->PushChild(output_expr_);
     output_expr_ = insert_expr;
+    return;
+  }
+  // column_objects represents the columns for the current table as defined in
+  // its schema
+  auto column_objects = target_table->GetColumnObjects();
+  // INSERT INTO table_name VALUES (val1, val2, ...), (val_a, val_b, ...), ...
+  if (op->columns.empty()) {
+    for (const auto &values : op->insert_values) {
+      if (values.size() > column_objects.size()) {
+        throw CatalogException(
+            "ERROR:  INSERT has more expressions than target columns");
+      } else if (values.size() < column_objects.size()) {
+        for (oid_t i = values.size(); i != column_objects.size(); ++i) {
+          // check whether null values are allowed in the rest of the columns
+          if (column_objects[i]->IsNotNull()) {
+            // TODO: Add check for default value's existence for the current
+            // column
+            throw CatalogException(
+                StringUtil::Format("ERROR:  null value in column \"%s\" "
+                                   "violates not-null constraint",
+                                   column_objects[i]->GetColumnName().c_str()));
+          }
+        }
+      }
+    }
   } else {
-    auto insert_expr = std::make_shared<OperatorExpression>(
-        LogicalInsert::make(target_table, &op->columns, &op->insert_values));
-    output_expr_ = insert_expr;
+    // INSERT INTO table_name (col1, col2, ...) VALUES (val1, val2, ...), ...
+    auto num_columns = op->columns.size();
+    for (const auto &tuple : op->insert_values) {  // check size of each tuple
+      if (tuple.size() > num_columns) {
+        throw CatalogException(
+            "ERROR:  INSERT has more expressions than target columns");
+      } else if (tuple.size() < num_columns) {
+        throw CatalogException(
+            "ERROR:  INSERT has more target columns than expressions");
+      }
+    }
+
+    // set below contains names of columns mentioned in the insert statement
+    std::unordered_set<std::string> specified;
+    auto column_names = target_table->GetColumnNames();
+
+    for (const auto col : op->columns) {
+      if (column_names.find(col) == column_names.end()) {
+        throw CatalogException(StringUtil::Format(
+            "ERROR:  column \"%s\" of relation \"%s\" does not exist",
+            col.c_str(), target_table->GetTableName().c_str()));
+      }
+      specified.insert(col);
+    }
+
+    for (auto column : column_names) {
+      // this loop checks not null constraint for unspecified columns
+      if (specified.find(column.first) == specified.end() &&
+          column.second->IsNotNull()) {
+        // TODO: Add check for default value's existence for the current column
+        throw CatalogException(StringUtil::Format(
+            "ERROR:  null value in column \"%s\" violates not-null constraint",
+            column.first.c_str()));
+      }
+    }
   }
+  auto insert_expr = std::make_shared<OperatorExpression>(
+      LogicalInsert::make(target_table, &op->columns, &op->insert_values));
+  output_expr_ = insert_expr;
 }
 
 void QueryToOperatorTransformer::Visit(parser::DeleteStatement *op) {

src/planner/insert_plan.cpp

@@ -66,6 +66,11 @@ InsertPlan::InsertPlan(
           is_prepared_stmt = true;
         }
       }
+      // for remaining columns, insert defaults
+      for (uint32_t column_id = values.size(); column_id != schema_col_count;
+           ++column_id) {
+        SetDefaultValue(column_id);
+      }
     }
   } else {
     // INSERT INTO table_name (col1, col2, ...) VALUES (val1, val2, ...);

test/sql/insert_sql_test.cpp

@@ -13,6 +13,7 @@
 #include <memory>
 
 #include "catalog/catalog.h"
+#include "common/exception.h"
 #include "common/harness.h"
 #include "concurrency/transaction_manager_factory.h"
 #include "executor/create_executor.h"
@@ -103,6 +104,12 @@ void CreateAndLoadTable7() {
   TestingSQLUtil::ExecuteSQLQuery("INSERT INTO test7 VALUES (55, 8, 999);");
 }
 
+void CreateAndLoadTable8() {
+  // Create a table with some defaults
+  TestingSQLUtil::ExecuteSQLQuery(
+      "CREATE TABLE test8(num1 int, num2 int, num3 int not null);");
+}
+
 TEST_F(InsertSQLTests, InsertOneValue) {
   auto &txn_manager = concurrency::TransactionManagerFactory::GetInstance();
   auto txn = txn_manager.BeginTransaction();
@@ -204,7 +211,7 @@ TEST_F(InsertSQLTests, InsertSpecifyColumns) {
   catalog::Catalog::GetInstance()->CreateDatabase(DEFAULT_DB_NAME, txn);
   txn_manager.CommitTransaction(txn);
 
-  CreateAndLoadTable();
+  CreateAndLoadTable8();
 
   std::vector<ResultValue> result;
   std::vector<FieldInfo> tuple_descriptor;
@@ -214,7 +221,8 @@ TEST_F(InsertSQLTests, InsertSpecifyColumns) {
       new optimizer::Optimizer());
 
   // INSERT a tuple with columns specified
-  std::string query("INSERT INTO test (b, a, c) VALUES (99, 8, 111);");
+  std::string query(
+      "INSERT INTO test8 (num3, num2, num1) VALUES (99, 8, 111);");
 
   txn = txn_manager.BeginTransaction();
   auto plan = TestingSQLUtil::GeneratePlanWithOptimizer(optimizer, query, txn);
@@ -228,12 +236,12 @@ TEST_F(InsertSQLTests, InsertSpecifyColumns) {
 
   // SELECT to find out if the tuple is correctly inserted
   TestingSQLUtil::ExecuteSQLQueryWithOptimizer(
-      optimizer, "SELECT * FROM test WHERE a=8", result, tuple_descriptor,
+      optimizer, "SELECT * FROM test8 WHERE num2=8", result, tuple_descriptor,
       rows_changed, error_message);
   EXPECT_EQ(3, result.size());
-  EXPECT_EQ("8", TestingSQLUtil::GetResultValueAsString(result, 0));
-  EXPECT_EQ("99", TestingSQLUtil::GetResultValueAsString(result, 1));
-  EXPECT_EQ("111", TestingSQLUtil::GetResultValueAsString(result, 2));
+  EXPECT_EQ("111", TestingSQLUtil::GetResultValueAsString(result, 0));
+  EXPECT_EQ("8", TestingSQLUtil::GetResultValueAsString(result, 1));
+  EXPECT_EQ("99", TestingSQLUtil::GetResultValueAsString(result, 2));
 
   // free the database just created
   txn = txn_manager.BeginTransaction();
@@ -620,6 +628,87 @@ TEST_F(InsertSQLTests, NonExistentTable) {
   EXPECT_THROW(TestingSQLUtil::GeneratePlanWithOptimizer(optimizer, query, txn),
                peloton::CatalogException);
   txn_manager.CommitTransaction(txn);
+
+  // free the database just created
+  txn = txn_manager.BeginTransaction();
+  catalog::Catalog::GetInstance()->DropDatabaseWithName(DEFAULT_DB_NAME, txn);
+  txn_manager.CommitTransaction(txn);
+}
+
+TEST_F(InsertSQLTests, BadInserts) {
+  auto &txn_manager = concurrency::TransactionManagerFactory::GetInstance();
+  auto txn = txn_manager.BeginTransaction();
+  catalog::Catalog::GetInstance()->CreateDatabase(DEFAULT_DB_NAME, txn);
+  txn_manager.CommitTransaction(txn);
+
+  CreateAndLoadTable8();
+
+  std::vector<ResultValue> result;
+  std::vector<FieldInfo> tuple_descriptor;
+  std::string error_message;
+  std::unique_ptr<optimizer::AbstractOptimizer> optimizer(
+      new optimizer::Optimizer());
+  // Insert a tuple with more values than target columns
+  std::string query("INSERT INTO test8 VALUES(1, 2, 3, 4);");
+  txn = txn_manager.BeginTransaction();
+  EXPECT_THROW(TestingSQLUtil::GeneratePlanWithOptimizer(optimizer, query, txn),
+               peloton::CatalogException);
+  txn_manager.CommitTransaction(txn);
+
+  // Insert a tuple with more target columns (implicit) than values, violating
+  // not null constraint for num3
+  query = "INSERT INTO test VALUES(3, 4);";
+  txn = txn_manager.BeginTransaction();
+  EXPECT_THROW(TestingSQLUtil::GeneratePlanWithOptimizer(optimizer, query, txn),
+               peloton::CatalogException);
+  txn_manager.CommitTransaction(txn);
+
+  // Insert a tuple with more target columns than values
+  query = "INSERT INTO test8(num1, num3) VALUES(3);";
+  txn = txn_manager.BeginTransaction();
+  EXPECT_THROW(TestingSQLUtil::GeneratePlanWithOptimizer(optimizer, query, txn),
+               peloton::CatalogException);
+  txn_manager.CommitTransaction(txn);
+
+  // Insert a tuple with more values than target columns (multiple tuples)
+  query = "INSERT INTO test8(num1, num3) VALUES (1, 2), (3, 4), (3, 4, 5);";
+  txn = txn_manager.BeginTransaction();
+  EXPECT_THROW(TestingSQLUtil::GeneratePlanWithOptimizer(optimizer, query, txn),
+               peloton::CatalogException);
+  txn_manager.CommitTransaction(txn);
+
+  // Insert a tuple with more target columns than values (multiple tuples)
+  query = "INSERT INTO test8(num1, num3) VALUES (6, 7), (5);";
+  txn = txn_manager.BeginTransaction();
+  EXPECT_THROW(TestingSQLUtil::GeneratePlanWithOptimizer(optimizer, query, txn),
+               peloton::CatalogException);
+  txn_manager.CommitTransaction(txn);
+
+  // Insert a tuple with a nonexistent target column
+  query = "INSERT INTO test8(numx) VALUES(3);";
+  txn = txn_manager.BeginTransaction();
+  EXPECT_THROW(TestingSQLUtil::GeneratePlanWithOptimizer(optimizer, query, txn),
+               peloton::CatalogException);
+  txn_manager.CommitTransaction(txn);
+
+  // Insert a tuple with a nonexistent target column (non-singleton tuple)
+  query = "INSERT INTO test8(num1, num4) VALUES(3, 4);";
+  txn = txn_manager.BeginTransaction();
+  EXPECT_THROW(TestingSQLUtil::GeneratePlanWithOptimizer(optimizer, query, txn),
+               peloton::CatalogException);
+  txn_manager.CommitTransaction(txn);
+
+  // Insert a tuple row with not-null field (num3) unspecified
+  query = "INSERT INTO test8 VALUES(1, 2)";
+  txn = txn_manager.BeginTransaction();
+  EXPECT_THROW(TestingSQLUtil::GeneratePlanWithOptimizer(optimizer, query, txn),
+               peloton::CatalogException);
+  txn_manager.CommitTransaction(txn);
+
+  // free the database just created
+  txn = txn_manager.BeginTransaction();
+  catalog::Catalog::GetInstance()->DropDatabaseWithName(DEFAULT_DB_NAME, txn);
+  txn_manager.CommitTransaction(txn);
 }
 
 }  // namespace test