Allow submitting statement w/o catalog/database

src/codelens/flinkSqlProvider.test.ts

@@ -221,11 +221,16 @@ describe("codelens/flinkSqlProvider.ts FlinkSqlCodelensProvider", () => {
 
       const codeLenses: CodeLens[] = await provider.provideCodeLenses(fakeDocument);
 
-      assert.strictEqual(codeLenses.length, 3);
+      assert.strictEqual(codeLenses.length, 4);
 
-      const poolLens = codeLenses[0];
-      const dbLens = codeLenses[1];
-      const resetLens = codeLenses[2];
+      const submitLens = codeLenses[0];
+      const poolLens = codeLenses[1];
+      const dbLens = codeLenses[2];
+      const resetLens = codeLenses[3];
+
+      assert.strictEqual(submitLens.command?.command, "confluent.statements.create");
+      assert.strictEqual(submitLens.command?.title, "▶️ Submit Statement");
+      assert.deepStrictEqual(submitLens.command?.arguments, [fakeDocument.uri, pool, undefined]);
 
       assert.strictEqual(dbLens.command?.command, "confluent.document.flinksql.setCCloudDatabase");
       assert.strictEqual(dbLens.command?.title, "Set Catalog & Database");
@@ -296,6 +301,51 @@ describe("codelens/flinkSqlProvider.ts FlinkSqlCodelensProvider", () => {
     assert.strictEqual(resetLens.command?.title, "Clear Settings");
     assert.deepStrictEqual(resetLens.command?.arguments, [fakeDocument.uri]);
   });
+
+  it("should provide 'Submit Statement' codelens when only a compute pool is set (without database)", async () => {
+    const pool: CCloudFlinkComputePool = TEST_CCLOUD_FLINK_COMPUTE_POOL;
+    // simulate stored compute pool metadata (but no database)
+    resourceManagerStub.getUriMetadata.resolves({
+      [UriMetadataKeys.FLINK_COMPUTE_POOL_ID]: pool.id,
+      // no database metadata set
+    });
+    const envWithPool: CCloudEnvironment = new CCloudEnvironment({
+      ...TEST_CCLOUD_ENVIRONMENT,
+      flinkComputePools: [pool],
+    });
+    ccloudLoaderStub.getEnvironments.resolves([envWithPool]);
+
+    const codeLenses: CodeLens[] = await provider.provideCodeLenses(fakeDocument);
+
+    assert.strictEqual(codeLenses.length, 4);
+
+    const submitLens = codeLenses[0];
+    const poolLens = codeLenses[1];
+    const dbLens = codeLenses[2];
+    const resetLens = codeLenses[3];
+
+    assert.strictEqual(submitLens.command?.command, "confluent.statements.create");
+    assert.strictEqual(submitLens.command?.title, "▶️ Submit Statement");
+    assert.deepStrictEqual(submitLens.command?.arguments, [fakeDocument.uri, pool, undefined]);
+
+    assert.strictEqual(
+      poolLens.command?.command,
+      "confluent.document.flinksql.setCCloudComputePool",
+    );
+    assert.strictEqual(poolLens.command?.title, pool.name);
+    assert.deepStrictEqual(poolLens.command?.arguments, [fakeDocument.uri, undefined]);
+
+    assert.strictEqual(dbLens.command?.command, "confluent.document.flinksql.setCCloudDatabase");
+    assert.strictEqual(dbLens.command?.title, "Set Catalog & Database");
+    assert.deepStrictEqual(dbLens.command?.arguments, [fakeDocument.uri, pool]);
+
+    assert.strictEqual(
+      resetLens.command?.command,
+      "confluent.document.flinksql.resetCCloudMetadata",
+    );
+    assert.strictEqual(resetLens.command?.title, "Clear Settings");
+    assert.deepStrictEqual(resetLens.command?.arguments, [fakeDocument.uri]);
+  });
 });
 
 describe("codelens/flinkSqlProvider.ts getComputePoolFromMetadata()", () => {

src/codelens/flinkSqlProvider.ts

@@ -140,7 +140,7 @@ export class FlinkSqlCodelensProvider extends DisposableCollection implements Co
     };
     const resetLens = new CodeLens(range, resetCommand);
 
-    if (computePool && database) {
+    if (computePool) {
       const submitCommand: Command = {
         title: "▶️ Submit Statement",
         command: "confluent.statements.create",
@@ -151,7 +151,7 @@ export class FlinkSqlCodelensProvider extends DisposableCollection implements Co
       // show the "Submit Statement" | <current pool> | <current catalog+db> codelenses
       codeLenses.push(submitLens, computePoolLens, databaseLens, resetLens);
     } else {
-      // don't show the submit codelens if we don't have a compute pool and database
+      // don't show the submit codelens if we don't have a compute pool
       codeLenses.push(computePoolLens, databaseLens, resetLens);
     }
 

src/commands/flinkStatements.ts

@@ -70,7 +70,8 @@ export async function viewStatementSqlCommand(statement: FlinkStatement): Promis
  *  2) Create **statement name** (auto-generated from template pattern, but user can override)
  *  3) (If no `pool` is passed): show a quickpick to **choose a Flink compute pool** to send the statement
  *  4) (if no `database` is passed): show a quickpick to **choose a database** (Kafka cluster) to
