Support column properties with jdbc connector

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
 

plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/BaseJdbcClient.java

@@ -336,13 +336,15 @@ 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)
                         .setJdbcTypeHandle(typeHandle)
                         .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(

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));

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();
+}

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

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);
+    }
 }

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();
+    }
 }