Constraints in create table as

src/common/enum_util.cpp

@@ -1222,19 +1222,20 @@ const StringUtil::EnumStringLiteral *GetConstraintTypeValues() {
 		{ static_cast<uint32_t>(ConstraintType::NOT_NULL), "NOT_NULL" },
 		{ static_cast<uint32_t>(ConstraintType::CHECK), "CHECK" },
 		{ static_cast<uint32_t>(ConstraintType::UNIQUE), "UNIQUE" },
-		{ static_cast<uint32_t>(ConstraintType::FOREIGN_KEY), "FOREIGN_KEY" }
+		{ static_cast<uint32_t>(ConstraintType::FOREIGN_KEY), "FOREIGN_KEY" },
+		{ static_cast<uint32_t>(ConstraintType::COMPRESSION), "COMPRESSION" }
 	};
 	return values;
 }
 
 template<>
 const char* EnumUtil::ToChars<ConstraintType>(ConstraintType value) {
-	return StringUtil::EnumToString(GetConstraintTypeValues(), 5, "ConstraintType", static_cast<uint32_t>(value));
+	return StringUtil::EnumToString(GetConstraintTypeValues(), 6, "ConstraintType", static_cast<uint32_t>(value));
 }
 
 template<>
 ConstraintType EnumUtil::FromString<ConstraintType>(const char *value) {
-	return static_cast<ConstraintType>(StringUtil::StringToEnum(GetConstraintTypeValues(), 5, "ConstraintType", value));
+	return static_cast<ConstraintType>(StringUtil::StringToEnum(GetConstraintTypeValues(), 6, "ConstraintType", value));
 }
 
 const StringUtil::EnumStringLiteral *GetCoordinateReferenceSystemTypeValues() {

src/include/duckdb/common/constants.hpp

@@ -80,7 +80,7 @@ struct LogicalIndex {
 	inline bool operator<(const LogicalIndex &rhs) const {
 		return index < rhs.index;
 	};
-	bool IsValid() {
+	bool IsValid() const {
 		return index != DConstants::INVALID_INDEX;
 	}
 };

src/include/duckdb/parser/constraint.hpp

@@ -26,6 +26,7 @@ enum class ConstraintType : uint8_t {
 	CHECK = 2,       // CHECK constraint
 	UNIQUE = 3,      // UNIQUE constraint
 	FOREIGN_KEY = 4, // FOREIGN KEY constraint
+	COMPRESSION = 5, // COMPRESSION constraint
 };
 
 enum class ForeignKeyType : uint8_t {

src/include/duckdb/parser/constraints/compression_constraint.hpp

@@ -0,0 +1,38 @@
+//===----------------------------------------------------------------------===//
+//                         DuckDB
+//
+// duckdb/parser/constraints/compression_constraint.hpp
+//
+//
+//===----------------------------------------------------------------------===//
+
+#pragma once
+
+#include "duckdb/parser/constraint.hpp"
+#include "duckdb/common/enums/compression_type.hpp"
+
+namespace duckdb {
+
+class CompressionConstraint : public Constraint {
+public:
+	static constexpr const ConstraintType TYPE = ConstraintType::COMPRESSION;
+
+public:
+	DUCKDB_API CompressionConstraint(string column_name, CompressionType compression_type);
+	DUCKDB_API ~CompressionConstraint() override;
+
+	//! Column name this constraint pertains to
+	string column_name;
+	//! The compression type
+	CompressionType compression_type;
+
+public:
+	DUCKDB_API string ToString() const override;
+
+	DUCKDB_API unique_ptr<Constraint> Copy() const override;
+
+	DUCKDB_API void Serialize(Serializer &serializer) const override;
+	DUCKDB_API static unique_ptr<Constraint> Deserialize(Deserializer &deserializer);
+};
+
+} // namespace duckdb

src/include/duckdb/parser/constraints/list.hpp

@@ -2,3 +2,4 @@
 #include "duckdb/parser/constraints/not_null_constraint.hpp"
 #include "duckdb/parser/constraints/unique_constraint.hpp"
 #include "duckdb/parser/constraints/foreign_key_constraint.hpp"
+#include "duckdb/parser/constraints/compression_constraint.hpp"

src/include/duckdb/parser/constraints/not_null_constraint.hpp

@@ -17,11 +17,12 @@ class NotNullConstraint : public Constraint {
 	static constexpr const ConstraintType TYPE = ConstraintType::NOT_NULL;
 
 public:
-	DUCKDB_API explicit NotNullConstraint(LogicalIndex index);
+	DUCKDB_API explicit NotNullConstraint(LogicalIndex index, string column_name = "");
 	DUCKDB_API ~NotNullConstraint() override;
 
 	//! Column index this constraint pertains to
-	LogicalIndex index;
+	mutable LogicalIndex index;
+	string column_name;
 
 public:
 	DUCKDB_API string ToString() const override;

src/include/duckdb/parser/parsed_data/create_table_info.hpp

@@ -42,6 +42,7 @@ struct CreateTableInfo : public CreateInfo {
 	DUCKDB_API void Serialize(Serializer &serializer) const override;
 	DUCKDB_API static unique_ptr<CreateInfo> Deserialize(Deserializer &deserializer);
 
+	string ConstraintsToString() const;
 	string ToString() const override;
 };
 

src/include/duckdb/storage/serialization/constraint.json

@@ -22,9 +22,14 @@
         "id": 200,
         "name": "index",
         "type": "LogicalIndex"
+      },
+      {
+        "id": 201,
+        "name": "column_name",
+        "type": "string"
       }
     ],
-    "constructor": ["index"]
+    "constructor": ["index", "column_name"]
   },
   {
     "class": "CheckConstraint",
@@ -107,5 +112,23 @@
         "type": "vector<string>"
       }
     ]
