# Creating & deleting Kafka topics
Lenses supports the typical SQL commands supported by a relational database:
* CREATE
* DROP
* TRUNCATE
* DELETE
* SHOW TABLES
* DESCRIBE TABLE
* DESCRIBE FORMATTED
### CREATE TABLE
```sql
CREATE TABLE
table_name(
$field $fieldType [, $field $fieldType,...]
)
FORMAT ($keyStorageFormat, $valueStorageFormat)
[PROPERTIES(
partitions= *,
replication=$replication,
compacted=true/false)
];
```
The `CREATE` statement has the following parts:
* **CREATE TABLE** - Instructs the construction of a table
* **$Table** - The actual name given to the table.
* **Schema** - Constructed as a list of (field, type) tuple, it describes the data each record in the table contains
* **FORMAT** - Defines the storage format. Since it is an Apache Kafka topic, both the Key and the Value formats are required. Valid values are STRING, INT, LONG, JSON, AVRO.
* **PROPERTIES** - Specifies the number of partitions the final Kafka topic should have, the replication factor in order to ensure high availability (it cannot be a number higher than the current Kafka Brokers number) and if the topic should be compacted.
A Kafka topic which is compacted is a special type of topic with a finer-grained retention mechanism that retains the last update record for each key.
A compacted topic (once the compaction has been completed) contains a full snapshot of the final record values for every record key and not just the recently changed keys. They are useful for in-memory services, persistent data stores, reloading caches, etc.
For more details on the subject, you should look at [Kafka Documentation](https://kafka.apache.org/documentation/#design_compactionbasics).
**Example**:
```sql
CREATE TABLE customer (
id STRING
, address.line STRING
, address.city STRING,
, address.postcode INT
, email STRING
)
FORMAT (string, json)
PROPERTIES (
partitions=1,
compacted=true
);
```
Best practices dictate to use Avro as a storage format over other formats. In this case, the key can still be stored as STRING but the value can be Avro.
```sql
CREATE TABLE customer_avro (
id STRING
, address.line STRING
, address.city STRING
, address.postcode int
, email string
)
FORMAT (string, avro)
PROPERTIES (
partitions=1,
compacted=true
);
```
### SHOW TABLES
To list all tables:
```sql
SHOW TABLES;
```
### DESCRIBE TABLE
To examine the schema an metadata for a topic:
```sql
DESCRIBE TABLE $tableName
```
The `$tableName` should contain the name of the table to describe.
Given the two tables created earlier, a user can run the following SQL to get the information on each table:
```sql
DESCRIBE TABLE customer_avro
```
the following information will be displayed:
```sql
/* Output:
# Column Name # Data Type
_key String
_value.address.postcode Int
_value.address.city String
_value.address.line String
_value.email String
_value.id String
# Config Key # Config Value
cleanup.policy compact
compression.type producer
delete.retention.ms 86400000
file.delete.delay.ms 60000
flush.messages 9223372036854775807
flush.ms 9223372036854775807
index.interval.bytes 4096
max.message.bytes 1000012
message.format.version 1.1-IV0
message.timestamp.difference.max.ms 9223372036854775807
message.timestamp.type CreateTime
min.cleanable.dirty.ratio 0.5
min.compaction.lag.ms 0
min.insync.replicas 1
preallocate false
retention.bytes 2147483648
retention.ms 604800000
segment.bytes 1073741824
segment.index.bytes 10485760
segment.jitter.ms 0
segment.ms 604800000
unclean.leader.election.enable false
*/
```
### DROP TABLE
To drop a table:
```sql
DROP TABLE $Table;
```
Dropping a table results in the underlying Kafka topics being removed.
### System virtual tables (\_\_tables, \_\_fields)
Lenses provides a set of virtual tables that contain information about all the fields in all the tables.
Using the virtual table, you can quickly search for a table name but also see the table type.
The `__table` has a `table_name` column containing the table name, and a `table_type` column describing the table type (system, user, etc).
```sql
SELECT * FROM __tables;
SELECT *
FROM __tables
WHERE table_name LIKE '%customer%';
```
To see all the tables fields select from the `_fields` virtual table.
```sql
SELECT *
FROM __fields;
SELECT *
FROM __fields
WHERE table_name LIKE '%customer%'
```
### Record metadata
Each Kafka message contains information related to partition, offset, timestamp, and topic. Additionally, the engine adds the key and value raw byte size.
Create a topic and insert a few entries.
```sql
CREATE TABLE tutorial(
_key STRING
, name STRING
, difficulty INT
)
FORMAT (Avro, Avro);
INSERT INTO tutorial(_key, name, difficulty)
VALUES
("1", "Learn Lenses SQL", 3),
("2", "Learn Quantum Physics", 10),
("3", "Learn French", 5);
```
Now we can query for specific metadata related to the records.
To query for metadata such as the underlying Kafka topic offset, partition and timestamp prefix your desired fields with `_meta`.
Run the following query to see each tutorial name along with its metadata information:
```sql
SELECT name
, _meta.offset
, _meta.timestamp
, _meta.partition
, _meta.__keysize
, _meta.__valsize
FROM tutorial
/* The output is (timestamp will be different)
Learn Lenses SQL 0 1540575169198 0 7 23
Learn Quantum Physics 1 1540575169226 0 7 28
Learn French 2 1540575169247 0 7 19
*/
```
---
# Agent Instructions: Querying This Documentation
If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.
Perform an HTTP GET request on the current page URL with the `ask` query parameter:
```
GET https://docs.lenses.io/latest/user-guide/using/using-sql-to-query-kafka/creating-and-deleting-kafka-topics.md?ask=
```
The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.
Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.