fix: Preserve applied IMPORT hints on importing another .sqrl script

sqrl-cli/src/main/java/com/datasqrl/compile/CompilationProcess.java

@@ -66,7 +66,7 @@ public Pair<PhysicalPlan, TestPlan> executeCompilation(Optional<Path> testsPath)
             buildPath,
             (FlinkStreamEngine) planner.getStreamStage().engine(),
             config.getCompilerConfig());
-    planner.planMain(mainScript, environment);
+    planner.planMain(mainScript, Optional.empty(), environment);
     var dagBuilder = planner.getDagBuilder();
     var dag = dagPlanner.optimize(dagBuilder.getDag());
     var physicalPlan = dagPlanner.assemble(dag, environment);

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

@@ -211,14 +211,19 @@ public SqlScriptPlanner(
   }
 
   /**
-   * Main entry method for parsing a SQRL script. The bulk of this method ensure that exceptions and
-   * errors are correctly mapped to the source so that users can easily understand what the issue is
-   * and what's causing it.
+   * Main entry method for parsing a SQRL script. The bulk of this method ensures that exceptions
+   * and errors are correctly mapped to the source so that users can easily understand what the
+   * issue is and what's causing it.
    *
-   * @param mainScript
-   * @param sqrlEnv
+   * @param mainScript SQRL script to plan
+   * @param inheritedHints inherited hints if the planned script is coming from an IMPORT
+   * @param sqrlEnv the SQRL compilation environment
    */
