Add subscriptions to test infra (draft)

sqrl-planner/src/main/java/com/datasqrl/graphql/generate/GraphqlSchemaFactory.java

@@ -130,7 +130,7 @@ public Optional<GraphQLSchema> generate(ExecutionGoal goal) {
     GraphQLSchema.Builder builder = GraphQLSchema.newSchema()
         .query(queryType);
     if (goal != ExecutionGoal.TEST) {
-      if (logManager.hasLogEngine() && System.getenv().get("ENABLE_SUBSCRIPTIONS") != null) {
+      if (true) {
         Optional<GraphQLObjectType.Builder> subscriptions = createSubscriptionTypes(schema);
         subscriptions.map(builder::subscription);
       }
@@ -226,6 +226,7 @@ public Optional<Builder> createSubscriptionTypes(SqrlSchema schema) {
 
       GraphQLFieldDefinition subscriptionField = GraphQLFieldDefinition.newFieldDefinition()
           .name(tableName)
+          .arguments(createArgumentsWithOptionalScalars(schema.getTableFunctions().get(0)))
           .type(createOutputTypeForRelDataType(table.getRowType(), NamePath.of(tableName), seen).get())
           .build();
 
@@ -242,6 +243,20 @@ public Optional<Builder> createSubscriptionTypes(SqrlSchema schema) {
   }
 
 
+  private List<GraphQLArgument> createArgumentsWithOptionalScalars(SqrlTableMacro field) {
+    if (!allowedArguments(field)) {
+      return List.of();
+    }
+
+    return field.getRowType().getFieldList().stream()
+        .filter(p -> getInputType(p.getType(), NamePath.of(p.getName()), seen).isPresent())
+        .map(parameter -> GraphQLArgument.newArgument()
+            .name(parameter.getName())
+            .type(getInputType(parameter.getType(), NamePath.of(parameter.getName()), seen).get()) // No nonNull here
+            .build())
+        .collect(Collectors.toList());
+  }
+
   private GraphQLObjectType createQueryType(ExecutionGoal goal, List<SqrlTableMacro> relationships) {
 
     List<GraphQLFieldDefinition> fields = new ArrayList<>();

sqrl-planner/src/main/java/com/datasqrl/graphql/generate/GraphqlSchemaUtil.java

@@ -30,6 +30,7 @@
 import lombok.extern.slf4j.Slf4j;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.sql.type.SqlTypeName;
 import org.apache.flink.table.planner.plan.schema.RawRelDataType;
 
 @Slf4j
@@ -59,8 +60,15 @@ public static boolean isValidGraphQLName(String name) {
   }
 
   public static Optional<GraphQLInputType> getInputType(RelDataType type, NamePath namePath, Set<String> seen) {
-    return getInOutType(type, namePath, seen)
-        .map(f->(GraphQLInputType)f);
+    if (type.getSqlTypeName() == SqlTypeName.ARRAY || type.getSqlTypeName() == SqlTypeName.MULTISET ||
+    type.getSqlTypeName() == SqlTypeName.ROW
+    ) {
+      return Optional.empty(); // Exclude arrays
+    }
+
+    Optional<GraphQLInputType> graphQLInputType = getInOutType(type, namePath, seen)
+          .map(f -> (GraphQLInputType) f);
+      return graphQLInputType;
   }
 
   public static Optional<GraphQLOutputType> getOutputType(RelDataType type, NamePath namePath, Set<String> seen) {
@@ -185,18 +193,20 @@ private static String toName(NamePath namePath, String postfix) {
 
   public static Optional<GraphQLOutputType> createOutputTypeForRelDataType(RelDataType type,
       NamePath namePath, Set<String> seen) {
+    Optional<GraphQLOutputType> outputType = getOutputType(type, namePath, seen);
     if (!type.isNullable()) {
-      return getOutputType(type, namePath, seen).map(GraphQLNonNull::nonNull);
+      return outputType.map(GraphQLNonNull::nonNull);
     }
-    return getOutputType(type, namePath, seen);
+    return outputType;
   }
 
   public static Optional<GraphQLInputType> createInputTypeForRelDataType(RelDataType type, NamePath namePath, Set<String> seen) {
     if (namePath.getLast().isHidden()) return Optional.empty();
+    Optional<GraphQLInputType> graphQLInputType = getGraphQLInputType(type, namePath, seen);
     if (!type.isNullable()) {
-      return getGraphQLInputType(type, namePath, seen).map(GraphQLNonNull::nonNull);
+      return graphQLInputType.map(GraphQLNonNull::nonNull);
     }
-    return getGraphQLInputType(type, namePath, seen);
+    return graphQLInputType;
   }
 
   private static Optional<GraphQLInputType> getGraphQLInputType(RelDataType type, NamePath namePath, Set<String> seen) {
@@ -220,7 +230,10 @@ private static Optional<GraphQLInputType> getGraphQLInputType(RelDataType type,
         return Optional.of(CustomScalars.DATE);
       case TIME:
         return Optional.of(CustomScalars.TIME);
+      case TIME_WITH_LOCAL_TIME_ZONE:
+        break;
       case TIMESTAMP:
+      case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
         return Optional.of(CustomScalars.DATETIME);
       case BINARY:
       case VARBINARY:
@@ -233,9 +246,36 @@ private static Optional<GraphQLInputType> getGraphQLInputType(RelDataType type,
         return createGraphQLInputObjectType(type, namePath, seen);
       case MAP:
         return Optional.of(CustomScalars.JSON);
+      case INTERVAL_YEAR:
+      case INTERVAL_YEAR_MONTH:
+      case INTERVAL_MONTH:
+      case INTERVAL_DAY:
+      case INTERVAL_DAY_HOUR:
+      case INTERVAL_DAY_MINUTE:
+      case INTERVAL_DAY_SECOND:
+      case INTERVAL_HOUR:
+      case INTERVAL_HOUR_MINUTE:
+      case INTERVAL_HOUR_SECOND:
+      case INTERVAL_MINUTE:
+      case INTERVAL_MINUTE_SECOND:
+      case INTERVAL_SECOND:
+      case NULL:
+      case UNKNOWN:
+      case ANY:
+      case SYMBOL:
+      case MULTISET:
+      case DISTINCT:
+      case STRUCTURED:
+      case OTHER:
+      case CURSOR:
+      case COLUMN_LIST:
+      case DYNAMIC_STAR:
+      case GEOMETRY:
+      case SARG:
       default:
-        return Optional.empty(); // Unsupported types are omitted
+        break;
     }
+    return Optional.empty(); // Unsupported types are omitted
   }
 
   /**

sqrl-server/sqrl-server-core/src/main/java/com/datasqrl/graphql/server/CustomScalars.java

@@ -41,7 +41,7 @@ public Object parseLiteral(Object input) {
       .build();
 
 
-  public static final GraphQLScalarType DATETIME = DateTimeScalar.INSTANCE;
+  public static final GraphQLScalarType DATETIME = ExtendedScalars.DateTime;
   public static final GraphQLScalarType DATE =  ExtendedScalars.Date;
   public static final GraphQLScalarType TIME =  ExtendedScalars.LocalTime;
   public static final GraphQLScalarType JSON =  ExtendedScalars.Json;

sqrl-server/sqrl-server-vertx/pom.xml

@@ -245,6 +245,10 @@
       <groupId>org.mockito</groupId>
       <artifactId>mockito-core</artifactId>
     </dependency>
+    <dependency>
+      <groupId>com.fasterxml.jackson.datatype</groupId>
+      <artifactId>jackson-datatype-jdk8</artifactId>
+    </dependency>
   </dependencies>
 
   <build>

sqrl-server/sqrl-server-vertx/src/main/java/com/datasqrl/graphql/SqrlObjectMapper.java

@@ -2,13 +2,27 @@
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
 import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
 
 public class SqrlObjectMapper {
 
-  public static final ObjectMapper mapper = new ObjectMapper();
-  static {
-    mapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
-    mapper.registerModule(new JavaTimeModule());
+  public static ObjectMapper mapper = createObjectMapper();
+  public static ObjectMapper createObjectMapper() {
+    ObjectMapper objectMapper = new ObjectMapper();
+    registerModules(objectMapper);
+    return objectMapper;
   }
+
+  private static void registerModules(ObjectMapper mapper) {
+    SimpleModule module = new SimpleModule();
+
+    mapper.registerModule(new JavaTimeModule())
+        .registerModule((new Jdk8Module()).configureAbsentsAsNulls(true))
+        .registerModule(module)
+        .disable(SerializationFeature.WRITE_DURATIONS_AS_TIMESTAMPS)
+        .disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
+  }
+
 }

sqrl-server/sqrl-server-vertx/src/main/java/com/datasqrl/graphql/SubscriptionConfigurationImpl.java

@@ -8,6 +8,7 @@
 
 import com.datasqrl.graphql.config.ServerConfig;
 import com.datasqrl.graphql.io.SinkConsumer;
+import com.datasqrl.graphql.kafka.JsonDeserializer;
 import com.datasqrl.graphql.kafka.KafkaDataFetcherFactory;
 import com.datasqrl.graphql.kafka.KafkaSinkConsumer;
 import com.datasqrl.graphql.postgres_log.PostgresDataFetcherFactory;
@@ -83,8 +84,8 @@ public Map<String, String> getSourceConfig() {
     Map<String, String> conf = new HashMap<>();
     conf.put(BOOTSTRAP_SERVERS_CONFIG, config.getEnvironmentVariable("PROPERTIES_BOOTSTRAP_SERVERS"));
     conf.put(GROUP_ID_CONFIG, UUID.randomUUID().toString());
-    conf.put(KEY_DESERIALIZER_CLASS_CONFIG, "com.datasqrl.graphql.kafka.JsonDeserializer");
-    conf.put(VALUE_DESERIALIZER_CLASS_CONFIG, "com.datasqrl.graphql.kafka.JsonDeserializer");
+    conf.put(KEY_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class.getName());
+    conf.put(VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class.getName());
     conf.put(AUTO_OFFSET_RESET_CONFIG, "latest");
     return conf;
   }

sqrl-server/sqrl-server-vertx/src/main/java/com/datasqrl/graphql/kafka/KafkaDataFetcherFactory.java

@@ -26,7 +26,7 @@ public Publisher<Object> get(DataFetchingEnvironment env) throws Exception {
       }
 
       private boolean filterSubscription(Object data, Map<String, Object> args) {
-        if (args == null) {
+        if (args == null || args.isEmpty()) {
           return false;
         }
         for (Map.Entry<String, String> filter : coords.getFilters().entrySet()) {