Handle dotted table names in quoted identifiers (#248)

A quoted identifier segment may legitimately contain a dot, e.g. the Kafka
topic "KAFKA"."my.event". Several code paths took a bare identifier string and
naively split("\\."), shredding "my.event" into "my" + "event" and breaking
!describe / !graph / !resolve and graph / custom-resource lookups.

Add IdentifierUtils.parseIdentifier, which splits unquoted input directly on
'.' (exact, since an unquoted segment cannot contain a dot — this also covers
the unquoted hyphenated names the CLI commands accept, e.g.
LOGICAL.testevent-graph) and uses the SQL parser only for quoted input so a
dot inside a quoted segment is preserved. Malformed quoted identifiers now
error rather than silently mis-splitting.

Wire it into GraphService.resolve, the !resolve/!describe CLI and quidem
paths, and PipelineGraphBuilder. K8s object names already permit dots, so
canonicalization is left unchanged (backwards compatible with existing dotted
resource names).

Add unit tests for IdentifierUtils, dot-preservation tests for
K8sUtils.canonicalizeName, a GraphService test, and a Kafka integration
script that creates/describes/drops a topic whose name contains a dot.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: jogrogan

hoptimator-cli/src/main/java/sqlline/HoptimatorAppConfig.java

@@ -11,6 +11,7 @@
 import com.linkedin.hoptimator.jdbc.ResolvedTable;
 import com.linkedin.hoptimator.jdbc.ddl.SqlCreateMaterializedView;
 import com.linkedin.hoptimator.util.DeploymentService;
+import com.linkedin.hoptimator.util.IdentifierUtils;
 import com.linkedin.hoptimator.util.planner.PipelineRel;
 import org.apache.calcite.plan.RelOptTable;
 import org.apache.calcite.rel.RelRoot;
@@ -24,7 +25,6 @@
 
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
@@ -205,7 +205,7 @@ public void execute(String line, DispatchCallback dispatchCallback) {
         dispatchCallback.setToFailure();
         return;
       }
-      List<String> tablePath = Arrays.asList(split[1].split("\\."));
+      List<String> tablePath = IdentifierUtils.parseIdentifier(split[1]);
       HoptimatorConnection conn = (HoptimatorConnection) sqlline.getConnection();
       try {
         ResolvedTable resolved = conn.resolve(tablePath, Collections.emptyMap());

hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/GraphService.java

@@ -1,7 +1,6 @@
 package com.linkedin.hoptimator.jdbc;
 
 import java.sql.SQLException;
-import java.util.Arrays;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.ServiceLoader;
@@ -15,6 +14,7 @@
 import com.linkedin.hoptimator.graph.GraphRenderer;
 import com.linkedin.hoptimator.graph.GraphTarget;
 import com.linkedin.hoptimator.graph.PipelineGraph;
+import com.linkedin.hoptimator.util.IdentifierUtils;
 import com.linkedin.hoptimator.util.planner.HoptimatorJdbcSchema;
 import org.apache.calcite.util.Util;
 
@@ -76,7 +76,7 @@ public static PipelineGraph buildGraph(String identifier, int depth, HoptimatorC
    * {@code HoptimatorJdbcSchema} lazily walks its downstream connection to find the marker.
    */
   static GraphTarget resolve(String identifier, HoptimatorConnection connection) throws SQLException {
-    List<String> fullPath = Arrays.asList(identifier.split("\\."));
+    List<String> fullPath = IdentifierUtils.parseIdentifier(identifier);
     SchemaPlus schema = connection.calciteConnection().getRootSchema();
     List<String> schemaPath = Util.skipLast(fullPath);
     for (String segment : schemaPath) {

hoptimator-jdbc/src/test/java/com/linkedin/hoptimator/jdbc/GraphServiceTest.java

@@ -81,6 +81,21 @@ void resolveTwoLevelResourceFindsDatabaseCrName() throws SQLException {
     assertEquals(Arrays.asList("ADS", "AD_CLICKS"), out.path());
   }
 
+  @Test
+  void resolveQuotedDottedTableNameTreatsQuotedSegmentAsOneIdentity() throws SQLException {
+    // User typed "VENICE"."my.table". The dot inside the quoted segment is part of the table
+    // name, NOT a path separator — so this must resolve to schema VENICE, table "my.table".
+    // A naive identifier.split("\\.") shreds this into [VENICE, my, table] and fails.
+    SchemaPlus venice = schemaWithDatabaseAndTable("venice-database", "my.table", plainTable());
+    stubRootSchema(schemaWithSubs("VENICE", venice));
+
+    GraphTarget.Resource out = (GraphTarget.Resource) GraphService.resolve(
+        "\"VENICE\".\"my.table\"", connection);
+
+    assertEquals("venice-database", out.database());
+    assertEquals(Arrays.asList("VENICE", "my.table"), out.path());
+  }
+
   @Test
   void resolveThreeLevelResourceWalksCatalogAndSchema() throws SQLException {
     // User typed MYSQL.testdb.orders. MYSQL is the catalog sub-schema; testdb is the Database;

hoptimator-jdbc/src/testFixtures/java/com/linkedin/hoptimator/jdbc/QuidemTestBase.java

@@ -13,6 +13,7 @@
 
 import com.linkedin.hoptimator.graph.PipelineGraph;
 import com.linkedin.hoptimator.graph.mermaid.MermaidRenderer;
+import com.linkedin.hoptimator.util.IdentifierUtils;
 
 import java.io.File;
 import java.io.FileReader;
@@ -98,9 +99,10 @@ public void execute(Context context, boolean execute) throws Exception {
                 throw new IllegalArgumentException("Usage: !describe \"SCHEMA\".\"TABLE\" or \"CATALOG\".\"SCHEMA\".\"TABLE\"");
               }
 
-              String tablePath = parts[1].trim().replaceAll("\"", "");
-              String[] pathParts = tablePath.split("\\.");
-              if (pathParts.length < 2) {
+              // Parse quote-aware so a dot inside a quoted segment (e.g. "KAFKA"."my.event") stays
+              // part of the name rather than being treated as a path separator.
+              List<String> pathParts = IdentifierUtils.parseIdentifier(parts[1].trim());
+              if (pathParts.size() < 2) {
                 throw new IllegalArgumentException("Table path must be at least SCHEMA.TABLE");
               }
 
@@ -109,24 +111,24 @@ public void execute(Context context, boolean execute) throws Exception {
               SchemaPlus schema;
               String tableName;
 
-              if (pathParts.length == 3) {
+              if (pathParts.size() == 3) {
                 // 3-level path: CATALOG.SCHEMA.TABLE (e.g., MySQL)
-                SchemaPlus catalog = rootSchema.subSchemas().get(pathParts[0]);
+                SchemaPlus catalog = rootSchema.subSchemas().get(pathParts.get(0));
                 if (catalog == null) {
-                  throw new IllegalArgumentException("Catalog not found: " + pathParts[0]);
+                  throw new IllegalArgumentException("Catalog not found: " + pathParts.get(0));
                 }
-                schema = catalog.subSchemas().get(pathParts[1]);
+                schema = catalog.subSchemas().get(pathParts.get(1));
                 if (schema == null) {
-                  throw new IllegalArgumentException("Schema not found: " + pathParts[1] + " in catalog " + pathParts[0]);
+                  throw new IllegalArgumentException("Schema not found: " + pathParts.get(1) + " in catalog " + pathParts.get(0));
                 }
-                tableName = pathParts[2];
+                tableName = pathParts.get(2);
               } else {
                 // 2-level path: SCHEMA.TABLE (e.g., Kafka, Venice)
-                schema = rootSchema.subSchemas().get(pathParts[0]);
+                schema = rootSchema.subSchemas().get(pathParts.get(0));
                 if (schema == null) {
-                  throw new IllegalArgumentException("Schema not found: " + pathParts[0]);
+                  throw new IllegalArgumentException("Schema not found: " + pathParts.get(0));
                 }
-                tableName = pathParts[pathParts.length - 1];
+                tableName = pathParts.get(pathParts.size() - 1);
               }
 
               Table table = schema.tables().get(tableName);

hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/PipelineGraphBuilder.java

@@ -23,6 +23,7 @@
 import com.linkedin.hoptimator.graph.GraphEdge;
 import com.linkedin.hoptimator.graph.GraphNode;
 import com.linkedin.hoptimator.graph.PipelineGraph;
+import com.linkedin.hoptimator.util.IdentifierUtils;
 import com.linkedin.hoptimator.k8s.models.V1alpha1LogicalTable;
 import com.linkedin.hoptimator.k8s.models.V1alpha1LogicalTableList;
 import com.linkedin.hoptimator.k8s.models.V1alpha1LogicalTableSpec;
@@ -84,7 +85,7 @@ public PipelineGraph forView(String name) throws SQLException {
     // stored under a canonicalized name (lowercase, {@code _} stripped, {@code $} → {@code -},
     // dot-separated parts joined with {@code -}). Canonicalization is idempotent, so passing the
     // already-canonical custom resource name still works.
-    String crName = K8sUtils.canonicalizeName(Arrays.asList(name.split("\\.")));
+    String crName = K8sUtils.canonicalizeName(IdentifierUtils.parseIdentifier(name));
     V1alpha1View view = viewApi.get(crName);
     if (view.getSpec() == null) {
       throw new SQLException("view " + crName + " not found");
@@ -118,7 +119,7 @@ public PipelineGraph forView(String name) throws SQLException {
   public PipelineGraph forLogicalTable(String name) throws SQLException {
     // Same canonicalization as forView — accept SQL-side identifiers and resolve to the
     // canonicalized custom resource name.
-    String crName = K8sUtils.canonicalizeName(Arrays.asList(name.split("\\.")));
+    String crName = K8sUtils.canonicalizeName(IdentifierUtils.parseIdentifier(name));
     V1alpha1LogicalTable lt = logicalTableApi.get(crName);
     Map<String, String> tierMap = tierMap(lt.getSpec());
     GraphNode.LogicalTable root = new GraphNode.LogicalTable(crName, tierMap);

hoptimator-k8s/src/test/java/com/linkedin/hoptimator/k8s/K8sUtilsTest.java

@@ -41,6 +41,20 @@ void canonicalizeNameReplacesUnderscoresAndDollarSigns() {
     assertEquals("my-table", K8sUtils.canonicalizeName("my_$table"));
   }
 
+  @Test
+  void canonicalizeNamePreservesDots() {
+    // Dots are intentionally NOT rewritten: K8s object names here are DNS-1123 subdomains, which
+    // permit dots, and existing deployed resources already carry dotted names. Rewriting dots
+    // would change the derived name and make the operator create duplicates instead of updating
+    // the pre-existing resource. Keeping dots keeps create-time and lookup-time names in agreement.
+    assertEquals("my.event", K8sUtils.canonicalizeName("my.event"));
+  }
+
+  @Test
+  void canonicalizeNameWithDottedTableAndDatabasePreservesDot() {
+    assertEquals("kafka-database-my.event", K8sUtils.canonicalizeName("kafka-database", "my.event"));
+  }
+
   @Test
   void canonicalizeNameWithDatabaseAndTable() {
     assertEquals("mydb-mytable", K8sUtils.canonicalizeName("myDB", "myTable"));

hoptimator-kafka/src/test/java/com/linkedin/hoptimator/kafka/TestSqlScripts.java

@@ -22,4 +22,9 @@ public void kafkaDdlScriptBeamJob() throws Exception {
   public void kafkaDdlCreateTableScript() throws Exception {
     run("kafka-ddl-create-table.id");
   }
+
+  @Test
+  public void kafkaDdlDottedTableScript() throws Exception {
+    run("kafka-ddl-dotted.id");
+  }
 }

hoptimator-kafka/src/test/resources/kafka-ddl-dotted.id

@@ -0,0 +1,30 @@
+!set outputformat mysql
+!use k8s
+
+# ─────────────────────────────────────────────────────────────────────────────
+# A Kafka topic whose *name* contains a dot. Quoted as "KAFKA"."my.event", the
+# dot is part of the topic identity, NOT a catalog/schema separator. Creating,
+# describing, and dropping it must all treat "my.event" as a single identifier.
+# Before the quote-aware identifier fix, !describe split this into a bogus
+# KAFKA.my.event three-level path and failed with "Schema not found: my".
+# ─────────────────────────────────────────────────────────────────────────────
+
+create or replace table "KAFKA"."my.event" ("KEY" VARCHAR, "VALUE" BINARY) WITH ("kafka.partitions" '1');
+(0 rows modified)
+
+!update
+
++------------+-----------+------------+------------+
+| columnName | typeName  | columnSize | isNullable |
++------------+-----------+------------+------------+
+| KEY        | VARCHAR   | null       | YES        |
+| VALUE      | BINARY(1) | 1          | YES        |
++------------+-----------+------------+------------+
+(2 rows)
+
+!describe "KAFKA"."my.event"
+
+drop table "KAFKA"."my.event";
+(0 rows modified)
+
+!update

hoptimator-util/src/main/java/com/linkedin/hoptimator/util/IdentifierUtils.java

@@ -0,0 +1,83 @@
+package com.linkedin.hoptimator.util;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.calcite.avatica.util.Casing;
+import org.apache.calcite.avatica.util.Quoting;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.parser.SqlParser;
+import org.apache.calcite.sql.parser.SqlParseException;
+import org.apache.calcite.sql.validate.SqlConformanceEnum;
+
+
+/** Helpers for turning user-typed SQL identifier strings into their component parts. */
+public final class IdentifierUtils {
+
+  private static final SqlParser.Config PARSER_CONFIG = SqlParser.config()
+      .withQuoting(Quoting.DOUBLE_QUOTE)
+      // Preserve case exactly as typed — Hoptimator schema/table names are case-sensitive.
+      .withUnquotedCasing(Casing.UNCHANGED)
+      .withQuotedCasing(Casing.UNCHANGED)
+      .withConformance(SqlConformanceEnum.BABEL);
+
+  private IdentifierUtils() {
+  }
+
+  /**
+   * Splits a possibly-quoted, dot-separated SQL identifier into its component parts, honoring
+   * double-quote quoting so that a dot <em>inside</em> a quoted segment is treated as part of the
+   * name rather than a path separator. For example {@code "VENICE"."my.table"} yields
+   * {@code [VENICE, my.table]}, not {@code [VENICE, my, table]}.
+   *
+   * <p>Hoptimator accepts bare identifier strings on several paths (the {@code !graph}/
+   * {@code !resolve}/{@code !describe} CLI commands, graph resolution, custom-resource lookups).
+   * Those paths must not naively {@code split("\\.")}, or they corrupt names that legitimately
+   * contain a dot.
+   *
+   * <p>The work splits cleanly by quoting:
+   * <ul>
+   *   <li><b>Unquoted input</b> ({@code no '"'}) is split directly on {@code .}. This is exact: an
+   *       unquoted segment cannot itself contain a dot, so there is nothing for the SQL parser to
+   *       disambiguate. It also covers names that are <em>not</em> valid standalone SQL identifiers
+   *       but that Hoptimator accepts on its bare-string CLI commands — e.g. the unquoted
+   *       hyphenated {@code LOGICAL.testevent-graph} (which the SQL grammar would read as the
+   *       subtraction {@code testevent - graph}).</li>
+   *   <li><b>Quoted input</b> is the <em>only</em> reason this helper exists: it runs the SQL
+   *       parser so a dot inside a quoted segment ({@code "my.table"}) stays intact. If the quoted
+   *       input does not parse to a single identifier — e.g. a malformed or mixed form like
+   *       {@code "KAFKA".my-topic}, where an unquoted hyphenated segment is illegal — it throws
+   *       rather than silently mis-splitting.</li>
+   * </ul>
+   *
+   * @throws IllegalArgumentException if {@code identifier} is quoted but is not a well-formed
+   *     dotted identifier.
+   */
+  public static List<String> parseIdentifier(String identifier) {
+    if (identifier == null) {
+      return new ArrayList<>();
+    }
+    // Unquoted: a plain dot-split is exact (no segment can contain a dot). This is also the path
+    // for the unquoted hyphenated names Hoptimator's CLI commands accept but the SQL grammar does
+    // not, so it must NOT go through the parser.
+    if (identifier.indexOf('"') < 0) {
+      return new ArrayList<>(Arrays.asList(identifier.split("\\.")));
+    }
+    // Quoted: use the parser so a dot inside a quoted segment is preserved.
+    try {
+      SqlNode node = SqlParser.create(identifier, PARSER_CONFIG).parseExpression();
+      if (node instanceof SqlIdentifier) {
+        SqlIdentifier id = (SqlIdentifier) node;
+        if (!id.isStar() && !id.names.isEmpty()) {
+          return new ArrayList<>(id.names);
+        }
+      }
+      // Parsed, but not a plain identifier (e.g. "KAFKA".my-topic reads as a subtraction).
+      throw new IllegalArgumentException("Not a well-formed quoted table identifier: " + identifier);
+    } catch (SqlParseException e) {
+      throw new IllegalArgumentException("Not a well-formed quoted table identifier: " + identifier, e);
+    }
+  }
+}

hoptimator-util/src/test/java/com/linkedin/hoptimator/util/IdentifierUtilsTest.java

@@ -0,0 +1,97 @@
+package com.linkedin.hoptimator.util;
+
+import java.util.List;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+
+class IdentifierUtilsTest {
+
+  @Test
+  void quotedDottedSegmentStaysWhole() {
+    // The dot inside "my.table" is part of the name, not a separator.
+    assertThat(IdentifierUtils.parseIdentifier("\"VENICE\".\"my.table\""))
+        .containsExactly("VENICE", "my.table");
+  }
+
+  @Test
+  void quotedDottedTopicStaysWhole() {
+    assertThat(IdentifierUtils.parseIdentifier("\"KAFKA\".\"my.event\""))
+        .containsExactly("KAFKA", "my.event");
+  }
+
+  @Test
+  void unquotedIdentifierSplitsOnDots() {
+    assertThat(IdentifierUtils.parseIdentifier("ADS.AD_CLICKS"))
+        .containsExactly("ADS", "AD_CLICKS");
+  }
+
+  @Test
+  void threeLevelUnquotedIdentifierSplits() {
+    assertThat(IdentifierUtils.parseIdentifier("MYSQL.testdb.orders"))
+        .containsExactly("MYSQL", "testdb", "orders");
+  }
+
+  @Test
+  void unquotedHyphenatedNameSplitsOnDots() {
+    // Calcite would read "LOGICAL.testevent-graph" as arithmetic (subtraction), but unquoted input
+    // never reaches the parser: a plain dot-split is exact because an unquoted segment cannot
+    // contain a dot. This is the form the !graph/!resolve/!describe CLI commands accept.
+    assertThat(IdentifierUtils.parseIdentifier("LOGICAL.testevent-graph"))
+        .containsExactly("LOGICAL", "testevent-graph");
+  }
+
+  @Test
+  void malformedQuotedIdentifierThrows() {
+    // Quoted input must be a well-formed identifier. A mixed form — quoted schema with an unquoted
+    // hyphenated segment — is illegal (the SQL grammar reads it as subtraction), so surface it
+    // rather than silently mis-splitting.
+    assertThatThrownBy(() -> IdentifierUtils.parseIdentifier("\"KAFKA\".my-topic"))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("\"KAFKA\".my-topic");
+  }
+
+  @Test
+  void unterminatedQuoteThrows() {
+    assertThatThrownBy(() -> IdentifierUtils.parseIdentifier("\"KAFKA\".\"my.event"))
+        .isInstanceOf(IllegalArgumentException.class);
+  }
+
+  @Test
+  void quotedHyphenDollarNamePreservesSpecialChars() {
+    assertThat(IdentifierUtils.parseIdentifier("VENICE.\"test-store$insert-partial\""))
+        .containsExactly("VENICE", "test-store$insert-partial");
+  }
+
+  @Test
+  void casePreservedForUnquotedAndQuoted() {
+    assertThat(IdentifierUtils.parseIdentifier("profile.Members"))
+        .containsExactly("profile", "Members");
+  }
+
+  @Test
+  void singleSegmentReturnsOneElement() {
+    assertThat(IdentifierUtils.parseIdentifier("audience")).containsExactly("audience");
+  }
+
+  @Test
+  void singleQuotedSegmentWithDotStaysWhole() {
+    assertThat(IdentifierUtils.parseIdentifier("\"my.event\"")).containsExactly("my.event");
+  }
+
+  @Test
+  void escapedQuotesInsideSegmentAreUnescaped() {
+    // "a""b" is the SQL-quoted form of the identifier a"b.
+    assertThat(IdentifierUtils.parseIdentifier("\"a\"\"b\".\"c.d\""))
+        .containsExactly("a\"b", "c.d");
+  }
+
+  @Test
+  void nullReturnsEmptyList() {
+    List<String> parts = IdentifierUtils.parseIdentifier(null);
+    assertThat(parts).isEmpty();
+  }
+}