-  public void planMain(MainScript mainScript, Sqrl2FlinkSQLTranslator sqrlEnv) {
+  public void planMain(
+      MainScript mainScript,
+      Optional<PlannerHints> inheritedHints,
+      Sqrl2FlinkSQLTranslator sqrlEnv) {
+
     var scriptErrors = errorCollector.withScript(mainScript.getPath(), mainScript.getContent());
     var statements = sqrlParser.parseScript(mainScript.getContent(), scriptErrors);
     var statementStack = new ArrayList<StackableStatement>();
@@ -229,7 +234,7 @@ public void planMain(MainScript mainScript, Sqrl2FlinkSQLTranslator sqrlEnv) {
       var sqlStatement = statement.get();
 
       try {
-        planStatement(sqlStatement, statementStack, sqrlEnv, lineErrors);
+        planStatement(sqlStatement, statementStack, inheritedHints, sqrlEnv, lineErrors);
 
       } catch (CollectedException e) {
         throw e;
@@ -302,14 +307,15 @@ public void planMain(MainScript mainScript, Sqrl2FlinkSQLTranslator sqrlEnv) {
   private void planStatement(
       SQLStatement stmt,
       List<StackableStatement> statementStack,
+      Optional<PlannerHints> inheritedHints,
       Sqrl2FlinkSQLTranslator sqrlEnv,
       ErrorCollector errors) {
     // Process hints & doc
     var hints = PlannerHints.EMPTY;
     Optional<String> documentation = Optional.empty();
     if (stmt instanceof SqrlStatement statement) {
       var comments = statement.getComments();
-      hints = PlannerHints.from(comments, errors);
+      hints = PlannerHints.from(comments, inheritedHints, errors);
       if (!comments.documentation().isEmpty()) {
         documentation =
             Optional.of(
@@ -842,7 +848,7 @@ private void addImport(
     NamePath aliasPath = null;
     if (importStmt.getAlias().isPresent()) {
       aliasPath = importStmt.getAlias().get();
-      ;
+
       checkFatal(
           aliasPath.size() == 1,
           ErrorCode.INVALID_IMPORT,
@@ -937,7 +943,11 @@ private void addImport(
         completeScript.append(stmts);
       }
 
-      planMain(scriptObject.getScript(), sqrlEnv);
+      var hints =
+          hintsAndDoc.hints().isEmpty()
+              ? Optional.<PlannerHints>empty()
+              : Optional.of(hintsAndDoc.hints());
+      planMain(scriptObject.getScript(), hints, sqrlEnv);
 
       if (priorContext.hasDifferentDatabase(scriptContext)) {
         // Switch it back

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

@@ -19,6 +19,7 @@
 import com.datasqrl.error.ErrorLabel;
 import com.datasqrl.planner.parser.SqrlComments;
 import com.datasqrl.planner.parser.StatementParserException;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
 import java.util.function.Function;
@@ -34,12 +35,15 @@ public class PlannerHints {
 
   private final List<PlannerHint> hints;
 
-  public static PlannerHints from(SqrlComments comments, ErrorCollector errors) {
-    var hints =
-        comments.hints().stream()
-            .map(c -> PlannerHint.from(c, errors))
-            .flatMap(Optional::stream)
-            .toList();
+  public static PlannerHints from(
+      SqrlComments comments, Optional<PlannerHints> inheritedHints, ErrorCollector errors) {
+    var hints = new ArrayList<PlannerHint>();
+    inheritedHints.ifPresent(inherited -> hints.addAll(inherited.hints));
+
+    comments.hints().stream()
+        .map(c -> PlannerHint.from(c, errors))
+        .flatMap(Optional::stream)
+        .forEach(hints::add);
 
     return new PlannerHints(hints);
   }
@@ -56,6 +60,10 @@ public Optional<ColumnNamesHint> getQueryByHint() {
     return queryBy.stream().map(ColumnNamesHint.class::cast).findFirst();
   }
 
+  public boolean isEmpty() {
+    return hints.isEmpty();
+  }
+
   public boolean isTest() {
     return getHint(TestHint.class).isPresent();
   }

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("unknownHint-warn.sqrl");
+    var script = SCRIPT_DIR.resolve("imports.sqrl");
     scripts(script);
   }
 

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

@@ -1,4 +1,10 @@
 IMPORT ecommerceTs.customer;
-/*+query_by_all(productid) */
+
+/*+ query_by_all(productid) */
 IMPORT ecommerceTs.product;
 
+/*+ ttl(14 days) */
+IMPORT mutation.events.*;
+
+-- Export mutation table explicitly for test snapshot
+EXPORT MyEvents TO logger.myevents;

sqrl-testing/sqrl-testing-integration/src/test/resources/dagplanner/mutation/events.sqrl

@@ -0,0 +1,7 @@
+/*+ no_query, engine(kafka) */
+CREATE TABLE MyEvents (
+    `id` STRING NOT NULL METADATA FROM 'uuid',
+    `payload` STRING,
+    `timestamp` TIMESTAMP_LTZ(3) NOT NULL METADATA FROM 'timestamp',
+    WATERMARK FOR `timestamp` AS `timestamp` - INTERVAL '0.001' SECOND
+);

sqrl-testing/sqrl-testing-integration/src/test/resources/snapshots/com/datasqrl/DAGPlannerTest/imports.txt

@@ -43,6 +43,42 @@ WITH (
   'connector' = 'filesystem'
 )
 LIKE `Customer__schema`
+=== MyEvents
+ID:          default_catalog.default_database.MyEvents
+Type:        stream
+Stage:       flink
+Primary key: id
+Timestamp:   timestamp
+Row count:   ~1e8
+---
+Schema:
+ - id: VARCHAR(2147483647) CHARACTER SET "UTF-16LE" NOT NULL
+ - payload: VARCHAR(2147483647) CHARACTER SET "UTF-16LE"
+ - timestamp: TIMESTAMP_LTZ(3) *ROWTIME* NOT NULL
+Inputs:
+ - default_catalog.default_database.MyEvents__base
+Annotations:
+ - stream-root: MyEvents
+Plan:
+LogicalWatermarkAssigner(rowtime=[timestamp], watermark=[-($2, 1:INTERVAL SECOND)])
+  LogicalProject(id=[$0], payload=[$1], timestamp=[CAST($2):TIMESTAMP_LTZ(3) *ROWTIME* NOT NULL])
+    LogicalTableScan(table=[[default_catalog, default_database, MyEvents, metadata=[timestamp]]])
+SQL:
+CREATE TABLE `MyEvents` (
+  `id` STRING NOT NULL,
+  `payload` STRING,
+  `timestamp` TIMESTAMP_LTZ(3) NOT NULL METADATA FROM 'timestamp',
+  WATERMARK FOR `timestamp` AS `timestamp` - INTERVAL '0.001' SECOND
+)
+WITH (
+  'connector' = 'kafka',
+  'format' = 'flexible-json',
+  'properties.auto.offset.reset' = 'earliest',
+  'properties.bootstrap.servers' = '${KAFKA_BOOTSTRAP_SERVERS}',
+  'properties.compression.type' = 'zstd',
+  'properties.group.id' = '${KAFKA_GROUP_ID}',
+  'topic' = 'kafka-mutation-MyEvents'
+)
 === Product
 ID:          default_catalog.default_database.Product
 Type:        stream
@@ -86,6 +122,14 @@ WITH (
   'connector' = 'filesystem'
 )
 LIKE `Product__schema`
+=== myevents
+ID:          logger.myevents
+Type:        export
+Stage:       flink
+---
+Inputs:
+ - default_catalog.default_database.MyEvents
+
 >>>flink-sql-no-functions.sql
 CREATE TEMPORARY TABLE `Customer__schema` (
   `customerid` BIGINT NOT NULL,
@@ -129,6 +173,21 @@ WITH (
   'connector' = 'filesystem'
 )
 LIKE `Product__schema`;
+CREATE TABLE `MyEvents` (
+  `id` STRING NOT NULL,
+  `payload` STRING,
+  `timestamp` TIMESTAMP_LTZ(3) NOT NULL METADATA FROM 'timestamp',
+  WATERMARK FOR `timestamp` AS `timestamp` - INTERVAL '0.001' SECOND
+)
+WITH (
+  'connector' = 'kafka',
+  'format' = 'flexible-json',
+  'properties.auto.offset.reset' = 'earliest',
+  'properties.bootstrap.servers' = '${KAFKA_BOOTSTRAP_SERVERS}',
+  'properties.compression.type' = 'zstd',
+  'properties.group.id' = '${KAFKA_GROUP_ID}',
+  'topic' = 'kafka-mutation-MyEvents'
+);
 CREATE TABLE `Customer_1` (
   `customerid` BIGINT NOT NULL,
   `email` VARCHAR(2147483647) CHARACTER SET `UTF-16LE` NOT NULL,
@@ -146,7 +205,17 @@ WITH (
   'url' = 'jdbc:postgresql://${POSTGRES_AUTHORITY}',
   'username' = '${POSTGRES_USERNAME}'
 );
-CREATE TABLE `Product_2` (
+CREATE TABLE `myevents_2` (
+  `id` VARCHAR(2147483647) CHARACTER SET `UTF-16LE` NOT NULL,
+  `payload` VARCHAR(2147483647) CHARACTER SET `UTF-16LE`,
+  `timestamp` TIMESTAMP(3) WITH LOCAL TIME ZONE NOT NULL,
+  PRIMARY KEY (`id`) NOT ENFORCED
+)
+WITH (
+  'connector' = 'print',
+  'print-identifier' = 'myevents'
+);
+CREATE TABLE `Product_3` (
   `productid` BIGINT NOT NULL,
   `name` VARCHAR(2147483647) CHARACTER SET `UTF-16LE` NOT NULL,
   `description` VARCHAR(2147483647) CHARACTER SET `UTF-16LE` NOT NULL,
@@ -159,7 +228,7 @@ WITH (
   'driver' = 'org.postgresql.Driver',
   'password' = '${POSTGRES_PASSWORD}',
   'sink.on-conflict.action' = 'IGNORE',
-  'table-name' = 'Product_2',
+  'table-name' = 'Product_3',
   'url' = 'jdbc:postgresql://${POSTGRES_AUTHORITY}',
   'username' = '${POSTGRES_USERNAME}'
 );
@@ -168,14 +237,32 @@ INSERT INTO `default_catalog`.`default_database`.`Customer_1`
 SELECT *
  FROM `default_catalog`.`default_database`.`Customer`
 ;
-INSERT INTO `default_catalog`.`default_database`.`Product_2`
+INSERT INTO `default_catalog`.`default_database`.`myevents_2`
  SELECT *
-  FROM `default_catalog`.`default_database`.`Product`
+  FROM `default_catalog`.`default_database`.`MyEvents`
  ;
- END
+ INSERT INTO `default_catalog`.`default_database`.`Product_3`
+  SELECT *
+   FROM `default_catalog`.`default_database`.`Product`
+  ;
+  END
 >>>kafka.json
 {
-  "topics" : [ ],
+  "topics" : [
+    {
+      "topicName" : "kafka-mutation-MyEvents",
+      "tableName" : "MyEvents",
+      "format" : "flexible-json",
+      "numPartitions" : 1,
+      "replicationFactor" : 3,
+      "type" : "MUTATION",
+      "messageKeys" : [ ],
+      "messageSchema" : "",
+      "config" : {
+        "retention.ms" : "1209600000"
+      }
+    }
+  ],
   "testRunnerTopics" : [ ]
 }
 >>>postgres.json
@@ -222,9 +309,9 @@ INSERT INTO `default_catalog`.`default_database`.`Product_2`
       "ttl" : 0.0
     },
     {
-      "name" : "Product_2",
+      "name" : "Product_3",
       "type" : "TABLE",
-      "sql" : "CREATE TABLE IF NOT EXISTS \"Product_2\" (\"productid\" BIGINT NOT NULL, \"name\" TEXT NOT NULL, \"description\" TEXT NOT NULL, \"category\" TEXT NOT NULL, \"_ingest_time\" TIMESTAMP WITH TIME ZONE NOT NULL, PRIMARY KEY (\"productid\",\"name\",\"description\",\"category\"))",
+      "sql" : "CREATE TABLE IF NOT EXISTS \"Product_3\" (\"productid\" BIGINT NOT NULL, \"name\" TEXT NOT NULL, \"description\" TEXT NOT NULL, \"category\" TEXT NOT NULL, \"_ingest_time\" TIMESTAMP WITH TIME ZONE NOT NULL, PRIMARY KEY (\"productid\",\"name\",\"description\",\"category\"))",
       "fields" : [
         {
           "name" : "productid",
@@ -349,7 +436,7 @@ INSERT INTO `default_catalog`.`default_database`.`Product_2`
             ],
             "query" : {
               "type" : "SqlQuery",
-              "sql" : "SELECT *\nFROM \"Product_2\"\nWHERE \"productid\" = $1",
+              "sql" : "SELECT *\nFROM \"Product_3\"\nWHERE \"productid\" = $1",
               "parameters" : [
                 {
                   "type" : "arg",
@@ -364,7 +451,29 @@ INSERT INTO `default_catalog`.`default_database`.`Product_2`
           }
         }
       ],
-      "mutations" : [ ],
+      "mutations" : [
+        {
+          "type" : "kafka",
+          "fieldName" : "MyEvents",
+          "returnList" : false,
+          "topic" : "kafka-mutation-MyEvents",
+          "keyColumns" : [ ],
+          "computedColumns" : {
+            "id" : {
+              "metadataType" : "UUID",
+              "name" : "",
+              "required" : true
+            },
+            "timestamp" : {
+              "metadataType" : "TIMESTAMP",
+              "name" : "",
+              "required" : true
+            }
+          },
+          "transactional" : false,
+          "sinkConfig" : { }
+        }
+      ],
       "subscriptions" : [ ],
       "operations" : [
         {
@@ -423,11 +532,34 @@ INSERT INTO `default_catalog`.`default_database`.`Product_2`
           "mcpMethod" : "TOOL",
           "restMethod" : "GET",
           "uriTemplate" : "queries/Product{?productid,offset,limit}"
+        },
+        {
+          "function" : {
+            "name" : "AddMyEvents",
+            "parameters" : {
+              "type" : "object",
+              "properties" : {
+                "payload" : {
+                  "type" : "string"
+                }
+              },
+              "required" : [ ]
+            }
+          },
+          "format" : "JSON",
+          "apiQuery" : {
+            "query" : "mutation MyEvents($payload: String) {\nMyEvents(event: { payload: $payload }) {\nid\npayload\ntimestamp\n}\n\n}",
+            "queryName" : "MyEvents",
+            "operationType" : "MUTATION"
+          },
+          "mcpMethod" : "TOOL",
+          "restMethod" : "POST",
+          "uriTemplate" : "mutations/MyEvents"
         }
       ],
       "schema" : {
         "type" : "string",
-        "schema" : "type Customer {\n  customerid: Long!\n  email: String!\n  name: String!\n  lastUpdated: Long!\n  timestamp: DateTime!\n}\n\n\"An RFC-3339 compliant Full Date Scalar\"\nscalar Date\n\n\"A DateTime scalar that handles both full RFC3339 and shorter timestamp formats\"\nscalar DateTime\n\n\"A JSON scalar\"\nscalar JSON\n\n\"24-hour clock time value string in the format `hh:mm:ss` or `hh:mm:ss.sss`.\"\nscalar LocalTime\n\n\"A 64-bit signed integer\"\nscalar Long\n\ntype Product {\n  productid: Long!\n  name: String!\n  description: String!\n  category: String!\n}\n\ntype Query {\n  Customer(limit: Int = 10, offset: Int = 0): [Customer!]\n  Product(productid: Long!, limit: Int = 10, offset: Int = 0): [Product!]\n}\n\nenum _McpMethodType {\n  NONE\n  TOOL\n  RESOURCE\n}\n\nenum _RestMethodType {\n  NONE\n  GET\n  POST\n}\n\ndirective @api(mcp: _McpMethodType, rest: _RestMethodType, uri: String) on QUERY | MUTATION | FIELD_DEFINITION\n"
+        "schema" : "type Customer {\n  customerid: Long!\n  email: String!\n  name: String!\n  lastUpdated: Long!\n  timestamp: DateTime!\n}\n\n\"An RFC-3339 compliant Full Date Scalar\"\nscalar Date\n\n\"A DateTime scalar that handles both full RFC3339 and shorter timestamp formats\"\nscalar DateTime\n\n\"A JSON scalar\"\nscalar JSON\n\n\"24-hour clock time value string in the format `hh:mm:ss` or `hh:mm:ss.sss`.\"\nscalar LocalTime\n\n\"A 64-bit signed integer\"\nscalar Long\n\ntype Mutation {\n  MyEvents(event: MyEventsInput!): MyEventsResultOutput!\n}\n\ninput MyEventsInput {\n  payload: String\n}\n\ntype MyEventsResultOutput {\n  id: String!\n  payload: String\n  timestamp: DateTime!\n}\n\ntype Product {\n  productid: Long!\n  name: String!\n  description: String!\n  category: String!\n}\n\ntype Query {\n  Customer(limit: Int = 10, offset: Int = 0): [Customer!]\n  Product(productid: Long!, limit: Int = 10, offset: Int = 0): [Product!]\n}\n\nenum _McpMethodType {\n  NONE\n  TOOL\n  RESOURCE\n}\n\nenum _RestMethodType {\n  NONE\n  GET\n  POST\n}\n\ndirective @api(mcp: _McpMethodType, rest: _RestMethodType, uri: String) on QUERY | MUTATION | FIELD_DEFINITION\n"
       }
     }
   }