Normalize ColumnMetadata to support case-sensitive column names

presto-accumulo/src/main/java/com/facebook/presto/accumulo/AccumuloMetadata.java

@@ -57,6 +57,7 @@
 import static com.facebook.presto.spi.StandardErrorCode.NOT_SUPPORTED;
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkState;
+import static com.google.common.collect.ImmutableList.toImmutableList;
 import static java.lang.String.format;
 import static java.util.Objects.requireNonNull;
 
@@ -282,7 +283,7 @@ public ConnectorTableMetadata getTableMetadata(ConnectorSession session, Connect
         AccumuloTableHandle handle = (AccumuloTableHandle) table;
         checkArgument(handle.getConnectorId().equals(connectorId), "table is not for this connector");
         SchemaTableName tableName = new SchemaTableName(handle.getSchema(), handle.getTable());
-        ConnectorTableMetadata metadata = getTableMetadata(tableName);
+        ConnectorTableMetadata metadata = getTableMetadata(session, tableName);
         if (metadata == null) {
             throw new TableNotFoundException(tableName);
         }
@@ -353,7 +354,7 @@ public Map<SchemaTableName, List<ColumnMetadata>> listTableColumns(ConnectorSess
         requireNonNull(prefix, "prefix is null");
         ImmutableMap.Builder<SchemaTableName, List<ColumnMetadata>> columns = ImmutableMap.builder();
         for (SchemaTableName tableName : listTables(session, prefix)) {
-            ConnectorTableMetadata tableMetadata = getTableMetadata(tableName);
+            ConnectorTableMetadata tableMetadata = getTableMetadata(session, tableName);
             // table can disappear during listing operation
             if (tableMetadata != null) {
                 columns.put(tableName, tableMetadata.getColumns());
@@ -385,7 +386,7 @@ public void rollback()
         }
     }
 
-    private ConnectorTableMetadata getTableMetadata(SchemaTableName tableName)
+    private ConnectorTableMetadata getTableMetadata(ConnectorSession session, SchemaTableName tableName)
     {
         if (!client.getSchemaNames().contains(tableName.getSchemaName())) {
             return null;
@@ -398,7 +399,13 @@ private ConnectorTableMetadata getTableMetadata(SchemaTableName tableName)
                 return null;
             }
 
-            return new ConnectorTableMetadata(tableName, table.getColumnsMetadata());
+            List<ColumnMetadata> columns = table.getColumnsMetadata().stream()
+                    .map(column -> column.toBuilder()
+                            .setName(normalizeIdentifier(session, column.getName()))
+                            .build())
+                    .collect(toImmutableList());
+
+            return new ConnectorTableMetadata(tableName, columns);
         }
 
         return null;

presto-atop/src/main/java/com/facebook/presto/atop/AtopMetadata.java

@@ -121,7 +121,7 @@ public ConnectorTableMetadata getTableMetadata(ConnectorSession session, Connect
         ImmutableList.Builder<ColumnMetadata> columns = ImmutableList.builder();
         for (AtopColumn column : atopTableHandle.getTable().getColumns()) {
             columns.add(ColumnMetadata.builder()
-                    .setName(column.getName())
+                    .setName(normalizeIdentifier(session, column.getName()))
                     .setType(typeManager.getType(column.getType()))
                     .build());
         }

presto-base-arrow-flight/src/main/java/com/facebook/plugin/arrow/ArrowMetadata.java

@@ -148,7 +148,7 @@ public ConnectorTableMetadata getTableMetadata(ConnectorSession session, Connect
         for (Field field : columnList) {
             String columnName = field.getName();
             Type fieldType = getPrestoTypeFromArrowField(field);
-            meta.add(ColumnMetadata.builder().setName(columnName).setType(fieldType).build());
+            meta.add(ColumnMetadata.builder().setName(normalizeIdentifier(session, columnName)).setType(fieldType).build());
         }
         return new ConnectorTableMetadata(new SchemaTableName(((ArrowTableHandle) table).getSchema(), ((ArrowTableHandle) table).getTable()), meta);
     }

presto-base-arrow-flight/src/test/java/com/facebook/plugin/arrow/TestArrowColumnHandle.java

@@ -17,8 +17,6 @@
 import com.facebook.presto.spi.ColumnMetadata;
 import org.testng.annotations.Test;
 
-import java.util.Locale;
-
 import static com.facebook.presto.testing.assertions.Assert.assertEquals;
 import static org.testng.Assert.assertNotNull;
 
@@ -67,7 +65,7 @@ public void testGetColumnMetadata()
 
         // Then
         assertNotNull(columnMetadata, "ColumnMetadata should not be null");
-        assertEquals(columnMetadata.getName(), columnName.toLowerCase(Locale.ENGLISH), "ColumnMetadata name should match the column name");
+        assertEquals(columnMetadata.getName(), columnName, "ColumnMetadata name should match the column name");
         assertEquals(columnMetadata.getType(), IntegerType.INTEGER, "ColumnMetadata type should match the column type");
     }
 

presto-base-jdbc/src/main/java/com/facebook/presto/plugin/jdbc/JdbcColumnHandle.java

@@ -16,6 +16,7 @@
 import com.facebook.presto.common.type.Type;
 import com.facebook.presto.spi.ColumnHandle;
 import com.facebook.presto.spi.ColumnMetadata;
+import com.facebook.presto.spi.ConnectorSession;
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonProperty;
 
@@ -98,6 +99,16 @@ public ColumnMetadata getColumnMetadata()
                 .build();
     }
 
+    public ColumnMetadata getColumnMetadata(ConnectorSession session, JdbcClient jdbcClient)
+    {
+        return ColumnMetadata.builder()
+                .setName(jdbcClient.normalizeIdentifier(session, columnName))
+                .setType(columnType)
+                .setNullable(nullable)
+                .setComment(comment.orElse(null))
+                .build();
+    }
+
     @Override
     public boolean equals(Object obj)
     {

presto-base-jdbc/src/main/java/com/facebook/presto/plugin/jdbc/JdbcMetadata.java

@@ -104,7 +104,7 @@ public ConnectorTableMetadata getTableMetadata(ConnectorSession session, Connect
 
         ImmutableList.Builder<ColumnMetadata> columnMetadata = ImmutableList.builder();
         for (JdbcColumnHandle column : jdbcMetadataCache.getColumns(session, handle)) {
-            columnMetadata.add(column.getColumnMetadata());
+            columnMetadata.add(column.getColumnMetadata(session, jdbcClient));
         }
         return new ConnectorTableMetadata(handle.getSchemaTableName(), columnMetadata.build());
     }
@@ -122,7 +122,7 @@ public Map<String, ColumnHandle> getColumnHandles(ConnectorSession session, Conn
 
         ImmutableMap.Builder<String, ColumnHandle> columnHandles = ImmutableMap.builder();
         for (JdbcColumnHandle column : jdbcMetadataCache.getColumns(session, jdbcTableHandle)) {
-            columnHandles.put(column.getColumnMetadata().getName(), column);
+            columnHandles.put(column.getColumnMetadata(session, jdbcClient).getName(), column);
         }
         return columnHandles.build();
     }

presto-base-jdbc/src/test/java/com/facebook/presto/plugin/jdbc/TestingDatabase.java

@@ -117,7 +117,7 @@ public Map<String, JdbcColumnHandle> getColumnHandles(String schemaName, String
 
         ImmutableMap.Builder<String, JdbcColumnHandle> columnHandles = ImmutableMap.builder();
         for (JdbcColumnHandle column : columns) {
-            columnHandles.put(column.getColumnMetadata().getName(), column);
+            columnHandles.put(column.getColumnMetadata(session, jdbcClient).getName(), column);
         }
         return columnHandles.build();
     }

presto-bigquery/src/main/java/com/facebook/presto/plugin/bigquery/BigQueryMetadata.java

@@ -48,6 +48,7 @@
 import java.util.Set;
 
 import static com.facebook.presto.plugin.bigquery.BigQueryErrorCode.BIGQUERY_TABLE_DISAPPEAR_DURING_LIST;
+import static com.facebook.presto.plugin.bigquery.Conversions.toColumnMetadata;
 import static com.google.cloud.bigquery.TableDefinition.Type.TABLE;
 import static com.google.cloud.bigquery.TableDefinition.Type.VIEW;
 import static com.google.common.collect.ImmutableList.toImmutableList;
@@ -179,7 +180,7 @@ public ConnectorTableMetadata getTableMetadata(ConnectorSession session, Connect
         List<ColumnMetadata> columns = schema == null ?
                 ImmutableList.of() :
                 schema.getFields().stream()
-                        .map(Conversions::toColumnMetadata)
+                        .map(field -> toColumnMetadata(field, normalizeIdentifier(session, field.getName())))
                         .collect(toImmutableList());
         return new ConnectorTableMetadata(schemaTableName, columns);
     }

presto-bigquery/src/main/java/com/facebook/presto/plugin/bigquery/Conversions.java

@@ -47,10 +47,10 @@ static BigQueryColumnHandle toColumnHandle(Field field)
                 field.getDescription());
     }
 