+  },
+  {
+    "class": "CompressionConstraint",
+    "base": "Constraint",
+    "enum": "COMPRESSION",
+    "members": [
+      {
+        "id": 200,
+        "name": "column_name",
+        "type": "string"
+      },
+      {
+        "id": 201,
+        "name": "compression_type",
+        "type": "CompressionType"
+      }
+    ],
+    "constructor": ["column_name", "compression_type"]
   }
 ]

src/parser/constraints/CMakeLists.txt

@@ -1,6 +1,11 @@
 add_library_unity(
-  duckdb_constraints OBJECT check_constraint.cpp not_null_constraint.cpp
-  unique_constraint.cpp foreign_key_constraint.cpp)
+  duckdb_constraints
+  OBJECT
+  check_constraint.cpp
+  not_null_constraint.cpp
+  unique_constraint.cpp
+  foreign_key_constraint.cpp
+  compression_constraint.cpp)
 set(ALL_OBJECT_FILES
     ${ALL_OBJECT_FILES} $<TARGET_OBJECTS:duckdb_constraints>
     PARENT_SCOPE)

src/parser/constraints/compression_constraint.cpp

@@ -0,0 +1,22 @@
+#include "duckdb/parser/constraints/compression_constraint.hpp"
+#include "duckdb/common/string_util.hpp"
+#include "duckdb/common/enum_util.hpp"
+
+namespace duckdb {
+
+CompressionConstraint::CompressionConstraint(string column_name, CompressionType compression_type)
+    : Constraint(ConstraintType::COMPRESSION), column_name(std::move(column_name)), compression_type(compression_type) {
+}
+
+CompressionConstraint::~CompressionConstraint() {
+}
+
+string CompressionConstraint::ToString() const {
+	return StringUtil::Format("%s USING COMPRESSION %s", column_name, CompressionTypeToString(compression_type));
+}
+
+unique_ptr<Constraint> CompressionConstraint::Copy() const {
+	return make_uniq<CompressionConstraint>(column_name, compression_type);
+}
+
+} // namespace duckdb

src/parser/constraints/not_null_constraint.cpp

@@ -1,20 +1,30 @@
 #include "duckdb/parser/constraints/not_null_constraint.hpp"
