diff --git a/docs/src/main/sphinx/connector/mysql.md b/docs/src/main/sphinx/connector/mysql.md
index 0c9568b220d5..df445d5cc4cc 100644
--- a/docs/src/main/sphinx/connector/mysql.md
+++ b/docs/src/main/sphinx/connector/mysql.md
@@ -105,37 +105,6 @@ creates a catalog named `sales` using the configured connector.
 The connector supports {doc}`/admin/fault-tolerant-execution` of query
 processing. Read and write operations are both supported with any retry policy.
 
-## Table properties
-
-Table property usage example:
-
-```
-CREATE TABLE person (
-  id INT NOT NULL,
-  name VARCHAR,
-  age INT,
-  birthday DATE 
-)
-WITH (
-  primary_key = ARRAY['id']
-);
-```
-
-The following are supported MySQL table properties: 
-
-:::{list-table}
-:widths: 30, 10, 60
-:header-rows: 1
-
-* - Property name
-  - Required
-  - Description
-* - `primary_key`
-  - No
-  - The primary key of the table, can choose multi columns as the table primary key.
-    All key columns must be defined as `NOT NULL`.
-:::
-
 (mysql-type-mapping)=
 ## Type mapping
 
@@ -376,6 +345,7 @@ following features:
 - [](/sql/drop-table)
 - [](/sql/create-schema)
 - [](/sql/drop-schema)
+- [](mysql-schema-and-table-management)
 - [](mysql-procedures)
 - [](mysql-table-functions)
 
@@ -395,6 +365,73 @@ following features:
 ```{include} non-transactional-merge.fragment
 ```
 
+(mysql-schema-and-table-management)=
+### Schema and table management
+
+#### Table properties
+
+Table property usage example:
+
+```
+CREATE TABLE person (
+  id INT NOT NULL,
+  name VARCHAR,
+  age INT,
+  birthday DATE 
+)
+WITH (
+  primary_key = ARRAY['id']
+);
+```
+
+The following are supported MySQL table properties:
+
+:::{list-table}
+:widths: 30, 10, 60
+:header-rows: 1
+
+* - Property name
+  - Required
+  - Description
+* - `primary_key`
+  - No
+  - The primary key of the table, can choose multi columns as the table primary key.
+    All key columns must be defined as `NOT NULL`.
+:::
+
+#### Column properties
+
+Column property usage example:
+
+```
+CREATE TABLE person (
+  id INT NOT NULL WITH (auto_increment = true),
+  name VARCHAR,
+  age INT,
+  birthday DATE
+)
+WITH (
+  primary_key = ARRAY['id']
+);
+```
+
+The following are supported MySQL column properties:
+
+:::{list-table}
+:widths: 30, 10, 60
+:header-rows: 1
+
+* - Property name
+  - Required
+  - Description
+* - `auto_increment`
+  - No
+  - Auto generate a unique identity for new rows. There can be only one auto increment column
+    and must be defined as the first key. Only applies to integer types (`TINYINT`,
+    `SMALLINT`, `INTEGER`, `BIGINT`) column. The value must be set to `true`. If set to `false`,
+    it will throw an exception.
+:::
+
 (mysql-procedures)=
 ### Procedures
 
