Repository: kayak/pypika
Pull request #489 (2020-09-24 16:39:34)

Add DDL for temporal tables
The following changes add data definitions for temporal tables along with a bunch of other stuff for data definition.

## Columns

* nullable - an optional `bool` argument which (if set) will emit `NULL` or `NOT NULL`,
* default - an optional `Term` which (if set) will emit `DEFAULT <value>`

## Unique

A new builder method taking a list of Union[str, Column]; e.g.

```python
Query.create_table('t').columns(Column('c1', 'INT'),Column( 'c2', 'VARCHAR(100)')).unique('c1', 'c2').unique('c2')
```

This will emit:

```sql
CREATE TABLE t (
  c1 INT,
  c2 VARCHAR(100),

  UNIQUE (c1, c2),
  UNIQUE (c2)
)
```

## Primary Key

The primary key can be set as follows:

```python
Query.create_table('t').columns(Column('c1', 'INT'),Column( 'c2', 'VARCHAR(100)')).primary_key('c1')
```

This will emit:

```sql
CREATE TABLE t (
  c1 INT,
  c2 VARCHAR(100),

  PRIMARY KEY (c1)
)
```

## Period For

Multiple period for clauses can be added as follows.

```python
Query.create_table('t').columns(Column('valid_from', 'DATETIME'),Column( 'valid_to', DATETIME')).period_for('valid_period', 'valid_from', 'valid_to')
```

This will emit:

```sql
CREATE TABLE t (
  valid_from DATETIME,
  valid_to DATETIME,

  PERIOD FOR valid_period (valid_from, valid_to)
)
```

## With system versioning

System versioning can be added as follows.

```python
Query.create_table('t').columns(Column('c1', 'INT'),Column( 'c2', 'VARCHAR(100)')).with_system_versioning()
```

This will emit:

```sql
CREATE TABLE t (
  c1 INT,
  c2 VARCHAR(100),
) WITH SYSTEM VERSIONING
```



Patch:
diff --git a/pypika/dialects.py b/pypika/dialects.py
index 51cb01c..83c97c8 100644
--- a/pypika/dialects.py
+++ b/pypika/dialects.py
@@ -1,5 +1,5 @@
 from copy import copy
-from typing import Any, Union, Optional
+from typing import Any, List, Union, Optional
 
 from pypika.enums import Dialects
 from pypika.queries import (
@@ -221,11 +221,10 @@ class VerticaCreateQueryBuilder(CreateQueryBuilder):
               table=self._create_table.get_sql(**kwargs),
         )
 