-
+#include "duckdb/common/constants.hpp"
+#include "duckdb/common/string_util.hpp"
 #include "duckdb/common/helper.hpp"
 
 namespace duckdb {
 
-NotNullConstraint::NotNullConstraint(LogicalIndex index) : Constraint(ConstraintType::NOT_NULL), index(index) {
+NotNullConstraint::NotNullConstraint(LogicalIndex index, string column_name)
+    : Constraint(ConstraintType::NOT_NULL), index(index), column_name(column_name) {
 }
 
 NotNullConstraint::~NotNullConstraint() {
 }
 
 string NotNullConstraint::ToString() const {
+	if (!index.IsValid()) {
+		D_ASSERT(!column_name.empty());
+		return StringUtil::Format("%s NOT NULL", column_name);
+	}
 	return "NOT NULL";
 }
 
 unique_ptr<Constraint> NotNullConstraint::Copy() const {
+	if (!index.IsValid()) {
+		D_ASSERT(!column_name.empty());
+		return make_uniq<NotNullConstraint>(index, column_name);
+	}
 	return make_uniq<NotNullConstraint>(index);
 }
 

src/parser/parsed_data/create_table_info.cpp

@@ -38,12 +38,27 @@ unique_ptr<CreateInfo> CreateTableInfo::Copy() const {
 	return std::move(result);
 }
 
+string CreateTableInfo::ConstraintsToString() const {
+	if (constraints.empty()) {
+		return "";
+	}
+	string ret = " CONSTRAINTS (";
+	for (auto &constraint : constraints) {
+		ret += constraint->ToString();
+		ret += ",";
+	}
+	ret.pop_back();
+	ret += ") ";
+	return ret;
+}
+
 string CreateTableInfo::ToString() const {
 	string ret = GetCreatePrefix("TABLE");
 	ret += QualifierToString(temporary ? "" : catalog, schema, table);
 
 	if (query != nullptr) {
 		ret += TableCatalogEntry::ColumnNamesToSQL(columns);
+		ret += ConstraintsToString();
 		ret += " AS " + query->ToString();
 	} else {
 		ret += TableCatalogEntry::ColumnsToSQL(columns, constraints);

src/parser/transform/constraint/transform_constraint.cpp

@@ -52,6 +52,12 @@ TransformForeignKeyConstraint(duckdb_libpgquery::PGConstraint &constraint,
 			auto value = Transformer::PGPointerCast<duckdb_libpgquery::PGValue>(kc->data.ptr_value);
 			fk_columns.emplace_back(value->val.str);
 		}
+	} else if (!constraint.fk_attrs && constraint.keys && constraint.keys->length > 0) {
+		// For CTAS statement, fk_columns is empty, and it needs to be pulled from the constraint.keys
+		for (auto kc = constraint.keys->head; kc; kc = kc->next) {
+			auto value = Transformer::PGPointerCast<duckdb_libpgquery::PGValue>(kc->data.ptr_value);
+			fk_columns.emplace_back(value->val.str);
+		}
 	}
 
 	if (constraint.pk_attrs) {
@@ -93,8 +99,48 @@ unique_ptr<Constraint> Transformer::TransformConstraint(duckdb_libpgquery::PGCon
 	}
 	case duckdb_libpgquery::PG_CONSTR_FOREIGN:
 		return TransformForeignKeyConstraint(constraint);
+	case duckdb_libpgquery::PG_CONSTR_NOTNULL: {
+		// get column name
+		vector<string> columns;
+		for (auto kc = constraint.keys->head; kc; kc = kc->next) {
+			auto value = PGPointerCast<duckdb_libpgquery::PGValue>(kc->data.ptr_value);
+			columns.emplace_back(value->val.str);
+		}
+		D_ASSERT(columns.size() == 1);
+		// set the Not NULL and use the column name.
+		return make_uniq<NotNullConstraint>(LogicalIndex(DConstants::INVALID_INDEX), columns[0]);
+	}
+	case duckdb_libpgquery::PG_CONSTR_COMPRESSION: {
+		auto compression_type = EnumUtil::FromString<CompressionType>(constraint.compression_name);
+		switch (compression_type) {
+		case CompressionType::COMPRESSION_AUTO:
+		case CompressionType::COMPRESSION_CONSTANT:
+		case CompressionType::COMPRESSION_EMPTY:
+			throw InvalidInputException("Compression method %s cannot be forced", constraint.compression_name);
+		default:
+			break;
+		}
+		// get column name
+		vector<string> columns;
+		for (auto kc = constraint.keys->head; kc; kc = kc->next) {
+			auto value = PGPointerCast<duckdb_libpgquery::PGValue>(kc->data.ptr_value);
+			columns.emplace_back(value->val.str);
+		}
+		D_ASSERT(columns.size() == 1);
+		return make_uniq<CompressionConstraint>(columns[0], compression_type);
+	}
+	case duckdb_libpgquery::PG_CONSTR_GENERATED_STORED:
+	case duckdb_libpgquery::PG_CONSTR_GENERATED_VIRTUAL: {
+		throw NotImplementedException("Generated Columns not handled here");
+	}
+	case duckdb_libpgquery::PG_CONSTR_DEFAULT: {
+		throw NotImplementedException("Default Column values not handled here");
+	}
+	case duckdb_libpgquery::PG_CONSTR_NULL: {
+		throw NotImplementedException("Default NULL values not handled here");
+	}
 	default:
-		throw NotImplementedException("Constraint type not handled yet!");
+		throw NotImplementedException("Constraint not implemented");
 	}
 }
 
@@ -130,7 +176,7 @@ unique_ptr<Constraint> Transformer::TransformConstraint(duckdb_libpgquery::PGCon
 		case CompressionType::COMPRESSION_AUTO:
 		case CompressionType::COMPRESSION_CONSTANT:
 		case CompressionType::COMPRESSION_EMPTY:
-			throw InvalidInputException("Compression method %d cannot be forced", constraint.compression_name);
+			throw InvalidInputException("Compression method %s cannot be forced", constraint.compression_name);
 		default:
 			break;
 		}

src/parser/transform/statement/transform_create_table_as.cpp

@@ -47,6 +47,13 @@ unique_ptr<CreateStatement> Transformer::TransformCreateTableAs(duckdb_libpgquer
 			info->columns.AddColumn(ColumnDefinition(cols[i], LogicalType::UNKNOWN));
 		}
 	}
+
+	if (stmt.into->constraints) {
+		for (auto cell = stmt.into->constraints->head; cell != nullptr; cell = cell->next) {
+			auto pg_constraint = PGPointerCast<duckdb_libpgquery::PGConstraint>(cell->data.ptr_value);
+			info->constraints.push_back(TransformConstraint(*pg_constraint));
+		}
+	}
 	info->catalog = qname.catalog;
 	info->schema = qname.schema;
 	info->table = qname.name;

src/planner/binder/statement/bind_create_table.cpp

@@ -233,6 +233,22 @@ unique_ptr<BoundConstraint> Binder::BindConstraint(const Constraint &constraint,
 	}
 	case ConstraintType::NOT_NULL: {
 		auto &not_null = constraint.Cast<NotNullConstraint>();
+		if (!not_null.index.IsValid()) {
+			// this came from a create table as statement
+			auto logical_columns = columns.Logical();
+			bool column_found = false;
+			for (auto &column_def : logical_columns) {
+				auto name = column_def.GetName();
+				if (name == not_null.column_name) {
+					not_null.index = LogicalIndex(column_def.Logical());
+					column_found = true;
+					break;
+				}
+			}
+			if (!column_found) {
+				throw BinderException("Column %s does not exist", not_null.column_name);
+			}
+		}
 		auto &col = columns.GetColumn(not_null.index);
 		return make_uniq<BoundNotNullConstraint>(col.Physical());
 	}
@@ -590,6 +606,31 @@ static void BindCreateTableConstraints(CreateTableInfo &create_info, CatalogEntr
 		D_ASSERT(fk.info.pk_keys.size() == fk.pk_columns.size());
 		D_ASSERT(fk.info.fk_keys.size() == fk.fk_columns.size());
 	}
+
+	// Handle compression constraints: apply compression type to columns and remove from constraint list
+	auto &columns = create_info.columns;
+	for (auto &constraint : create_info.constraints) {
+		if (constraint->type != ConstraintType::COMPRESSION) {
+			continue;
+		}
+		auto &compression = constraint->Cast<CompressionConstraint>();
+		bool column_found = false;
+		for (idx_t col_idx = 0; col_idx < columns.LogicalColumnCount(); col_idx++) {
+			auto &column_def = columns.GetColumnMutable(LogicalIndex(col_idx));
+			if (column_def.GetName() == compression.column_name) {
+				column_def.SetCompressionType(compression.compression_type);
+				column_found = true;
+				break;
+			}
+		}
+		if (!column_found) {
+			throw BinderException("Column %s does not exist", compression.column_name);
+		}
+	}
+	auto &constraints = create_info.constraints;
+	auto it = std::remove_if(constraints.begin(), constraints.end(),
+	                         [](const unique_ptr<Constraint> &c) { return c->type == ConstraintType::COMPRESSION; });
+	constraints.erase(it, constraints.end());
 }
 
 unique_ptr<BoundCreateTableInfo> Binder::BindCreateTableInfo(unique_ptr<CreateInfo> info, SchemaCatalogEntry &schema,