-    static ColumnMetadata toColumnMetadata(Field field)
+    static ColumnMetadata toColumnMetadata(Field field, String name)
     {
         return ColumnMetadata.builder()
-                .setName(field.getName())
+                .setName(name)
                 .setType(adapt(field).getPrestoType())
                 .setNullable(getMode(field) == NULLABLE)
                 .setComment(field.getDescription())

presto-bigquery/src/test/java/com/facebook/presto/plugin/bigquery/TestTypeConversions.java

@@ -134,7 +134,7 @@ public void testConvertOneLevelRecordField()
                 RECORD,
                 Field.of("sub_s", STRING),
                 Field.of("sub_i", INTEGER));
-        ColumnMetadata metadata = Conversions.toColumnMetadata(field);
+        ColumnMetadata metadata = Conversions.toColumnMetadata(field, field.getName());
         RowType targetType = RowType.from(ImmutableList.of(
                 RowType.field("sub_s", VarcharType.VARCHAR),
                 RowType.field("sub_i", BigintType.BIGINT)));
@@ -152,7 +152,7 @@ public void testConvertTwoLevelsRecordField()
                         Field.of("sub_sub_i", INTEGER)),
                 Field.of("sub_s", STRING),
                 Field.of("sub_i", INTEGER));
-        ColumnMetadata metadata = Conversions.toColumnMetadata(field);
+        ColumnMetadata metadata = Conversions.toColumnMetadata(field, field.getName());
         RowType targetType = RowType.from(ImmutableList.of(
                 RowType.field("sub_rec", RowType.from(ImmutableList.of(
                         RowType.field("sub_sub_s", VarcharType.VARCHAR),
@@ -168,13 +168,13 @@ public void testConvertStringArrayField()
         Field field = Field.newBuilder("test", STRING)
                 .setMode(REPEATED)
                 .build();
-        ColumnMetadata metadata = Conversions.toColumnMetadata(field);
+        ColumnMetadata metadata = Conversions.toColumnMetadata(field, field.getName());
         assertThat(metadata.getType()).isEqualTo(new ArrayType(VarcharType.VARCHAR));
     }
 
     void assertSimpleFieldTypeConversion(LegacySQLTypeName from, Type to)
     {
-        ColumnMetadata metadata = Conversions.toColumnMetadata(createField(from));
+        ColumnMetadata metadata = Conversions.toColumnMetadata(createField(from), createField(from).getName());
         assertThat(metadata.getType()).isEqualTo(to);
     }
 

presto-blackhole/src/main/java/com/facebook/presto/plugin/blackhole/BlackHoleColumnHandle.java

@@ -61,6 +61,14 @@ public ColumnMetadata toColumnMetadata()
                 .build();
     }
 
+    public ColumnMetadata toColumnMetadata(String name)
+    {
+        return ColumnMetadata.builder()
+                .setName(name)
+                .setType(columnType)
+                .build();
+    }
+
     @Override
     public int hashCode()
     {

presto-blackhole/src/main/java/com/facebook/presto/plugin/blackhole/BlackHoleMetadata.java

@@ -54,6 +54,7 @@
 import static com.facebook.presto.plugin.blackhole.BlackHoleConnector.SPLIT_COUNT_PROPERTY;
 import static com.facebook.presto.spi.StandardErrorCode.ALREADY_EXISTS;
 import static com.facebook.presto.spi.StandardErrorCode.INVALID_TABLE_PROPERTY;
+import static com.google.common.collect.ImmutableList.toImmutableList;
 import static java.lang.String.format;
 import static java.util.stream.Collectors.toList;
 import static java.util.stream.Collectors.toMap;
@@ -97,7 +98,16 @@ public ConnectorTableHandle getTableHandle(ConnectorSession session, SchemaTable
     public ConnectorTableMetadata getTableMetadata(ConnectorSession session, ConnectorTableHandle tableHandle)
     {
         BlackHoleTableHandle blackHoleTableHandle = (BlackHoleTableHandle) tableHandle;
-        return blackHoleTableHandle.toTableMetadata();
+        return toTableMetadata(blackHoleTableHandle, session);
+    }
+
+    public ConnectorTableMetadata toTableMetadata(BlackHoleTableHandle blackHoleTableHandle, ConnectorSession session)
+    {
+        List<ColumnMetadata> columns = blackHoleTableHandle.getColumnHandles().stream()
+                .map(column -> column.toColumnMetadata(normalizeIdentifier(session, column.getName())))
+                .collect(toImmutableList());
+
+        return new ConnectorTableMetadata(blackHoleTableHandle.toSchemaTableName(), columns);
     }
 
     @Override
@@ -129,7 +139,7 @@ public Map<SchemaTableName, List<ColumnMetadata>> listTableColumns(ConnectorSess
     {
         return tables.values().stream()
                 .filter(table -> prefix.matches(table.toSchemaTableName()))
-                .collect(toMap(BlackHoleTableHandle::toSchemaTableName, handle -> handle.toTableMetadata().getColumns()));
+                .collect(toMap(BlackHoleTableHandle::toSchemaTableName, handle -> toTableMetadata(handle, session).getColumns()));
     }
 
     @Override

presto-blackhole/src/main/java/com/facebook/presto/plugin/blackhole/BlackHoleTableHandle.java

@@ -126,13 +126,6 @@ public Duration getPageProcessingDelay()
         return pageProcessingDelay;
     }
 
-    public ConnectorTableMetadata toTableMetadata()
-    {
-        return new ConnectorTableMetadata(
-                toSchemaTableName(),
-                columnHandles.stream().map(BlackHoleColumnHandle::toColumnMetadata).collect(toList()));
-    }
-
     public SchemaTableName toSchemaTableName()
     {
         return new SchemaTableName(schemaName, tableName);

presto-cassandra/src/main/java/com/facebook/presto/cassandra/CassandraColumnHandle.java

@@ -138,6 +138,15 @@ public ColumnMetadata getColumnMetadata()
                 .build();
     }
 
+    public ColumnMetadata getColumnMetadata(String name)
+    {
+        return ColumnMetadata.builder()
+                .setName(name)
+                .setType(cassandraType.getNativeType())
+                .setHidden(hidden)
+                .build();
+    }
+
     public Type getType()
     {
         return cassandraType.getNativeType();

presto-cassandra/src/main/java/com/facebook/presto/cassandra/CassandraMetadata.java

@@ -53,6 +53,7 @@
 import java.util.stream.Collectors;
 
 import static com.facebook.presto.cassandra.CassandraType.toCassandraType;
+import static com.facebook.presto.cassandra.util.CassandraCqlUtils.cqlNameToSqlName;
 import static com.facebook.presto.cassandra.util.CassandraCqlUtils.validSchemaName;
 import static com.facebook.presto.cassandra.util.CassandraCqlUtils.validTableName;
 import static com.facebook.presto.spi.StandardErrorCode.NOT_SUPPORTED;
@@ -62,7 +63,6 @@
 import static com.google.common.collect.ImmutableList.toImmutableList;
 import static java.util.Locale.ENGLISH;
 import static java.util.Objects.requireNonNull;
-import static java.util.stream.Collectors.toList;
 
 public class CassandraMetadata
         implements ConnectorMetadata
@@ -121,15 +121,15 @@ private static SchemaTableName getTableName(ConnectorTableHandle tableHandle)
     public ConnectorTableMetadata getTableMetadata(ConnectorSession session, ConnectorTableHandle tableHandle)
     {
         requireNonNull(tableHandle, "tableHandle is null");
-        return getTableMetadata(getTableName(tableHandle));
+        return getTableMetadata(session, getTableName(tableHandle));
     }
 
-    private ConnectorTableMetadata getTableMetadata(SchemaTableName tableName)
+    private ConnectorTableMetadata getTableMetadata(ConnectorSession session, SchemaTableName tableName)
     {
         CassandraTable table = cassandraSession.getTable(tableName);
         List<ColumnMetadata> columns = table.getColumns().stream()
-                .map(CassandraColumnHandle::getColumnMetadata)
-                .collect(toList());
+                .map(column -> column.getColumnMetadata(normalizeIdentifier(session, cqlNameToSqlName(column.getName()))))
+                .collect(toImmutableList());
         return new ConnectorTableMetadata(tableName, columns);
     }
 
@@ -178,7 +178,7 @@ public Map<SchemaTableName, List<ColumnMetadata>> listTableColumns(ConnectorSess
         ImmutableMap.Builder<SchemaTableName, List<ColumnMetadata>> columns = ImmutableMap.builder();
         for (SchemaTableName tableName : listTables(session, prefix)) {
             try {
-                columns.put(tableName, getTableMetadata(tableName).getColumns());
+                columns.put(tableName, getTableMetadata(session, tableName).getColumns());
             }
             catch (NotFoundException e) {
                 // table disappeared during listing operation

presto-clickhouse/src/main/java/com/facebook/presto/plugin/clickhouse/ClickHouseColumnHandle.java

@@ -142,6 +142,17 @@ public ColumnMetadata getColumnMetadata()
                 .build();
     }
 
+    public ColumnMetadata getColumnMetadata(String name)
+    {
+        return ColumnMetadata.builder()
+                .setName(name)
+                .setType(columnType)
+                .setNullable(nullable)
+                .setHidden(false)
+                .setProperties(emptyMap())
+                .build();
+    }
+
     @Override
     public boolean equals(Object obj)
     {

presto-clickhouse/src/main/java/com/facebook/presto/plugin/clickhouse/ClickHouseMetadata.java

@@ -104,7 +104,7 @@ public ConnectorTableMetadata getTableMetadata(ConnectorSession session, Connect
 
         ImmutableList.Builder<ColumnMetadata> columnMetadata = ImmutableList.builder();
         for (ClickHouseColumnHandle column : clickHouseClient.getColumns(session, handle)) {
-            columnMetadata.add(column.getColumnMetadata());
+            columnMetadata.add(column.getColumnMetadata(normalizeIdentifier(session, column.getColumnName())));
         }
         return new ConnectorTableMetadata(handle.getSchemaTableName(), columnMetadata.build());
     }
@@ -122,7 +122,7 @@ public Map<String, ColumnHandle> getColumnHandles(ConnectorSession session, Conn
 
         ImmutableMap.Builder<String, ColumnHandle> columnHandles = ImmutableMap.builder();
         for (ClickHouseColumnHandle column : clickHouseClient.getColumns(session, clickHouseTableHandle)) {
-            columnHandles.put(column.getColumnMetadata().getName(), column);
+            columnHandles.put(normalizeIdentifier(session, column.getColumnMetadata(column.getColumnName()).getName()), column);
         }
         return columnHandles.build();
     }

presto-delta/src/main/java/com/facebook/presto/delta/DeltaMetadata.java

@@ -333,7 +333,7 @@ private ConnectorTableMetadata getTableMetadata(ConnectorSession session, Schema
         }
 
         List<ColumnMetadata> columnMetadata = tableHandle.getDeltaTable().getColumns().stream()
-                .map(this::getColumnMetadata)
+                .map(column -> getColumnMetadata(session, column))
                 .collect(Collectors.toList());
 
         return new ConnectorTableMetadata(tableName, columnMetadata);
@@ -362,10 +362,10 @@ private List<SchemaTableName> listTables(ConnectorSession session, SchemaTablePr
         return ImmutableList.of(new SchemaTableName(prefix.getSchemaName(), prefix.getTableName()));
     }
 
-    private ColumnMetadata getColumnMetadata(DeltaColumn deltaColumn)
+    private ColumnMetadata getColumnMetadata(ConnectorSession session, DeltaColumn deltaColumn)
     {
         return ColumnMetadata.builder()
-                .setName(deltaColumn.getName())
+                .setName(normalizeIdentifier(session, deltaColumn.getName()))
                 .setType(typeManager.getType(deltaColumn.getType()))
                 .build();
     }

presto-docs/src/main/sphinx/connector/hana.rst

@@ -91,6 +91,10 @@ Property Name                                      Description
                                                    cached. Set to ``0ms`` to disable the cache.                         ``1m``
 
 ``list-schemas-ignored-schemas``                   List of schemas to ignore when listing schemas.                      ``information_schema``
+
+``case-sensitive-name-matching``                   Enable case sensitive identifier support for schema and table        ``false``
+                                                   names for the connector. When disabled, names are matched
+                                                   case-insensitively using lowercase normalization.
 ================================================== ==================================================================== ===========
 
 Querying HANA