diff --git a/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/BaseJdbcClient.java b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/BaseJdbcClient.java
index 49577f824f50..a717457d306f 100644
--- a/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/BaseJdbcClient.java
+++ b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/BaseJdbcClient.java
@@ -336,6 +336,7 @@ public List<JdbcColumnHandle> getColumns(ConnectorSession session, SchemaTableNa
                 boolean nullable = (resultSet.getInt("NULLABLE") != columnNoNulls);
                 // Note: some databases (e.g. SQL Server) do not return column remarks/comment here.
                 Optional<String> comment = Optional.ofNullable(emptyToNull(resultSet.getString("REMARKS")));
+                Map<String, Object> columnProperties = getColumnProperties(resultSet);
                 // skip unsupported column types
                 columnMapping.ifPresent(mapping -> columns.add(JdbcColumnHandle.builder()
                         .setColumnName(columnName)
@@ -343,6 +344,7 @@ public List<JdbcColumnHandle> getColumns(ConnectorSession session, SchemaTableNa
                         .setColumnType(mapping.getType())
                         .setNullable(nullable)
                         .setComment(comment)
+                        .setColumnProperties(columnProperties)
                         .build()));
                 if (columnMapping.isEmpty()) {
                     UnsupportedTypeHandling unsupportedTypeHandling = getUnsupportedTypeHandling(session);
@@ -366,6 +368,16 @@ public List<JdbcColumnHandle> getColumns(ConnectorSession session, SchemaTableNa
         }
     }
 
+    /**
+     * Extract column property from DatabaseMetaData.
+     * The default implementation returns an empty map and each connector needs to provide its own implementation.
+     */
+    public Map<String, Object> getColumnProperties(ResultSet resultSet)
+            throws SQLException
+    {
+        return ImmutableMap.of();
+    }
+
     @Override
     public Iterator<RelationColumnsMetadata> getAllTableColumns(ConnectorSession session, Optional<String> schema)
     {
@@ -1285,7 +1297,7 @@ protected Map<Integer, JdbcOutputTableHandle> beginUpdate(
             }
 
             List<ColumnMetadata> columnMetadata = updatedColumnHandles.build().stream()
-                    .map(JdbcColumnHandle::getColumnMetadata)
+                    .map(column -> toColumnMetadata(column))
                     .collect(toImmutableList());
 
             JdbcOutputTableHandle temporaryTableHandle = createTable(
diff --git a/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/CachingJdbcClient.java b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/CachingJdbcClient.java
index ef559f6a0dad..6b514cda4bc0 100644
--- a/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/CachingJdbcClient.java
+++ b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/CachingJdbcClient.java
@@ -652,6 +652,12 @@ public List<JdbcColumnHandle> getPrimaryKeys(ConnectorSession session, RemoteTab
         return get(tablePrimaryKeysCache, remoteTableName, () -> delegate.getPrimaryKeys(session, remoteTableName));
     }
 
+    @Override
+    public ColumnMetadata toColumnMetadata(JdbcColumnHandle columnHandle)
+    {
+        return delegate.toColumnMetadata(columnHandle);
+    }
+
     public void onDataChanged(SchemaTableName table)
     {
         invalidateAllIf(statisticsCache, key -> key.mayReference(table));
diff --git a/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/ColumnPropertiesProvider.java b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/ColumnPropertiesProvider.java
new file mode 100644
index 000000000000..63819d07958f
--- /dev/null
+++ b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/ColumnPropertiesProvider.java
@@ -0,0 +1,23 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.trino.plugin.jdbc;
+
+import io.trino.spi.session.PropertyMetadata;
+
+import java.util.List;
+
+public interface ColumnPropertiesProvider
+{
+    List<PropertyMetadata<?>> getColumnProperties();
+}
diff --git a/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/DefaultJdbcMetadata.java b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/DefaultJdbcMetadata.java
index 2ab2b5b772f3..51bba8898fce 100644
--- a/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/DefaultJdbcMetadata.java
+++ b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/DefaultJdbcMetadata.java
@@ -1056,7 +1056,7 @@ public ConnectorTableMetadata getTableMetadata(ConnectorSession session, Connect
         return new ConnectorTableMetadata(
                 schemaTableName,
                 jdbcClient.getColumns(session, schemaTableName, remoteTableName).stream()
-                        .map(JdbcColumnHandle::getColumnMetadata)
+                        .map(column -> jdbcClient.toColumnMetadata(column))
                         .collect(toImmutableList()),
                 jdbcClient.getTableProperties(session, handle),
                 getTableComment(handle));
@@ -1078,11 +1078,11 @@ public Map<String, ColumnHandle> getColumnHandles(ConnectorSession session, Conn
     {
         if (tableHandle instanceof JdbcProcedureHandle procedureHandle) {
             return procedureHandle.getColumns().orElseThrow().stream()
-                    .collect(toImmutableMap(columnHandle -> columnHandle.getColumnMetadata().getName(), identity()));
+                    .collect(toImmutableMap(columnHandle -> jdbcClient.toColumnMetadata(columnHandle).getName(), identity()));
         }
 
         return getColumns(session, jdbcClient, (JdbcTableHandle) tableHandle).stream()
-                .collect(toImmutableMap(columnHandle -> columnHandle.getColumnMetadata().getName(), identity()));
+                .collect(toImmutableMap(columnHandle -> jdbcClient.toColumnMetadata(columnHandle).getName(), identity()));
     }
 
     @Override
@@ -1110,7 +1110,7 @@ public List<ColumnMetadata> getColumnMetadata(ConnectorSession session, JdbcTabl
         return getColumnHandles(session, tableHandle).values()
                 .stream()
                 .map(JdbcColumnHandle.class::cast)
-                .map(JdbcColumnHandle::getColumnMetadata)
+                .map(column -> jdbcClient.toColumnMetadata(column))
                 .collect(toImmutableList());
     }
 
@@ -1140,7 +1140,7 @@ public Iterator<RelationCommentMetadata> streamRelationComments(ConnectorSession
     @Override
     public ColumnMetadata getColumnMetadata(ConnectorSession session, ConnectorTableHandle tableHandle, ColumnHandle columnHandle)
     {
-        return ((JdbcColumnHandle) columnHandle).getColumnMetadata();
+        return jdbcClient.toColumnMetadata((JdbcColumnHandle) columnHandle);
     }
 
     @Override
diff --git a/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/ForwardingJdbcClient.java b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/ForwardingJdbcClient.java
index 2f0ae837f095..4e3416c64541 100644
--- a/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/ForwardingJdbcClient.java
+++ b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/ForwardingJdbcClient.java
@@ -535,4 +535,10 @@ public List<JdbcColumnHandle> getPrimaryKeys(ConnectorSession session, RemoteTab
     {
         return delegate().getPrimaryKeys(session, remoteTableName);
     }
+
+    @Override
+    public ColumnMetadata toColumnMetadata(JdbcColumnHandle columnHandle)
+    {
+        return delegate().toColumnMetadata(columnHandle);
+    }
 }
diff --git a/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/JdbcClient.java b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/JdbcClient.java
index e0287046f2b8..93260da96fb0 100644
--- a/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/JdbcClient.java
+++ b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/JdbcClient.java
@@ -294,4 +294,9 @@ default List<JdbcColumnHandle> getPrimaryKeys(ConnectorSession session, RemoteTa
     {
         return List.of();
     }
+
+    default ColumnMetadata toColumnMetadata(JdbcColumnHandle columnHandle)
+    {
+        return columnHandle.getColumnMetadata();
+    }
 }
diff --git a/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/JdbcColumnHandle.java b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/JdbcColumnHandle.java
index 0696f6fdae09..830a528b82f9 100644
--- a/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/JdbcColumnHandle.java
+++ b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/JdbcColumnHandle.java
@@ -16,18 +16,23 @@
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.google.common.base.Joiner;
+import com.google.common.collect.ImmutableMap;
 import io.airlift.slice.SizeOf;
+import io.trino.spi.TrinoException;
 import io.trino.spi.connector.ColumnHandle;
 import io.trino.spi.connector.ColumnMetadata;
 import io.trino.spi.connector.ColumnSchema;
 import io.trino.spi.type.Type;
 
+import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 
 import static io.airlift.slice.SizeOf.estimatedSizeOf;
 import static io.airlift.slice.SizeOf.instanceSize;
 import static io.airlift.slice.SizeOf.sizeOf;
+import static io.trino.spi.StandardErrorCode.INVALID_COLUMN_PROPERTY;
 import static java.util.Objects.requireNonNull;
 
 public final class JdbcColumnHandle
@@ -40,11 +45,12 @@ public final class JdbcColumnHandle
     private final Type columnType;
     private final boolean nullable;
     private final Optional<String> comment;
+    private final Map<String, Object> columnProperties;
 
     // All and only required fields
     public JdbcColumnHandle(String columnName, JdbcTypeHandle jdbcTypeHandle, Type columnType)
     {
-        this(columnName, jdbcTypeHandle, columnType, true, Optional.empty());
+        this(columnName, jdbcTypeHandle, columnType, true, Optional.empty(), ImmutableMap.of());
     }
 
     /**
@@ -57,13 +63,15 @@ public JdbcColumnHandle(
             @JsonProperty("jdbcTypeHandle") JdbcTypeHandle jdbcTypeHandle,
             @JsonProperty("columnType") Type columnType,
             @JsonProperty("nullable") boolean nullable,
-            @JsonProperty("comment") Optional<String> comment)
+            @JsonProperty("comment") Optional<String> comment,
+            @JsonProperty("columnProperties") Map<String, Object> columnProperties)
     {
         this.columnName = requireNonNull(columnName, "columnName is null");
         this.jdbcTypeHandle = requireNonNull(jdbcTypeHandle, "jdbcTypeHandle is null");
         this.columnType = requireNonNull(columnType, "columnType is null");
         this.nullable = nullable;
         this.comment = requireNonNull(comment, "comment is null");
+        this.columnProperties = ImmutableMap.copyOf(verifyColumnProperties(columnProperties));
     }
 
     @JsonProperty
@@ -96,6 +104,12 @@ public Optional<String> getComment()
         return comment;
     }
 
+    @JsonProperty
+    public Map<String, Object> getColumnProperties()
+    {
+        return columnProperties;
+    }
+
     public ColumnMetadata getColumnMetadata()
     {
         return ColumnMetadata.builder()
@@ -149,7 +163,42 @@ public long getRetainedSizeInBytes()
                 + sizeOf(nullable)
                 + estimatedSizeOf(columnName)
                 + sizeOf(comment, SizeOf::estimatedSizeOf)
-                + jdbcTypeHandle.getRetainedSizeInBytes();
+                + jdbcTypeHandle.getRetainedSizeInBytes()
+                + estimatedSizeOf(columnProperties, SizeOf::estimatedSizeOf, JdbcColumnHandle::estimatedObjectSizeOf);
+    }
+
+    private static long estimatedObjectSizeOf(Object value)
+    {
+        return switch (value) {
+            case String stringValue -> estimatedSizeOf(stringValue);
+            case Boolean _, Integer _, Long _, Double _ -> instanceSize(value.getClass());
+            case List<?> list -> estimatedSizeOf(list, JdbcColumnHandle::estimatedObjectSizeOf);
+            case Map<?, ?> map -> estimatedSizeOf(map, JdbcColumnHandle::estimatedObjectSizeOf, JdbcColumnHandle::estimatedObjectSizeOf);
+            default -> throw new TrinoException(INVALID_COLUMN_PROPERTY, "Unsupported property value type: " + value.getClass().getName());
+        };
+    }
+
+    private static Map<String, Object> verifyColumnProperties(Map<String, Object> columnProperties)
+    {
+        requireNonNull(columnProperties, "columnProperties is null");
+        columnProperties.values().forEach(JdbcColumnHandle::verifyPropertyValueType);
+        return columnProperties;
+    }
+
+    private static void verifyPropertyValueType(Object value)
+    {
+        switch (value) {
+            case String _, Boolean _, Integer _, Long _, Double _ -> {
+                return;
+            }
+            case List<?> list -> list.stream().forEach(JdbcColumnHandle::verifyPropertyValueType);
+            case Map<?, ?> map -> map.entrySet().stream().forEach(entry -> {
+                verifyPropertyValueType(entry.getKey());
+                verifyPropertyValueType(entry.getValue());
+            });
+            case null -> throw new TrinoException(INVALID_COLUMN_PROPERTY, "Property value is null");
+            default -> throw new TrinoException(INVALID_COLUMN_PROPERTY, "Unsupported property value type: " + value.getClass().getName());
+        }
     }
 
     public static Builder builder()
@@ -169,6 +218,7 @@ public static final class Builder
         private Type columnType;
         private boolean nullable = true;
         private Optional<String> comment = Optional.empty();
+        private Map<String, Object> columnProperties = ImmutableMap.of();
 
         public Builder() {}
 
@@ -179,6 +229,7 @@ private Builder(JdbcColumnHandle handle)
             this.columnType = handle.getColumnType();
             this.nullable = handle.isNullable();
             this.comment = handle.getComment();
+            this.columnProperties = handle.getColumnProperties();
         }
 
         public Builder setColumnName(String columnName)
@@ -211,6 +262,12 @@ public Builder setComment(Optional<String> comment)
             return this;
         }
 
+        public Builder setColumnProperties(Map<String, Object> columnProperties)
+        {
+            this.columnProperties = columnProperties;
+            return this;
+        }
+
         public JdbcColumnHandle build()
         {
             return new JdbcColumnHandle(
@@ -218,7 +275,8 @@ public JdbcColumnHandle build()
                     jdbcTypeHandle,
                     columnType,
                     nullable,
-                    comment);
+                    comment,
+                    columnProperties);
         }
     }
 }
diff --git a/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/JdbcConnector.java b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/JdbcConnector.java
index 59a45526b8e7..7733ddb1b294 100644
--- a/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/JdbcConnector.java
+++ b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/JdbcConnector.java
@@ -53,6 +53,7 @@ public class JdbcConnector
     private final Set<ConnectorTableFunction> connectorTableFunctions;
     private final List<PropertyMetadata<?>> sessionProperties;
     private final List<PropertyMetadata<?>> tableProperties;
+    private final List<PropertyMetadata<?>> columnProperties;
     private final JdbcTransactionManager transactionManager;
 
     @Inject
@@ -66,6 +67,7 @@ public JdbcConnector(
             Set<ConnectorTableFunction> connectorTableFunctions,
             Set<SessionPropertiesProvider> sessionProperties,
             Set<TablePropertiesProvider> tableProperties,
+            Set<ColumnPropertiesProvider> columnProperties,
             JdbcTransactionManager transactionManager)
     {
         this.lifeCycleManager = requireNonNull(lifeCycleManager, "lifeCycleManager is null");
@@ -81,6 +83,9 @@ public JdbcConnector(
         this.tableProperties = tableProperties.stream()
                 .flatMap(tablePropertiesProvider -> tablePropertiesProvider.getTableProperties().stream())
                 .collect(toImmutableList());
+        this.columnProperties = requireNonNull(columnProperties, "columnProperties is null").stream()
+                .flatMap(columnPropertiesProvider -> columnPropertiesProvider.getColumnProperties().stream())
+                .collect(toImmutableList());
         this.transactionManager = requireNonNull(transactionManager, "transactionManager is null");
     }
 
@@ -156,6 +161,12 @@ public List<PropertyMetadata<?>> getTableProperties()
         return tableProperties;
     }
 
+    @Override
+    public List<PropertyMetadata<?>> getColumnProperties()
+    {
+        return columnProperties;
+    }
+
     @Override
     public final void shutdown()
     {
diff --git a/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/JdbcModule.java b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/JdbcModule.java
index 140caed15e96..24597b63827a 100644
--- a/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/JdbcModule.java
+++ b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/JdbcModule.java
@@ -62,6 +62,7 @@ public void setup(Binder binder)
 
         procedureBinder(binder);
         tablePropertiesProviderBinder(binder);
+        columnPropertiesProviderBinder(binder);
 
         newOptionalBinder(binder, JdbcMetadataFactory.class).setDefault().to(DefaultJdbcMetadataFactory.class).in(Scopes.SINGLETON);
         newOptionalBinder(binder, IdentityCacheMapping.class).setDefault().to(SingletonIdentityCacheMapping.class).in(Scopes.SINGLETON);
@@ -146,4 +147,14 @@ public static void bindTablePropertiesProvider(Binder binder, Class<? extends Ta
     {
         tablePropertiesProviderBinder(binder).addBinding().to(type).in(Scopes.SINGLETON);
     }
+
+    public static Multibinder<ColumnPropertiesProvider> columnPropertiesProviderBinder(Binder binder)
+    {
+        return newSetBinder(binder, ColumnPropertiesProvider.class);
+    }
+
+    public static void bindColumnPropertiesProvider(Binder binder, Class<? extends ColumnPropertiesProvider> type)
+    {
+        columnPropertiesProviderBinder(binder).addBinding().to(type).in(Scopes.SINGLETON);
+    }
 }
diff --git a/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/RetryingJdbcClient.java b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/RetryingJdbcClient.java
index fa695fb4485c..014e83bb1fa7 100644
--- a/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/RetryingJdbcClient.java
+++ b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/RetryingJdbcClient.java
@@ -563,4 +563,10 @@ public List<JdbcColumnHandle> getPrimaryKeys(ConnectorSession session, RemoteTab
     {
         return retry(policy, () -> delegate.getPrimaryKeys(session, remoteTableName));
     }
+
+    @Override
+    public ColumnMetadata toColumnMetadata(JdbcColumnHandle columnHandle)
+    {
+        return delegate.toColumnMetadata(columnHandle);
+    }
 }
diff --git a/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/jmx/StatisticsAwareJdbcClient.java b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/jmx/StatisticsAwareJdbcClient.java
index d522fa8b3cf7..c90cfb39bde6 100644
--- a/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/jmx/StatisticsAwareJdbcClient.java
+++ b/plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/jmx/StatisticsAwareJdbcClient.java
@@ -556,4 +556,10 @@ public List<JdbcColumnHandle> getPrimaryKeys(ConnectorSession session, RemoteTab
     {
         return stats.getGetPrimaryKeys().wrap(() -> delegate().getPrimaryKeys(session, remoteTableName));
     }
+
+    @Override
+    public ColumnMetadata toColumnMetadata(JdbcColumnHandle columnHandle)
+    {
+        return delegate().toColumnMetadata(columnHandle);
+    }
 }
diff --git a/plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/TestCachingJdbcClient.java b/plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/TestCachingJdbcClient.java
index 42b35f1cb087..a59a356cc2e8 100644
--- a/plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/TestCachingJdbcClient.java
+++ b/plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/TestCachingJdbcClient.java
@@ -1096,7 +1096,7 @@ private JdbcColumnHandle addColumn(JdbcClient client, JdbcTableHandle tableHandl
         client.addColumn(SESSION, tableHandle, columnMetadata, new ColumnPosition.Last());
         return getColumns(SESSION, client, tableHandle)
                 .stream()
-                .filter(jdbcColumnHandle -> jdbcColumnHandle.getColumnMetadata().equals(columnMetadata))
+                .filter(jdbcColumnHandle -> jdbcClient.toColumnMetadata(jdbcColumnHandle).equals(columnMetadata))
                 .collect(onlyElement());
     }
 
diff --git a/plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/TestJdbcColumnHandle.java b/plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/TestJdbcColumnHandle.java
index dffac4e56588..9914c4cf60c6 100644
--- a/plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/TestJdbcColumnHandle.java
+++ b/plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/TestJdbcColumnHandle.java
@@ -13,9 +13,13 @@
  */
 package io.trino.plugin.jdbc;
 
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
 import io.airlift.testing.EquivalenceTester;
 import org.junit.jupiter.api.Test;
 
+import java.util.HashMap;
+import java.util.Map;
 import java.util.Optional;
 
 import static io.trino.plugin.jdbc.MetadataUtil.COLUMN_CODEC;
@@ -24,6 +28,8 @@
 import static io.trino.plugin.jdbc.TestingJdbcTypeHandle.JDBC_VARCHAR;
 import static io.trino.spi.type.BigintType.BIGINT;
 import static io.trino.spi.type.VarcharType.VARCHAR;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 public class TestJdbcColumnHandle
 {
@@ -56,4 +62,49 @@ public void testEquivalence()
                         new JdbcColumnHandle("columnNameX", JDBC_VARCHAR, VARCHAR))
                 .check();
     }
+
+    @Test
+    public void testValidColumnPropertyType()
+    {
+        JdbcColumnHandle.Builder builder = generateBaseColumnHandleBuilder();
+        ImmutableMap.Builder<String, Object> columnPropertiesBuilder = ImmutableMap.builder();
+        columnPropertiesBuilder.put("int_property", 1);
+        columnPropertiesBuilder.put("boolean_property", true);
+        columnPropertiesBuilder.put("long_property", 1L);
+        columnPropertiesBuilder.put("double_property", 1.0);
+        columnPropertiesBuilder.put("string_property", "string");
+        columnPropertiesBuilder.put("list_properties", ImmutableList.of(1, 2, 3));
+        columnPropertiesBuilder.put("map_properties", ImmutableMap.of("key", "value"));
+        ImmutableMap<String, Object> validColumnProperties = columnPropertiesBuilder.buildOrThrow();
+        builder.setColumnProperties(validColumnProperties);
+        JdbcColumnHandle jdbcColumnHandle = builder.build();
+
+        assertThat(jdbcColumnHandle.getColumnProperties()).hasSize(7);
+        assertThat(jdbcColumnHandle.getColumnProperties()).containsAllEntriesOf(validColumnProperties);
+    }
+
+    @Test
+    public void testInvalidColumnPropertyType()
+    {
+        JdbcColumnHandle.Builder nullValueBuilder = generateBaseColumnHandleBuilder();
+        Map<String, Object> nullColumnProperties = new HashMap<>();
+        nullColumnProperties.put("null_property", null);
+        nullValueBuilder.setColumnProperties(nullColumnProperties);
+        assertThatThrownBy(() -> nullValueBuilder.build())
+                .hasMessageContaining("Property value is null");
+
+        JdbcColumnHandle.Builder mapValueBuilder = generateBaseColumnHandleBuilder();
+        mapValueBuilder.setColumnProperties(ImmutableMap.of("invalid_property", Optional.empty()));
+        assertThatThrownBy(() -> mapValueBuilder.build())
+                .hasMessageContaining("Unsupported property value type: java.util.Optional");
+    }
+
+    private JdbcColumnHandle.Builder generateBaseColumnHandleBuilder()
+    {
+        JdbcColumnHandle.Builder columnBuilder = JdbcColumnHandle.builder();
+        columnBuilder.setColumnName("columnName");
+        columnBuilder.setColumnType(VARCHAR);
+        columnBuilder.setJdbcTypeHandle(JDBC_VARCHAR);
+        return columnBuilder;
+    }
 }
diff --git a/plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/TestingDatabase.java b/plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/TestingDatabase.java
index 46a0e28f06a6..1aa032638c68 100644
--- a/plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/TestingDatabase.java
+++ b/plugin/trino-base-jdbc/src/test/java/io/trino/plugin/jdbc/TestingDatabase.java
@@ -109,6 +109,6 @@ public JdbcSplit getSplit(ConnectorSession session, JdbcTableHandle table)
     public Map<String, JdbcColumnHandle> getColumnHandles(ConnectorSession session, JdbcTableHandle table)
     {
         return jdbcClient.getColumns(session, table.getRequiredNamedRelation().getSchemaTableName(), table.getRequiredNamedRelation().getRemoteTableName()).stream()
-                .collect(toImmutableMap(column -> column.getColumnMetadata().getName(), identity()));
+                .collect(toImmutableMap(column -> jdbcClient.toColumnMetadata(column).getName(), identity()));
     }
 }
diff --git a/plugin/trino-ignite/src/main/java/io/trino/plugin/ignite/IgniteMetadata.java b/plugin/trino-ignite/src/main/java/io/trino/plugin/ignite/IgniteMetadata.java
index 9b015ba15222..58937373772b 100644
--- a/plugin/trino-ignite/src/main/java/io/trino/plugin/ignite/IgniteMetadata.java
+++ b/plugin/trino-ignite/src/main/java/io/trino/plugin/ignite/IgniteMetadata.java
@@ -199,7 +199,7 @@ public List<ColumnMetadata> getColumnMetadata(ConnectorSession session, JdbcTabl
     {
         return getColumns(session, igniteClient, handle).stream()
                 .filter(column -> !IGNITE_DUMMY_ID.equalsIgnoreCase(column.getColumnName()))
-                .map(JdbcColumnHandle::getColumnMetadata)
+                .map(column -> igniteClient.toColumnMetadata(column))
                 .collect(toImmutableList());
     }
 
diff --git a/plugin/trino-mysql/src/main/java/io/trino/plugin/mysql/MySqlClient.java b/plugin/trino-mysql/src/main/java/io/trino/plugin/mysql/MySqlClient.java
index c9519b799587..78475eeb9065 100644
--- a/plugin/trino-mysql/src/main/java/io/trino/plugin/mysql/MySqlClient.java
+++ b/plugin/trino-mysql/src/main/java/io/trino/plugin/mysql/MySqlClient.java
@@ -83,14 +83,18 @@
 import io.trino.spi.statistics.ColumnStatistics;
 import io.trino.spi.statistics.Estimate;
 import io.trino.spi.statistics.TableStatistics;
+import io.trino.spi.type.BigintType;
 import io.trino.spi.type.CharType;
 import io.trino.spi.type.DecimalType;
 import io.trino.spi.type.Decimals;
+import io.trino.spi.type.IntegerType;
 import io.trino.spi.type.LongTimestampWithTimeZone;
+import io.trino.spi.type.SmallintType;
 import io.trino.spi.type.StandardTypes;
 import io.trino.spi.type.TimeType;
 import io.trino.spi.type.TimestampType;
 import io.trino.spi.type.TimestampWithTimeZoneType;
+import io.trino.spi.type.TinyintType;
 import io.trino.spi.type.Type;
 import io.trino.spi.type.TypeManager;
 import io.trino.spi.type.TypeSignature;
@@ -114,6 +118,7 @@
 import java.time.OffsetDateTime;
 import java.util.AbstractMap.SimpleEntry;
 import java.util.Collection;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -182,6 +187,7 @@
 import static io.trino.plugin.jdbc.StandardColumnMappings.varcharWriteFunction;
 import static io.trino.plugin.jdbc.TypeHandlingJdbcSessionProperties.getUnsupportedTypeHandling;
 import static io.trino.plugin.jdbc.UnsupportedTypeHandling.CONVERT_TO_VARCHAR;
+import static io.trino.plugin.mysql.MySqlColumnProperties.AUTO_INCREMENT;
 import static io.trino.plugin.mysql.MySqlTableProperties.PRIMARY_KEY_PROPERTY;
 import static io.trino.spi.StandardErrorCode.ALREADY_EXISTS;
 import static io.trino.spi.StandardErrorCode.INVALID_TABLE_PROPERTY;
@@ -487,15 +493,35 @@ protected List<String> createTableSqls(RemoteTableName remoteTableName, List<Str
         columnDefinitions.addAll(columns);
 
         List<String> primaryKeys = MySqlTableProperties.getPrimaryKey(tableMetadata.getProperties());
+        verifyColumnMetadata(primaryKeys, tableMetadata.getColumns());
         if (!primaryKeys.isEmpty()) {
-            verifyPrimaryKey(primaryKeys, tableMetadata.getColumns());
             columnDefinitions.add("PRIMARY KEY (" + primaryKeys.stream().map(this::quoted).collect(joining(", ")) + ")");
         }
         return ImmutableList.of(format("CREATE TABLE %s (%s) COMMENT %s", quoted(remoteTableName), join(", ", columnDefinitions.build()), mysqlVarcharLiteral(tableMetadata.getComment().orElse(NO_COMMENT))));
     }
 
-    private static void verifyPrimaryKey(List<String> primaryKeys, List<ColumnMetadata> columns)
+    private static void verifyColumnMetadata(List<String> primaryKeys, List<ColumnMetadata> columns)
     {
+        Set<String> autoIncrementColumnNames = columns.stream()
+                .filter(column -> column.getProperties().containsKey(AUTO_INCREMENT))
+                .map(column -> {
+                    if (!(boolean) column.getProperties().get(AUTO_INCREMENT)) {
+                        throw new TrinoException(NOT_SUPPORTED, "Column property value for AUTO_INCREMENT must be true");
+                    }
+                    return column.getName();
+                })
+                .collect(toImmutableSet());
+
+        if (autoIncrementColumnNames.size() > 1) {
+            throw new TrinoException(NOT_SUPPORTED, "There can be only one auto increment column in MySQL");
+        }
+
+        if (autoIncrementColumnNames.size() == 1) {
+            if (primaryKeys.isEmpty() || !autoIncrementColumnNames.contains(primaryKeys.getFirst())) {
+                throw new TrinoException(NOT_SUPPORTED, "Auto increment column must be defined as the first key in MySQL");
+            }
+        }
+
         Set<String> columnNames = columns.stream()
                 .map(column -> {
                     String columnName = column.getName();
@@ -521,10 +547,23 @@ protected String getColumnDefinitionSql(ConnectorSession session, ColumnMetadata
             throw new TrinoException(NOT_SUPPORTED, "This connector does not support creating tables with column comment");
         }
 
-        return "%s %s %s".formatted(
+        return "%s %s %s %s".formatted(
                 quoted(columnName),
                 toWriteMapping(session, column.getType()).getDataType(),
-                column.isNullable() ? "NULL" : "NOT NULL");
+                column.isNullable() ? "NULL" : "NOT NULL",
+                isAutoIncrement(column) ? "AUTO_INCREMENT" : "");
+    }
+
+    private static boolean isAutoIncrement(ColumnMetadata column)
+    {
+        boolean isAutoIncrement = (boolean) column.getProperties().getOrDefault(AUTO_INCREMENT, false);
+        if (isAutoIncrement) {
+            Type type = column.getType();
+            if (!(type instanceof TinyintType || type instanceof SmallintType || type instanceof IntegerType || type instanceof BigintType)) {
+                throw new TrinoException(NOT_SUPPORTED, "Unsupported column type for AUTO_INCREMENT: " + type);
+            }
+        }
+        return isAutoIncrement;
     }
 
     private static String mysqlVarcharLiteral(String value)
@@ -644,6 +683,18 @@ public Optional<ColumnMapping> toColumnMapping(ConnectorSession session, Connect
         return Optional.empty();
     }
 
+    @Override
+    public Map<String, Object> getColumnProperties(ResultSet resultSet)
+            throws SQLException
+    {
+        ImmutableMap.Builder<String, Object> columnPropertiesBuilder = ImmutableMap.builder();
+        boolean autoIncrement = "YES".equals(resultSet.getString("IS_AUTOINCREMENT"));
+        if (autoIncrement) {
+            columnPropertiesBuilder.put(AUTO_INCREMENT, autoIncrement);
+        }
+        return columnPropertiesBuilder.buildOrThrow();
+    }
+
     private static ColumnMapping mySqlDefaultVarcharColumnMapping(int columnSize, Optional<CaseSensitivity> caseSensitivity)
     {
         if (columnSize > VarcharType.MAX_LENGTH) {
@@ -963,6 +1014,10 @@ private void addColumn(ConnectorSession session, RemoteTableName table, ColumnMe
             throw new TrinoException(NOT_SUPPORTED, "This connector does not support adding columns with comments");
         }
 
+        if (!column.getProperties().isEmpty()) {
+            throw new TrinoException(NOT_SUPPORTED, "This connector does not support adding columns with column properties");
+        }
+
         try (Connection connection = connectionFactory.openConnection(session)) {
             verify(connection.getAutoCommit());
             String columnName = column.getName();
@@ -1320,6 +1375,36 @@ public List<JdbcColumnHandle> getPrimaryKeys(ConnectorSession session, RemoteTab
         }
     }
 
+    @Override
+    public ColumnMetadata toColumnMetadata(JdbcColumnHandle handle)
+    {
+        ColumnMetadata columnMetadata = handle.getColumnMetadata();
+        ColumnMetadata.Builder columnMetadataBuilder = ColumnMetadata.builderFrom(columnMetadata);
+
+        Map<String, Object> properties = new LinkedHashMap<>(columnMetadata.getProperties());
+
+        StringBuilder extraBuilder = new StringBuilder();
+        String extraInfo = columnMetadata.getExtraInfo();
+        if (extraInfo != null && !extraInfo.isEmpty()) {
+            extraBuilder.append(extraInfo);
+        }
+
+        if ((boolean) handle.getColumnProperties().getOrDefault(AUTO_INCREMENT, false)) {
+            properties.put(AUTO_INCREMENT, true);
+            if (!extraBuilder.isEmpty()) {
+                extraBuilder.append(", ");
+            }
+            extraBuilder.append(AUTO_INCREMENT).append("=true");
+        }
+
+        columnMetadataBuilder.setProperties(properties);
+
+        if (!extraBuilder.isEmpty()) {
+            columnMetadataBuilder.setExtraInfo(Optional.of(extraBuilder.toString()));
+        }
+        return columnMetadataBuilder.build();
+    }
+
     private static Optional<ColumnHistogram> getColumnHistogram(Map<String, String> columnHistograms, String columnName)
     {
         return Optional.ofNullable(columnHistograms.get(columnName))
diff --git a/plugin/trino-mysql/src/main/java/io/trino/plugin/mysql/MySqlClientModule.java b/plugin/trino-mysql/src/main/java/io/trino/plugin/mysql/MySqlClientModule.java
index 90dbefa3138a..02defd85f60a 100644
--- a/plugin/trino-mysql/src/main/java/io/trino/plugin/mysql/MySqlClientModule.java
+++ b/plugin/trino-mysql/src/main/java/io/trino/plugin/mysql/MySqlClientModule.java
@@ -40,6 +40,7 @@
 import static com.google.inject.multibindings.Multibinder.newSetBinder;
 import static com.google.inject.multibindings.OptionalBinder.newOptionalBinder;
 import static io.airlift.configuration.ConfigBinder.configBinder;
+import static io.trino.plugin.jdbc.JdbcModule.bindColumnPropertiesProvider;
 import static io.trino.plugin.jdbc.JdbcModule.bindTablePropertiesProvider;
 
 public class MySqlClientModule
@@ -55,6 +56,7 @@ protected void setup(Binder binder)
         configBinder(binder).bindConfig(MySqlConfig.class);
         configBinder(binder).bindConfig(JdbcStatisticsConfig.class);
         bindTablePropertiesProvider(binder, MySqlTableProperties.class);
+        bindColumnPropertiesProvider(binder, MySqlColumnProperties.class);
         install(new DecimalModule());
         install(new JdbcJoinPushdownSupportModule());
         newSetBinder(binder, ConnectorTableFunction.class).addBinding().toProvider(Query.class).in(Scopes.SINGLETON);
diff --git a/plugin/trino-mysql/src/main/java/io/trino/plugin/mysql/MySqlColumnProperties.java b/plugin/trino-mysql/src/main/java/io/trino/plugin/mysql/MySqlColumnProperties.java
new file mode 100644
index 000000000000..1c44c1d97ca1
--- /dev/null
+++ b/plugin/trino-mysql/src/main/java/io/trino/plugin/mysql/MySqlColumnProperties.java
@@ -0,0 +1,49 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.trino.plugin.mysql;
+
+import com.google.common.collect.ImmutableList;
+import com.google.inject.Inject;
+import io.trino.plugin.jdbc.ColumnPropertiesProvider;
+import io.trino.spi.session.PropertyMetadata;
+
+import java.util.List;
+
+import static io.trino.spi.session.PropertyMetadata.booleanProperty;
+
+public final class MySqlColumnProperties
+        implements ColumnPropertiesProvider
+{
+    public static final String AUTO_INCREMENT = "auto_increment";
+
+    private final List<PropertyMetadata<?>> columnProperties;
+
+    @Inject
+    public MySqlColumnProperties()
+    {
+        columnProperties = ImmutableList.<PropertyMetadata<?>>builder()
+                .add(booleanProperty(
+                        AUTO_INCREMENT,
+                        "Auto generate a unique identity for new rows",
+                        null,
+                        false))
+                .build();
+    }
+
+    @Override
+    public List<PropertyMetadata<?>> getColumnProperties()
+    {
+        return columnProperties;
+    }
+}
diff --git a/plugin/trino-mysql/src/test/java/io/trino/plugin/mysql/BaseMySqlConnectorTest.java b/plugin/trino-mysql/src/test/java/io/trino/plugin/mysql/BaseMySqlConnectorTest.java
index 910666787f0e..2ef489c3feb0 100644
--- a/plugin/trino-mysql/src/test/java/io/trino/plugin/mysql/BaseMySqlConnectorTest.java
+++ b/plugin/trino-mysql/src/test/java/io/trino/plugin/mysql/BaseMySqlConnectorTest.java
@@ -259,7 +259,7 @@ public void testCreateTableWithPrimaryKey()
 
     private void verifyCreateTableDefinition(String tableDefinition, String showCreateTableFormat)
     {
-        try (TestTable table = newTrinoTable("test_create_with_primary_key_", tableDefinition)) {
+        try (TestTable table = newTrinoTable("test_create_table_definition_", tableDefinition)) {
             assertThat(computeScalar("SHOW CREATE TABLE " + table.getName()))
                     .isEqualTo(format(showCreateTableFormat, getSession().getCatalog().orElseThrow(), getSession().getSchema().orElseThrow(), table.getName()));
         }
@@ -355,6 +355,149 @@ private void verifyTableDefinitionWithUnsupportedKey(String tableDefinition, Str
         }
     }
 
+    @Test
+    public void testCreateTableWithAutoIncrementColumn()
+    {
+        verifyCreateTableDefinition(
+                "(a tinyint NOT NULL WITH (auto_increment = true), b bigint, c bigint) WITH (primary_key = ARRAY['a'])",
+                """
+                CREATE TABLE %s.%s.%s (
+                   a tinyint NOT NULL WITH (auto_increment = true),
+                   b bigint,
+                   c bigint
+                )
+                WITH (
+                   primary_key = ARRAY['a']
+                )\
+                """
+        );
+
+        verifyCreateTableDefinition(
+                "(a smallint NOT NULL WITH (auto_increment = true), b bigint, c bigint) WITH (primary_key = ARRAY['a'])",
+                """
+                CREATE TABLE %s.%s.%s (
+                   a smallint NOT NULL WITH (auto_increment = true),
+                   b bigint,
+                   c bigint
+                )
+                WITH (
+                   primary_key = ARRAY['a']
+                )\
+                """
+        );
+
+        verifyCreateTableDefinition(
+                "(a int NOT NULL WITH (auto_increment = true), b bigint, c bigint) WITH (primary_key = ARRAY['a'])",
+                """
+                CREATE TABLE %s.%s.%s (
+                   a integer NOT NULL WITH (auto_increment = true),
+                   b bigint,
+                   c bigint
+                )
+                WITH (
+                   primary_key = ARRAY['a']
+                )\
+                """
+        );
+
+        verifyCreateTableDefinition(
+                "(a bigint NOT NULL WITH (auto_increment = true), b bigint NOT NULL, c bigint) WITH (primary_key = ARRAY['a', 'b'])",
+                """
+                CREATE TABLE %s.%s.%s (
+                   a bigint NOT NULL WITH (auto_increment = true),
+                   b bigint NOT NULL,
+                   c bigint
+                )
+                WITH (
+                   primary_key = ARRAY['a','b']
+                )\
+                """
+        );
+
+        verifyCreateTableDefinition(
+                "(a bigint NOT NULL, b bigint NOT NULL WITH (auto_increment = true), c bigint) WITH (primary_key = ARRAY['b', 'a'])",
+                """
+                CREATE TABLE %s.%s.%s (
+                   a bigint NOT NULL,
+                   b bigint NOT NULL WITH (auto_increment = true),
+                   c bigint
+                )
+                WITH (
+                   primary_key = ARRAY['b','a']
+                )\
+                """
+        );
+    }
+
+    @Test
+    public void testCreateTableWithInvalidAutoIncrementColumn()
+    {
+        String tableName = "test_create_with_invalid_auto_increment_column";
+
+        assertQueryFails("CREATE TABLE " + tableName + " (a varchar NOT NULL WITH (auto_increment = true), b bigint, c bigint) WITH (primary_key = ARRAY['a'])",
+                "Unsupported column type for AUTO_INCREMENT: varchar");
+        assertQueryFails("CREATE TABLE " + tableName + " (a date NOT NULL WITH (auto_increment = true), b bigint, c bigint)  WITH (primary_key = ARRAY['a'])",
+                "Unsupported column type for AUTO_INCREMENT: date");
+        assertQueryFails("CREATE TABLE " + tableName + " (a double NOT NULL WITH (auto_increment = true), b bigint, c bigint)  WITH (primary_key = ARRAY['a'])",
+                "Unsupported column type for AUTO_INCREMENT: double");
+
+        assertQueryFails("CREATE TABLE " + tableName + " (a double NOT NULL WITH (auto_increment = false), b bigint, c bigint)  WITH (primary_key = ARRAY['a'])",
+                "Column property value for AUTO_INCREMENT must be true");
+        assertQueryFails("CREATE TABLE " + tableName + " (a bigint NOT NULL WITH (auto_increment = false), b bigint NOT NULL WITH (auto_increment = true), c bigint) WITH (primary_key = ARRAY['a', 'b'])",
+                "Column property value for AUTO_INCREMENT must be true");
+
+        assertQueryFails("CREATE TABLE " + tableName + " (a bigint NOT NULL WITH (auto_increment = true), b bigint NOT NULL WITH (auto_increment = true), c bigint) WITH (primary_key = ARRAY['a', 'b'])",
+                "There can be only one auto increment column in MySQL");
+
+        assertQueryFails("CREATE TABLE " + tableName + " (a bigint NOT NULL, b bigint WITH (auto_increment = true), c bigint) WITH (primary_key = ARRAY['a', 'b'])",
+                "Auto increment column must be defined as the first key in MySQL");
+        assertQueryFails("CREATE TABLE " + tableName + " (a bigint NOT NULL, b bigint WITH (auto_increment = true), c bigint) WITH (primary_key = ARRAY['a'])",
+                "Auto increment column must be defined as the first key in MySQL");
+        assertQueryFails("CREATE TABLE " + tableName + " (a bigint WITH (auto_increment = true), b bigint, c bigint)",
+                "Auto increment column must be defined as the first key in MySQL");
+    }
+
+    @Test
+    public void testInsertWithAutoIncrementColumn()
+    {
+        try (TestTable table = newTrinoTable("test_insert_with_auto_increment_column_", "(a int NOT NULL WITH (auto_increment = true), b int) WITH (primary_key = ARRAY['a'])")) {
+            assertUpdate("INSERT INTO " + table.getName() + " (b) VALUES (1), (2)", 2);
+            assertThat(query("SELECT a, b FROM " + table.getName()))
+                    .matches("VALUES (1, 1), (2, 2)");
+
+            assertUpdate("INSERT INTO " + table.getName() + " (a, b) VALUES (4, 4)", 1);
+            assertThat(query("SELECT a, b FROM " + table.getName()))
+                    .matches("VALUES (1, 1), (2, 2), (4, 4)");
+
+            assertUpdate("INSERT INTO " + table.getName() + " (b) VALUES (5)", 1);
+            assertThat(query("SELECT a, b FROM " + table.getName()))
+                    .matches("VALUES (1, 1), (2, 2), (4, 4), (5, 5)");
+        }
+    }
+
+
+    @Test
+    public void testShowColumnsWithAutoIncrementColumn()
+    {
+        String expected = "VALUES (VARCHAR 'a', VARCHAR 'bigint', VARCHAR 'auto_increment=true', VARCHAR ''), (VARCHAR 'b', VARCHAR 'bigint', VARCHAR '', VARCHAR '')";
+        try (TestTable table = newTrinoTable("test_show_columns_with_auto_increment_column_", "(a bigint NOT NULL WITH (auto_increment = true), b bigint) WITH (primary_key = ARRAY['a'])")) {
+            assertThat(query("SHOW COLUMNS FROM " + table.getName()))
+                    .matches(expected);
+
+            assertThat(query("DESCRIBE " + table.getName()))
+                    .matches(expected);
+        }
+    }
+
+    @Test
+    public void testAddColumnWithColumnProperties()
+    {
+        try (TestTable table = newTrinoTable("test_add_column_with_properties_", "(a bigint NOT NULL, b bigint) WITH (primary_key = ARRAY['a'])")) {
+            assertQueryFails("ALTER TABLE " + table.getName() + " ADD COLUMN test_add_col_properties int WITH (auto_increment=true)",
+                    "This connector does not support adding columns with column properties");
+        }
+    }
+
     @Test
     public void testViews()
     {