feat(plugin-kafka): Enable case-senstive support for Kafka connector (#26023)

## Description
This PR adds support for case-sensitive identifiers in the Kafka
connector.

## Motivation and Context
<!---Why is this change required? What problem does it solve?-->
<!---If it fixes an open issue, please link to the issue here.-->

## Impact
<!---Describe any public API or user-facing feature change or any
performance impact-->

## Test Plan
<!---Please fill in how you tested your change-->

## Contributor checklist

- [ ] Please make sure your submission complies with our [contributing
guide](https://github.com/prestodb/presto/blob/master/CONTRIBUTING.md),
in particular [code
style](https://github.com/prestodb/presto/blob/master/CONTRIBUTING.md#code-style)
and [commit
standards](https://github.com/prestodb/presto/blob/master/CONTRIBUTING.md#commit-standards).
- [ ] PR description addresses the issue accurately and concisely. If
the change is non-trivial, a GitHub Issue is referenced.
- [ ] Documented new properties (with its default value), SQL syntax,
functions, or other functionality.
- [ ] If release notes are required, they follow the [release notes
guidelines](https://github.com/prestodb/presto/wiki/Release-Notes-Guidelines).
- [ ] Adequate tests were added if applicable.
- [ ] CI passed.

## Release Notes
Please follow [release notes
guidelines](https://github.com/prestodb/presto/wiki/Release-Notes-Guidelines)
and fill in the release notes below.
```
==RELEASE NOTE ==

Kafka Connector Changes
* Add mixed case support for Kafka connector.
```

Author: Mariamalmesfer

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

@@ -73,6 +73,10 @@ Property Name                       Description
 ``kafka.truststore.password``       Password for the truststore file
 ``kafka.truststore.type``           File format of the truststore file, defaults to ``JKS``
 ``kafka.config.resources``          A comma-separated list of Kafka client configuration files. If a specialized authentication method is required, it can be specified in these additional Kafka client properties files. Example: `/etc/kafka-configuration.properties`
+``case-sensitive-name-matching``    Enable case-sensitive identifier support for schema,
+                                    table, and column names for the connector. When disabled,
+                                    names are matched case-insensitively using lowercase
+                                    normalization. Default is ``false``.
 =================================== ==============================================================
 
 ``kafka.table-names``

presto-kafka/src/main/java/com/facebook/presto/kafka/KafkaConnectorConfig.java

@@ -70,6 +70,8 @@ public class KafkaConnectorConfig
      */
     private List<File> resourceConfigFiles = ImmutableList.of();
 
+    private boolean caseSensitiveNameMatching;
+
     @NotNull
     public String getDefaultSchema()
     {
@@ -173,4 +175,18 @@ public KafkaConnectorConfig setResourceConfigFiles(String files)
                 .collect(toImmutableList());
         return this;
     }
+
+    public boolean isCaseSensitiveNameMatching()
+    {
+        return caseSensitiveNameMatching;
+    }
+
+    @Config("case-sensitive-name-matching")
+    @ConfigDescription("Enable case-sensitive matching of schema, table names across the connector. " +
+            "When disabled, names are matched case-insensitively using lowercase normalization.")
+    public KafkaConnectorConfig setCaseSensitiveNameMatching(boolean caseSensitiveNameMatchingEnabled)
+    {
+        this.caseSensitiveNameMatching = caseSensitiveNameMatchingEnabled;
+        return this;
+    }
 }

presto-kafka/src/main/java/com/facebook/presto/kafka/KafkaMetadata.java

@@ -50,6 +50,7 @@
 import static com.facebook.presto.kafka.KafkaHandleResolver.convertTableHandle;
 import static com.google.common.collect.ImmutableList.toImmutableList;
 import static java.lang.String.format;
+import static java.util.Locale.ROOT;
 import static java.util.Objects.requireNonNull;
 
 /**
@@ -63,6 +64,7 @@ public class KafkaMetadata
     private final String connectorId;
     private final boolean hideInternalColumns;
     private final TableDescriptionSupplier tableDescriptionSupplier;
+    private final boolean caseSensitiveNameMatching;
 
     @Inject
     public KafkaMetadata(
@@ -75,6 +77,7 @@ public KafkaMetadata(
         requireNonNull(kafkaConnectorConfig, "kafkaConfig is null");
         this.hideInternalColumns = kafkaConnectorConfig.isHideInternalColumns();
         this.tableDescriptionSupplier = requireNonNull(tableDescriptionSupplier, "tableDescriptionSupplier is null");
+        this.caseSensitiveNameMatching = kafkaConnectorConfig.isCaseSensitiveNameMatching();
     }
 
     @Override
@@ -305,4 +308,9 @@ private Optional<KafkaTopicDescription> getTopicDescription(SchemaTableName sche
     {
         return tableDescriptionSupplier.getTopicDescription(schemaTableName);
     }
+    @Override
+    public String normalizeIdentifier(ConnectorSession session, String identifier)
+    {
+        return caseSensitiveNameMatching ? identifier : identifier.toLowerCase(ROOT);
+    }
 }

presto-kafka/src/test/java/com/facebook/presto/kafka/KafkaQueryRunner.java

@@ -58,28 +58,29 @@ private KafkaQueryRunner()
     public static DistributedQueryRunner createKafkaQueryRunner(EmbeddedKafka embeddedKafka, TpchTable<?>... tables)
             throws Exception
     {
-        return createKafkaQueryRunner(embeddedKafka, ImmutableList.copyOf(tables));
+        List<SchemaTableName> extraTables = ImmutableList.of();
+        return createKafkaQueryRunner(embeddedKafka, ImmutableList.copyOf(tables), ImmutableMap.of(), extraTables);
     }
 
-    public static DistributedQueryRunner createKafkaQueryRunner(EmbeddedKafka embeddedKafka, Iterable<TpchTable<?>> tables)
+    public static DistributedQueryRunner createKafkaQueryRunner(EmbeddedKafka embeddedKafka, Iterable<TpchTable<?>> tables, Map<String, String> connectorProperties, List<SchemaTableName> extraTables)
             throws Exception
     {
         DistributedQueryRunner queryRunner = null;
         try {
             queryRunner = new DistributedQueryRunner(createSession(), 2);
 
             queryRunner.installPlugin(new TpchPlugin());
-            queryRunner.createCatalog("tpch", "tpch");
+            queryRunner.createCatalog("tpch", "tpch", connectorProperties);
 
             embeddedKafka.start();
 
             for (TpchTable<?> table : tables) {
                 embeddedKafka.createTopics(kafkaTopicName(table));
             }
 
-            Map<SchemaTableName, KafkaTopicDescription> topicDescriptions = createTpchTopicDescriptions(queryRunner.getCoordinator().getMetadata(), tables, embeddedKafka);
+            Map<SchemaTableName, KafkaTopicDescription> topicDescriptions = createTpchTopicDescriptions(queryRunner.getCoordinator().getMetadata(), tables, embeddedKafka, extraTables);
 
-            installKafkaPlugin(embeddedKafka, queryRunner, topicDescriptions);
+            installKafkaPlugin(embeddedKafka, queryRunner, topicDescriptions, connectorProperties);
 
             TestingPrestoClient prestoClient = queryRunner.getRandomClient();
 
@@ -111,7 +112,7 @@ private static String kafkaTopicName(TpchTable<?> table)
         return TPCH_SCHEMA + "." + table.getTableName().toLowerCase(ENGLISH);
     }
 
-    private static Map<SchemaTableName, KafkaTopicDescription> createTpchTopicDescriptions(Metadata metadata, Iterable<TpchTable<?>> tables, EmbeddedKafka embeddedKafka)
+    private static Map<SchemaTableName, KafkaTopicDescription> createTpchTopicDescriptions(Metadata metadata, Iterable<TpchTable<?>> tables, EmbeddedKafka embeddedKafka, List<SchemaTableName> extraTables)
             throws Exception
     {
         JsonCodec<KafkaTopicDescription> topicDescriptionJsonCodec = new CodecSupplier<>(KafkaTopicDescription.class, metadata).get();
@@ -121,7 +122,15 @@ private static Map<SchemaTableName, KafkaTopicDescription> createTpchTopicDescri
             String tableName = table.getTableName();
             SchemaTableName tpchTable = new SchemaTableName(TPCH_SCHEMA, tableName);
 
-            topicDescriptions.put(loadTpchTopicDescription(topicDescriptionJsonCodec, tpchTable.toString(), tpchTable));
+            topicDescriptions.put(loadTpchTopicDescription(topicDescriptionJsonCodec, tpchTable.toString(), tpchTable, table.getTableName()));
+        }
+
+        for (SchemaTableName extra : extraTables) {
+            topicDescriptions.put(loadTpchTopicDescription(
+                    topicDescriptionJsonCodec,
+                    extra.getTableName(),
+                    extra,
+                    extra.getTableName().toLowerCase() + "_upper"));
         }
 
         List<String> tableNames = new ArrayList<>(4);
@@ -173,7 +182,7 @@ public static void main(String[] args)
             throws Exception
     {
         Logging.initialize();
-        DistributedQueryRunner queryRunner = createKafkaQueryRunner(EmbeddedKafka.createEmbeddedKafka(), TpchTable.getTables());
+        DistributedQueryRunner queryRunner = createKafkaQueryRunner(EmbeddedKafka.createEmbeddedKafka(), TpchTable.getTables(), ImmutableMap.of(), ImmutableList.of());
         Thread.sleep(10);
         Logger log = Logger.get(KafkaQueryRunner.class);
         log.info("======== SERVER STARTED ========");

presto-kafka/src/test/java/com/facebook/presto/kafka/TestKafkaConnectorConfig.java

@@ -33,7 +33,8 @@ public void testDefaults()
                 .setHideInternalColumns(true)
                 .setMaxPartitionFetchBytes(1048576)
                 .setMaxPollRecords(500)
-                .setResourceConfigFiles(""));
+                .setResourceConfigFiles("")
+                .setCaseSensitiveNameMatching(false));
     }
 
     @Test
@@ -51,6 +52,7 @@ public void testExplicitPropertyMappings()
                 .put("kafka.max-partition-fetch-bytes", "1024")
                 .put("kafka.max-poll-records", "1000")
                 .put("kafka.config.resources", tempFile1 + "," + tempFile2)
+                .put("case-sensitive-name-matching", "true")
                 .build();
 
         KafkaConnectorConfig expected = new KafkaConnectorConfig()
@@ -61,7 +63,8 @@ public void testExplicitPropertyMappings()
                 .setHideInternalColumns(false)
                 .setMaxPartitionFetchBytes(1024)
                 .setMaxPollRecords(1000)
-                .setResourceConfigFiles(tempFile1 + "," + tempFile2);
+                .setResourceConfigFiles(tempFile1 + "," + tempFile2)
+                .setCaseSensitiveNameMatching(true);
 
         ConfigAssertions.assertFullMapping(properties, expected);
     }

presto-kafka/src/test/java/com/facebook/presto/kafka/TestKafkaDistributed.java

@@ -16,6 +16,8 @@
 import com.facebook.presto.kafka.util.EmbeddedKafka;
 import com.facebook.presto.testing.QueryRunner;
 import com.facebook.presto.tests.AbstractTestQueries;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
 import io.airlift.tpch.TpchTable;
 import org.testng.annotations.AfterClass;
 import org.testng.annotations.Test;
@@ -36,7 +38,7 @@ protected QueryRunner createQueryRunner()
             throws Exception
     {
         this.embeddedKafka = createEmbeddedKafka();
-        return createKafkaQueryRunner(embeddedKafka, TpchTable.getTables());
+        return createKafkaQueryRunner(embeddedKafka, TpchTable.getTables(), ImmutableMap.of(), ImmutableList.of());
     }
 
     @AfterClass(alwaysRun = true)