fix: Add no_source hint to prevent a sink-only table marked as source

sqrl-planner/src/main/java/com/datasqrl/error/ErrorCode.java

@@ -53,7 +53,8 @@ public enum ErrorCode implements ErrorLabel {
   NOT_YET_IMPLEMENTED,
   NO_API_ENDPOINTS,
   MISSING_SORT_COLUMN,
-  ROWTIME_IS_NULLABLE;
+  ROWTIME_IS_NULLABLE,
+  UNBOUNDED_BATCH_SOURCE;
 
   @Override
   public String getLabel() {

sqrl-planner/src/main/java/com/datasqrl/error/ValidationExceptionHandler.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright © 2021 DataSQRL (contact@datasqrl.com)
+ *
+ * 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 com.datasqrl.error;
+
+import com.google.auto.service.AutoService;
+import org.apache.flink.table.api.ValidationException;
+
+@AutoService(ErrorHandler.class)
+public class ValidationExceptionHandler implements ErrorHandler<ValidationException> {
+
+  @Override
+  public ErrorMessage handle(ValidationException e, ErrorLocation baseLocation) {
+    var errCode = ErrorCode.GENERIC;
+    var msg = e.getMessage();
+
+    if (msg != null && msg.contains("table source is unbounded")) {
+      errCode = ErrorCode.UNBOUNDED_BATCH_SOURCE;
+    }
+
+    return new ErrorMessage.Implementation(errCode, msg, baseLocation, ErrorMessage.Severity.FATAL);
+  }
+
+  @Override
+  public Class<ValidationException> getHandleClass() {
+    return ValidationException.class;
+  }
+}

sqrl-planner/src/main/java/com/datasqrl/planner/Sqrl2FlinkSQLTranslator.java

@@ -740,10 +740,16 @@ public Optional<TableAnalysis> createTable(
       Optional<MutationBuilder> mutationBuilder,
       SchemaLoader schemaLoader,
       HintsAndDoc hintsAndDoc) {
+
     var result = addTable(Function.identity(), tableDefinition, schemaLoader, mutationBuilder);
     hintsAndDoc = updateDocumentationFromLike(result, hintsAndDoc);
-    if (result.isSourceTable()) return Optional.of(addSourceTable(result, hintsAndDoc));
-    else return Optional.empty();
+
+    if (!result.isSourceTable() || hintsAndDoc.hints().isNoSource()) {
+      return Optional.empty();
+    }
+
+    var srcTable = addSourceTable(result, hintsAndDoc);
+    return Optional.of(srcTable);
   }
 
   private HintsAndDoc updateDocumentationFromLike(

sqrl-planner/src/main/java/com/datasqrl/planner/hint/NoSourceHint.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright © 2021 DataSQRL (contact@datasqrl.com)
+ *
+ * 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 com.datasqrl.planner.hint;
+
+import com.datasqrl.planner.parser.ParsedObject;
+import com.datasqrl.planner.parser.SqrlHint;
+import com.google.auto.service.AutoService;
+
+/** Makes sure a table will not be registered as a source. */
+public class NoSourceHint extends PlannerHint {
+
+  public static final String HINT_NAME = "no_source";
+
+  protected NoSourceHint(ParsedObject<SqrlHint> source) {
+    super(source, Type.DAG);
+  }
+
+  @AutoService(Factory.class)
+  public static class NoSourceHintFactory implements Factory {
+
+    @Override
+    public PlannerHint create(ParsedObject<SqrlHint> source) {
+      return new NoSourceHint(source);
+    }
+
+    @Override
+    public String getName() {
+      return HINT_NAME;
+    }
+  }
+}

sqrl-planner/src/main/java/com/datasqrl/planner/hint/PlannerHints.java

@@ -76,6 +76,10 @@ public boolean isWorkload() {
     return getHint(WorkloadHint.class).isPresent();
   }
 
+  public boolean isNoSource() {
+    return getHint(NoSourceHint.class).isPresent();
+  }
+
   public <H> Optional<H> getHint(Class<H> hintClass) {
     return getHints(hintClass).findFirst();
   }

sqrl-planner/src/main/resources/com/datasqrl/error/errorCodes/unbounded_batch_source.md

@@ -0,0 +1 @@
+In case this table is solely used as a sink, apply the /*+ no_source */ hint to avoid marking it as a source.

sqrl-testing/sqrl-testing-integration/src/test/java/com/datasqrl/DAGPlannerTest.java

@@ -78,7 +78,7 @@ void scripts(Path script) {
   @Disabled
   @Test
   void specificScript() {
-    var script = SCRIPT_DIR.resolve("mutationNestedTypeTest.sqrl");
+    var script = SCRIPT_DIR.resolve("noSourceHint.sqrl");
     scripts(script);
   }
 

sqrl-testing/sqrl-testing-integration/src/test/java/com/datasqrl/UseCaseCompileTest.java

@@ -66,7 +66,7 @@ void testUseCase(Path packageFile) {
   @Test
   @Disabled("Intended for manual usage")
   void runTestCaseByName() {
-    var pkg = USECASE_DIR.resolve("banking").resolve("package.json");
+    var pkg = USECASE_DIR.resolve("batch-to-kafka-compile").resolve("package-fail.json");
     UseCaseTestHelper.testUseCase(
         snapshotExtension,
         getClass(),

sqrl-testing/sqrl-testing-integration/src/test/resources/dagplanner/noSourceHint.sqrl

@@ -0,0 +1,17 @@
+IMPORT ecommerceTs.orders;
+
+MaterializedOrders := SELECT * FROM Orders;
+
+/*+ no_source */
+CREATE TABLE KafkaSink (
+    PRIMARY KEY (id, `time`) NOT ENFORCED
+) WITH (
+    'connector' = 'upsert-kafka',
+    'properties.bootstrap.servers' = '${KAFKA_BOOTSTRAP_SERVERS}',
+    'key.format' = 'json',
+    'value.format' = 'json',
+    'value.fields-include' = 'ALL',
+    'topic' = 'customers'
+) LIKE Orders__schema;
+
+EXPORT MaterializedOrders TO KafkaSink;