- *     submit along with the **catalog name** (the environment, inferable from the chosen database).
+ *     submit along with the **catalog name** (the environment, inferable from the chosen database),
+ *     user can choose to skip this, their query will only work if the table name is fully qualified.
  *  5) Submit!
  *  6) Show error notification for any submission errors.
  *  7) Refresh the statements view if the view is focused on the chosen compute pool.
@@ -121,11 +122,8 @@ export async function submitFlinkStatementCommand(
   const currentDatabaseKafkaCluster: KafkaCluster | undefined = validDatabaseProvided
     ? database
     : await flinkDatabaseQuickpick(computePool);
-  if (!currentDatabaseKafkaCluster) {
-    funcLogger.debug("User canceled the default database quickpick");
-    return;
-  }
-  const currentDatabase = currentDatabaseKafkaCluster.name;
+  // Database selection is now optional - user can dismiss the quickpick and submit without it
+  const currentDatabase = currentDatabaseKafkaCluster?.name;
 
   // 5. Prep to submit, submit.
   const submission: IFlinkStatementSubmitParameters = {
@@ -135,7 +133,7 @@ export async function submitFlinkStatementCommand(
     hidden: false, // Do not create a hidden statement, the user authored it.
     properties: new FlinkSpecProperties({
       currentDatabase,
-      currentCatalog: currentDatabaseKafkaCluster.environmentId,
+      currentCatalog: currentDatabaseKafkaCluster?.environmentId,
       localTimezone: localTimezoneOffset(),
     }),
   };

src/quickpicks/kafkaClusters.test.ts

@@ -220,12 +220,17 @@ describe("flinkDatabaseQuickPick", () => {
     await flinkDatabaseQuickpick(computePool);
     const itemsCalledWith = showQuickPickStub.getCall(0).args[0];
     // one separator for the single environment, one for the single cloud
-    // cluster in same provider/region.
-    assert.strictEqual(itemsCalledWith.length, 2);
+    // cluster in same provider/region, plus separator with message and skip option
+    assert.strictEqual(itemsCalledWith.length, 4);
     assert.strictEqual(itemsCalledWith[0].kind, QuickPickItemKind.Separator);
-    // other two items are the clusters. Their description should be the id.
-    // and their .value should be the cluster.
+    // cluster item should be the second item (after environment separator)
     assert.strictEqual(itemsCalledWith[1].description, ccloudClusters[0].id);
     assert.strictEqual(itemsCalledWith[1].value, TEST_CCLOUD_KAFKA_CLUSTER);
+    // third item should be the separator before skip with message
+    assert.strictEqual(itemsCalledWith[2].kind, QuickPickItemKind.Separator);
+    assert.strictEqual(itemsCalledWith[2].label, "Must use fully qualified table name");
+    // fourth item should be the skip option
+    assert.strictEqual(itemsCalledWith[3].label, "Skip");
+    assert.strictEqual(itemsCalledWith[3].value, null);
   });
 });

src/quickpicks/kafkaClusters.ts

@@ -34,6 +34,8 @@
   placeHolder?: string;
   /** Function to filter the list of Kafka clusters before making quickpick items. */
   filter?: KafkaClusterFilter;
+  /** When true, adds a "Skip" option at the bottom of the list that returns undefined when selected */
+  allowSkip?: boolean;
 };
 
 /**
@@ -56,7 +58,7 @@
  * ---------------------------------- Other: directEnv1
  * direct-cluster1 (direct-cluster-id1)
  */
 export async function kafkaClusterQuickPick(
   options: KafkaClusterQuickPickOptions = {},
 ): Promise<KafkaCluster | undefined> {
   const environments: Environment[] = [];
@@ -157,6 +159,20 @@
     });
   }
 
+  // Add skip option at the bottom if requested
+  if (options.allowSkip) {
+    clusterItems.push({
+      kind: QuickPickItemKind.Separator,
+      label: "Must use fully qualified table name",
+    });
+    clusterItems.push({
+      label: "Skip",
+      description: "Continue without selecting a database",
+      iconPath: new ThemeIcon("testing-skipped-icon"),
+      value: null as any,
+    });
+  }
+
   // prompt the user to select a Kafka Cluster
   const chosenClusterItem: QuickPickItemWithValue<KafkaCluster> | undefined =
     await window.showQuickPick(clusterItems, {
@@ -171,14 +187,15 @@
  * @param computePool The compute pool to use as the context for the quickpick (limits to clusters in same cloud provider/region).
  * @param placeholder Optionally override the placeholder text for the quickpick.
  *                    Defaults to "Select the Kafka cluster to use as the default database for the statement".
- * @returns chosen Kafka cluster or undefined if the user cancels the quickpick.
+ * @returns chosen Kafka cluster or undefined if the user cancels the quickpick or selects "Skip".
  */
 export async function flinkDatabaseQuickpick(
   computePool: CCloudFlinkComputePool,
   placeholder: string = "Select the Kafka cluster to use as the default database for the statement",
 ): Promise<KafkaCluster | undefined> {
   return await kafkaClusterQuickPick({
     placeHolder: placeholder,
+    allowSkip: true,
     filter: (cluster: KafkaCluster) => {
       if (!isCCloud(cluster)) {
         // Only CCloud clusters are supported for Flink compute pools.