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,72 @@ 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.
+:::
+
 (mysql-procedures)=
 ### Procedures
 

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

@@ -334,6 +334,7 @@ public List<JdbcColumnHandle> getColumns(ConnectorSession session, SchemaTableNa
                 Optional<ColumnMapping> columnMapping = toColumnMapping(session, connection, typeHandle);
                 log.debug("Mapping data type of '%s' column '%s': %s mapped to %s", schemaTableName, columnName, typeHandle, columnMapping);
                 boolean nullable = (resultSet.getInt("NULLABLE") != columnNoNulls);
+                boolean autoIncrement = "YES".equals(resultSet.getString("IS_AUTOINCREMENT"));
                 // Note: some databases (e.g. SQL Server) do not return column remarks/comment here.
                 Optional<String> comment = Optional.ofNullable(emptyToNull(resultSet.getString("REMARKS")));
                 // skip unsupported column types
@@ -343,6 +344,7 @@ public List<JdbcColumnHandle> getColumns(ConnectorSession session, SchemaTableNa
                         .setColumnType(mapping.getType())
                         .setNullable(nullable)
                         .setComment(comment)
+                        .setAutoIncrement(autoIncrement)
                         .build()));
                 if (columnMapping.isEmpty()) {
                     UnsupportedTypeHandling unsupportedTypeHandling = getUnsupportedTypeHandling(session);

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/JdbcColumnHandle.java

@@ -40,11 +40,12 @@ public final class JdbcColumnHandle
     private final Type columnType;
     private final boolean nullable;
     private final Optional<String> comment;
+    private final boolean autoIncrement;
 
     // 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(), false);
     }
 
     /**
@@ -57,13 +58,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("autoIncrement") boolean autoIncrement)
     {
         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.autoIncrement = autoIncrement;
     }
 
     @JsonProperty
@@ -96,6 +99,12 @@ public Optional<String> getComment()
         return comment;
     }
 
+    @JsonProperty
+    public boolean isAutoIncrement()
+    {
+        return autoIncrement;
+    }
+
     public ColumnMetadata getColumnMetadata()
     {
         return ColumnMetadata.builder()
@@ -147,6 +156,7 @@ public long getRetainedSizeInBytes()
         // columnType is not accounted for as the instances are cached (by TypeRegistry) and shared
         return INSTANCE_SIZE
                 + sizeOf(nullable)
+                + sizeOf(autoIncrement)
                 + estimatedSizeOf(columnName)
                 + sizeOf(comment, SizeOf::estimatedSizeOf)
                 + jdbcTypeHandle.getRetainedSizeInBytes();
@@ -169,6 +179,7 @@ public static final class Builder
         private Type columnType;
         private boolean nullable = true;
         private Optional<String> comment = Optional.empty();
+        private boolean autoIncrement;
 
         public Builder() {}
 
@@ -179,6 +190,7 @@ private Builder(JdbcColumnHandle handle)
             this.columnType = handle.getColumnType();
             this.nullable = handle.isNullable();
             this.comment = handle.getComment();
+            this.autoIncrement = handle.isAutoIncrement();
         }
 
         public Builder setColumnName(String columnName)
@@ -211,14 +223,21 @@ public Builder setComment(Optional<String> comment)
             return this;
         }
 
+        public Builder setAutoIncrement(boolean autoIncrement)
+        {
+            this.autoIncrement = autoIncrement;
+            return this;
+        }
+
         public JdbcColumnHandle build()
         {
             return new JdbcColumnHandle(
                     columnName,
                     jdbcTypeHandle,
                     columnType,
                     nullable,
-                    comment);
+                    comment,
+                    autoIncrement);
         }
     }
 }

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()
     {

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

@@ -64,6 +64,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);
@@ -150,6 +151,16 @@ 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);
+    }
+
     @ProvidesIntoOptional(DEFAULT)
     @Inject
     @Singleton