-    def _columns_sql(self, **kwargs: Any) -> str:
-        return " ({columns}){preserve_rows}".format(
-              columns=",".join(column.get_sql(**kwargs) for column in self._columns),
-              preserve_rows=self._preserve_rows_sql(),
-        )
+    def _table_options_sql(self, **kwargs) -> str:
+        table_options = super()._table_options_sql(**kwargs)
+        table_options += self._preserve_rows_sql()
+        return table_options
 
     def _as_select_sql(self, **kwargs: Any) -> str:
         return "{preserve_rows} AS ({query})".format(
diff --git a/pypika/queries.py b/pypika/queries.py
index 392d3f9..4d68ad7 100644
--- a/pypika/queries.py
+++ b/pypika/queries.py
@@ -92,9 +92,9 @@ class Schema:
 
     def __eq__(self, other: "Schema") -> bool:
         return (
-              isinstance(other, Schema)
-              and self._name == other._name
-              and self._parent == other._parent
+            isinstance(other, Schema)
+            and self._name == other._name
+            and self._parent == other._parent
         )
 
     def __ne__(self, other: "Schema") -> bool:
@@ -110,8 +110,8 @@ class Schema:
 
         if self._parent is not None:
             return "{parent}.{schema}".format(
-                  parent=self._parent.get_sql(quote_char=quote_char, **kwargs),
-                  schema=schema_sql,
+                parent=self._parent.get_sql(quote_char=quote_char, **kwargs),
+                schema=schema_sql,
             )
 
         return schema_sql
@@ -132,7 +132,7 @@ class Table(Selectable):
             return schema
         if isinstance(schema, (list, tuple)):
             return reduce(
-                  lambda obj, s: Schema(s, parent=obj), schema[1:], Schema(schema[0])
+                lambda obj, s: Schema(s, parent=obj), schema[1:], Schema(schema[0])
             )
         if schema is not None:
             return Schema(schema)
@@ -156,7 +156,7 @@ class Table(Selectable):
 
         if self._schema is not None:
             table_sql = "{schema}.{table}".format(
-                  schema=self._schema.get_sql(**kwargs), table=table_sql
+                schema=self._schema.get_sql(**kwargs), table=table_sql
             )
 
         return format_alias_sql(table_sql, self.alias, **kwargs)
@@ -235,29 +235,46 @@ def make_tables(*names: Union[TypedTuple[str, str], str], **kwargs: Any) -> List
     for name in names:
         if isinstance(name, tuple) and len(name) == 2:
             t = Table(
-                  name=name[0], alias=name[1], schema=kwargs.get("schema"),
-                  query_cls=kwargs.get("query_cls"),
+                name=name[0], alias=name[1], schema=kwargs.get("schema"),
+                query_cls=kwargs.get("query_cls"),
             )
         else:
             t = Table(
-                  name=name, schema=kwargs.get("schema"),
-                  query_cls=kwargs.get("query_cls"),
+                name=name, schema=kwargs.get("schema"),
+                query_cls=kwargs.get("query_cls"),
             )
         tables.append(t)
     return tables
 
 
 class Column:
-    def __init__(self, column_name: str, column_type: Optional[str] = None) -> None:
+    def __init__(
+            self,
+            column_name: str,
+            column_type: Optional[str] = None,
+            nullable: Optional[bool] = None,
+            default: Optional[Term] = None
+    ) -> None:
         self.name = column_name
         self.type = column_type
+        self.nullable = nullable
+        self.default = default
 
-    def get_sql(self, **kwargs: Any) -> str:
+    def get_name_sql(self, **kwargs: Any) -> str:
         quote_char = kwargs.get("quote_char")
 
-        column_sql = "{name}{type}".format(
-              name=format_quotes(self.name, quote_char),
-              type=" {}".format(self.type) if self.type else "",
+        column_sql = "{name}".format(
+            name=format_quotes(self.name, quote_char),
+        )
+
+        return column_sql
+
+    def get_sql(self, **kwargs: Any) -> str:
+        column_sql = "{name}{type}{nullable}{default}".format(
+            name=self.get_name_sql(**kwargs),
+            type=" {}".format(self.type) if self.type else "",
+            nullable=" {}".format("NULL" if self.nullable else "NOT NULL") if self.nullable is not None else "",
+            default=" {}".format("DEFAULT " + self.default.get_sql(**kwargs)) if self.default else "",
         )
 
         return column_sql
@@ -283,6 +300,24 @@ def make_columns(*names: Union[TypedTuple[str, str], str]) -> List[Column]:
     return columns
 
 
+class PeriodFor:
+    def __init__(self, name: str, start_column: Union[str, Column], end_column: Union[str, Column]) -> None:
+        self.name = name
+        self.start_column = start_column if isinstance(start_column, Column) else Column(start_column)
+        self.end_column = end_column if isinstance(end_column, Column) else Column(end_column)
+
+    def get_sql(self, **kwargs: Any) -> str:
+        quote_char = kwargs.get("quote_char")
+
+        period_for_sql = "PERIOD FOR {name} ({start_column_name},{end_column_name})".format(
+            name=format_quotes(self.name, quote_char),
+            start_column_name=self.start_column.get_name_sql(**kwargs),
+            end_column_name=self.end_column.get_name_sql(**kwargs)
+        )
+
+        return period_for_sql
+
+
 # for typing in Query's methods
 _TableClass = Table
 
@@ -419,7 +454,7 @@ class _SetOperation(Selectable, Term):
     """
 
     def __init__(
-          self, base_query: "QueryBuilder", set_operation_query: "QueryBuilder", set_operation: SetOperation, alias: Optional[str] = None, wrapper_cls: Type[ValueWrapper] = ValueWrapper,
+        self, base_query: "QueryBuilder", set_operation_query: "QueryBuilder", set_operation: SetOperation, alias: Optional[str] = None, wrapper_cls: Type[ValueWrapper] = ValueWrapper,
     ):
         super().__init__(alias)
         self.base_query = base_query
@@ -491,25 +526,25 @@ class _SetOperation(Selectable, Term):
         kwargs.setdefault("quote_char", self.base_query.QUOTE_CHAR)
 
         base_querystring = self.base_query.get_sql(
-              subquery=self.base_query.wrap_set_operation_queries, **kwargs
+            subquery=self.base_query.wrap_set_operation_queries, **kwargs
         )
 
         querystring = base_querystring
         for set_operation, set_operation_query in self._set_operation:
             set_operation_querystring = set_operation_query.get_sql(
-                  subquery=self.base_query.wrap_set_operation_queries, **kwargs
+                subquery=self.base_query.wrap_set_operation_queries, **kwargs
             )
 
             if len(self.base_query._selects) != len(set_operation_query._selects):
                 raise SetOperationException(
-                      "Queries must have an equal number of select statements in a set operation."
-                      "\n\nMain Query:\n{query1}\n\nSet Operations Query:\n{query2}".format(
-                            query1=base_querystring, query2=set_operation_querystring
-                      )
+                    "Queries must have an equal number of select statements in a set operation."
+                    "\n\nMain Query:\n{query1}\n\nSet Operations Query:\n{query2}".format(
+                        query1=base_querystring, query2=set_operation_querystring
+                    )
                 )
 
             querystring += set_operation_template.format(
-                  type=set_operation.value, query_string=set_operation_querystring
+                type=set_operation.value, query_string=set_operation_querystring
             )
 
         if self._orderbys:
@@ -550,9 +585,9 @@ class _SetOperation(Selectable, Term):
             )
 
             clauses.append(
-                  "{term} {orient}".format(term=term, orient=directionality.value)
-                  if directionality is not None
-                  else term
+                "{term} {orient}".format(term=term, orient=directionality.value)
+                if directionality is not None
+                else term
             )
 
         return " ORDER BY {orderby}".format(orderby=",".join(clauses))
@@ -576,12 +611,12 @@ class QueryBuilder(Selectable, Term):
     QUERY_ALIAS_QUOTE_CHAR = None
 
     def __init__(
-          self,
-          dialect: Optional[Dialects] = None,
-          wrap_set_operation_queries: bool = True,
-          wrapper_cls: Type[ValueWrapper] = ValueWrapper,
-          immutable: bool = True,
-          as_keyword: bool = False,
+        self,
+        dialect: Optional[Dialects] = None,
+        wrap_set_operation_queries: bool = True,
+        wrapper_cls: Type[ValueWrapper] = ValueWrapper,
+        immutable: bool = True,
+        as_keyword: bool = False,
     ):
         super().__init__(None)
 
@@ -663,12 +698,12 @@ class QueryBuilder(Selectable, Term):
         """
 
         self._from.append(
-              Table(selectable) if isinstance(selectable, str) else selectable
+            Table(selectable) if isinstance(selectable, str) else selectable
         )
 
         if (
-              isinstance(selectable, (QueryBuilder, _SetOperation))
-              and selectable.alias is None
+            isinstance(selectable, (QueryBuilder, _SetOperation))
+            and selectable.alias is None
         ):
             if isinstance(selectable, QueryBuilder):
                 sub_query_count = selectable._subquery_count
@@ -910,8 +945,8 @@ class QueryBuilder(Selectable, Term):
             # MySQL rolls up all of the dimensions always
             if not terms and not self._groupbys:
                 raise RollupException(
-                      "At least one group is required. Call Query.groupby(term) or pass"
-                      "as parameter to rollup."
+                    "At least one group is required. Call Query.groupby(term) or pass"
+                    "as parameter to rollup."
                 )
 
             self._mysql_rollup = True
@@ -981,7 +1016,7 @@ class QueryBuilder(Selectable, Term):
     @builder
     def union(self, other: "QueryBuilder") -> _SetOperation:
         return _SetOperation(
-              self, other, SetOperation.union, wrapper_cls=self._wrapper_cls)
+            self, other, SetOperation.union, wrapper_cls=self._wrapper_cls)
 
     @builder
     def union_all(self, other: "QueryBuilder") -> _SetOperation:
@@ -1032,7 +1067,7 @@ class QueryBuilder(Selectable, Term):
     def _select_field_str(self, term: str) -> None:
         if 0 == len(self._from):
             raise QueryException(
-                  "Cannot select {term}, no FROM table specified.".format(term=term)
+                "Cannot select {term}, no FROM table specified.".format(term=term)
             )
 
         if term == "*":
@@ -1073,8 +1108,8 @@ class QueryBuilder(Selectable, Term):
         join.validate(base_tables, self._joins)
 
         table_in_query = any(
-              isinstance(clause, Table) and join.item in base_tables
-              for clause in base_tables
+            isinstance(clause, Table) and join.item in base_tables
+            for clause in base_tables
         )
         if isinstance(join.item, Table) and join.item.alias is None and table_in_query:
             # On the odd chance that we join the same table as the FROM table and don't set an alias
@@ -1097,12 +1132,12 @@ class QueryBuilder(Selectable, Term):
             table_in_base_tables = field.table in base_tables
             table_in_joins = field.table in [join.item for join in self._joins]
             if all(
-                  [
-                      field.table is not None,
-                      not table_in_base_tables,
-                      not table_in_joins,
-                      field.table != self._update_table,
-                  ]
+                [
+                    field.table is not None,
+                    not table_in_base_tables,
+                    not table_in_joins,
+                    field.table != self._update_table,
+                ]
             ):
                 return False
         return True
@@ -1121,10 +1156,10 @@ class QueryBuilder(Selectable, Term):
 
         for values in terms:
             self._values.append(
-                  [
-                      value if isinstance(value, Term) else self.wrap_constant(value)
-                      for value in values
-                  ]
+                [
+                    value if isinstance(value, Term) else self.wrap_constant(value)
+                    for value in values
+                ]
             )
 
     def __str__(self) -> str:
@@ -1158,10 +1193,10 @@ class QueryBuilder(Selectable, Term):
     def get_sql(self, with_alias: bool = False, subquery: bool = False, **kwargs: Any) -> str:
         self._set_kwargs_defaults(kwargs)
         if not (
-              self._selects
-              or self._insert_table
-              or self._delete_from
-              or self._update_table
+            self._selects
+            or self._insert_table
+            or self._delete_from
+            or self._update_table
         ):
             return ""
         if self._insert_table and not (self._selects or self._values):
@@ -1195,7 +1230,7 @@ class QueryBuilder(Selectable, Term):
 
             if self._joins:
                 querystring += " " + " ".join(
-                      join.get_sql(**kwargs) for join in self._joins
+                    join.get_sql(**kwargs) for join in self._joins
                 )
 
             querystring += self._set_sql(**kwargs)
@@ -1256,7 +1291,7 @@ class QueryBuilder(Selectable, Term):
 
         if self._joins:
             querystring += " " + " ".join(
-                  join.get_sql(**kwargs) for join in self._joins
+                join.get_sql(**kwargs) for join in self._joins
             )
 
         if self._prewheres:
@@ -1328,22 +1363,22 @@ class QueryBuilder(Selectable, Term):
 
     def _select_sql(self, **kwargs: Any) -> str:
         return "SELECT {distinct}{select}".format(
-              distinct=self._distinct_sql(**kwargs),
-              select=",".join(
-                    term.get_sql(with_alias=True, subquery=True, **kwargs)
-                    for term in self._selects
-              ),
+            distinct=self._distinct_sql(**kwargs),
+            select=",".join(
+                term.get_sql(with_alias=True, subquery=True, **kwargs)
+                for term in self._selects
+            ),
         )
 
     def _insert_sql(self, **kwargs: Any) -> str:
         return "INSERT {ignore}INTO {table}".format(
-              table=self._insert_table.get_sql(**kwargs),
-              ignore="IGNORE " if self._ignore else "",
+            table=self._insert_table.get_sql(**kwargs),
+            ignore="IGNORE " if self._ignore else "",
         )
 
     def _replace_sql(self, **kwargs: Any) -> str:
         return "REPLACE INTO {table}".format(
-              table=self._insert_table.get_sql(**kwargs),
+            table=self._insert_table.get_sql(**kwargs),
         )
 
     @staticmethod
@@ -1360,55 +1395,55 @@ class QueryBuilder(Selectable, Term):
             Remove from kwargs, never format the column terms with namespaces since only one table can be inserted into
         """
         return " ({columns})".format(
-              columns=",".join(
-                    term.get_sql(with_namespace=False, **kwargs) for term in self._columns
-              )
+            columns=",".join(
+                term.get_sql(with_namespace=False, **kwargs) for term in self._columns
+            )
         )
 
     def _values_sql(self, **kwargs: Any) -> str:
         return " VALUES ({values})".format(
-              values="),(".join(
-                    ",".join(
-                          term.get_sql(with_alias=True, subquery=True, **kwargs)
-                          for term in row
-                    )
-                    for row in self._values
-              )
+            values="),(".join(
+                ",".join(
+                    term.get_sql(with_alias=True, subquery=True, **kwargs)
+                    for term in row
+                )
+                for row in self._values
+            )
         )
 
     def _into_sql(self, **kwargs: Any) -> str:
         return " INTO {table}".format(
-              table=self._insert_table.get_sql(with_alias=False, **kwargs),
+            table=self._insert_table.get_sql(with_alias=False, **kwargs),
         )
 
     def _from_sql(self, with_namespace: bool = False, **kwargs: Any) -> str:
         return " FROM {selectable}".format(
-              selectable=",".join(
-                    clause.get_sql(subquery=True, with_alias=True, **kwargs)
-                    for clause in self._from
-              )
+            selectable=",".join(
+                clause.get_sql(subquery=True, with_alias=True, **kwargs)
+                for clause in self._from
+            )
         )
 
     def _force_index_sql(self, **kwargs: Any) -> str:
         return " FORCE INDEX ({indexes})".format(
-              indexes=",".join(index.get_sql(**kwargs) for index in self._force_indexes),
+            indexes=",".join(index.get_sql(**kwargs) for index in self._force_indexes),
         )
 
     def _use_index_sql(self, **kwargs: Any) -> str:
         return " USE INDEX ({indexes})".format(
-              indexes=",".join(index.get_sql(**kwargs) for index in self._use_indexes),
+            indexes=",".join(index.get_sql(**kwargs) for index in self._use_indexes),
         )
 
     def _prewhere_sql(self, quote_char: Optional[str] = None, **kwargs: Any) -> str:
         return " PREWHERE {prewhere}".format(
-              prewhere=self._prewheres.get_sql(
-                    quote_char=quote_char, subquery=True, **kwargs
-              )
+            prewhere=self._prewheres.get_sql(
+                quote_char=quote_char, subquery=True, **kwargs
+            )
         )
 
     def _where_sql(self, quote_char: Optional[str] = None, **kwargs: Any) -> str:
         return " WHERE {where}".format(
-              where=self._wheres.get_sql(quote_char=quote_char, subquery=True, **kwargs)
+            where=self._wheres.get_sql(quote_char=quote_char, subquery=True, **kwargs)
         )
 
     def _group_sql(
@@ -1428,15 +1463,15 @@ class QueryBuilder(Selectable, Term):
         for field in self._groupbys:
             if groupby_alias and field.alias and field.alias in selected_aliases:
                 clauses.append(
-                      format_quotes(field.alias, alias_quote_char or quote_char)
+                    format_quotes(field.alias, alias_quote_char or quote_char)
                 )
             else:
                 clauses.append(
-                      field.get_sql(
-                            quote_char=quote_char,
-                            alias_quote_char=alias_quote_char,
-                            **kwargs
-                      )
+                    field.get_sql(
+                        quote_char=quote_char,
+                        alias_quote_char=alias_quote_char,
+                        **kwargs
+                    )
                 )
 
         sql = " GROUP BY {groupby}".format(groupby=",".join(clauses))
@@ -1466,14 +1501,14 @@ class QueryBuilder(Selectable, Term):
                 format_quotes(field.alias, alias_quote_char or quote_char)
                 if orderby_alias and field.alias and field.alias in selected_aliases
                 else field.get_sql(
-                      quote_char=quote_char, alias_quote_char=alias_quote_char, **kwargs
+                    quote_char=quote_char, alias_quote_char=alias_quote_char, **kwargs
                 )
             )
 
             clauses.append(
-                  "{term} {orient}".format(term=term, orient=directionality.value)
-                  if directionality is not None
-                  else term
+                "{term} {orient}".format(term=term, orient=directionality.value)
+                if directionality is not None
+                else term
             )
 
         return " ORDER BY {orderby}".format(orderby=",".join(clauses))
@@ -1483,7 +1518,7 @@ class QueryBuilder(Selectable, Term):
 
     def _having_sql(self, quote_char: Optional[str] = None, **kwargs: Any) -> str:
         return " HAVING {having}".format(
-              having=self._havings.get_sql(quote_char=quote_char, **kwargs)
+            having=self._havings.get_sql(quote_char=quote_char, **kwargs)
         )
 
     def _offset_sql(self) -> str:
@@ -1494,12 +1529,12 @@ class QueryBuilder(Selectable, Term):
 
     def _set_sql(self, **kwargs: Any) -> str:
         return " SET {set}".format(
-              set=",".join(
-                    "{field}={value}".format(
-                          field=field.get_sql(**dict(kwargs, with_namespace=False)), value=value.get_sql(**kwargs)
-                    )
-                    for field, value in self._updates
-              )
+            set=",".join(
+                "{field}={value}".format(
+                    field=field.get_sql(**dict(kwargs, with_namespace=False)), value=value.get_sql(**kwargs)
+                )
+                for field, value in self._updates
+            )
         )
 
 
@@ -1513,8 +1548,8 @@ class Joiner:
     def on(self, criterion: Optional[Criterion], collate: Optional[str] = None) -> QueryBuilder:
         if criterion is None:
             raise JoinException(
-                  "Parameter 'criterion' is required for a "
-                  "{type} JOIN but was not supplied.".format(type=self.type_label)
+                "Parameter 'criterion' is required for a "
+                "{type} JOIN but was not supplied.".format(type=self.type_label)
             )
 
         self.query.do_join(JoinOn(self.item, self.how, criterion, collate))
@@ -1523,14 +1558,14 @@ class Joiner:
     def on_field(self, *fields: Any) -> QueryBuilder:
         if not fields:
             raise JoinException(
-                  "Parameter 'fields' is required for a "
-                  "{type} JOIN but was not supplied.".format(type=self.type_label)
+                "Parameter 'fields' is required for a "
+                "{type} JOIN but was not supplied.".format(type=self.type_label)
             )
 
         criterion = None
         for field in fields:
             consituent = Field(field, table=self.query._from[0]) == Field(
-                  field, table=self.item
+                field, table=self.item
             )
             criterion = consituent if criterion is None else criterion & consituent
 
@@ -1540,12 +1575,12 @@ class Joiner:
     def using(self, *fields: Any) -> QueryBuilder:
         if not fields:
             raise JoinException(
-                  "Parameter 'fields' is required when joining with "
-                  "a using clause but was not supplied.".format(type=self.type_label)
+                "Parameter 'fields' is required when joining with "
+                "a using clause but was not supplied.".format(type=self.type_label)
             )
 
         self.query.do_join(
-              JoinUsing(self.item, self.how, [Field(field) for field in fields])
+            JoinUsing(self.item, self.how, [Field(field) for field in fields])
         )
         return self.query
 
@@ -1598,9 +1633,9 @@ class JoinOn(Join):
     def get_sql(self, **kwargs: Any) -> str:
         join_sql = super().get_sql(**kwargs)
         return "{join} ON {criterion}{collate}".format(
-              join=join_sql,
-              criterion=self.criterion.get_sql(subquery=True, **kwargs),
-              collate=" COLLATE {}".format(self.collate) if self.collate else "",
+            join=join_sql,
+            criterion=self.criterion.get_sql(subquery=True, **kwargs),
+            collate=" COLLATE {}".format(self.collate) if self.collate else "",
         )
 
     def validate(self, _from: Sequence[Table], _joins: Sequence[Table]) -> None:
@@ -1609,10 +1644,10 @@ class JoinOn(Join):
         missing_tables = criterion_tables - available_tables
         if missing_tables:
             raise JoinException(
-                  "Invalid join criterion. One field is required from the joined item and "
-                  "another from the selected table or an existing join.  Found [{tables}]".format(
-                        tables=", ".join(map(str, missing_tables))
-                  )
+                "Invalid join criterion. One field is required from the joined item and "
+                "another from the selected table or an existing join.  Found [{tables}]".format(
+                    tables=", ".join(map(str, missing_tables))
+                )
             )
 
     @builder
@@ -1640,8 +1675,8 @@ class JoinUsing(Join):
     def get_sql(self, **kwargs: Any) -> str:
         join_sql = super().get_sql(**kwargs)
         return "{join} USING ({fields})".format(
-              join=join_sql,
-              fields=",".join(field.get_sql(**kwargs) for field in self.fields),
+            join=join_sql,
+            fields=",".join(field.get_sql(**kwargs) for field in self.fields),
         )
 
     def validate(self, _from: Sequence[Table], _joins: Sequence[Table]) -> None:
@@ -1680,6 +1715,10 @@ class CreateQueryBuilder:
         self._temporary = False
         self._as_select = None
         self._columns = []
+        self._period_fors = []
+        self._with_system_versioning = False
+        self._primary_key = None
+        self._uniques = []
         self.dialect = dialect
 
     def _set_kwargs_defaults(self, kwargs: dict) -> None:
@@ -1687,24 +1726,6 @@ class CreateQueryBuilder:
         kwargs.setdefault("secondary_quote_char", self.SECONDARY_QUOTE_CHAR)
         kwargs.setdefault("dialect", self.dialect)
 
-    def get_sql(self, **kwargs: Any) -> str:
-        self._set_kwargs_defaults(kwargs)
-
-        if not self._create_table:
-            return ""
-
-        if not self._columns and not self._as_select:
-            return ""
-
-        querystring = self._create_table_sql(**kwargs)
-
-        if self._columns:
-            querystring += self._columns_sql(**kwargs)
-        else:
-            querystring += self._as_select_sql(**kwargs)
-
-        return querystring
-
     @builder
     def create_table(self, table: Union[Table, str]) -> "CreateQueryBuilder":
         if self._create_table:
@@ -1716,6 +1737,10 @@ class CreateQueryBuilder:
     def temporary(self) -> "CreateQueryBuilder":
         self._temporary = True
 
+    @builder
+    def with_system_versioning(self) -> "CreateQueryBuilder":
+        self._with_system_versioning = True
+
     @builder
     def columns(self, *columns: Union[str, TypedTuple[str, str], Column]) -> "CreateQueryBuilder":
         if self._as_select:
@@ -1728,6 +1753,26 @@ class CreateQueryBuilder:
                 column = Column(column_name=column[0], column_type=column[1])
             self._columns.append(column)
 
+    @builder
+    def period_for(self, name, start_column: Union[str, Column], end_column: Union[str, Column]) -> "CreateQueryBuilder":
+        self._period_fors.append(PeriodFor(name, start_column, end_column))
+
+    @builder
+    def unique(self, *columns: Union[str, Column]) -> "CreateQueryBuilder":
+        self._uniques.append([
+            (column if isinstance(column, Column) else Column(column))
+            for column in columns
+        ])
+
+    @builder
+    def primary_key(self, *columns: Union[str, Column]) -> "CreateQueryBuilder":
+        if self._primary_key:
+            raise AttributeError("'Query' object already has attribute primary_key")
+        self._primary_key = [
+            (column if isinstance(column, Column) else Column(column))
+            for column in columns
+        ]
+
     @builder
     def as_select(self, query_builder: QueryBuilder) -> "CreateQueryBuilder":
         if self._columns:
@@ -1738,17 +1783,84 @@ class CreateQueryBuilder:
 
         self._as_select = query_builder
 
+    def get_sql(self, **kwargs: Any) -> str:
+        self._set_kwargs_defaults(kwargs)
+
+        if not self._create_table:
+            return ""
+
+        if not self._columns and not self._as_select:
+            return ""
+
+        create_table = self._create_table_sql(**kwargs)
+
+        if self._as_select:
+            return create_table + self._as_select_sql(**kwargs)
+
+        body = self._body_sql(**kwargs)
+        table_options = self._table_options_sql(**kwargs)
+
+        return "{create_table} ({body}){table_options}".format(
+            create_table=create_table,
+            body=body,
+            table_options=table_options
+        )
+
     def _create_table_sql(self, **kwargs: Any) -> str:
         return "CREATE {temporary}TABLE {table}".format(
-              temporary="TEMPORARY " if self._temporary else "",
-              table=self._create_table.get_sql(**kwargs),
+            temporary="TEMPORARY " if self._temporary else "",
+            table=self._create_table.get_sql(**kwargs),
         )
 
-    def _columns_sql(self, **kwargs: Any) -> str:
-        return " ({columns})".format(
-              columns=",".join(column.get_sql(**kwargs) for column in self._columns)
+    def _table_options_sql(self, **kwargs) -> str:
+        table_options = ""
+
+        if self._with_system_versioning:
+            table_options += ' WITH SYSTEM VERSIONING'
+            
+        return table_options
+
+    def _column_clauses(self, **kwargs) -> List[str]:
+        return [
+            column.get_sql(**kwargs)
+            for column in self._columns
+        ]
+
+    def _period_for_clauses(self, **kwargs) -> List[str]:
+        return [
+            period_for.get_sql(**kwargs)
+            for period_for in self._period_fors
+        ]
+
+    def _unique_key_clauses(self, **kwargs) -> List[str]:
+        return [
+            "UNIQUE ({unique})".format(
+                unique=",".join(
+                    column.get_name_sql(**kwargs)
+                    for column in unique)
+            )
+            for unique in self._uniques
+        ]
+
+    def _primary_key_clause(self, **kwargs) -> str:
+        return "PRIMARY KEY ({columns})".format(
+            columns=",".join(
+                column.get_name_sql(**kwargs)
+                for column in self._primary_key
+            )
         )
 
+    def _body_sql(self, **kwargs) -> str:
+        clauses = self._column_clauses(**kwargs)
+        clauses += self._period_for_clauses(**kwargs)
+        clauses += self._unique_key_clauses(**kwargs)
+
+        # Primary keys
+        if self._primary_key:
+            clauses.append(self._primary_key_clause(**kwargs))
+
+        return ",".join(clauses)
+
     def _as_select_sql(self, **kwargs: Any) -> str:
         return " AS ({query})".format(query=self._as_select.get_sql(**kwargs), )
 


Test patch:
diff --git a/pypika/tests/test_create.py b/pypika/tests/test_create.py
index ae3a840..75968cf 100644
--- a/pypika/tests/test_create.py
+++ b/pypika/tests/test_create.py
@@ -2,9 +2,11 @@ import unittest
 
 from pypika import (
     Tables,
+    Column,
     Columns,
     Query,
 )
+from pypika.terms import ValueWrapper
 
 
 class CreateTableTests(unittest.TestCase):
@@ -12,6 +14,35 @@ class CreateTableTests(unittest.TestCase):
     foo, bar = Columns(("a", "INT"), ("b", "VARCHAR(100)"))
 
     def test_create_table_with_columns(self):
+        with self.subTest("with nullable"):
+            a = Column("a", "INT", True)
+            b = Column("b", "VARCHAR(100)", False)
+            q = Query.create_table(self.new_table).columns(a, b)
+
+            self.assertEqual('CREATE TABLE "abc" ("a" INT NULL,"b" VARCHAR(100) NOT NULL)', str(q))
+
+        with self.subTest("with defaults"):
+            a = Column("a", "INT", default=ValueWrapper(42))
+            b = Column("b", "VARCHAR(100)", default=ValueWrapper("foo"))
+            q = Query.create_table(self.new_table).columns(a, b)
+
+            self.assertEqual('CREATE TABLE "abc" ("a" INT DEFAULT 42,"b" VARCHAR(100) DEFAULT \'foo\')', str(q))
+
+        with self.subTest("with period for"):
+            a = Column("id", "INT")
+            b = Column("valid_from", "DATETIME")
+            c = Column("valid_to", "DATETIME")
+            q = Query.create_table(self.new_table).columns(a, b, c).period_for('valid_period', b, c)
+
+            self.assertEqual(
+                'CREATE TABLE "abc" ('
+                '"id" INT,'
+                '"valid_from" DATETIME,'
+                '"valid_to" DATETIME,'
+                'PERIOD FOR "valid_period" ("valid_from","valid_to"))',
+                str(q)
+            )
+
         with self.subTest("without temporary keyword"):
             q = Query.create_table(self.new_table).columns(self.foo, self.bar)
 
@@ -28,6 +59,35 @@ class CreateTableTests(unittest.TestCase):
                 'CREATE TEMPORARY TABLE "abc" ("a" INT,"b" VARCHAR(100))', str(q)
             )
 
+        with self.subTest("with primary key"):
+            q = Query.create_table(
+                self.new_table
+            ).columns(
+                self.foo, self.bar
+            ).primary_key(
+                self.foo, self.bar
+            )
+
+            self.assertEqual('CREATE TABLE "abc" ("a" INT,"b" VARCHAR(100),PRIMARY KEY ("a","b"))', str(q))
+
+        with self.subTest("with unique keys"):
+            q = Query.create_table(
+                self.new_table
+            ).columns(
+                self.foo, self.bar
+            ).unique(
+                self.foo, self.bar
+            ).unique(
+                self.foo
+            )
+
+            self.assertEqual('CREATE TABLE "abc" ("a" INT,"b" VARCHAR(100),UNIQUE ("a","b"),UNIQUE ("a"))', str(q))
+
+        with self.subTest("with system versioning"):
+            q = Query.create_table(self.new_table).columns(self.foo, self.bar).with_system_versioning()
+
+            self.assertEqual('CREATE TABLE "abc" ("a" INT,"b" VARCHAR(100)) WITH SYSTEM VERSIONING', str(q))
+
     def test_create_table_with_select(self):
         select = Query.from_(self.existing_table).select(
             self.existing_table.foo, self.existing_table.bar